测试：覆盖失败治理与上下文性能边界

agent/README.md

@@ -2,6 +2,12 @@
 
 通用、可扩展的 Agent 执行框架。`0.4.0` 在保留 `legacy_auto` 行为的同时，新增可选的 Planner → Worker Attempt → 独立 Validator → Planner Decision 闭环。
 
+## 失败和上下文治理
+
+- Tool 用 `FailureDetail` 返回结构化失败，并通过 `retry_call`、`repair_attempt`、`replan_task` 或 `abort_run` 声明处置语义。业务错误的分类由 Host 边界完成，通用框架不解析业务 payload。
+- 显式 Planner 连续 3 轮没有持久化语义进展，或任一角色连续 2 次返回同一失败指纹时，Runner 以 `NO_PROGRESS` 收敛；计数保存在 `Trace.runtime_state`。
+- `RunConfig.compression` 定义硬 token 预算、75% 触发线和 50% 压缩目标。预算包含工具 schema，并用上一次供应商返回的 `prompt_tokens` 保守校准；任何超硬上限的请求都会在发给模型前失败。
+
 - 框架说明：[`agent/README.md`](agent/README.md)
 - V1 显式任务闭环：[`agent/docs/orchestration-v1.md`](agent/docs/orchestration-v1.md)
 - Preset 与角色权限：[`agent/docs/presets.md`](agent/docs/presets.md)

agent/agent/core/runner.py

@@ -39,7 +39,6 @@ from agent.trace.compaction import (
     CompressionConfig,
     calibrate_prompt_estimate,
     compress_completed_goals,
-    estimate_tokens,
     measure_prompt_tokens,
 )
 from agent.skill.models import Skill
@@ -2277,34 +2276,25 @@ class AgentRunner:
                                 set_trace_context(trace_id)
                             except ImportError:
                                 pass
-                        try:
-                            tool_result = await self._execute_authorized_tool(
-                                tool_name,
-                                tool_args,
-                                tc["id"],
-                                config,
-                                trace,
-                                goal_tree,
-                                sequence,
-                                side_branch={
-                                    "type": side_branch_ctx.type,
-                                    "branch_id": side_branch_ctx.branch_id,
-                                    "is_side_branch": True,
-                                    "max_turns": side_branch_ctx.max_turns,
-                                    "trigger_event": trigger_event_for_tool,
-                                }
-                                if side_branch_ctx
-                                else None,
-                            )
-                            return (tc, tool_args, tool_result)
-                        except Exception as e:
-                            import traceback
-
-                            return (
-                                tc,
-                                tool_args,
-                                f"Error executing tool {tool_name}: {str(e)}\n{traceback.format_exc()}",
-                            )
+                        tool_result = await self._execute_tool_safely(
+                            tool_name,
+                            tool_args,
+                            tc["id"],
+                            config,
+                            trace,
+                            goal_tree,
+                            sequence,
+                            side_branch={
+                                "type": side_branch_ctx.type,
+                                "branch_id": side_branch_ctx.branch_id,
+                                "is_side_branch": True,
+                                "max_turns": side_branch_ctx.max_turns,
+                                "trigger_event": trigger_event_for_tool,
+                            }
+                            if side_branch_ctx
+                            else None,
+                        )
+                        return (tc, tool_args, tool_result)
 
                     tasks = [_execute_single_tool(tc) for tc in tool_calls]
                     results = await asyncio.gather(*tasks)
@@ -2572,7 +2562,7 @@ class AgentRunner:
                             except ImportError:
                                 pass
 
