增加 Validator 真实产物引用协议

cyber_agent/core/artifacts.py

@@ -0,0 +1,227 @@
+"""Recursive Validator 的最小只读产物协议。
+
+框架只保存不可变 ``ArtifactRef``，并通过应用注入的 ``ArtifactResolver``
+读取当前任务获授权的真实材料；上传、ACL、索引和生命周期不属于本模块。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from hashlib import sha256
+import json
+from typing import Any, Literal, Protocol, runtime_checkable
+
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
+
+
+class _StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+
+class ArtifactRef(_StrictModel):
+    """任务报告或标准 Tool Result 中的不可变产物句柄。"""
+
+    artifact_id: str = Field(min_length=1, max_length=500)
+    version: str = Field(min_length=1, max_length=200)
+    content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    kind: str = Field(min_length=1, max_length=100)
+    mime_type: str | None = Field(default=None, max_length=200)
+
+
+class ValidationMaterial(_StrictModel):
+    """Resolver 返回、可直接放入 Validator packet 的只读材料。"""
+
+    artifact_id: str = Field(min_length=1, max_length=500)
+    version: str = Field(min_length=1, max_length=200)
+    content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    kind: str = Field(min_length=1, max_length=100)
+    mime_type: str | None = Field(default=None, max_length=200)
+    root_trace_id: str = Field(min_length=1)
+    uid: str | None = None
+    content: Any
+
+    @field_validator("content")
+    @classmethod
+    def validate_content(cls, value: Any) -> Any:
+        canonical_material_content(value)
+        return value
+
+    def to_ref(self) -> ArtifactRef:
+        return ArtifactRef(
+            artifact_id=self.artifact_id,
+            version=self.version,
+            content_hash=self.content_hash,
+            kind=self.kind,
+            mime_type=self.mime_type,
+        )
+
+
+@runtime_checkable
+class ArtifactResolver(Protocol):
+    """由业务应用实现的可信、只读产物解析端口。"""
+
+    async def resolve(
+        self,
+        ref: ArtifactRef,
+        root_trace_id: str,
+        uid: str | None,
+    ) -> ValidationMaterial:
+        ...
+
+
+class ArtifactResolutionError(RuntimeError):
+    """带确定性语义的产物解析错误。"""
+
+    def __init__(
+        self,
+        outcome: Literal["failed", "unknown", "error"],
+        message: str,
+    ) -> None:
+        super().__init__(message)
+        self.outcome = outcome
+
+
+class ArtifactNotFound(ArtifactResolutionError):
+    def __init__(self, message: str = "Artifact does not exist") -> None:
+        super().__init__("failed", message)
+
+
+class ArtifactTemporarilyUnavailable(ArtifactResolutionError):
+    def __init__(self, message: str = "Artifact resolver is unavailable") -> None:
+        super().__init__("unknown", message)
+
+
+class ArtifactAccessDenied(ArtifactResolutionError):
+    def __init__(self, message: str = "Artifact is not authorized") -> None:
+        super().__init__("error", message)
+
+
+class MaterialIssue(_StrictModel):
+    """框架在 LLM 调用前产生的材料确定性检查结果。"""
+
+    artifact_id: str
+    outcome: Literal["failed", "unknown", "error"]
+    reason: str = Field(min_length=1)
+    scopes: list[Literal["evidence", "hypothesis", "output", "task", "root"]] = (
+        Field(default_factory=lambda: [
+            "evidence", "hypothesis", "output", "task", "root",
+        ], min_length=1)
+    )
+
+    @field_validator("scopes")
+    @classmethod
+    def normalize_scopes(cls, value: list[str]) -> list[str]:
+        order = ("evidence", "hypothesis", "output", "task", "root")
+        selected = set(value)
+        return [scope for scope in order if scope in selected]
+
+
+def canonical_material_content(content: Any) -> str:
+    """稳定序列化材料内容，供字符计数和 SHA-256 校验共用。"""
+    if isinstance(content, str):
+        return content
+    try:
+        return json.dumps(
+            content,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"Artifact content must be JSON serializable: {exc}") from exc
+
+
+def material_content_hash(content: Any) -> str:
+    return sha256(canonical_material_content(content).encode("utf-8")).hexdigest()
+
+
+def material_chars(material: ValidationMaterial) -> int:
+    return len(canonical_material_content(material.content))
+
+
+def extract_artifact_refs(value: Any) -> list[ArtifactRef]:
+    """只读取标准对象顶层 ``artifact_refs``，非法元数据直接失败关闭。"""
+    if isinstance(value, BaseModel):
+        value = value.model_dump(mode="json")
+    if not isinstance(value, Mapping) or "artifact_refs" not in value:
+        return []
+    raw_refs = value.get("artifact_refs")
+    if not isinstance(raw_refs, Sequence) or isinstance(
+        raw_refs, (str, bytes, bytearray)
+    ):
+        raise ValueError("artifact_refs must be an array")
+    found = [ArtifactRef.model_validate(raw) for raw in raw_refs]
+    unique: dict[tuple[str, str, str], ArtifactRef] = {}
+    for ref in found:
+        unique[(ref.artifact_id, ref.version, ref.content_hash)] = ref
+    return list(unique.values())
+
+
+async def resolve_artifact_refs(
+    refs: Sequence[ArtifactRef],
+    *,
+    resolver: ArtifactResolver | None,
+    root_trace_id: str,
+    uid: str | None,
+) -> tuple[list[ValidationMaterial], list[MaterialIssue]]:
+    """逐个解析引用并把缺失、暂不可用和越权映射为稳定结果。"""
+    materials: list[ValidationMaterial] = []
+    issues: list[MaterialIssue] = []
+    for ref in refs:
+        issue_scopes = _artifact_validation_scopes(ref.kind)
+        if resolver is None:
+            issues.append(MaterialIssue(
+                artifact_id=ref.artifact_id,
+                outcome="unknown",
+                reason="No ArtifactResolver is configured for this artifact",
+                scopes=issue_scopes,
+            ))
+            continue
+        try:
+            material = ValidationMaterial.model_validate(
+                await resolver.resolve(ref, root_trace_id, uid)
+            )
+            if material.root_trace_id != root_trace_id or material.uid != uid:
+                raise ArtifactAccessDenied(
+                    "Resolved artifact belongs to another task tree or owner"
+                )
+            if material.to_ref() != ref:
+                raise ArtifactAccessDenied(
+                    "Resolved artifact metadata does not match the authorized reference"
+                )
+            if material_content_hash(material.content) != ref.content_hash:
+                raise ArtifactNotFound("Resolved artifact content hash does not match")
+            materials.append(material)
+        except ArtifactResolutionError as exc:
+            issues.append(MaterialIssue(
+                artifact_id=ref.artifact_id,
+                outcome=exc.outcome,
+                reason=str(exc),
+                scopes=issue_scopes,
+            ))
+        except (ValidationError, TypeError, ValueError) as exc:
+            issues.append(MaterialIssue(
+                artifact_id=ref.artifact_id,
+                outcome="error",
+                reason=f"Artifact resolver returned invalid data: {exc}",
+                scopes=issue_scopes,
+            ))
+        except Exception as exc:
+            issues.append(MaterialIssue(
+                artifact_id=ref.artifact_id,
+                outcome="unknown",
+                reason=f"Artifact resolver failed: {exc}",
+                scopes=issue_scopes,
+            ))
+    return materials, issues
+
+
+def _artifact_validation_scopes(
+    kind: str,
+) -> list[Literal["evidence", "hypothesis", "output", "task", "root"]]:
+    """按产物类型限制确定性失败的影响面，未受影响 Scope 仍继续验收。"""
+    normalized = kind.strip().lower()
+    if any(marker in normalized for marker in ("evidence", "source", "reference")):
+        return ["evidence", "hypothesis", "task", "root"]
+    return ["output", "task", "root"]

cyber_agent/tools/models.py

@@ -40,6 +40,7 @@ class ToolResult:
 	# 附件支持（用于浏览器自动化等场景）
 	attachments: List[str] = field(default_factory=list)  # 文件路径列表
 	images: List[Dict[str, Any]] = field(default_factory=list)  # 图片列表
+	artifact_refs: List[Dict[str, Any]] = field(default_factory=list)  # Validator可解析的标准只读产物引用
 
 	# Token追踪（用于工具内部LLM调用）
 	tool_usage: Optional[Dict[str, Any]] = None  # 格式：{"model": "...", "prompt_tokens": 100, "completion_tokens": 50, "cost": 0.0}

cyber_agent/tools/registry.py

@@ -351,6 +351,11 @@ class ToolRegistry:
 				if result.tool_usage:
 					ret["tool_usage"] = result.tool_usage
 
+				# 仅透传标准 artifact_refs；不把任意 metadata 暴露给模型或 Validator。
+				artifact_refs = result.artifact_refs or result.metadata.get("artifact_refs", [])
+				if artifact_refs:
+					ret["artifact_refs"] = artifact_refs
+
 				# 向后兼容：只有text时返回字符串
 				if len(ret) == 1:
 					return ret["text"]

tests/test_recursive_validation_artifacts.py

@@ -0,0 +1,199 @@
+import json
+import unittest
+
+from pydantic import ValidationError
+
+from cyber_agent.core.artifacts import (
+    ArtifactAccessDenied,
+    ArtifactNotFound,
+    ArtifactRef,
+    ArtifactTemporarilyUnavailable,
+    ValidationMaterial,
+    extract_artifact_refs,
+    material_content_hash,
+    resolve_artifact_refs,
+)
+from cyber_agent.tools.models import ToolResult
+from cyber_agent.tools.registry import ToolRegistry
+
+
+CONTENT = {"script": "official value is 12%", "revision": 2}
+CONTENT_HASH = material_content_hash(CONTENT)
+REF = ArtifactRef(
+    artifact_id="script:42",
+    version="v2",
+    content_hash=CONTENT_HASH,
+    kind="script",
+    mime_type="application/json",
+)
+
+
+def material(**overrides):
+    values = {
+        **REF.model_dump(mode="json"),
+        "root_trace_id": "root-1",
+        "uid": "user-1",
+        "content": CONTENT,
+    }
+    values.update(overrides)
+    return ValidationMaterial.model_validate(values)
+
+
+class FakeResolver:
+    def __init__(self, value):
+        self.value = value
+        self.calls = []
+
+    async def resolve(self, ref, root_trace_id, uid):
+        self.calls.append((ref, root_trace_id, uid))
+        if isinstance(self.value, Exception):
+            raise self.value
+        return self.value
+
+
+class ArtifactProtocolTest(unittest.IsolatedAsyncioTestCase):
+    def test_ref_is_strict_and_hash_is_lowercase_sha256(self):
+        with self.assertRaises(ValidationError):
+            ArtifactRef.model_validate({
+                **REF.model_dump(),
+                "content_hash": "A" * 64,
+            })
+        with self.assertRaises(ValidationError):
+            ArtifactRef.model_validate({**REF.model_dump(), "extra": True})
+
+    def test_hash_is_stable_for_json_key_order(self):
+        self.assertEqual(
+            material_content_hash({"b": 2, "a": 1}),
+            material_content_hash({"a": 1, "b": 2}),
+        )
+
+    def test_extract_only_reads_top_level_standard_metadata(self):
+        self.assertEqual([], extract_artifact_refs({
+            "result": {"artifact_refs": [REF.model_dump()]},
+        }))
+        refs = extract_artifact_refs({
+            "artifact_refs": [REF.model_dump(), REF.model_dump()],
+            "untrusted": {"artifact_refs": []},
+        })
+        self.assertEqual([REF], refs)
+
+    def test_malformed_artifact_metadata_fails_closed(self):
+        for payload in (
+            {"artifact_refs": "not-an-array"},
+            {"artifact_refs": [{"artifact_id": "missing-fields"}]},
+        ):
+            with self.subTest(payload=payload), self.assertRaises(
+                (ValueError, ValidationError)
+            ):
+                extract_artifact_refs(payload)
+
+    async def test_valid_resolver_material_is_accepted(self):
+        resolver = FakeResolver(material())
+        materials, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=resolver,
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual([], issues)
+        self.assertEqual([material()], materials)
+        self.assertEqual((REF, "root-1", "user-1"), resolver.calls[0])
+
+    async def test_missing_resolver_is_unknown(self):
+        materials, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=None,
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual([], materials)
+        self.assertEqual("unknown", issues[0].outcome)
+
+    async def test_explicit_not_found_is_failed(self):
+        _, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=FakeResolver(ArtifactNotFound("gone")),
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual("failed", issues[0].outcome)
+
+    async def test_temporary_failure_is_unknown(self):
+        _, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=FakeResolver(ArtifactTemporarilyUnavailable("retry later")),
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual("unknown", issues[0].outcome)
+
+    async def test_access_denied_and_cross_tree_are_errors(self):
+        for value in (
+            ArtifactAccessDenied("denied"),
+            material(root_trace_id="other-root"),
+            material(uid="other-user"),
+        ):
+            with self.subTest(value=value):
+                _, issues = await resolve_artifact_refs(
+                    [REF],
+                    resolver=FakeResolver(value),
+                    root_trace_id="root-1",
+                    uid="user-1",
+                )
+                self.assertEqual("error", issues[0].outcome)
+
+    async def test_version_or_metadata_mismatch_is_error(self):
+        changed = material(version="v3")
+        _, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=FakeResolver(changed),
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual("error", issues[0].outcome)
+
+    async def test_content_hash_mismatch_is_failed(self):
+        forged = ValidationMaterial.model_validate({
+            **REF.model_dump(mode="json"),
+            "root_trace_id": "root-1",
+            "uid": "user-1",
+            "content": {"script": "forged 30%"},
+        })
+        _, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=FakeResolver(forged),
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual("failed", issues[0].outcome)
+
+    async def test_untyped_resolver_exception_is_unknown(self):
+        _, issues = await resolve_artifact_refs(
+            [REF],
+            resolver=FakeResolver(RuntimeError("database unavailable")),
+            root_trace_id="root-1",
+            uid="user-1",
+        )
+        self.assertEqual("unknown", issues[0].outcome)
+
+    async def test_tool_result_preserves_only_standard_artifact_refs(self):
+        registry = ToolRegistry()
+
+        async def produce_script():
+            return ToolResult(
+                title="script",
+                output="created",
+                metadata={
+                    "artifact_refs": [REF.model_dump(mode="json")],
+                    "private": "must-not-leak",
+                },
+            )
+
+        registry.register(produce_script)
+        result = await registry.execute("produce_script", {})
+        self.assertEqual([REF.model_dump(mode="json")], result["artifact_refs"])
+        self.assertNotIn("private", json.dumps(result))
+
+
+if __name__ == "__main__":
+    unittest.main()