框架：按真实令牌预算提前压缩上下文

agent/agent/core/prompts/compression.py

@@ -45,6 +45,13 @@ SINGLE_TURN_PROMPT = """请对以上对话历史进行压缩总结。
 (此处填写结构化的摘要内容)
 """
 
+EXPLICIT_COMPRESSION_RULES = """
+
+当前是显式任务模式。摘要只保留：当前 Root 状态、活跃 Task、已接受 Decision 与
+Artifact digest、当前 focused Task、最后一个结构化失败，以及必须保留的系统策略。
+已 cancel、supersede 或 reject 的历史只保留数量，不要重复它们的完整合同。
+"""
+
 SUMMARY_HEADER_TEMPLATE = """## 对话历史摘要（自动压缩）
 
 {summary_text}
@@ -58,15 +65,18 @@ SUMMARY_HEADER_TEMPLATE = """## 对话历史摘要（自动压缩）
 def build_compression_eval_prompt(
     execution_context: str,
     ex_reference_list: str = "",
+    explicit: bool = False,
 ) -> str:
-    return COMPRESSION_EVAL_PROMPT_TEMPLATE.format(
+    prompt = COMPRESSION_EVAL_PROMPT_TEMPLATE.format(
         execution_context=execution_context,
         ex_reference_list=ex_reference_list,
     )
+    return prompt + (EXPLICIT_COMPRESSION_RULES if explicit else "")
 
 
-def build_single_turn_prompt(execution_context: str) -> str:
-    return SINGLE_TURN_PROMPT.format(execution_context=execution_context)
+def build_single_turn_prompt(execution_context: str, explicit: bool = False) -> str:
+    prompt = SINGLE_TURN_PROMPT.format(execution_context=execution_context)
+    return prompt + (EXPLICIT_COMPRESSION_RULES if explicit else "")
 
 
 def build_summary_header(summary_text: str) -> str:

agent/agent/core/runner.py

@@ -35,9 +35,12 @@ from agent.failures import FailureDetail, FailureDisposition
 from agent.trace.protocols import TraceAttachmentStore, TraceStore
 from agent.trace.goal_models import GoalTree
 from agent.trace.compaction import (
+    PromptTokenEstimate,
     CompressionConfig,
+    calibrate_prompt_estimate,
     compress_completed_goals,
     estimate_tokens,
+    measure_prompt_tokens,
 )
 from agent.skill.models import Skill
 from agent.tools import ToolRegistry, get_tool_registry
@@ -93,6 +96,14 @@ class ContextUsage:
     max_tokens: int
     usage_percent: float
     image_count: int = 0
+    estimated_prompt_tokens: int = 0
+    calibrated_prompt_tokens: int = 0
+    actual_prompt_tokens: Optional[int] = None
+    tool_schema_tokens: int = 0
+    calibration_factor: float = 1.0
+    compression_count: int = 0
+    trigger_tokens: int = 0
+    target_tokens: int = 0
 
 
 @dataclass
@@ -173,6 +184,7 @@ class RunConfig:
         None  # explicit Planner 新 Trace 的 Root Task 草案
     )
     no_progress: NoProgressPolicy = field(default_factory=NoProgressPolicy)
+    compression: CompressionConfig = field(default_factory=CompressionConfig)
 
     # --- Trace 控制 ---
     trace_id: Optional[str] = None  # None = 新建
@@ -872,18 +884,23 @@ class AgentRunner:
         config: RunConfig,
         sequence: int,
         head_seq: int,
+        tool_schemas: Optional[List[Dict[str, Any]]] = None,
     ) -> Tuple[List[Dict], int, int, bool]:
-        """
-        管理 context 用量：检查、预警、压缩
-
-        Returns:
-            (updated_history, new_head_seq, next_sequence, needs_enter_compression_branch)
-        """
-        compression_config = CompressionConfig()
-        token_count = estimate_tokens(history)
-        max_tokens = compression_config.get_max_tokens(config.model)
+        """Measure calibrated prompt size and start compression before overflow."""
 