-                        tool_result = await self._execute_authorized_tool(
+                        tool_result = await self._execute_tool_safely(
                             tool_name,
                             tool_args,
                             tc["id"],
@@ -3857,6 +3847,43 @@ class AgentRunner:
             side_branch=side_branch,
         )
 
+    async def _execute_tool_safely(
+        self,
+        tool_name: str,
+        tool_args: Dict[str, Any],
+        tool_call_id: str,
+        config: RunConfig,
+        trace: Trace,
+        goal_tree: Optional[GoalTree],
+        sequence: int,
+        side_branch: Optional[Dict[str, Any]] = None,
+    ) -> Any:
+        """Keep unexpected framework exceptions out of model-visible feedback."""
+
+        try:
+            return await self._execute_authorized_tool(
+                tool_name,
+                tool_args,
+                tool_call_id,
+                config,
+                trace,
+                goal_tree,
+                sequence,
+                side_branch=side_branch,
+            )
+        except Exception:
+            self.log.exception("Unexpected failure while executing tool %s", tool_name)
+            failure = FailureDetail(
+                code="UNEXPECTED_TOOL_ERROR",
+                message="The tool failed unexpectedly",
+                disposition=FailureDisposition.ABORT_RUN,
+                source_tool=tool_name,
+            )
+            return {
+                "text": failure.message,
+                "_control": {"failure": failure.to_dict()},
+            }
+
     def _get_tool_schemas(
         self,
         tools: Optional[List[str]] = None,

agent/tests/test_failure_contract.py

@@ -5,7 +5,15 @@ import logging
 
 import pytest
 
-from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolResult
+from agent import (
+    AgentRunner,
+    FailureDetail,
+    FailureDisposition,
+    RunConfig,
+    ToolExecutionError,
+    ToolResult,
+)
+from agent.trace.store import FileSystemTraceStore
 from agent.tools.registry import ToolRegistry
 
 
@@ -103,3 +111,63 @@ async def test_registry_hides_unexpected_error_but_logs_traceback(caplog) -> Non
     assert failure["code"] == "UNEXPECTED_TOOL_ERROR"
     assert "secret implementation detail" not in result["text"]
     assert any(record.exc_info for record in caplog.records)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("parallel", [False, True])
+async def test_runner_hides_unexpected_errors_in_both_tool_paths(
+    tmp_path,
+    caplog,
+    parallel,
+) -> None:
+    registry = ToolRegistry()
+
+    async def sample() -> str:
+        return "unused"
+
+    registry.register(sample, schema=_schema("sample"))
+
+    async def llm_call(**_kwargs):
+        return {
+            "content": "",
+            "tool_calls": [{
+                "id": "call-sample",
+                "type": "function",
+                "function": {"name": "sample", "arguments": "{}"},
+            }],
+            "finish_reason": "tool_calls",
+        }
+
+    runner = AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        tool_registry=registry,
+        llm_call=llm_call,
+    )
+
+    async def broken_executor(*_args, **_kwargs):
+        raise RuntimeError("private framework traceback")
+
+    runner._execute_authorized_tool = broken_executor  # type: ignore[method-assign]
+    from agent.tools.builtin.knowledge import KnowledgeConfig
+
+    with caplog.at_level(logging.ERROR):
+        result = await runner.run_result(
+            [{"role": "user", "content": "run"}],
+            RunConfig(
+                max_iterations=2,
+                tools=["sample"],
+                tool_groups=[],
+                parallel_tool_execution=parallel,
+                knowledge=KnowledgeConfig(
+                    enable_extraction=False,
+                    enable_completion_extraction=False,
+                    enable_injection=False,
+                ),
+            ),
+        )
+
+    assert result["status"] == "failed"
+    assert result["failure"]["code"] == "UNEXPECTED_TOOL_ERROR"
+    messages = await runner.trace_store.get_trace_messages(result["trace_id"])
+    assert "private framework traceback" not in "\n".join(str(item.content) for item in messages)
+    assert any(record.exc_info for record in caplog.records)

script_build_host/src/script_build_host/tools/failures.py

@@ -6,10 +6,8 @@ from collections.abc import Mapping
 from typing import Any
 
 from agent import FailureDetail, FailureDisposition
-
 from script_build_host.domain.errors import ScriptBuildError
 
-
 _REPLAN_CODES = frozenset({
     "INPUT_SCOPE_MISMATCH",
     "CHILD_DECISION_INVALID",

script_build_host/src/script_build_host/tools/registry.py

@@ -8,9 +8,9 @@ import json
 from functools import wraps
 from typing import Any
 
-from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolRegistry
 from agent.tools.models import ToolResult
 
+from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolRegistry
 from script_build_host.domain.errors import ScriptBuildError
 
 from .failures import classify_script_tool_failure
@@ -807,7 +807,11 @@ def _strongest_cycle_failure(results: Any) -> FailureDetail | None:
     }
     strongest: FailureDetail | None = None
     for result in results:
-        value = result.get("failure") if isinstance(result, dict) else getattr(result, "failure", None)
+        value = (
+            result.get("failure")
+            if isinstance(result, dict)
+            else getattr(result, "failure", None)
+        )
         if value is None:
             continue
         failure = FailureDetail.from_dict(value)

script_build_host/tests/test_phase_two_agent_tools.py

@@ -6,16 +6,16 @@ from types import SimpleNamespace
 from typing import Any
 
 import pytest
-from agent import ToolRegistry
 from agent.orchestration import ArtifactRef, ValidationVerdict
 from agent.orchestration.evidence import EvidenceResponse
-
 from script_build_host.domain.artifacts import ArtifactState
 from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
 from script_build_host.tools.contracts import AttemptManifest
 from script_build_host.tools.gateway import LegacyScriptToolGateway, RetrievalResult
 from script_build_host.tools.registry import register_script_tools
 
+from agent import ToolRegistry
+
 
 class _CandidateTools:
     def __init__(self, artifact_ref: ArtifactRef) -> None:
@@ -526,5 +526,12 @@ async def test_view_frozen_images_returns_multimodal_without_url_or_data_in_text
         {"raw_artifact_refs": ["script-build://raw-artifacts/sha256/abc"]},
         context={"root_trace_id": "root-a", "task_id": "task-1"},
     )
-    assert isinstance(rejected, str)
-    assert "must not contain network URLs" in rejected
+    assert isinstance(rejected, dict)
+    assert rejected["_control"]["failure"] == {
+        "code": "INVALID_TOOL_ARGUMENT",
+        "message": "frozen images must not contain network URLs",
+        "disposition": "retry_call",
+        "source_tool": "view_frozen_images",
+        "details": {"task_id": "task-1"},
+    }
+    assert "must not contain network URLs" in rejected["text"]

script_build_host/tests/test_tool_failure_bridge.py

@@ -0,0 +1,317 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+from agent.orchestration import CompletionPolicy
+from agent.trace.store import FileSystemTraceStore
+from script_build_host.agents.presets import register_script_presets
+from script_build_host.application.phase_two_candidates import PhaseTwoCandidateError
+from script_build_host.domain.errors import (
+    ArtifactDigestMismatch,
+    MissionFencingTokenStale,
+    PublicationLockTimeout,
+    ScriptBuildError,
+)
+from script_build_host.tools.failures import classify_script_tool_failure
+from script_build_host.tools.registry import register_script_tools
+
+from agent import (
+    AgentRunner,
+    FailureDetail,
+    FailureDisposition,
+    RunConfig,
+    ToolRegistry,
+)
+
+
+class _Coordinator:
+    async def ensure_ledger(self, *_args, **_kwargs):
+        return None
+
+
+def _knowledge_off():
+    from agent.tools.builtin.knowledge import KnowledgeConfig
+
+    return KnowledgeConfig(
+        enable_extraction=False,
+        enable_completion_extraction=False,
+        enable_injection=False,
+    )
+
+
+@pytest.mark.parametrize(
+    ("error", "expected"),
+    [
+        (
+            PhaseTwoCandidateError("INPUT_SCOPE_MISMATCH", "scope is invalid"),
+            FailureDisposition.REPLAN_TASK,
+        ),
+        (
+            PhaseTwoCandidateError("GOAL_COVERAGE_INCOMPLETE", "goal is missing"),
+            FailureDisposition.REPLAN_TASK,
+        ),
+        (
+            PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "workspace is invalid"),
+            FailureDisposition.REPAIR_ATTEMPT,
+        ),
+        (PublicationLockTimeout(), FailureDisposition.RETRY_CALL),
+        (MissionFencingTokenStale(), FailureDisposition.ABORT_RUN),
+        (ArtifactDigestMismatch(), FailureDisposition.ABORT_RUN),
+        (
+            ScriptBuildError("UNCLASSIFIED_DOMAIN_FAILURE", "fail closed"),
+            FailureDisposition.ABORT_RUN,
+        ),
+        (ValueError("bad argument"), FailureDisposition.RETRY_CALL),
+    ],
+)
+def test_script_failure_classification_is_central_and_fail_closed(error, expected):
+    failure = classify_script_tool_failure(
+        error,
+        source_tool="test_tool",
+        context={"task_id": "task-1", "attempt_id": "attempt-1"},
+    )
+
+    assert failure.disposition is expected
+    assert failure.source_tool == "test_tool"
+    assert failure.details["task_id"] == "task-1"
+
+
+@pytest.mark.asyncio
+async def test_compose_scope_failure_exits_worker_and_keeps_exact_domain_error(tmp_path):
+    class Gateway:
+        async def save_structured_script_candidate(self, _notes, _context):
+            raise PhaseTwoCandidateError(
+                "INPUT_SCOPE_MISMATCH",
+                "Paragraph has no adopted covering Structure",
+            )
+
+    register_script_presets()
+    registry = ToolRegistry()
+    register_script_tools(registry, Gateway())  # type: ignore[arg-type]
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {
+            "content": "",
+            "tool_calls": [{
+                "id": "save-compose",
+                "type": "function",
+                "function": {
+                    "name": "save_structured_script_candidate",
+                    "arguments": "{}",
+                },
+            }],
+            "finish_reason": "tool_calls",
+        }
+
+    result = await AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        tool_registry=registry,
+        llm_call=llm_call,
+        task_coordinator=_Coordinator(),
+    ).run_result(
+        [{"role": "user", "content": "compose"}],
+        RunConfig(
+            agent_type="script_compose_worker",
+            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            tools=["save_structured_script_candidate"],
+            tool_groups=[],
+            context={
+                "root_trace_id": "root",
+                "task_id": "compose-task",
+                "attempt_id": "attempt-1",
+            },
+            knowledge=_knowledge_off(),
+        ),
+    )
+
+    assert calls == 1
+    assert result["status"] == "failed"
+    assert result["failure"] == {
+        "code": "INPUT_SCOPE_MISMATCH",
+        "message": "Paragraph has no adopted covering Structure",
+        "disposition": "replan_task",
+        "source_tool": "save_structured_script_candidate",
+        "details": {"attempt_id": "attempt-1", "task_id": "compose-task"},
+    }
+
+
+@pytest.mark.asyncio
+async def test_workspace_error_can_be_repaired_in_same_attempt(tmp_path):
+    class Gateway:
+        def __init__(self):
+            self.write_calls = 0
+
+        async def candidate_command(self, name, _payload, _context):
+            assert name == "create_script_paragraph"
+            self.write_calls += 1
+            if self.write_calls == 1:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_WRITE_INVALID",
+                    "paragraph dimension is duplicated",
+                )
+            return {"created": True}
+
+        async def submit_current_attempt(self, _context):
+            return {"attempt_id": "attempt-1", "status": "awaiting_validation"}
+
+    register_script_presets()
+    gateway = Gateway()
+    registry = ToolRegistry()
+    register_script_tools(registry, gateway)  # type: ignore[arg-type]
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        if calls <= 2:
+            tool_name = "create_script_paragraph"
+            arguments = json.dumps({
+                "paragraph_index": 1,
+                "name": "opening",
+                "content_range": {"start": 0, "end": 100},
+            })
+        else:
+            tool_name = "submit_attempt"
+            arguments = "{}"
+        return {
+            "content": "",
+            "tool_calls": [{
+                "id": f"call-{calls}",
+                "type": "function",
+                "function": {"name": tool_name, "arguments": arguments},
+            }],
+            "finish_reason": "tool_calls",
+        }
+
+    result = await AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        tool_registry=registry,
+        llm_call=llm_call,
+        task_coordinator=_Coordinator(),
+    ).run_result(
+        [{"role": "user", "content": "write paragraph"}],
+        RunConfig(
+            agent_type="script_paragraph_worker",
+            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            tools=["create_script_paragraph", "submit_attempt"],
+            tool_groups=[],
+            context={"task_id": "paragraph-task", "attempt_id": "attempt-1"},
+            knowledge=_knowledge_off(),
+        ),
+    )
+
+    assert gateway.write_calls == 2
+    assert calls == 3
+    assert result["status"] == "completed"
+    assert result["failure"] is None
+
+
+@pytest.mark.asyncio
+async def test_dispatch_preserves_batch_and_surfaces_child_failure():
+    child_failure = FailureDetail(
+        code="INPUT_SCOPE_MISMATCH",
+        message="scope mismatch",
+        disposition=FailureDisposition.REPLAN_TASK,
+        source_tool="save_structured_script_candidate",
+    )
+
+    class Gateway:
+        async def dispatch_script_tasks(self, *, task_ids, context):
+            assert task_ids == ["compose-task", "other-task"]
+            assert context["root_trace_id"] == "root"
+            return [
+                {
+                    "task_id": "compose-task",
+                    "task_status": "needs_replan",
+                    "attempt_id": "attempt-1",
+                    "failure": child_failure.to_dict(),
+                },
+                {"task_id": "other-task", "task_status": "completed"},
+            ]
+
+    registry = ToolRegistry()
+    register_script_tools(registry, Gateway())  # type: ignore[arg-type]
+    result = await registry._tools["dispatch_script_tasks"]["func"](
+        task_ids=["compose-task", "other-task"],
+        context={"root_trace_id": "root"},
+    )
+
+    assert "other-task" in result.output
+    assert result.failure.code == "INPUT_SCOPE_MISMATCH"
+    assert result.failure.source_tool == "save_structured_script_candidate"
+    assert result.failure.details == {
+        "attempt_id": "attempt-1",
+        "task_id": "compose-task",
+    }
+
+
+@pytest.mark.asyncio
+async def test_planner_breaks_after_same_dispatched_failure_twice(tmp_path):
+    child_failure = FailureDetail(
+        code="INPUT_SCOPE_MISMATCH",
+        message="scope mismatch",
+        disposition=FailureDisposition.REPLAN_TASK,
+        source_tool="save_structured_script_candidate",
+    )
+
+    class Gateway:
+        async def dispatch_script_tasks(self, *, task_ids, context):
+            return [{
+                "task_id": task_ids[0],
+                "task_status": "needs_replan",
+                "attempt_id": "volatile-attempt-id",
+                "failure": child_failure.to_dict(),
+            }]
+
+    register_script_presets()
+    registry = ToolRegistry()
+    register_script_tools(registry, Gateway())  # type: ignore[arg-type]
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {
+            "content": "",
+            "tool_calls": [{
+                "id": f"dispatch-{calls}",
+                "type": "function",
+                "function": {
+                    "name": "dispatch_script_tasks",
+                    "arguments": json.dumps({"task_ids": ["compose-task"]}),
+                },
+            }],
+            "finish_reason": "tool_calls",
+        }
+
+    result = await AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        tool_registry=registry,
+        llm_call=llm_call,
+        task_coordinator=_Coordinator(),
+    ).run_result(
+        [{"role": "user", "content": "plan"}],
+        RunConfig(
+            agent_type="script_planner",
+            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            tools=["dispatch_script_tasks"],
+            tool_groups=[],
+            root_task_spec={
+                "objective": "deliver script",
+                "acceptance_criteria": [{"description": "script is accepted"}],
+            },
+            context={"root_trace_id": "root"},
+            knowledge=_knowledge_off(),
+        ),
+    )
+
+    assert calls == 2
+    assert result["status"] == "incomplete"
+    assert result["failure"]["code"] == "NO_PROGRESS"
+    assert result["failure"]["details"]["last_failure"]["code"] == (
+        "INPUT_SCOPE_MISMATCH"
+    )