主机：扩展受保护 API 的分段运行能力

更新启动、恢复、阶段推进与诊断路由的参数和权限校验，保持内部 E2E checkpoint 与正式 stop 语义分离，并同步安全边界测试。

script_build_host/src/script_build_host/api/routes.py

@@ -18,7 +18,9 @@ from script_build_host.application.mission_service import (
     StartScriptBuildCommand,
 )
 from script_build_host.application.observation_views import (
+    input_summary_view,
     mission_snapshot_view,
+    task_contract_summary_view,
     task_detail_view,
 )
 from script_build_host.domain.ports import (
@@ -66,6 +68,8 @@ def create_script_build_router(
     legacy_projection: Any | None = None,
     legacy_api: Any | None = None,
     http_command_journal: Any | None = None,
+    task_contract_store: Any | None = None,
+    input_snapshots: Any | None = None,
 ) -> APIRouter:
     router = APIRouter(tags=["script-build"])
 
@@ -596,6 +600,67 @@ def create_script_build_router(
             raise problem(404, "TASK_NOT_FOUND", "task does not exist in this build")
         return task_detail_view(task)
 
+    @router.get("/api/pattern/script_builds/{script_build_id}/tasks/{task_id}/contract")
+    async def task_contract_summary(
+        script_build_id: int,
+        task_id: str,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        if task_contract_store is None:
+            raise problem(503, "TASK_CONTRACT_STORE_NOT_CONFIGURED", "contract view is unavailable")
+        binding = await scope(script_build_id, actor)
+        ledger = await coordinator.task_store.load(binding.root_trace_id)
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise problem(404, "TASK_NOT_FOUND", "task does not exist in this build")
+        versions: list[dict[str, Any]] = []
+        specs = getattr(task, "specs", (task.current_spec,))
+        for spec in specs:
+            contract_ref = next(
+                (
+                    value
+                    for value in spec.context_refs
+                    if value.startswith("script-build://task-contracts/sha256/")
+                ),
+                None,
+            )
+            if contract_ref is None:
+                continue
+            frozen = await task_contract_store.read(binding.root_trace_id, contract_ref)
+            versions.append(
+                task_contract_summary_view(
+                    task_id,
+                    frozen,
+                    spec_version=int(spec.version),
+                    spec_created_at=getattr(spec, "created_at", None),
+                )
+            )
+        if not versions:
+            raise problem(404, "TASK_CONTRACT_NOT_FOUND", "task has no frozen contract")
+        current_version = int(getattr(task, "current_spec_version", task.current_spec.version))
+        current = next(
+            (value for value in versions if value["spec_version"] == current_version),
+            versions[-1],
+        )
+        return {
+            **current,
+            "current_spec_version": current_version,
+            "versions": versions,
+        }
+
+    @router.get("/api/pattern/script_builds/{script_build_id}/input-summary")
+    async def input_summary(
+        script_build_id: int,
+        actor: PrincipalDependency,
+    ) -> dict[str, Any]:
+        if input_snapshots is None:
+            raise problem(503, "INPUT_SNAPSHOT_STORE_NOT_CONFIGURED", "input view is unavailable")
+        binding = await scope(script_build_id, actor)
+        snapshot = await input_snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=script_build_id
+        )
+        return input_summary_view(snapshot)
+
     @router.get("/api/pattern/script_builds/{script_build_id}/artifacts/{artifact_version_id}")
     async def artifact_detail(
         script_build_id: int,
@@ -700,11 +765,16 @@ def create_script_build_router(
         script_build_id: int,
         trace_id: str,
         actor: PrincipalDependency,
+        after_sequence: int = Query(0, ge=0),
     ) -> dict[str, Any]:
         binding = await scope(script_build_id, actor)
         await _owned_trace(trace_store, binding.root_trace_id, trace_id)
         messages = await trace_store.get_trace_messages(trace_id)
-        return {"messages": [item.to_dict() for item in messages]}
+        return {
+            "messages": [
+                item.to_dict() for item in messages if int(item.sequence) > after_sequence
+            ]
+        }
 
     @router.get("/api/pattern/script_builds/{script_build_id}/trace_messages")
     async def legacy_trace_messages(

script_build_host/tests/test_api_security.py

@@ -78,7 +78,14 @@ class _MissionService:
 
 class _Bindings:
     async def get_by_build(self, script_build_id: int) -> object:
-        return SimpleNamespace(script_build_id=script_build_id, root_trace_id="root-7")
+        return SimpleNamespace(
+            script_build_id=script_build_id,
+            root_trace_id="root-7",
+            input_snapshot_id=11,
+        )
+
+
+_CONTRACT_REF = "script-build://task-contracts/sha256/" + "c" * 64
 
 
 class _TaskStore:
@@ -88,7 +95,7 @@ class _TaskStore:
             version=1,
             objective="bounded task",
             acceptance_criteria=(criterion,),
-            context_refs=("script-build://task-kinds/paragraph",),
+            context_refs=("script-build://task-kinds/paragraph", _CONTRACT_REF),
         )
         self.task = SimpleNamespace(
             task_id="task-7",
@@ -96,6 +103,8 @@ class _TaskStore:
             display_path="Root/task-7",
             status="completed",
             current_spec=current_spec,
+            current_spec_version=1,
+            specs=[current_spec],
             child_task_ids=(),
             attempt_ids=(),
             validation_ids=(),
@@ -118,12 +127,78 @@ class _TaskStore:
         )
 
 
+class _Contract:
+    def to_payload(self) -> dict[str, object]:
+        return {
+            "schema_version": "script-task-contract/v1",
+            "task_kind": "paragraph",
+            "scope_ref": "script-build://scopes/body",
+            "intent_class": "explore",
+            "objective": "write the evidence paragraph",
+            "input_decision_refs": [],
+            "base_artifact_ref": None,
+            "write_scope": ["script-build://scopes/body"],
+            "gap_ref": None,
+            "output_schema": "paragraph-artifact/v1",
+            "criteria": [
+                {"criterion_id": "grounded", "description": "grounded", "hard": True}
+            ],
+            "budget": {"max_tokens": 1},
+            "goal_ids": ["goal-1"],
+            "supersedes_decision_ids": [],
+            "candidate_closure_decision_refs": [],
+            "adopted_decision_ids": [],
+            "held_or_rejected_decision_ids": [],
+            "compose_order": [],
+            "comparison_decision_refs": [],
+        }
+
+
+class _Contracts:
+    async def read(self, root_trace_id: str, uri: str) -> object:
+        assert root_trace_id == "root-7"
+        assert uri == _CONTRACT_REF
+        return SimpleNamespace(
+            uri=uri,
+            digest="sha256:" + "c" * 64,
+            contract=_Contract(),
+        )
+
+
+class _InputSnapshots:
+    async def get(self, snapshot_id: str, *, script_build_id: int) -> object:
+        assert snapshot_id == "11"
+        assert script_build_id == 7
+        return SimpleNamespace(
+            snapshot_id="11",
+            script_build_id=7,
+            execution_id=1,
+            topic_build_id=2,
+            topic_id=3,
+            topic={"topic": {"result": "How one reversal changes a belief"}},
+            account={"account_name": "acct", "resolved_account_name": "acct"},
+            persona_points=({"point_type": "voice"},),
+            section_patterns=({}, {}),
+            strategies=(
+                {
+                    "id": 5,
+                    "name": "contrast",
+                    "description": "show the turn clearly",
+                    "content": "private strategy body",
+                },
+            ),
+            canonical_sha256="sha256:" + "d" * 64,
+        )
+
+
 def _app(
     principal: Principal | None,
     *,
     trace_store: object | None = None,
     uploaded_topics: object | None = None,
     business_artifacts: object | None = None,
+    task_contract_store: object | None = None,
+    input_snapshots: object | None = None,
 ) -> tuple[object, _MissionService]:
     mission = _MissionService()
     app = create_app(
@@ -143,6 +218,8 @@ def _app(
             frozenset({"data.example"}),
             resolver=_public_resolver,
         ),
+        task_contract_store=task_contract_store,
+        input_snapshots=input_snapshots,
     )
     return app, mission
 
@@ -333,6 +410,13 @@ class _TraceStore:
         del since
         return []
 
+    async def get_trace_messages(self, trace_id: str) -> list[object]:
+        assert trace_id in {"root-7", "worker-7"}
+        return [
+            SimpleNamespace(sequence=value, to_dict=lambda value=value: {"sequence": value})
+            for value in (1, 2, 3)
+        ]
+
 
 class _BusinessArtifacts:
     def __init__(self) -> None:
@@ -432,6 +516,14 @@ async def test_phase_two_observation_hides_foreign_resources_and_returns_owned_a
         assert owned_trace.json()["trace"]["trace_id"] == "worker-7"
         assert owned_trace.json()["goal_tree"] is None
 
+        incremental_messages = await client.get(
+            "/api/pattern/script_builds/7/traces/worker-7/messages?after_sequence=1"
+        )
+        assert incremental_messages.status_code == 200
+        assert incremental_messages.json() == {
+            "messages": [{"sequence": 2}, {"sequence": 3}]
+        }
+
         foreign_trace = await client.get("/api/pattern/script_builds/7/traces/worker-8")
         assert foreign_trace.status_code == 404
         assert foreign_trace.json()["detail"]["error_code"] == "TRACE_NOT_FOUND"
@@ -441,6 +533,38 @@ async def test_phase_two_observation_hides_foreign_resources_and_returns_owned_a
         assert forged_root.json()["detail"]["error_code"] == "ROOT_NOT_FOUND"
 
 
+@pytest.mark.asyncio
+async def test_safe_contract_and_input_summaries_are_build_scoped() -> None:
+    app, _ = _app(
+        Principal("owner"),
+        task_contract_store=_Contracts(),
+        input_snapshots=_InputSnapshots(),
+    )
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        contract = await client.get(
+            "/api/pattern/script_builds/7/tasks/task-7/contract"
+        )
+        assert contract.status_code == 200, contract.text
+        assert contract.json()["task_kind"] == "paragraph"
+        assert contract.json()["goal_ids"] == ["goal-1"]
+        assert contract.json()["current_spec_version"] == 1
+        assert len(contract.json()["versions"]) == 1
+        assert "budget" not in contract.json()
+
+        summary = await client.get("/api/pattern/script_builds/7/input-summary")
+        assert summary.status_code == 200, summary.text
+        payload = summary.json()
+        assert payload["topic"] == "How one reversal changes a belief"
+        assert payload["persona"]["point_types"] == {"voice": 1}
+        assert payload["strategies"][0]["name"] == "contrast"
+        assert "private strategy body" not in summary.text
+
+        denied = await client.get("/api/pattern/script_builds/8/input-summary")
+        assert denied.status_code == 404
+
+
 class _UploadedTopics:
     async def parse(self, _value: object) -> dict[str, object]:
         return {