-        # 计算使用率
+        trace = await self.trace_store.get_trace(trace_id) if self.trace_store else None
+        budget_state = (
+            trace.runtime_state.setdefault("context_budget", {}) if trace else {}
+        )
+        measurement = measure_prompt_tokens(
+            history,
+            tool_schemas or [],
+            config.compression,
+            config.model,
+            budget_state.get("calibration_factor"),
+        )
+        token_count = measurement.calibrated_tokens
+        max_tokens = measurement.hard_limit
         progress_pct = (token_count / max_tokens * 100) if max_tokens > 0 else 0
         msg_count = len(history)
         img_count = sum(
@@ -902,13 +919,20 @@ class AgentRunner:
             max_tokens=max_tokens,
             usage_percent=progress_pct,
             image_count=img_count,
+            estimated_prompt_tokens=measurement.estimated_tokens,
+            calibrated_prompt_tokens=measurement.calibrated_tokens,
+            actual_prompt_tokens=budget_state.get("last_actual_prompt_tokens"),
+            tool_schema_tokens=measurement.tool_schema_tokens,
+            calibration_factor=measurement.calibration_factor,
+            compression_count=int(budget_state.get("compression_count", 0)),
+            trigger_tokens=measurement.trigger_tokens,
+            target_tokens=measurement.target_tokens,
         )
 
-        # 阈值警告（30%, 50%, 80%）
         if trace_id not in self._context_warned:
             self._context_warned[trace_id] = set()
 
-        for threshold in [30, 50, 80]:
+        for threshold in [30, 50, 75]:
             if (
                 progress_pct >= threshold
                 and threshold not in self._context_warned[trace_id]
@@ -918,47 +942,38 @@ class AgentRunner:
                     f"Context 使用率达到 {threshold}%: {token_count:,} / {max_tokens:,} tokens ({msg_count} 条消息)"
                 )
 
-        # 检查是否需要压缩（仅基于 token 数量）
-        needs_compression = token_count > max_tokens
+        needs_compression = (
+            token_count >= measurement.trigger_tokens
+            or (
+                config.compression.max_messages > 0
+                and msg_count >= config.compression.max_messages
+            )
+        )
 
         if not needs_compression:
             return history, head_seq, sequence, False
 
-        # 检查是否有待评估知识（压缩前必须先评估）
-        if self.trace_store and not config.force_side_branch:
-            pending = await self.trace_store.get_pending_knowledge_entries(trace_id)
-            if pending:
-                # 设置侧分支队列：反思 → 知识评估 → 压缩
-                # 反思放在前面，确保反思期间完成的 goal 产生的新知识也能在压缩前被评估
-                if config.knowledge.enable_extraction:
-                    config.force_side_branch = [
-                        "reflection",
-                        "knowledge_eval",
-                        "compression",
-                    ]
-                else:
-                    config.force_side_branch = ["knowledge_eval", "compression"]
-
-                # 在 trace.context 中设置触发事件
-                trace = await self.trace_store.get_trace(trace_id)
-                if trace:
-                    if not trace.context:
-                        trace.context = {}
-                    trace.context["knowledge_eval_trigger"] = "compression"
-                    await self.trace_store.update_trace(trace_id, context=trace.context)
-
-                self.log.info(
-                    f"[Knowledge Eval] 压缩前触发知识评估，待评估: {len(pending)} 条"
-                )
-                return history, head_seq, sequence, True
-
-        # 知识提取：在任何压缩发生前，用完整 history 做反思（进入反思侧分支）
-        if config.knowledge.enable_extraction and not config.force_side_branch:
-            # 设置侧分支队列：先反思，再压缩
-            config.force_side_branch = ["reflection", "compression"]
-            return history, head_seq, sequence, True
+        if trace and not budget_state.get("compression_pending"):
+            budget_state["compression_pending"] = True
+            budget_state["compression_started_at"] = datetime.now().isoformat()
+            started_payload = self._context_event_payload(
+                measurement,
+                message_count=msg_count,
+                actual_prompt_tokens=budget_state.get("last_actual_prompt_tokens"),
+                reason="trigger_ratio" if token_count >= measurement.trigger_tokens else "message_limit",
+            )
+            budget_state["compression_before"] = started_payload
+            trace.runtime_state["context_budget"] = budget_state
+            await self.trace_store.update_trace(
+                trace_id,
+                runtime_state=trace.runtime_state,
+            )
+            await self.trace_store.append_event(
+                trace_id,
+                "context_compression_started",
+                started_payload,
+            )
 
-        # Level 1 是 legacy GoalTree 的确定性压缩；explicit 直接保留 Level 2。
         is_legacy = (
             CompletionPolicy(config.completion_policy) == CompletionPolicy.LEGACY_AUTO
         )
@@ -991,41 +1006,192 @@ class AgentRunner:
                 token_count,
             )
 
