恢复：严格校验工作台合同编译清单

未完成 Build 恢复前逐一读取冻结合同并验证 compiler_manifest_digest；缺少新清单的旧任务不再猜测重放，明确分类为 MANUAL_RECONCILIATION。已发布 Build 保持只读一致性路径，不新增数据库状态。

script_build_host/src/script_build_host/application/recovery.py

@@ -103,6 +103,7 @@ class PhaseThreeRecoveryService:
         artifacts: ScriptBusinessArtifactRepository | None = None,
         legacy_projection: Any | None = None,
         mission_service: Any | None = None,
+        contract_reader: Any | None = None,
     ) -> None:
         self.coordinator = coordinator
         self.bindings = bindings
@@ -114,6 +115,7 @@ class PhaseThreeRecoveryService:
         self.artifacts = artifacts
         self.legacy_projection = legacy_projection
         self.mission_service = mission_service
+        self.contract_reader = contract_reader
 
     async def resume(self, script_build_id: int, principal: Principal) -> ResumeResult:
         await self.authorizer.require_access(principal, script_build_id)
@@ -179,6 +181,23 @@ class PhaseThreeRecoveryService:
             )
         ledger = await self.coordinator.task_store.load(binding.root_trace_id)
         root = ledger.tasks[ledger.root_task_id]
+        if self.contract_reader is not None and status in {
+            BuildStatus.RUNNING,
+            BuildStatus.PARTIAL,
+        }:
+            try:
+                for task in ledger.tasks.values():
+                    if task.task_id != ledger.root_task_id:
+                        await self.contract_reader.contract_for_task(binding.root_trace_id, task)
+            except Exception as exc:
+                if getattr(exc, "code", "") == "TASK_CONTRACT_INVALID":
+                    return ResumeResult(
+                        script_build_id,
+                        RecoveryClassification.MANUAL_RECONCILIATION,
+                        status,
+                        "unfinished build predates the Mission Workbench compiler manifest",
+                    )
+                raise
         if root.status is TaskStatus.COMPLETED:
             if publication is not None and publication.state is PublicationState.PUBLISHING:
                 raise PublicationStateInconsistent()

script_build_host/tests/test_recovery_workbench.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from agent.orchestration import TaskStatus
+
+from script_build_host.application.recovery import MissionRecoveryService
+from script_build_host.domain.records import (
+    BuildStatus,
+    Principal,
+    RecoveryClassification,
+)
+
+
+class _Repository:
+    def __init__(self, value: Any) -> None:
+        self.value = value
+
+    async def get_by_build(self, *_args: Any, **_kwargs: Any) -> Any:
+        return self.value
+
+    async def get_status(self, _script_build_id: int) -> BuildStatus:
+        return self.value
+
+
+class _Authorizer:
+    async def require_access(self, _principal: Principal, _script_build_id: int) -> None:
+        return None
+
+
+class _TaskStore:
+    async def load(self, _root_trace_id: str) -> Any:
+        root = SimpleNamespace(task_id="root", status=TaskStatus.RUNNING)
+        child = SimpleNamespace(task_id="legacy-child")
+        return SimpleNamespace(
+            root_task_id="root",
+            tasks={"root": root, "legacy-child": child},
+            operations={},
+        )
+
+
+class _LegacyContractReader:
+    async def contract_for_task(self, _root_trace_id: str, _task: Any) -> Any:
+        error = RuntimeError("compiler manifest is missing")
+        error.code = "TASK_CONTRACT_INVALID"  # type: ignore[attr-defined]
+        raise error
+
+
+@pytest.mark.asyncio
+async def test_unfinished_legacy_contract_requires_manual_reconciliation() -> None:
+    binding = SimpleNamespace(root_trace_id="root-trace")
+    coordinator = SimpleNamespace(task_store=_TaskStore())
+    service = MissionRecoveryService(
+        coordinator=coordinator,
+        bindings=_Repository(binding),
+        publications=_Repository(None),
+        legacy_state=_Repository(BuildStatus.RUNNING),
+        authorizer=_Authorizer(),
+        continuation=SimpleNamespace(),
+        finalization=SimpleNamespace(),
+        contract_reader=_LegacyContractReader(),
+    )
+
+    result = await service.resume(900049, Principal("owner"))
+
+    assert result.classification is RecoveryClassification.MANUAL_RECONCILIATION
+    assert result.status is BuildStatus.RUNNING
+    assert "compiler manifest" in result.detail