主机：集中映射脚本工具的失败处置

script_build_host/src/script_build_host/tools/failures.py

@@ -0,0 +1,100 @@
+"""Map script-build domain errors to the framework's business-neutral protocol."""
+
+from __future__ import annotations
+
+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",
+    "PHASE_POLICY_VIOLATION",
+    "PHASE_TWO_BOUNDARY_NOT_READY",
+    "PHASE_THREE_BOUNDARY_NOT_READY",
+    "PROTOCOL_VIOLATION",
+    "TASK_CHILD_BLOCKED",
+    "TASK_BUDGET_EXCEEDED",
+    "ATTEMPT_WORKSPACE_FROZEN",
+    "WRITE_SCOPE_VIOLATION",
+    "SUPERSESSION_CYCLE",
+})
+_REPAIR_CODES = frozenset({
+    "LEGACY_WRITE_INVALID",
+    "LEGACY_REFERENCE_INVALID",
+    "ARTIFACT_NOT_FOUND",
+})
+_RETRY_CODES = frozenset({
+    "PUBLICATION_LOCK_TIMEOUT",
+    "PUBLICATION_DEADLOCK_RETRYABLE",
+    "STALE_BASE_REVISION",
+})
+_ABORT_MARKERS = (
+    "STOP",
+    "FENCING",
+    "OWNERSHIP",
+    "ALREADY_OWNED",
+    "DIGEST",
+    "RECOVERY",
+    "MIGRATION",
+    "RECONCILIATION",
+)
+
+
+def classify_script_tool_failure(
+    error: ScriptBuildError | ValueError,
+    *,
+    source_tool: str,
+    context: Mapping[str, Any] | None = None,
+) -> FailureDetail:
+    """Return the sole Host classification for an Agent-visible tool error."""
+
+    if isinstance(error, ValueError) and not isinstance(error, ScriptBuildError):
+        return FailureDetail(
+            code="INVALID_TOOL_ARGUMENT",
+            message=str(error),
+            disposition=FailureDisposition.RETRY_CALL,
+            source_tool=source_tool,
+            details=_identity_details(context),
+        )
+
+    code = error.code
+    if (
+        code in _REPLAN_CODES
+        or code.startswith("GOAL_")
+        or code.startswith("TASK_CONTRACT_")
+        or code.startswith("ADOPTION_")
+    ):
+        disposition = FailureDisposition.REPLAN_TASK
+    elif code in _REPAIR_CODES:
+        disposition = FailureDisposition.REPAIR_ATTEMPT
+    elif code in _RETRY_CODES:
+        disposition = FailureDisposition.RETRY_CALL
+    elif any(marker in code for marker in _ABORT_MARKERS):
+        disposition = FailureDisposition.ABORT_RUN
+    else:
+        disposition = FailureDisposition.ABORT_RUN
+    return FailureDetail(
+        code=code,
+        message=error.summary,
+        disposition=disposition,
+        source_tool=source_tool,
+        details=_identity_details(context),
+    )
+
+
+def _identity_details(context: Mapping[str, Any] | None) -> dict[str, str]:
+    if not isinstance(context, Mapping):
+        return {}
+    return {
+        key: str(context[key])
+        for key in ("task_id", "attempt_id", "validation_id", "operation_id")
+        if context.get(key) not in (None, "")
+    }
+
+
+__all__ = ["classify_script_tool_failure"]

script_build_host/src/script_build_host/tools/registry.py

@@ -5,11 +5,15 @@ from __future__ import annotations
 import base64
 import hashlib
 import json
+from functools import wraps
 from typing import Any
 
-from agent import ToolRegistry
+from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolRegistry
 from agent.tools.models import ToolResult
 
+from script_build_host.domain.errors import ScriptBuildError
+
+from .failures import classify_script_tool_failure
 from .gateway import LegacyScriptToolGateway
 
 TASK_PRESET_BY_KIND = {
@@ -169,14 +173,22 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     async def dispatch_script_tasks(
         task_ids: list[str], context: dict[str, Any] | None = None
-    ) -> str:
+    ) -> str | ToolResult:
         """Dispatch allowlisted script Tasks through durable operations."""
 
         results = await gateway.dispatch_script_tasks(
             task_ids=task_ids,
             context=context or {},
         )
-        return _json(results)
+        result_json = _json(results)
+        failure = _strongest_cycle_failure(results)
+        if failure is None:
+            return result_json
+        return ToolResult(
+            title="One or more dispatched script tasks failed",
+            output=result_json,
+            failure=failure,
+        )
 
     _register(registry, dispatch_script_tasks, capabilities=["task_control"])
 
@@ -754,8 +766,29 @@ def _register(
     capabilities: list[str],
     schema: dict[str, Any] | None = None,
 ) -> None:
+    @wraps(func)
+    async def guarded(*args: Any, **kwargs: Any) -> Any:
+        try:
+            return await func(*args, **kwargs)
+        except ScriptBuildError as exc:
+            raise ToolExecutionError(
+                classify_script_tool_failure(
+                    exc,
+                    source_tool=func.__name__,
+                    context=kwargs.get("context"),
+                )
+            ) from exc
+        except ValueError as exc:
+            raise ToolExecutionError(
+                classify_script_tool_failure(
+                    exc,
+                    source_tool=func.__name__,
+                    context=kwargs.get("context"),
+                )
+            ) from exc
+
     registry.register(
-        func,
+        guarded,
         schema=schema,
         hidden_params=["context"],
         groups=["script_build"],
@@ -763,6 +796,42 @@ def _register(
     )
 
 
+def _strongest_cycle_failure(results: Any) -> FailureDetail | None:
+    if not isinstance(results, (list, tuple)):
+        return None
+    priority = {
+        FailureDisposition.RETRY_CALL: 0,
+        FailureDisposition.REPAIR_ATTEMPT: 1,
+        FailureDisposition.REPLAN_TASK: 2,
+        FailureDisposition.ABORT_RUN: 3,
+    }
+    strongest: FailureDetail | None = None
+    for result in results:
+        value = result.get("failure") if isinstance(result, dict) else getattr(result, "failure", None)
+        if value is None:
+            continue
+        failure = FailureDetail.from_dict(value)
+        result_value = result if isinstance(result, dict) else result.to_dict()
+        details = {
+            **dict(failure.details),
+            **{
+                key: result_value[key]
+                for key in ("task_id", "attempt_id", "validation_id")
+                if result_value.get(key) not in (None, "")
+            },
+        }
+        failure = FailureDetail(
+            code=failure.code,
+            message=failure.message,
+            disposition=failure.disposition,
+            source_tool=failure.source_tool,
+            details=details,
+        )
+        if strongest is None or priority[failure.disposition] > priority[strongest.disposition]:
+            strongest = failure
+    return strongest
+
+
 def _decode_schema() -> dict[str, Any]:
     return {
         "type": "function",