-        # Level 2 压缩：检查 Level 1 后是否仍超阈值
-        # 注意：Level 1 压缩后需要重新优化图片并计算 token
         optimized_history_after = await self._optimize_images(history, config.model)
-        token_count_after = estimate_tokens(optimized_history_after)
-        needs_level2 = token_count_after > max_tokens
+        after = measure_prompt_tokens(
+            optimized_history_after,
+            tool_schemas or [],
+            config.compression,
+            config.model,
+            measurement.calibration_factor,
+        )
+        needs_level2 = after.calibrated_tokens > after.target_tokens
 
         if needs_level2:
             source = "Level 1 后" if is_legacy else "explicit 模式"
             self.log.info(
                 "%s仍超阈值 (token=%d/%d)，需要进入压缩侧分支",
                 source,
-                token_count_after,
-                max_tokens,
+                after.calibrated_tokens,
+                after.hard_limit,
             )
-            # 如果还没有设置侧分支（说明没有启用知识提取），直接进入压缩
             if not config.force_side_branch:
                 config.force_side_branch = ["compression"]
-            # 返回标志，让主循环进入侧分支
             return history, head_seq, sequence, True
 
-        # 压缩完成后，输出最终发给模型的消息列表
-        self.log.info("上下文压缩检查完成，发送给模型的消息列表:")
-        for idx, msg in enumerate(history):
-            role = msg.get("role", "unknown")
-            content = msg.get("content", "")
-            if isinstance(content, str):
-                preview = content[:100] + ("..." if len(content) > 100 else "")
-            elif isinstance(content, list):
-                preview = f"[{len(content)} blocks]"
-            else:
-                preview = str(content)[:100]
-            self.log.info(f"  [{idx}] {role}: {preview}")
-
+        await self._finish_context_compression(
+            trace,
+            after,
+            before_message_count=msg_count,
+            after_message_count=len(history),
+        )
         return history, head_seq, sequence, False
 
