封印Artifact内容并统一完整性复验

docs/adr/0001-source-asset-expectation-binding.md

@@ -37,6 +37,8 @@ Global Data 使用唯一身份链：
 - 只有 `reference-inspection` 可以满足 `source_identity`；生成媒体的 `source_uri`
   只是参考来源。
 - 一个 Artifact 可以满足多个 Expectation，但每条关系必须单独声明。
+- 本地正式 Artifact 必须保存内容 SHA-256 与字节数；Task 验收、Stage 收口和
+  completed-run 重验使用同一套完整性检查，路径存在但内容变化仍视为无效。
 - ExpectationGap 使用结构化身份和原因码，不解析错误文案。
 - Validator 逐项返回 `(expectation_id, verification_capability)` 的判断；Requirement
   重要性和总体结论由代码从 Plan 重算。

production_build_agents/agents/executor/agent.py

@@ -39,9 +39,15 @@ from ...contracts.models import (
 )
 from ...contracts.evaluation import evaluate_task_delivery
 from ...run.records import (
+    executor_delivery_artifact_text,
     save_executor_candidate,
     save_executor_delivery_artifact,
 )
+from ...run.artifacts import (
+    content_identity_for_bytes,
+    content_identity_for_uri,
+    verify_delivery_artifacts,
+)
 from ...tools.registry import (
     SIDE_EFFECT_TOOL_IDS,
     ToolRegistry,
@@ -509,6 +515,7 @@ def _build_artifacts(
     materialized_artifact_path: Path,
     task: TaskPackage,
     planned_task: PlannedTask,
+    records: list[ToolCallRecord],
 ) -> list[Artifact]:
     if candidate.artifacts:
         return [
@@ -520,6 +527,7 @@ def _build_artifacts(
                 uri=item.uri,
                 source_uri=item.source_uri,
                 description=item.description,
+                **content_identity_for_uri(item.uri, records),
             )
             for index, item in enumerate(candidate.artifacts, start=1)
         ]
@@ -533,12 +541,24 @@ def _build_artifacts(
         raise ExecutorOutputError(
             f"{planned_task.deliverable_type} 交付缺少真实媒体 Artifact"
         )
+    source_urls = sorted(
+        {
+            source
+            for finding in candidate.findings
+            for source in finding.source_urls
+        }
+    )
     return [
         Artifact(
             artifact_id=f"{task.task_id}-v{task.plan_version}-artifact-1",
             artifact_type=artifact_type,
             uri=str(materialized_artifact_path.resolve()),
+            source_uri=source_urls[0] if len(source_urls) == 1 else None,
             description=candidate.summary,
+            **content_identity_for_bytes(
+                executor_delivery_artifact_text(candidate).encode("utf-8"),
+                filename=materialized_artifact_path.name,
+            ),
         )
     ]
 
@@ -591,6 +611,7 @@ def _build_delivery(
         materialized_artifact_path=materialized_artifact_path,
         task=task,
         planned_task=planned_task,
+        records=records,
     )
     delivery = ExecutorDelivery(
         run_id=task.run_id,
@@ -716,4 +737,5 @@ def run_executor_agent(
         )
         if artifact_path.resolve() != expected_artifact_path.resolve():
             raise ExecutorOutputError("结构化 Artifact 落盘路径与 Delivery 不一致")
+    verify_delivery_artifacts(delivery.artifacts, delivery.tool_calls)
     return delivery

production_build_agents/agents/validator/task_context.py

@@ -24,6 +24,10 @@ from ...contracts.evaluation import (
     task_expectation_pairs,
 )
 from ...preprocess.production_brief import read_brief_paths
+from ...run.artifacts import (
+    ArtifactIntegrityError,
+    verify_delivery_artifacts,
+)
 from ..capabilities import CAPABILITIES
 from ..executor.context import (
     load_dependency_delivery_summaries,
@@ -121,7 +125,9 @@ def _load_materialized_artifacts(
             continue
         path = Path(artifact.uri)
         if not path.is_file():
-            continue
+            raise ValidatorContextError(
+                f"本地结构化 Artifact 不存在：{path}"
+            )
         try:
             content = json.loads(path.read_text(encoding="utf-8"))
         except (OSError, ValueError) as exc:
@@ -260,6 +266,10 @@ def build_validator_user_message(
 ) -> dict[str, Any]:
     """只传最终产物和显式证据，不传 Executor 的对话或自评。"""
 
+    try:
+        verify_delivery_artifacts(delivery.artifacts, delivery.tool_calls)
+    except ArtifactIntegrityError as exc:
+        raise ValidatorContextError(str(exc)) from exc
     evidence, image_blocks = _media_evidence(
         skill=skill,
         delivery=delivery,

production_build_agents/contracts/models.py

@@ -384,6 +384,9 @@ class Artifact(CandidateArtifact):
     """Executor 已验证并持久化的产物引用。"""
 
     artifact_id: ArtifactId
+    content_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
+    size_bytes: int = Field(ge=0)
+    mime_type: Optional[str] = None
 
 
 class ToolArtifactMapping(BaseModel):

production_build_agents/global_data_stage_assembly.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 from pathlib import Path
 from typing import Any, Mapping
-from urllib.parse import urlparse
 
 from .contracts.models import (
     ExecutorDelivery,
@@ -29,6 +28,10 @@ from .contracts.evaluation import (
 from .agents.capabilities import CAPABILITIES
 from .agents.validator.stage_agent import validate_global_data_stage_report
 from .agents.validator.task_context import validate_validation_report
+from .run.artifacts import (
+    ArtifactIntegrityError,
+    verify_delivery_artifacts,
+)
 
 
 class GlobalDataStageError(ValueError):
@@ -127,6 +130,13 @@ def _read_task_evidence(
                 f"{planned.task_id} 的 Delivery 合同无效："
                 + "；".join(item.message for item in delivery_issues)
             )
+        try:
+            verify_delivery_artifacts(
+                delivery.artifacts,
+                delivery.tool_calls,
+            )
+        except ArtifactIntegrityError as exc:
+            raise GlobalDataStageError(str(exc)) from exc
         validate_validation_report(package_plan, task, delivery, report)
         if report.verdict != "PASS":
             raise GlobalDataStageError(
@@ -295,14 +305,6 @@ def build_global_data_stage_delivery(
             if item.importance == "minor"
         ),
     }
-    for artifact in evaluation.active_artifacts:
-        parsed = urlparse(artifact.uri)
-        if parsed.scheme not in {"http", "https"} and not Path(
-            artifact.uri
-        ).is_file():
-            raise GlobalDataStageError(
-                f"Active Artifact 不存在：{artifact.artifact_id}"
-            )
     return GlobalDataStageDelivery(
         run_id=candidate.run_id,
         plan_id=candidate.plan_id,

production_build_agents/run/artifacts.py

@@ -0,0 +1,124 @@
+"""正式 Artifact 的内容封印与统一复验。"""
+
+from __future__ import annotations
+
+import hashlib
+import mimetypes
+from pathlib import Path
+from typing import Iterable
+from urllib.parse import urlparse
+
+from ..contracts.models import Artifact, ToolCallRecord
+
+
+class ArtifactIntegrityError(ValueError):
+    """正式 Artifact 缺失、无法定位或内容与 Delivery 不一致。"""
+
+
+def _local_path_for_uri(
+    uri: str,
+    tool_calls: Iterable[ToolCallRecord],
+) -> Path | None:
+    if urlparse(uri).scheme not in {"http", "https"}:
+        return Path(uri).resolve()
+    mapped = {
+        Path(mapping.local_path).resolve()
+        for record in tool_calls
+        for mapping in record.artifact_mappings
+        if mapping.source == uri
+    }
+    if not mapped:
+        return None
+    if len(mapped) != 1:
+        raise ArtifactIntegrityError(
+            f"远程 Artifact 对应多个本地内容：{uri}"
+        )
+    return mapped.pop()
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def content_identity_for_bytes(
+    content: bytes,
+    *,
+    filename: str,
+) -> dict[str, str | int | None]:
+    """为尚未落盘但字节已经确定的正式 Artifact 生成内容身份。"""
+
+    return {
+        "content_sha256": hashlib.sha256(content).hexdigest(),
+        "size_bytes": len(content),
+        "mime_type": mimetypes.guess_type(filename)[0],
+    }
+
+
+def content_identity_for_path(path: Path) -> dict[str, str | int | None]:
+    """读取一次本地文件，返回可持久化的内容身份。"""
+
+    resolved = path.resolve()
+    if not resolved.is_file():
+        raise ArtifactIntegrityError(f"正式 Artifact 不存在：{resolved}")
+    return {
+        "content_sha256": _sha256(resolved),
+        "size_bytes": resolved.stat().st_size,
+        "mime_type": mimetypes.guess_type(resolved.name)[0],
+    }
+
+
+def content_identity_for_uri(
+    uri: str,
+    tool_calls: Iterable[ToolCallRecord],
+) -> dict[str, str | int | None]:
+    """封印正式 Artifact；远程 URI 必须有本地工具映射。"""
+
+    path = _local_path_for_uri(uri, tool_calls)
+    if path is None:
+        raise ArtifactIntegrityError(
+            f"远程 Artifact 缺少可封印的本地内容：{uri}"
+        )
+    return content_identity_for_path(path)
+
+
+def verify_artifact(
+    artifact: Artifact,
+    tool_calls: Iterable[ToolCallRecord],
+) -> None:
+    """根据 Delivery 中保存的身份复验同一份 Artifact 内容。"""
+
+    path = _local_path_for_uri(artifact.uri, tool_calls)
+    if path is None:
+        raise ArtifactIntegrityError(
+            f"远程 Artifact 缺少可复验的本地映射：{artifact.artifact_id}"
+        )
+    actual = content_identity_for_path(path)
+    if (
+        artifact.content_sha256 != actual["content_sha256"]
+        or artifact.size_bytes != actual["size_bytes"]
+    ):
+        raise ArtifactIntegrityError(
+            f"Artifact 内容已变化：{artifact.artifact_id}"
+        )
+    if (
+        artifact.mime_type is not None
+        and artifact.mime_type != actual["mime_type"]
+    ):
+        raise ArtifactIntegrityError(
+            f"Artifact MIME 类型已变化：{artifact.artifact_id}"
+        )
+
+
+def verify_delivery_artifacts(
+    artifacts: Iterable[Artifact],
+    tool_calls: Iterable[ToolCallRecord],
+) -> None:
+    """在各运行边界复用同一套正式 Artifact 完整性规则。"""
+
+    records = tuple(tool_calls)
+    for artifact in artifacts:
+        verify_artifact(artifact, records)

production_build_agents/run/records.py

@@ -135,8 +135,13 @@ def save_executor_delivery_artifact(
 ) -> Path:
     """把结构化业务内容单独落盘，不让 Candidate 审计记录充当 Artifact。"""
 
-    return write_once_json(
-        path,
+    return write_once_text(path, executor_delivery_artifact_text(candidate))
+
+
+def executor_delivery_artifact_text(candidate: ExecutorCandidate) -> str:
+    """返回结构化 Artifact 将要落盘的稳定字节内容。"""
+
+    return _serialize_json(
         candidate.model_dump(
             mode="json",
             include={
@@ -150,7 +155,7 @@ def save_executor_delivery_artifact(
                 "findings",
                 "summary",
             },
-        ),
+        )
     )
 
 

production_build_agents/tools/media.py

@@ -636,6 +636,8 @@ def extract_frames(
         )
     return {
         "success": True,
+        "source": video,
+        "local_path": str(metadata["path"]),
         "duration_sec": duration,
         "frames": frames,
     }

tests/agents/test_executor.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import tempfile
 import unittest
@@ -61,6 +62,16 @@ IMAGE_URL = "https://example.test/generated-reference.png"
 SOURCE_URL = "https://example.test/research-source"
 VIDEO_URL = "https://example.test/generated-video.mp4"
 FRAME_URL = "https://example.test/generated-video-frame.jpg"
+_FAKE_MEDIA_TEMP = tempfile.TemporaryDirectory()
+_FAKE_MEDIA_DIR = Path(_FAKE_MEDIA_TEMP.name)
+
+
+def _fake_media_path(source: str, suffix: str) -> Path:
+    path = _FAKE_MEDIA_DIR / (
+        hashlib.sha256(source.encode("utf-8")).hexdigest() + suffix
+    )
+    path.write_bytes(source.encode("utf-8"))
+    return path
 
 
 def _fake_tool_registry() -> ToolRegistry:
@@ -108,6 +119,9 @@ def _fake_tool_registry() -> ToolRegistry:
             "images": [
                 {
                     "source": source,
+                    "local_path": str(
+                        _fake_media_path(source, ".png")
+                    ),
                     "data_url": (
                         "data:image/png;base64,"
                         "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
@@ -142,6 +156,8 @@ def _fake_tool_registry() -> ToolRegistry:
 
         return {
             "success": True,
+            "source": video,
+            "local_path": str(_fake_media_path(video, ".mp4")),
             "frames": [
                 {
                     "source": FRAME_URL,

tests/agents/test_task_validator.py

@@ -4,6 +4,7 @@ import json
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 from langchain_core.messages import AIMessage
 
@@ -36,6 +37,7 @@ from production_build_agents.contracts.models import (
     ValidatorCandidate,
 )
 from production_build_agents.tools.registry import ToolRegistry
+from production_build_agents.run.artifacts import content_identity_for_path
 from tests.support.fake_models import ToolAwareFakeChatModel
 from tests.support.planner_fixtures import (
     build_planned_task,
@@ -171,6 +173,11 @@ def _delivery(
                 artifact_id=artifact_id,
                 artifact_type=expectation.artifact_type,
                 uri=final_uri,
+                **(
+                    {}
+                    if media
+                    else content_identity_for_path(materialized_artifact)
+                ),
                 description="最终测试产物",
             )
         ],
@@ -375,6 +382,34 @@ class GeneralValidatorTest(unittest.TestCase):
                     tool_registry=ToolRegistry({}),
                 )
 
+    def test_missing_structured_artifact_is_rejected_before_model_execution(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            task = _task(root)
+            delivery = _delivery(root, task)
+            Path(delivery.artifacts[0].uri).unlink()
+            model = _model(_candidate(task, delivery))
+            with (
+                patch.object(
+                    model,
+                    "_generate",
+                    side_effect=AssertionError("Validator 模型不应被调用"),
+                ),
+                self.assertRaisesRegex(
+                    ValidatorContextError,
+                    "正式 Artifact 不存在",
+                ),
+            ):
+                run_validator_agent(
+                    task,
+                    delivery,
+                    run_dir=root,
+                    model=model,
+                    tool_registry=ToolRegistry({}),
+                )
+
     def test_invalid_target_retries_inside_same_validator_run(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             root = Path(temp_dir)

tests/agents/test_validator_media_evidence.py

@@ -27,6 +27,8 @@ def _artifact(artifact_type: str, uri: str, *, ordinal: int = 1) -> Artifact:
         artifact_id=f"Task1-v1-artifact-{ordinal}",
         artifact_type=artifact_type,
         uri=uri,
+        content_sha256="0" * 64,
+        size_bytes=1,
         description="媒体测试产物",
     )
 

tests/contracts/test_evaluation.py

@@ -188,6 +188,8 @@ def _delivery(
             artifact_id=artifact_id,
             artifact_type=artifact_type,
             uri=f"/cache/{artifact_id}.png",
+            content_sha256="0" * 64,
+            size_bytes=1,
             source_uri=(
                 source_by_id[source_id_by_artifact[artifact_id]].source_uri
                 if artifact_id in source_id_by_artifact
@@ -206,6 +208,8 @@ def _delivery(
                 ),
                 artifact_type=artifact_type,
                 uri="/cache/unbound.png",
+                content_sha256="0" * 64,
+                size_bytes=1,
                 description="辅助产物",
             )
         )

tests/fixtures/v18_replay/deliveries/delivery.Task1.v1.json

@@ -13,6 +13,9 @@
       "artifact_id": "Task1-v1-artifact-1",
       "artifact_type": "image",
       "uri": "/fixture/reference_cache/character.png",
+      "content_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
+      "size_bytes": 1,
+      "mime_type": "image/png",
       "source_uri": "https://res.cybertogether.net/crawler/image/d20b24da7f704398a15a46b3e3711858.png",
       "description": "v18 角色基准图的脱敏缓存引用"
     }

tests/fixtures/v18_replay/deliveries/delivery.Task2.v2.json

@@ -13,6 +13,9 @@
       "artifact_id": "Task2-v2-artifact-1",
       "artifact_type": "video",
       "uri": "/fixture/reference_cache/gesture.mp4",
+      "content_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
+      "size_bytes": 1,
+      "mime_type": "video/mp4",
       "source_uri": "https://res.cybertogether.net/crawler/skeleton/2973c6def09244579501ff4235711d74.mp4",
       "description": "v18 手势引导视频的脱敏缓存引用"
     }

tests/fixtures/v18_replay/stage_reports/expected_stage.v1.json

@@ -56,6 +56,9 @@
       "artifact_id": "Task1-v1-artifact-1",
       "artifact_type": "image",
       "uri": "/fixture/reference_cache/character.png",
+      "content_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
+      "size_bytes": 1,
+      "mime_type": "image/png",
       "source_uri": "https://res.cybertogether.net/crawler/image/d20b24da7f704398a15a46b3e3711858.png",
       "description": "v18 角色基准图的脱敏缓存引用"
     }

tests/fixtures/v18_replay/stage_reports/expected_stage.v2.json

@@ -53,6 +53,9 @@
       "artifact_id": "Task1-v1-artifact-1",
       "artifact_type": "image",
       "uri": "/fixture/reference_cache/character.png",
+      "content_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
+      "size_bytes": 1,
+      "mime_type": "image/png",
       "source_uri": "https://res.cybertogether.net/crawler/image/d20b24da7f704398a15a46b3e3711858.png",
       "description": "v18 角色基准图的脱敏缓存引用"
     },
@@ -60,6 +63,9 @@
       "artifact_id": "Task2-v2-artifact-1",
       "artifact_type": "video",
       "uri": "/fixture/reference_cache/gesture.mp4",
+      "content_sha256": "0000000000000000000000000000000000000000000000000000000000000000",
+      "size_bytes": 1,
+      "mime_type": "video/mp4",
       "source_uri": "https://res.cybertogether.net/crawler/skeleton/2973c6def09244579501ff4235711d74.mp4",
       "description": "v18 手势引导视频的脱敏缓存引用"
     }

tests/graph/test_recovery.py

@@ -512,6 +512,34 @@ class DurableRunTest(unittest.TestCase):
                         registry_dir=output_root / ".registry",
                     )
 
+    def test_completed_noop_rejects_tampered_artifact_content(self) -> None:
+        run_id = "Run-noop-artifact-tamper"
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_root = Path(temp_dir)
+            result = _completed_run(output_root, run_id)
+            delivery = ExecutorDelivery.model_validate_json(
+                Path(
+                    result["task_records"]["Task1"][
+                        "executor_delivery_path"
+                    ]
+                ).read_text(encoding="utf-8")
+            )
+            artifact_path = Path(delivery.artifacts[0].uri)
+            artifact_path.write_text(
+                '{"tampered": true}\n',
+                encoding="utf-8",
+            )
+            with self.assertRaisesRegex(
+                ValueError,
+                "Artifact 内容已变化",
+            ):
+                run_production(
+                    INPUT_PATH,
+                    output_root=output_root,
+                    thread_id=run_id,
+                    registry_dir=output_root / ".registry",
+                )
+
     def test_resume_rejects_input_drift(self) -> None:
         run_id = "Run-input-conflict"
         with tempfile.TemporaryDirectory() as temp_dir:

tests/support/executor_fixtures.py

@@ -52,13 +52,7 @@ def build_executor_candidate(
         findings = []
     elif deliverable_type == "research_result":
         payload = {}
-        artifacts = [
-            CandidateArtifact(
-                artifact_type="research_result",
-                uri=artifact_uri or "https://example.test/source",
-                description="可追溯的研究结果",
-            )
-        ]
+        artifacts = []
         findings = [
             Finding(
                 statement="检索得到一条可用参考资料",