输入：冻结阶段二提示词谱系并校验持久追踪存储

script_build_host/src/script_build_host/adapters/prompts.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from hashlib import sha256
 from pathlib import Path
 from typing import Any
 
@@ -8,7 +9,6 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.domain.ports import PromptRequest
-from script_build_host.infrastructure.canonical_json import canonical_sha256
 from script_build_host.infrastructure.legacy_tables import prompt
 
 
@@ -90,5 +90,5 @@ class DatabaseFirstPromptSource:
             "source_id": source_id,
             "version": version,
             "content": content,
-            "content_sha256": canonical_sha256(content).wire,
+            "content_sha256": f"sha256:{sha256(content.encode('utf-8')).hexdigest()}",
         }

script_build_host/src/script_build_host/agents/__init__.py

@@ -2,6 +2,7 @@
 
 from .model_resolver import ScriptRoleRunConfigResolver
 from .presets import register_script_presets
+from .prompt_resolver import SnapshotRoleSystemPromptResolver
 from .validation import (
     ScriptBuildArtifactEvidenceReader,
     ScriptBuildDeterministicValidator,
@@ -15,5 +16,6 @@ __all__ = [
     "ScriptBuildEvidenceProvider",
     "ScriptBuildValidationPolicy",
     "ScriptRoleRunConfigResolver",
+    "SnapshotRoleSystemPromptResolver",
     "register_script_presets",
 ]

script_build_host/src/script_build_host/agents/prompt_catalog.py

@@ -0,0 +1,34 @@
+"""Prompt catalog adapter shared by HTTP start and lifecycle continuation."""
+
+from __future__ import annotations
+
+from script_build_host.agents import prompts as prompt_contracts
+from script_build_host.domain.ports import PromptRequest
+
+PHASE_TWO_PRESETS = frozenset(
+    {
+        "script_structure_worker",
+        "script_paragraph_worker",
+        "script_element_set_worker",
+        "script_candidate_compare_worker",
+        "script_compose_worker",
+        "script_candidate_portfolio_worker",
+    }
+)
+
+
+def script_prompt_requests() -> tuple[PromptRequest, ...]:
+    catalog = getattr(prompt_contracts, "script_build_prompt_manifest", None)
+    manifest = catalog() if callable(catalog) else prompt_contracts.phase_one_prompt_manifest()
+    return tuple(
+        PromptRequest(
+            biz_type=str(item["biz_type"]),
+            filename=str(item["source_id"]),
+            preset=str(item["preset"]),
+            role=str(item["role"]),
+        )
+        for item in manifest
+    )
+
+
+__all__ = ["PHASE_TWO_PRESETS", "script_prompt_requests"]

script_build_host/src/script_build_host/agents/prompt_resolver.py

@@ -0,0 +1,75 @@
+"""Resolve Worker and Validator prompts from the Task's frozen input snapshot."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from hashlib import sha256
+from typing import Any
+
+from agent import AgentRole, RoleSystemPromptOverride
+
+from script_build_host.domain.errors import ProtocolViolation
+
+_INPUT_PREFIX = "script-build://inputs/"
+
+
+class SnapshotRoleSystemPromptResolver:
+    """Fail closed when a Task cannot prove the exact frozen role prompt."""
+
+    def __init__(self, bindings: Any, snapshots: Any) -> None:
+        self._bindings = bindings
+        self._snapshots = snapshots
+
+    async def resolve(
+        self,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Mapping[str, Any],
+    ) -> RoleSystemPromptOverride | None:
+        root_trace_id = _text(context.get("root_trace_id"), "root_trace_id")
+        binding = await self._bindings.get_by_root(root_trace_id)
+        task_spec = context.get("task_spec")
+        if not isinstance(task_spec, Mapping):
+            raise ProtocolViolation("TaskSpec is required to resolve a frozen role prompt")
+        raw_refs = task_spec.get("context_refs")
+        if not isinstance(raw_refs, Sequence) or isinstance(raw_refs, (str, bytes)):
+            raise ProtocolViolation("TaskSpec context_refs are invalid")
+        snapshot_ids = [
+            str(value).removeprefix(_INPUT_PREFIX)
+            for value in raw_refs
+            if isinstance(value, str) and value.startswith(_INPUT_PREFIX)
+        ]
+        if len(snapshot_ids) != 1 or not snapshot_ids[0].isdigit():
+            raise ProtocolViolation("TaskSpec must pin exactly one frozen input snapshot")
+        snapshot = await self._snapshots.get(
+            snapshot_ids[0], script_build_id=binding.script_build_id
+        )
+        matches = [
+            item
+            for item in snapshot.prompt_manifest
+            if item.get("preset") == preset and item.get("role") == role.value
+        ]
+        if not matches:
+            # Backward compatibility for pre-resolver Phase1 snapshots.  New
+            # Phase2 starts enforce a complete prompt catalog before dispatch.
+            return None
+        if len(matches) != 1:
+            raise ProtocolViolation("frozen prompt manifest contains duplicate role entries")
+        item = matches[0]
+        content = item.get("content")
+        identity = item.get("content_sha256")
+        if not isinstance(content, str) or not content.strip() or not isinstance(identity, str):
+            raise ProtocolViolation("frozen role prompt is incomplete")
+        if f"sha256:{sha256(content.encode('utf-8')).hexdigest()}" != identity:
+            raise ProtocolViolation("frozen role prompt digest does not match its content")
+        return RoleSystemPromptOverride(content=content, prompt_identity=identity)
+
+
+def _text(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise ProtocolViolation(f"protected {label} is required")
+    return value
+
+
+__all__ = ["SnapshotRoleSystemPromptResolver"]

script_build_host/src/script_build_host/application/input_snapshot_service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import Any
 
+from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import InputRelationMismatch
 from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
 from script_build_host.domain.ports import (
@@ -15,8 +16,7 @@ from script_build_host.domain.ports import (
     StrategySource,
 )
 from script_build_host.domain.records import Principal
-from script_build_host.infrastructure.canonical_json import canonical_sha256
-from script_build_host.infrastructure.redaction import redact, redact_text
+from script_build_host.domain.redaction import redact, redact_text
 
 
 @dataclass(frozen=True, slots=True)
@@ -121,6 +121,50 @@ class ScriptInputSnapshotService:
     async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
         return await self._snapshots.get(snapshot_id, script_build_id=script_build_id)
 
+    async def extend_prompt_lineage(
+        self,
+        snapshot: ScriptBuildInputSnapshotV1,
+        *,
+        prompt_requests: tuple[PromptRequest, ...],
+        model_manifest: dict[str, Any] | None = None,
+    ) -> ScriptBuildInputSnapshotV1:
+        """Append a v2 snapshot containing phase-two runtime inputs.
+
+        The source graph is copied from the verified parent.  Only prompt/model
+        manifests may be extended, and the stable business digest makes that
+        restriction independently checkable by every later consumer.
+        """
+
+        loaded = await self._prompt_source.load(prompt_requests)
+        merged: dict[tuple[str, str], dict[str, Any]] = {}
+        for value in (*snapshot.prompt_manifest, *loaded):
+            key = (str(value.get("preset") or ""), str(value.get("role") or ""))
+            if not all(key):
+                raise InputRelationMismatch()
+            merged[key] = value
+        parent_input = snapshot.to_input()
+        business_digest = canonical_sha256(parent_input.business_payload()).wire
+        if snapshot.business_input_sha256 and snapshot.business_input_sha256 != business_digest:
+            raise InputRelationMismatch()
+        extended = ScriptBuildInput(
+            script_build_id=snapshot.script_build_id,
+            execution_id=snapshot.execution_id,
+            topic_build_id=snapshot.topic_build_id,
+            topic_id=snapshot.topic_id,
+            topic=snapshot.topic,
+            account=snapshot.account,
+            persona_points=snapshot.persona_points,
+            section_patterns=snapshot.section_patterns,
+            strategies=snapshot.strategies,
+            prompt_manifest=tuple(merged[key] for key in sorted(merged)),
+            datasource_manifest=snapshot.datasource_manifest,
+            model_manifest=redact(model_manifest or snapshot.model_manifest),
+            parent_snapshot_id=snapshot.snapshot_id,
+            parent_snapshot_sha256=snapshot.canonical_sha256,
+            business_input_sha256=business_digest,
+        )
+        return await self._snapshots.freeze(extended, version=2)
+
 
 def _redacted_rows(
     values: tuple[dict[str, Any], ...],

script_build_host/src/script_build_host/application/mission_factory.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from hashlib import sha256
 from math import isfinite
 
 from agent import CompletionPolicy, RunConfig
@@ -111,6 +112,7 @@ class ScriptMissionFactory:
             model=planner_config["model"],
             temperature=planner_config["temperature"],
             max_iterations=planner_config["max_iterations"],
+            system_prompt=_frozen_prompt(snapshot, "script_planner"),
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
             new_trace_id=binding.root_trace_id,
             root_task_spec=self.build_root_task_spec(snapshot),
@@ -130,6 +132,79 @@ class ScriptMissionFactory:
             name=f"Script build {binding.script_build_id}",
         )
 
+    def build_phase_two_policy(self, snapshot: ScriptBuildInputSnapshotV1) -> str:
+        frozen = _frozen_prompt(snapshot, "script_planner") or ""
+        policy = json.dumps(
+            {
+                "policy": "script-build-phase-two/v1",
+                "phase": 2,
+                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
+                "input_sha256": snapshot.canonical_sha256,
+                "instruction": (
+                    "Dynamically create independently verifiable creative increments. "
+                    "Use only controlled phase-two task contracts and stop Root at "
+                    "PHASE_TWO_CANDIDATE_PORTFOLIO_READY."
+                ),
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+        return f"{frozen}\n\n{policy}".strip()
+
+    def build_phase_two_message(
+        self,
+        binding: MissionBinding,
+        snapshot: ScriptBuildInputSnapshotV1,
+        *,
+        direction_artifact_version_id: int,
+    ) -> str:
+        return json.dumps(
+            {
+                "mission": "continue_toward_root",
+                "script_build_id": binding.script_build_id,
+                "phase": 2,
+                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
+                "input_sha256": snapshot.canonical_sha256,
+                "accepted_direction_ref": (
+                    f"script-build://artifact-versions/{direction_artifact_version_id}"
+                ),
+                "phase_boundary": "PHASE_TWO_CANDIDATE_PORTFOLIO_READY",
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+
+    def build_phase_two_run_config(
+        self,
+        binding: MissionBinding,
+        snapshot: ScriptBuildInputSnapshotV1,
+    ) -> RunConfig:
+        planner_config = _planner_model_config(snapshot.model_manifest)
+        return RunConfig(
+            agent_type="script_planner",
+            model=planner_config["model"],
+            temperature=planner_config["temperature"],
+            max_iterations=planner_config["max_iterations"],
+            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            trace_id=binding.root_trace_id,
+            root_task_spec=None,
+            tool_groups=None,
+            parallel_tool_execution=False,
+            enable_memory=False,
+            enable_research_flow=False,
+            knowledge=KnowledgeConfig(
+                enable_extraction=False,
+                enable_completion_extraction=False,
+                enable_injection=False,
+            ),
+            context={
+                "script_build_id": binding.script_build_id,
+                "input_snapshot_id": snapshot.snapshot_id,
+                "phase": 2,
+            },
+            name=f"Script build {binding.script_build_id} phase two",
+        )
+
 
 def _planner_model_config(manifest: dict[str, object]) -> dict[str, object]:
     raw_presets = manifest.get("presets", manifest)
@@ -148,4 +223,20 @@ def _planner_model_config(manifest: dict[str, object]) -> dict[str, object]:
     }
 
 
+def _frozen_prompt(snapshot: ScriptBuildInputSnapshotV1, preset: str) -> str | None:
+    matches = [item for item in snapshot.prompt_manifest if item.get("preset") == preset]
+    if not matches:
+        return None
+    if len(matches) != 1:
+        raise ValueError(f"frozen prompt manifest has duplicate preset: {preset}")
+    value = matches[0]
+    content = value.get("content")
+    digest = value.get("content_sha256")
+    if not isinstance(content, str) or not isinstance(digest, str):
+        raise ValueError(f"frozen prompt manifest is incomplete: {preset}")
+    if f"sha256:{sha256(content.encode('utf-8')).hexdigest()}" != digest:
+        raise ValueError(f"frozen prompt manifest digest mismatch: {preset}")
+    return content
+
+
 __all__ = ["ScriptMissionFactory", "normalize_topic_summary"]

script_build_host/src/script_build_host/infrastructure/raw_artifacts.py

@@ -20,6 +20,25 @@ class FileRawArtifactStore:
         await asyncio.to_thread(self._write_once, digest, content)
         return f"script-build://raw-artifacts/sha256/{digest}"
 
+    async def read_bytes(self, ref: str) -> bytes:
+        """Read and re-verify one content-addressed raw artifact."""
+
+        prefix = "script-build://raw-artifacts/sha256/"
+        digest = ref.removeprefix(prefix) if ref.startswith(prefix) else ""
+        if len(digest) != 64 or any(value not in "0123456789abcdef" for value in digest):
+            raise ValueError("raw artifact reference has an invalid digest")
+        return await asyncio.to_thread(self._read_verified, digest)
+
+    def _read_verified(self, digest: str) -> bytes:
+        resolved_root = self.root.resolve()
+        target = (resolved_root / "raw-artifacts" / "sha256" / digest).resolve()
+        if not target.is_relative_to(resolved_root) or not target.is_file():
+            raise ValueError("raw artifact does not exist")
+        content = target.read_bytes()
+        if sha256(content).hexdigest() != digest:
+            raise ValueError("raw artifact digest verification failed")
+        return content
+
     def _write_once(self, digest: str, content: bytes) -> None:
         directory = (self.root / "raw-artifacts" / "sha256").resolve()
         resolved_root = self.root.resolve()

script_build_host/src/script_build_host/infrastructure/trace_verifier.py

@@ -0,0 +1,130 @@
+"""Startup verifier for deployment-owned durable Trace stores."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+from pathlib import Path
+from uuid import uuid4
+
+from agent import FileSystemTraceStore
+from agent.trace.models import Message, Trace
+
+from script_build_host.domain.errors import InvalidConfiguration
+
+
+class ApiRoundTripTraceStoreVerifier:
+    async def verify(self, trace_store: object, *, require_attachments: bool = False) -> None:
+        required = ("create_trace", "get_trace", "append_event", "get_events")
+        if any(not callable(getattr(trace_store, name, None)) for name in required):
+            raise InvalidConfiguration("trace store lacks durable read/write APIs")
+        if require_attachments and any(
+            not callable(getattr(trace_store, name, None))
+            for name in ("store_message_attachment", "read_message_attachment")
+        ):
+            raise InvalidConfiguration(
+                "image understanding requires durable Trace attachment storage"
+            )
+        trace_id = f"script-build-startup-probe-{uuid4()}"
+        trace = Trace(
+            trace_id=trace_id,
+            mode="agent",
+            task="durability probe",
+            agent_type="script_build_probe",
+            status="completed",
+        )
+        await trace_store.create_trace(trace)  # type: ignore[attr-defined]
+        loaded = await trace_store.get_trace(trace_id)  # type: ignore[attr-defined]
+        if loaded is None or loaded.trace_id != trace_id:
+            raise InvalidConfiguration("trace store failed the durable metadata round trip")
+        await trace_store.append_event(trace_id, "durability_probe", {"ok": True})  # type: ignore[attr-defined]
+        events = await trace_store.get_events(trace_id, 0)  # type: ignore[attr-defined]
+        if not events:
+            raise InvalidConfiguration("trace store failed the durable event round trip")
+
+
+class FileSystemTraceStoreVerifier(ApiRoundTripTraceStoreVerifier):
+    """Verify the built-in store across a fresh adapter instance and fsync boundary."""
+
+    async def verify(self, trace_store: object, *, require_attachments: bool = False) -> None:
+        if not isinstance(trace_store, FileSystemTraceStore):
+            raise InvalidConfiguration("FileSystem Trace verifier received another store type")
+        trace_id = f"script-build-startup-probe-{uuid4()}"
+        trace = Trace(
+            trace_id=trace_id,
+            mode="agent",
+            task="durability probe",
+            agent_type="script_build_probe",
+            status="completed",
+        )
+        await trace_store.create_trace(trace)
+        message = Message.create(
+            trace_id=trace_id,
+            role="user",
+            sequence=1,
+            content="durability probe",
+        )
+        await trace_store.add_message(message)
+        await trace_store.update_trace(trace_id, head_sequence=1)
+        await trace_store.append_event(trace_id, "durability_probe", {"ok": True})
+
+        attachment_ref = None
+        attachment_content = b"script-build-trace-attachment-probe"
+        if require_attachments:
+            attachment_ref = await trace_store.store_message_attachment(
+                trace_id=trace_id,
+                message_id=message.message_id,
+                media_type="application/octet-stream",
+                content=attachment_content,
+                sha256=f"sha256:{hashlib.sha256(attachment_content).hexdigest()}",
+            )
+
+        _fsync_tree(trace_store.base_path / trace_id)
+        reloaded = FileSystemTraceStore(str(trace_store.base_path))
+        loaded_trace = await reloaded.get_trace(trace_id)
+        loaded_message = await reloaded.get_message(message.message_id)
+        loaded_events = await reloaded.get_events(trace_id, 0)
+        if (
+            loaded_trace is None
+            or loaded_trace.trace_id != trace_id
+            or loaded_trace.head_sequence != 1
+            or loaded_message is None
+            or loaded_message.trace_id != trace_id
+            or loaded_message.content != "durability probe"
+            or not any(item.get("event") == "durability_probe" for item in loaded_events)
+        ):
+            raise InvalidConfiguration("FileSystem Trace store failed the reopen round trip")
+        if attachment_ref is not None:
+            try:
+                loaded_content = await reloaded.read_message_attachment(attachment_ref)
+            except Exception as exc:
+                raise InvalidConfiguration(
+                    "FileSystem Trace attachment failed the reopen round trip"
+                ) from exc
+            if loaded_content != attachment_content:
+                raise InvalidConfiguration(
+                    "FileSystem Trace attachment changed across the reopen boundary"
+                )
+
+
+def _fsync_tree(root: Path) -> None:
+    if not root.is_dir():
+        raise InvalidConfiguration("FileSystem Trace probe directory was not created")
+    for path in sorted(root.rglob("*"), reverse=True):
+        if path.is_file():
+            with path.open("rb") as handle:
+                os.fsync(handle.fileno())
+        elif path.is_dir():
+            _fsync_directory(path)
+    _fsync_directory(root)
+
+
+def _fsync_directory(path: Path) -> None:
+    descriptor = os.open(path, os.O_RDONLY)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+__all__ = ["ApiRoundTripTraceStoreVerifier", "FileSystemTraceStoreVerifier"]