+    @staticmethod
+    def _context_event_payload(
+        measurement: PromptTokenEstimate,
+        *,
+        message_count: int,
+        actual_prompt_tokens: Optional[int],
+        reason: str,
+    ) -> Dict[str, Any]:
+        return {
+            "message_count": message_count,
+            "estimated_prompt_tokens": measurement.estimated_tokens,
+            "calibrated_prompt_tokens": measurement.calibrated_tokens,
+            "actual_prompt_tokens": actual_prompt_tokens,
+            "tool_schema_tokens": measurement.tool_schema_tokens,
+            "calibration_factor": measurement.calibration_factor,
+            "trigger_tokens": measurement.trigger_tokens,
+            "target_tokens": measurement.target_tokens,
+            "hard_limit": measurement.hard_limit,
+            "reason": reason,
+        }
+
+    async def _record_prompt_measurement(
+        self,
+        trace: Trace,
+        config: RunConfig,
+        messages: List[Dict[str, Any]],
+        tool_schemas: List[Dict[str, Any]],
+        actual_prompt_tokens: int,
+    ) -> None:
+        budget = trace.runtime_state.setdefault("context_budget", {})
+        raw = measure_prompt_tokens(
+            messages,
+            tool_schemas,
+            config.compression,
+            config.model,
+            1.0,
+        )
+        previous = float(
+            budget.get(
+                "calibration_factor",
+                config.compression.fallback_safety_factor,
+            )
+        )
+        budget["calibration_factor"] = calibrate_prompt_estimate(
+            actual_prompt_tokens=actual_prompt_tokens,
+            estimated_prompt_tokens=raw.estimated_tokens,
+            previous_factor=previous,
+        )
+        budget["last_actual_prompt_tokens"] = actual_prompt_tokens
+        budget["last_estimated_prompt_tokens"] = raw.estimated_tokens
+        await self._persist_runtime_state(trace)
+
+    async def _finish_context_compression(
+        self,
+        trace: Optional[Trace],
+        measurement: PromptTokenEstimate,
+        *,
+        before_message_count: int,
+        after_message_count: int,
+    ) -> Optional[FailureDetail]:
+        if trace is None or self.trace_store is None:
+            return None
+        budget = trace.runtime_state.setdefault("context_budget", {})
+        started_at = budget.get("compression_started_at")
+        duration_ms = None
+        if isinstance(started_at, str):
+            try:
+                duration_ms = max(
+                    0,
+                    int((datetime.now() - datetime.fromisoformat(started_at)).total_seconds() * 1000),
+                )
+            except ValueError:
+                duration_ms = None
+        budget["compression_pending"] = False
+        budget.pop("compression_started_at", None)
+        before_measurement = budget.pop("compression_before", None)
+        budget["compression_count"] = int(budget.get("compression_count", 0)) + 1
+        await self._persist_runtime_state(trace)
+        payload = {
+            **self._context_event_payload(
+                measurement,
+                message_count=after_message_count,
+                actual_prompt_tokens=budget.get("last_actual_prompt_tokens"),
+                reason="compression_completed",
+            ),
+            "before_message_count": before_message_count,
+            "after_message_count": after_message_count,
+            "duration_ms": duration_ms,
+            "compression_count": budget["compression_count"],
+            "before": before_measurement,
+        }
+        usage = self._context_usage.get(trace.trace_id)
+        if usage is not None:
+            usage.message_count = after_message_count
+            usage.token_count = measurement.calibrated_tokens
+            usage.estimated_prompt_tokens = measurement.estimated_tokens
+            usage.calibrated_prompt_tokens = measurement.calibrated_tokens
+            usage.tool_schema_tokens = measurement.tool_schema_tokens
+            usage.calibration_factor = measurement.calibration_factor
+            usage.compression_count = budget["compression_count"]
+            usage.usage_percent = (
+                measurement.calibrated_tokens / measurement.hard_limit * 100
+                if measurement.hard_limit
+                else 0
+            )
+        if measurement.calibrated_tokens > measurement.target_tokens:
+            return await self._context_budget_exceeded(
+                trace,
+                measurement,
+                reason="compression_target_not_reached",
+                extra=payload,
+            )
+        await self.trace_store.append_event(
+            trace.trace_id,
+            "context_compression_completed",
+            payload,
+        )
+        return None
+
+    async def _context_budget_exceeded(
+        self,
+        trace: Trace,
+        measurement: PromptTokenEstimate,
+        *,
+        reason: str,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> FailureDetail:
+        payload = {
+            **self._context_event_payload(
+                measurement,
+                message_count=int((extra or {}).get("after_message_count", 0)),
+                actual_prompt_tokens=trace.runtime_state.get("context_budget", {}).get(
+                    "last_actual_prompt_tokens"
+                ),
+                reason=reason,
+            ),
+            **(extra or {}),
+        }
+        if self.trace_store:
+            await self.trace_store.append_event(
+                trace.trace_id,
+                "context_budget_exceeded",
+                payload,
+            )
+        return FailureDetail(
+            code="CONTEXT_BUDGET_EXCEEDED",
+            message="The minimum safe execution context exceeds the configured token budget",
+            disposition=FailureDisposition.ABORT_RUN,
+            details={
+                "reason": reason,
+                "calibrated_prompt_tokens": measurement.calibrated_tokens,
+                "hard_limit": measurement.hard_limit,
+                "target_tokens": measurement.target_tokens,
+            },
+        )
+
     async def _build_knowledge_eval_prompt(
         self, trace_id: str, goal_tree: Optional[GoalTree]
     ) -> str:
@@ -1115,7 +1281,13 @@ class AgentRunner:
             goal_tree,
             config,
         )
-        compress_prompt = build_single_turn_prompt(execution_context)
+        compress_prompt = build_single_turn_prompt(
+            execution_context,
+            explicit=(
+                CompletionPolicy(config.completion_policy)
+                == CompletionPolicy.EXPLICIT_VALIDATION
+            ),
+        )
         compress_messages = list(history) + [
             {"role": "user", "content": compress_prompt}
         ]
@@ -1124,6 +1296,27 @@ class AgentRunner:
         compress_messages = self._add_cache_control(
             compress_messages, config.model, config.enable_prompt_caching
         )
+        trace = await self.trace_store.get_trace(trace_id) if self.trace_store else None
+        measurement = measure_prompt_tokens(
+            compress_messages,
+            [],
+            config.compression,
+            config.model,
+            (
+                trace.runtime_state.get("context_budget", {}).get("calibration_factor")
+                if trace
+                else None
+            ),
+        )
+        if measurement.calibrated_tokens > measurement.hard_limit:
+            if trace:
+                await self._context_budget_exceeded(
+                    trace,
+                    measurement,
+                    reason="compression_fallback_would_exceed_hard_limit",
+                    extra={"after_message_count": len(compress_messages)},
+                )
+            return ""
 
         # 单次 LLM 调用（无工具）
         result = await self.llm_call(
@@ -1133,6 +1326,14 @@ class AgentRunner:
             temperature=config.temperature,
             **config.extra_llm_params,
         )
