框架：定义通用结构化失败协议

新增 FailureDetail、FailureDisposition 与 ToolExecutionError，约束错误文本和扩展详情的大小及 JSON 安全性。\n\n为 ToolResult 增加结构化失败载荷，并通过 Agent 与 tools 公共入口导出，给后续 Runner、编排层和业务 Host 提供统一且不包含业务概念的失败语言。\n\n补充值对象边界、指纹稳定性、序列化和模型反馈测试。

agent/agent/__init__.py

@@ -36,6 +36,7 @@ from agent.skill.models import Skill
 # 工具系统
 from agent.tools import tool, ToolRegistry, get_tool_registry
 from agent.tools.models import ToolCapability, ToolResult, ToolContext
+from agent.failures import FailureDetail, FailureDisposition, ToolExecutionError
 
 # 显式任务执行与独立验证
 from agent.orchestration import (
@@ -104,6 +105,9 @@ __all__ = [
     "ToolCapability",
     "ToolResult",
     "ToolContext",
+    "FailureDetail",
+    "FailureDisposition",
+    "ToolExecutionError",
     # Orchestration
     "CompletionPolicy",
     "AgentRole",

agent/agent/failures.py

@@ -0,0 +1,145 @@
+"""Business-neutral failure values shared by tools, runners and orchestration."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any, Mapping
+
+_MAX_CODE_LENGTH = 128
+_MAX_MESSAGE_LENGTH = 2_000
+_MAX_SOURCE_LENGTH = 200
+_MAX_DETAILS_BYTES = 16 * 1024
+_DETAIL_PREVIEW_LENGTH = 4_000
+_VOLATILE_DETAIL_KEYS = {
+    "attempt_id",
+    "operation_id",
+    "task_id",
+    "trace_id",
+    "validation_id",
+    "timestamp",
+    "updated_at",
+    "created_at",
+}
+
+
+class FailureDisposition(StrEnum):
+    """What the current role may safely do after a tool failure."""
+
+    RETRY_CALL = "retry_call"
+    REPAIR_ATTEMPT = "repair_attempt"
+    REPLAN_TASK = "replan_task"
+    ABORT_RUN = "abort_run"
+
+
+@dataclass(frozen=True)
+class FailureDetail:
+    """Bounded, JSON-safe failure information suitable for model feedback."""
+
+    code: str
+    message: str
+    disposition: FailureDisposition
+    source_tool: str | None = None
+    details: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        code = _bounded_text(self.code, _MAX_CODE_LENGTH, "code")
+        message = _bounded_text(self.message, _MAX_MESSAGE_LENGTH, "message")
+        source_tool = (
+            _bounded_text(self.source_tool, _MAX_SOURCE_LENGTH, "source_tool")
+            if self.source_tool is not None
+            else None
+        )
+        try:
+            disposition = FailureDisposition(self.disposition)
+        except (TypeError, ValueError) as exc:
+            raise ValueError("FailureDetail.disposition is invalid") from exc
+        safe_details = _bounded_details(self.details)
+        object.__setattr__(self, "code", code)
+        object.__setattr__(self, "message", message)
+        object.__setattr__(self, "source_tool", source_tool)
+        object.__setattr__(self, "disposition", disposition)
+        object.__setattr__(self, "details", safe_details)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "code": self.code,
+            "message": self.message,
+            "disposition": self.disposition.value,
+            "source_tool": self.source_tool,
+            "details": dict(self.details),
+        }
+
+    @classmethod
+    def from_dict(cls, value: "FailureDetail | Mapping[str, Any]") -> "FailureDetail":
+        if isinstance(value, cls):
+            return value
+        return cls(
+            code=str(value.get("code", "")),
+            message=str(value.get("message", "")),
+            disposition=FailureDisposition(value.get("disposition", "")),
+            source_tool=(
+                str(value["source_tool"])
+                if value.get("source_tool") is not None
+                else None
+            ),
+            details=(
+                value.get("details", {})
+                if isinstance(value.get("details", {}), Mapping)
+                else {}
+            ),
+        )
+
+    def fingerprint(self) -> str:
+        """Return a stable semantic fingerprint, excluding execution identities."""
+
+        semantic_details = {
+            key: value
+            for key, value in self.details.items()
+            if key not in _VOLATILE_DETAIL_KEYS and not key.endswith("_id")
+        }
+        payload = {
+            "code": self.code,
+            "disposition": self.disposition.value,
+            "source_tool": self.source_tool,
+            "details": semantic_details,
+        }
+        encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+class ToolExecutionError(Exception):
+    """Expected tool rejection carrying a safe, structured failure."""
+
+    def __init__(self, failure: FailureDetail) -> None:
+        self.failure = FailureDetail.from_dict(failure)
+        super().__init__(self.failure.message)
+
+
+def _bounded_text(value: object, limit: int, field_name: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"FailureDetail.{field_name} must be a non-blank string")
+    text = value.strip()
+    return text if len(text) <= limit else text[: limit - 1] + "…"
+
+
+def _bounded_details(value: Mapping[str, Any]) -> dict[str, Any]:
+    if not isinstance(value, Mapping):
+        raise ValueError("FailureDetail.details must be a mapping")
+    try:
+        normalized = json.loads(json.dumps(dict(value), ensure_ascii=False, sort_keys=True))
+    except (TypeError, ValueError) as exc:
+        raise ValueError("FailureDetail.details must contain JSON values") from exc
+    encoded = json.dumps(normalized, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    if len(encoded.encode("utf-8")) <= _MAX_DETAILS_BYTES:
+        return normalized
+    return {
+        "truncated": True,
+        "sha256": hashlib.sha256(encoded.encode("utf-8")).hexdigest(),
+        "preview": encoded[:_DETAIL_PREVIEW_LENGTH],
+    }
+
+
+__all__ = ["FailureDetail", "FailureDisposition", "ToolExecutionError"]

agent/agent/tools/__init__.py

@@ -5,6 +5,7 @@ Tools 包 - 工具注册和 Schema 生成
 from agent.tools.registry import ToolRegistry, tool, get_tool_registry
 from agent.tools.schema import SchemaGenerator
 from agent.tools.models import ToolCapability, ToolResult, ToolContext, ToolContextImpl
+from agent.failures import FailureDetail, FailureDisposition, ToolExecutionError
 
 # 导入 builtin 工具以触发 @tool 装饰器注册
 # noqa: F401 表示这是故意的副作用导入
@@ -19,4 +20,7 @@ __all__ = [
 	"ToolResult",
 	"ToolContext",
 	"ToolContextImpl",
+	"FailureDetail",
+	"FailureDisposition",
+	"ToolExecutionError",
 ]

agent/agent/tools/models.py

@@ -6,10 +6,13 @@ Tool Models - 工具系统核心数据模型
 2. ToolContext: 工具执行上下文（依赖注入）
 """
 
+import json
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, Dict, List, Optional, Protocol
 
+from agent.failures import FailureDetail
+
 
 class ToolCapability(str, Enum):
 	"""Security-relevant effects declared by every explicit-mode tool."""
@@ -52,6 +55,7 @@ class ToolResult:
 	# 状态标志
 	truncated: bool = False  # 输出是否被截断
 	error: Optional[str] = None  # 错误信息（如果执行失败）
+	failure: Optional[FailureDetail] = None  # 结构化失败（可传播到 Runner/Planner）
 	terminate_run: bool = False  # terminal tool：保存结果后结束当前 Agent Loop
 	result_summary: Optional[str] = None  # terminal tool 写入 Trace 的结构化摘要
 
@@ -73,6 +77,11 @@ class ToolResult:
 			给 LLM 的消息字符串
 		"""
 		# 如果有错误，优先返回错误
+		if self.failure:
+			payload: Dict[str, Any] = {"failure": self.failure.to_dict()}
+			if self.output:
+				payload["output"] = self.output
+			return json.dumps(payload, ensure_ascii=False, indent=2)
 		if self.error:
 			return f"Error: {self.error}"
 

agent/tests/test_failure_contract.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolResult
+
+
+def test_failure_detail_is_bounded_json_safe_and_round_trips() -> None:
+    failure = FailureDetail(
+        code="INPUT_SCOPE_MISMATCH",
+        message="x" * 3_000,
+        disposition=FailureDisposition.REPLAN_TASK,
+        source_tool="save_candidate",
+        details={"task_id": "task-1", "scope": ["paragraph", 3]},
+    )
+
+    assert len(failure.message) == 2_000
+    assert FailureDetail.from_dict(failure.to_dict()) == failure
+    assert failure.fingerprint() == FailureDetail(
+        code="INPUT_SCOPE_MISMATCH",
+        message="different wording",
+        disposition=FailureDisposition.REPLAN_TASK,
+        source_tool="save_candidate",
+        details={"task_id": "task-2", "scope": ["paragraph", 3]},
+    ).fingerprint()
+
+
+def test_failure_detail_rejects_non_json_details() -> None:
+    with pytest.raises(ValueError, match="JSON values"):
+        FailureDetail(
+            code="INVALID",
+            message="invalid",
+            disposition=FailureDisposition.RETRY_CALL,
+            details={"value": object()},
+        )
+
+
+def test_tool_result_exposes_structured_failure_to_model() -> None:
+    failure = FailureDetail(
+        code="REJECTED",
+        message="retry with another value",
+        disposition=FailureDisposition.RETRY_CALL,
+        source_tool="sample",
+    )
+    result = ToolResult(title="rejected", output="context", failure=failure)
+
+    payload = json.loads(result.to_llm_message())
+    assert payload["failure"] == failure.to_dict()
+    assert payload["output"] == "context"
+    assert ToolExecutionError(failure).failure == failure