+        if trace and int(result.get("prompt_tokens", 0)) > 0:
+            await self._record_prompt_measurement(
+                trace,
+                config,
+                compress_messages,
+                [],
+                int(result["prompt_tokens"]),
+            )
 
         summary_text = result.get("content", "").strip()
 
@@ -1362,7 +1563,13 @@ class AgentRunner:
                         sequence,
                         needs_enter_side_branch,
                     ) = await self._manage_context_usage(
-                        trace_id, history, goal_tree, config, sequence, head_seq
+                        trace_id,
+                        history,
+                        goal_tree,
+                        config,
+                        sequence,
+                        head_seq,
+                        tool_schemas,
                     )
 
             # 进入侧分支
@@ -1450,7 +1657,10 @@ class AgentRunner:
                         goal_tree,
                         config,
                     )
-                    prompt = build_compression_prompt(execution_context)
+                    prompt = build_compression_prompt(
+                        execution_context,
+                        explicit=(explicit_role is not None),
+                    )
 
                 branch_user_msg = Message.create(
                     trace_id=trace_id,
@@ -1487,6 +1697,23 @@ class AgentRunner:
                 llm_messages, config.model, config.enable_prompt_caching
             )
 
+            budget_state = trace.runtime_state.get("context_budget", {})
+            request_measurement = measure_prompt_tokens(
+                llm_messages,
+                tool_schemas,
+                config.compression,
+                config.model,
+                budget_state.get("calibration_factor"),
+            )
+            if request_measurement.calibrated_tokens > request_measurement.hard_limit:
+                terminal_failure = await self._context_budget_exceeded(
+                    trace,
+                    request_measurement,
+                    reason="request_would_exceed_hard_limit",
+                    extra={"after_message_count": len(llm_messages)},
+                )
+                break
+
             # 调用 LLM（等待完成后再检查 cancel 信号，不中断正在进行的调用）
             result = await self.llm_call(
                 messages=llm_messages,
@@ -1505,6 +1732,14 @@ class AgentRunner:
             step_cost = result.get("cost", 0)
             cache_creation_tokens = result.get("cache_creation_tokens")
             cache_read_tokens = result.get("cache_read_tokens")
+            if prompt_tokens > 0:
+                await self._record_prompt_measurement(
+                    trace,
+                    config,
+                    llm_messages,
+                    tool_schemas,
+                    prompt_tokens,
+                )
 
             # 周期性自动注入上下文（仅主路径）
             if (
@@ -1821,6 +2056,22 @@ class AgentRunner:
                         history = history[: side_branch_ctx.start_history_length]
                         head_seq = side_branch_ctx.start_head_seq
 
+                    compacted_measurement = measure_prompt_tokens(
+                        history,
+                        tool_schemas,
+                        config.compression,
+                        config.model,
+                        trace.runtime_state.get("context_budget", {}).get(
+                            "calibration_factor"
+                        ),
+                    )
+                    compression_failure = await self._finish_context_compression(
+                        trace,
+                        compacted_measurement,
+                        before_message_count=side_branch_ctx.start_history_length,
+                        after_message_count=len(history),
+                    )
+
                     # 清理
                     trace.context.pop("active_side_branch", None)
                     config.force_side_branch = None
@@ -1831,6 +2082,9 @@ class AgentRunner:
                             head_sequence=head_seq,
                         )
                     side_branch_ctx = None
+                    if compression_failure is not None:
+                        terminal_failure = compression_failure
+                        break
                     continue
 
                 elif should_exit and side_branch_ctx.type == "reflection":

agent/agent/orchestration/_task_context.py

@@ -2,28 +2,34 @@
 
 from __future__ import annotations
 
-from .models import TaskLedger
-from ._task_graph import path_key
+from .models import DecisionAction, TaskLedger, TaskStatus, ValidationVerdict
+from ._task_graph import TERMINAL_TASK_STATUSES, path_key
 
 
 def render_task_context(ledger: TaskLedger) -> str:
     """Render current task state for Planner refresh and context compression."""
 
     tasks = sorted(ledger.tasks.values(), key=lambda task: path_key(task.display_path))
+    root = ledger.tasks.get(ledger.root_task_id or "")
     focused = ledger.tasks.get(ledger.focused_task_id or "")
     lines = [
         "## Current Task Plan",
         "",
-        f"Root: {_one_line(ledger.root_objective)}",
+        "Root: " + (
+            f"[{root.status.value}] {_one_line(root.current_spec.objective)}"
+            if root
+            else _one_line(ledger.root_objective)
+        ),
         "Focused: " + (
             f"{focused.display_path} [{focused.status.value}] {_one_line(focused.current_spec.objective)}"
             if focused
             else "none"
         ),
         "",
-        "Tasks:",
+        "Active Tasks:",
     ]
-    for task in tasks:
+    active = [task for task in tasks if task.status not in TERMINAL_TASK_STATUSES]
+    for task in active:
         line = (
             f"- {task.display_path} [{task.status.value}] "
             f"{_one_line(task.current_spec.objective)}"
@@ -31,6 +37,58 @@ def render_task_context(ledger: TaskLedger) -> str:
         if task.blocked_reason:
             line += f" (blocked: {_one_line(task.blocked_reason)})"
         lines.append(line)
+    if not active:
+        lines.append("- none")
+
+    lines.extend(["", "Accepted Results:"])
+    accepted_count = 0
+    for decision in sorted(ledger.decisions.values(), key=lambda item: item.created_at):
+        if decision.action != DecisionAction.ACCEPT:
+            continue
+        task = ledger.tasks.get(decision.task_id)
+        attempt = ledger.attempts.get(decision.attempt_id or "")
+        if task is None or attempt is None or attempt.submission is None:
+            continue
+        artifacts = [
+            f"{ref.kind}:{ref.digest or ref.version}"
+            for ref in attempt.submission.artifact_refs
+        ]
+        lines.append(
+            f"- {task.display_path} {_one_line(task.current_spec.objective)}"
+            + (f" -> {', '.join(artifacts)}" if artifacts else "")
+        )
+        accepted_count += 1
+    if accepted_count == 0:
+        lines.append("- none")
+
+    retired = {
+        status.value: sum(task.status == status for task in tasks)
+        for status in (TaskStatus.CANCELLED, TaskStatus.SUPERSEDED)
+    }
+    retired["rejected_validations"] = sum(
+        report.verdict == ValidationVerdict.FAILED
+        for report in ledger.validations.values()
+    )
+    lines.extend([
+        "",
+        "Retired History: " + ", ".join(
+            f"{key}={value}" for key, value in retired.items()
+        ),
+    ])
+
+    failed_runs = [
+        item
+        for item in (*ledger.attempts.values(), *ledger.validations.values())
+        if item.failure is not None
+    ]
+    if failed_runs:
+        latest = max(failed_runs, key=lambda item: item.updated_at)
+        failure = latest.failure
+        lines.extend([
+            "",
+            "Last Failure: "
+            f"{failure.code} [{failure.disposition.value}] {_one_line(failure.message)}",
+        ])
     return "\n".join(lines)
 
 

agent/agent/trace/compaction.py

@@ -15,6 +15,7 @@ Level 2: LLM 总结（仅在 Level 1 后仍超限时触发）
 import copy
 import json
 import logging
+import math
 from dataclasses import dataclass
 from typing import List, Dict, Any, Optional, Set
 
@@ -112,9 +113,22 @@ class CompressionConfig:
     """压缩配置"""
     max_tokens: int = 0            # 最大 token 数（0 = 自动：context_window * 0.5）
     threshold_ratio: float = 0.5       # 触发压缩的阈值 = context_window 的比例
+    trigger_ratio: float = 0.75
+    target_ratio: float = 0.50
+    fallback_safety_factor: float = 1.25
     keep_recent_messages: int = 10     # Level 1 中始终保留最近 N 条消息
     max_messages: int = 0              # 最大消息数（超过此数量触发压缩，0 = 禁用，默认禁用）
 
+    def __post_init__(self) -> None:
+        if self.max_tokens < 0:
+            raise ValueError("max_tokens cannot be negative")
+        if not 0 < self.threshold_ratio <= 1:
+            raise ValueError("threshold_ratio must be in (0, 1]")
+        if not 0 < self.target_ratio < self.trigger_ratio < 1:
+            raise ValueError("target_ratio and trigger_ratio must satisfy 0 < target < trigger < 1")
+        if self.fallback_safety_factor < 1:
+            raise ValueError("fallback_safety_factor must be at least 1")
+
     def get_max_tokens(self, model: str) -> int:
         """获取实际的 max_tokens（如果为 0 则自动计算）"""
         if self.max_tokens > 0:
@@ -123,6 +137,67 @@ class CompressionConfig:
         return int(window * self.threshold_ratio)
 
 
+@dataclass(frozen=True)
+class PromptTokenEstimate:
+    estimated_tokens: int
+    tool_schema_tokens: int
+    calibrated_tokens: int
+    calibration_factor: float
+    hard_limit: int
+    trigger_tokens: int
+    target_tokens: int
+
+
+def estimate_tool_schema_tokens(tool_schemas: List[Dict[str, Any]]) -> int:
+    if not tool_schemas:
+        return 0
+    encoded = json.dumps(
+        tool_schemas,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return _estimate_text_tokens(encoded)
+
+
+def measure_prompt_tokens(
+    messages: List[Dict[str, Any]],
+    tool_schemas: List[Dict[str, Any]],
+    config: CompressionConfig,
+    model: str,
+    calibration_factor: Optional[float] = None,
+) -> PromptTokenEstimate:
+    tool_tokens = estimate_tool_schema_tokens(tool_schemas)
+    estimated = estimate_tokens(messages) + tool_tokens
+    factor = max(
+        1.0,
+        calibration_factor
+        if calibration_factor is not None
+        else config.fallback_safety_factor,
+    )
+    hard_limit = config.get_max_tokens(model)
+    return PromptTokenEstimate(
+        estimated_tokens=estimated,
+        tool_schema_tokens=tool_tokens,
+        calibrated_tokens=math.ceil(estimated * factor),
+        calibration_factor=factor,
+        hard_limit=hard_limit,
+        trigger_tokens=int(hard_limit * config.trigger_ratio),
+        target_tokens=int(hard_limit * config.target_ratio),
+    )
+
+
+def calibrate_prompt_estimate(
+    *,
+    actual_prompt_tokens: int,
+    estimated_prompt_tokens: int,
+    previous_factor: float,
+) -> float:
+    if actual_prompt_tokens <= 0 or estimated_prompt_tokens <= 0:
+        return max(1.0, previous_factor)
+    return max(1.0, previous_factor, actual_prompt_tokens / estimated_prompt_tokens)
+
+
 # ===== Level 1: Goal 完成压缩 =====
 
 def compress_completed_goals(
@@ -358,10 +433,13 @@ def needs_level2_compression(
 # COMPRESSION_EVAL_PROMPT 和 REFLECT_PROMPT 现在从 prompts.py 导入
 
 
-def build_compression_prompt(execution_context: str) -> str:
+def build_compression_prompt(execution_context: str, *, explicit: bool = False) -> str:
     """构建 Level 2 压缩 prompt，计划上下文由运行模式提供。"""
 
-    return build_compression_eval_prompt(execution_context=execution_context)
+    return build_compression_eval_prompt(
+        execution_context=execution_context,
+        explicit=explicit,
+    )
 
 
 def build_reflect_prompt() -> str:

agent/tests/test_context_budget.py

@@ -0,0 +1,195 @@
+import math
+
+import pytest
+
+from agent.core.runner import AgentRunner, RunConfig
+from agent.trace.compaction import (
+    CompressionConfig,
+    calibrate_prompt_estimate,
+    measure_prompt_tokens,
+)
+from agent.trace.models import Trace
+from agent.trace.store import FileSystemTraceStore
+
+
+def _knowledge_off():
+    from agent.tools.builtin.knowledge import KnowledgeConfig
+
+    return KnowledgeConfig(
+        enable_extraction=False,
+        enable_completion_extraction=False,
+        enable_injection=False,
+    )
+
+
+def test_prompt_measurement_includes_tools_and_provider_calibration():
+    config = CompressionConfig(max_tokens=100_000)
+    messages = [{"role": "user", "content": "x" * 4_000}]
+    schemas = [{
+        "type": "function",
+        "function": {
+            "name": "large_tool",
+            "description": "y" * 4_000,
+            "parameters": {"type": "object"},
+        },
+    }]
+
+    raw = measure_prompt_tokens(messages, [], config, "unknown-model", 1.0)
+    measured = measure_prompt_tokens(messages, schemas, config, "unknown-model", 1.0)
+    factor = calibrate_prompt_estimate(
+        actual_prompt_tokens=measured.estimated_tokens * 2,
+        estimated_prompt_tokens=measured.estimated_tokens,
+        previous_factor=1.25,
+    )
+    calibrated = measure_prompt_tokens(
+        messages,
+        schemas,
+        config,
+        "unknown-model",
+        factor,
+    )
+
+    assert measured.tool_schema_tokens > 0
+    assert measured.estimated_tokens > raw.estimated_tokens
+    assert factor == 2
+    assert calibrated.calibrated_tokens == measured.estimated_tokens * 2
+    assert calibrated.trigger_tokens == 75_000
+    assert calibrated.target_tokens == 50_000
+
+
+@pytest.mark.asyncio
+async def test_build_900036_growth_replay_compresses_before_hard_limit(tmp_path):
+    store = FileSystemTraceStore(str(tmp_path))
+    trace = Trace(trace_id="growth-replay", mode="agent", agent_role="legacy")
+    await store.create_trace(trace)
+    runner = AgentRunner(trace_store=store)
+    config = RunConfig(
+        model="unregistered-build-model",
+        compression=CompressionConfig(max_tokens=100_000),
+        knowledge=_knowledge_off(),
+    )
+    schemas = [{
+        "type": "function",
+        "function": {
+            "name": "offline_step",
+            "description": "append one deterministic fixture step",
+            "parameters": {"type": "object", "properties": {}},
+        },
+    }]
+    history = [
+        {"role": "system", "content": "keep system policy"},
+        {"role": "user", "content": "offline 97-round context fixture"},
+    ]
+    total_prompt_tokens = 0
+    max_prompt_tokens = 0
+
+    for turn in range(97):
+        history.extend([
+            {"role": "assistant", "content": f"turn {turn}", "tool_calls": []},
+            {"role": "tool", "content": "x" * 4_000},
+        ])
+        history, _, _, needs_compression = await runner._manage_context_usage(
+            trace.trace_id,
+            history,
+            None,
+            config,
+            sequence=turn + 1,
+            head_seq=0,
+            tool_schemas=schemas,
+        )
+        current = await store.get_trace(trace.trace_id)
+        measurement = measure_prompt_tokens(
+            history,
+            schemas,
+            config.compression,
+            config.model,
+            current.runtime_state.get("context_budget", {}).get("calibration_factor"),
+        )
+        simulated_actual = math.ceil(measurement.estimated_tokens * 1.10)
+        total_prompt_tokens += simulated_actual
+        max_prompt_tokens = max(max_prompt_tokens, simulated_actual)
+        await runner._record_prompt_measurement(
+            current,
+            config,
+            history,
+            schemas,
+            simulated_actual,
+        )
+
+        if needs_compression:
+            history = [
+                history[0],
+                history[1],
+                {"role": "user", "content": "compact execution summary"},
+            ]
+            current = await store.get_trace(trace.trace_id)
+            compacted = measure_prompt_tokens(
+                history,
+                schemas,
+                config.compression,
+                config.model,
+                current.runtime_state["context_budget"]["calibration_factor"],
+            )
+            failure = await runner._finish_context_compression(
+                current,
+                compacted,
+                before_message_count=turn * 2 + 4,
+                after_message_count=len(history),
+            )
+            assert failure is None
+            assert compacted.calibrated_tokens <= compacted.target_tokens
+
+    events = await store.get_events(trace.trace_id)
+    event_names = [event["event"] for event in events]
+    assert "context_compression_started" in event_names
+    assert "context_compression_completed" in event_names
+    assert max_prompt_tokens < 100_000
+    assert total_prompt_tokens < 6_327_689 * 0.50
+    usage = runner.get_context_usage(trace.trace_id)
+    assert usage.compression_count >= 1
+    assert usage.actual_prompt_tokens is not None
+
+
+@pytest.mark.asyncio
+async def test_oversized_minimum_context_fails_before_model_request(tmp_path):
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {"content": "must not be called", "tool_calls": None}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    result = await AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+    ).run_result(
+        [{"role": "user", "content": "x" * 8_000}],
+        RunConfig(
+            max_iterations=4,
+            tools=[],
+            tool_groups=[],
+            knowledge=_knowledge_off(),
+            compression=CompressionConfig(max_tokens=1_000),
+        ),
+    )
+
+    assert calls == 0
+    assert result["status"] == "failed"
+    assert result["failure"]["code"] == "CONTEXT_BUDGET_EXCEEDED"
+    events = await store.get_events(result["trace_id"])
+    assert "context_budget_exceeded" in [event["event"] for event in events]
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"trigger_ratio": 0.5, "target_ratio": 0.5},
+        {"trigger_ratio": 1.0},
+        {"fallback_safety_factor": 0.9},
+        {"max_tokens": -1},
+    ],
+)
+def test_compression_config_rejects_unsafe_budgets(kwargs):
+    with pytest.raises(ValueError):
+        CompressionConfig(**kwargs)