测试：覆盖阶段二合同、输入闭包、工作区与迁移约束

script_build_host/tests/test_input_snapshot_service.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from types import SimpleNamespace
 from typing import Any
 
 import pytest
@@ -9,8 +10,10 @@ from script_build_host.application.input_snapshot_service import (
     ScriptInputSnapshotService,
 )
 from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
-from script_build_host.domain.ports import PersonaInput
+from script_build_host.domain.ports import PersonaInput, PromptRequest
 from script_build_host.domain.records import Principal
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.repositories.sqlalchemy import SqlAlchemyInputSnapshotRepository
 
 
 class FakeLegacyInput:
@@ -91,3 +94,65 @@ async def test_validate_source_is_read_only_and_assemble_redacts_secrets() -> No
     assert legacy.calls == [(10, 20, 30), (10, 20, 30)]
     assert "hidden" not in str(assembled.canonical_payload())
     assert assembled.strategies[0] == {"always_on": [7], "on_demand": [8]}
+
+
+@pytest.mark.asyncio
+async def test_extend_prompt_lineage_is_append_only_and_preserves_business_digest(
+    database,
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyInputSnapshotRepository(sessions)
+    parent = await repository.freeze(
+        ScriptBuildInput(
+            script_build_id=1,
+            execution_id=10,
+            topic_build_id=20,
+            topic_id=30,
+            topic={"topic": {"id": 30}},
+            account={"account_name": "acct"},
+            prompt_manifest=(
+                {
+                    "preset": "script_planner",
+                    "role": "planner",
+                    "content": "phase one",
+                    "content_sha256": canonical_sha256("phase one").wire,
+                },
+            ),
+        )
+    )
+
+    class Prompts:
+        async def load(self, _requests):
+            return (
+                {
+                    "preset": "script_compose_worker",
+                    "role": "worker",
+                    "content": "phase two",
+                    "content_sha256": canonical_sha256("phase two").wire,
+                },
+            )
+
+    service = ScriptInputSnapshotService(
+        legacy_input=SimpleNamespace(),
+        persona_source=SimpleNamespace(),
+        strategy_source=SimpleNamespace(),
+        prompt_source=Prompts(),
+        snapshots=repository,
+    )
+    child = await service.extend_prompt_lineage(
+        parent,
+        prompt_requests=(
+            PromptRequest("compose", "compose.md", "script_compose_worker", "worker"),
+        ),
+        model_manifest={"presets": {"script_compose_worker": {"model": "fake"}}},
+    )
+    assert child.snapshot_id != parent.snapshot_id
+    assert child.parent_snapshot_id == parent.snapshot_id
+    assert child.parent_snapshot_sha256 == parent.canonical_sha256
+    assert (
+        child.business_input_sha256 == canonical_sha256(parent.to_input().business_payload()).wire
+    )
+    assert {item["preset"] for item in child.prompt_manifest} == {
+        "script_planner",
+        "script_compose_worker",
+    }

script_build_host/tests/test_migrations.py

@@ -54,6 +54,35 @@ def test_mysql_offline_sql_compiles_without_credentials(monkeypatch: object) ->
     assert sql.count("CREATE TABLE script_build_") == 4
     assert "script_build_paragraph" not in sql
     assert "script_build_element" not in sql
+    assert "DROP CHECK ck_artifact_phase_one_type" in sql
+    assert "ADD CONSTRAINT ck_artifact_business_type CHECK" in sql
+
+
+def test_phase_two_artifacts_make_downgrade_refuse_data_loss(
+    tmp_path: Path, monkeypatch: object
+) -> None:
+    root = Path(__file__).parents[1]
+    database = tmp_path / "migration-refusal.db"
+    url = f"sqlite+aiosqlite:///{database}"
+    monkeypatch.setenv("SCRIPT_BUILD_WRITE_DATABASE_URL", url)  # type: ignore[attr-defined]
+    config = Config(str(root / "alembic.ini"))
+    command.upgrade(config, "head")
+    sync_engine = create_engine(f"sqlite:///{database}")
+    with sync_engine.begin() as connection:
+        connection.execute(
+            text(
+                "INSERT INTO script_build_artifact_version "
+                "(id, script_build_id, task_id, attempt_id, spec_version, artifact_type, "
+                "legacy_branch_id, canonical_json, canonical_sha256, state, created_at, frozen_at) "
+                "VALUES (1, 1, 'task', 'attempt', 1, 'paragraph', 1, '{}', :digest, "
+                "'frozen', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
+            ),
+            {"digest": "a" * 64},
+        )
+    with pytest.raises(RuntimeError, match="cannot downgrade while phase-two artifacts exist"):
+        command.downgrade(config, "0001_phase_one")
+    assert inspect(sync_engine).has_table("script_build_artifact_version")
+    sync_engine.dispose()
 
 
 @pytest.mark.mysql

script_build_host/tests/test_phase_two_candidates.py

@@ -0,0 +1,753 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import UTC, datetime
+from hashlib import sha256
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+from agent.orchestration import ArtifactRef
+
+from script_build_host.application.phase_two_candidates import (
+    PhaseTwoCandidateError,
+    PhaseTwoCandidateService,
+)
+from script_build_host.application.phase_two_inputs import (
+    ActiveFrontierResolver,
+    PhaseTwoInputError,
+)
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ArtifactVersion,
+    EvidenceRecordV1,
+)
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.phase_two_artifacts import (
+    CandidatePortfolioArtifactV1,
+    ComparisonArtifactV1,
+    ScriptParagraphV1,
+    StructuredScriptArtifactV1,
+)
+from script_build_host.domain.task_contracts import (
+    AcceptedDecisionRef,
+    AcceptedInput,
+    AcceptedInputBundleV1,
+    ScriptCriterion,
+    ScriptIntentClass,
+    ScriptTaskBudget,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+)
+from script_build_host.domain.workspaces import CandidateWriteContext, WorkspaceError
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.repositories.sqlalchemy import (
+    SqlAlchemyScriptBusinessArtifactRepository,
+)
+from script_build_host.repositories.workspace import (
+    SqlAlchemyCandidateWorkspaceRepository,
+)
+
+DIGEST = "sha256:" + "a" * 64
+SCOPE = "script-build://scopes/opening"
+WRITE = "script-build://writes/paragraphs/opening"
+
+
+def _contract(
+    kind: ScriptTaskKind,
+    *,
+    refs: tuple[AcceptedDecisionRef, ...] = (),
+    comparison_refs: tuple[AcceptedDecisionRef, ...] = (),
+    adopted: tuple[str, ...] = (),
+    held: tuple[str, ...] = (),
+    order: tuple[str, ...] = (),
+) -> ScriptTaskContractV1:
+    schemas = {
+        ScriptTaskKind.PARAGRAPH: "paragraph-artifact/v1",
+        ScriptTaskKind.DECODE_RETRIEVAL: "evidence-record/v1",
+        ScriptTaskKind.COMPARE: "comparison-artifact/v1",
+        ScriptTaskKind.COMPOSE: "structured-script/v1",
+        ScriptTaskKind.CANDIDATE_PORTFOLIO: "candidate-portfolio/v1",
+    }
+    return ScriptTaskContractV1(
+        task_kind=kind,
+        scope_ref=SCOPE,
+        intent_class=(
+            ScriptIntentClass.COMPOSE
+            if kind is ScriptTaskKind.COMPOSE
+            else ScriptIntentClass.COMPARE
+            if kind is ScriptTaskKind.COMPARE
+            else ScriptIntentClass.EXPLORE
+        ),
+        objective="produce one bounded candidate increment",
+        input_decision_refs=refs if kind is ScriptTaskKind.PARAGRAPH else (),
+        base_artifact_ref=None,
+        write_scope=(WRITE,),
+        gap_ref=None,
+        output_schema=schemas[kind],
+        criteria=(ScriptCriterion("closed", "output is concretely complete"),),
+        budget=ScriptTaskBudget(),
+        candidate_closure_decision_refs=(
+            refs if kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO} else ()
+        ),
+        adopted_decision_ids=adopted,
+        held_or_rejected_decision_ids=held,
+        compose_order=order,
+        comparison_decision_refs=comparison_refs,
+    )
+
+
+class _Bindings:
+    async def get_by_root(self, root_trace_id: str) -> Any:
+        return SimpleNamespace(root_trace_id=root_trace_id, script_build_id=7, input_snapshot_id=11)
+
+
+class _TaskStore:
+    def __init__(self, ledger: Any) -> None:
+        self.ledger = ledger
+
+    async def load(self, root_trace_id: str) -> Any:
+        assert root_trace_id == "root"
+        return self.ledger
+
+
+class _AcceptedInputs:
+    def __init__(
+        self, contracts: dict[str, ScriptTaskContractV1], bundle: AcceptedInputBundleV1
+    ) -> None:
+        self.contracts = contracts
+        self.bundle = bundle
+
+    async def contract_for_task(self, *, task: Any, **_: Any) -> ScriptTaskContractV1:
+        return self.contracts[task.task_id]
+
+    async def resolve(self, **_: Any) -> AcceptedInputBundleV1:
+        return self.bundle
+
+
+class _FrameworkArtifacts:
+    async def get(self, *_: Any) -> Any:
+        raise AssertionError("framework snapshot should not be read")
+
+
+class _SnapshotArtifacts:
+    def __init__(self, ref: ArtifactRef) -> None:
+        self.ref = ref
+
+    async def get(self, root_trace_id: str, snapshot_id: str) -> Any:
+        assert (root_trace_id, snapshot_id) == ("root", "validation-snapshot")
+        return SimpleNamespace(artifact_refs=[self.ref], evidence_refs=[])
+
+
+class _RawStore:
+    def __init__(self, content: bytes) -> None:
+        self.content = content
+
+    async def read_bytes(self, ref: str) -> bytes:
+        assert ref.endswith(sha256(self.content).hexdigest())
+        return self.content
+
+
+class _Artifacts:
+    def __init__(self, versions: dict[str, ArtifactVersion]) -> None:
+        self.versions = versions
+        self.frozen: ArtifactVersion | None = None
+
+    async def read_by_ref(self, ref: ArtifactRef, **_: Any) -> ArtifactVersion:
+        return self.versions[ref.uri]
+
+    async def get_by_attempt(self, **owners: Any) -> ArtifactVersion:
+        for value in self.versions.values():
+            if value.attempt_id == owners["attempt_id"]:
+                return value
+        raise AssertionError("Attempt artifact not found")
+
+    async def freeze(self, *, artifact: Any, **owners: Any) -> tuple[ArtifactVersion, ArtifactRef]:
+        digest = canonical_sha256(artifact.content_payload()).wire
+        artifact = replace(artifact, canonical_sha256=digest)
+        artifact_kind = (
+            ArtifactKind.COMPARISON
+            if isinstance(artifact, ComparisonArtifactV1)
+            else ArtifactKind.CANDIDATE_PORTFOLIO
+        )
+        version = ArtifactVersion(
+            artifact_version_id=99,
+            script_build_id=owners["script_build_id"],
+            task_id=owners["task_id"],
+            attempt_id=owners["attempt_id"],
+            spec_version=owners["spec_version"],
+            artifact_type=artifact_kind,
+            canonical_sha256=digest,
+            state=ArtifactState.FROZEN,
+            artifact=artifact,
+            created_at=datetime.now(UTC),
+            frozen_at=datetime.now(UTC),
+        )
+        self.frozen = version
+        ref = ArtifactRef(
+            "script-build://artifact-versions/99",
+            artifact_kind.value,
+            "99",
+            digest,
+        )
+        self.versions[ref.uri] = version
+        return version, ref
+
+
+def _scope_fixture(
+    contract: ScriptTaskContractV1,
+    *,
+    bundle: AcceptedInputBundleV1 | None = None,
+    task_id: str = "task",
+    attempt_id: str = "attempt",
+) -> tuple[Any, dict[str, Any]]:
+    task = SimpleNamespace(task_id=task_id, attempt_ids=[attempt_id])
+    attempt = SimpleNamespace(
+        attempt_id=attempt_id,
+        task_id=task_id,
+        spec_version=1,
+        accepted_child_decision_ids=(),
+    )
+    ledger = SimpleNamespace(
+        tasks={task_id: task},
+        attempts={attempt_id: attempt},
+        decisions={},
+    )
+    accepted = _AcceptedInputs(
+        {task_id: contract},
+        bundle or AcceptedInputBundleV1((), DIGEST),
+    )
+    return (ledger, accepted), {
+        "root_trace_id": "root",
+        "task_id": task_id,
+        "attempt_id": attempt_id,
+        "spec_version": 1,
+    }
+
+
+@pytest.mark.asyncio
+async def test_phase_one_manifest_does_not_require_candidate_workspace() -> None:
+    contract = _contract(ScriptTaskKind.DECODE_RETRIEVAL)
+    (ledger, accepted), context = _scope_fixture(contract)
+    evidence = EvidenceRecordV1(
+        evidence_id="evidence",
+        source_type="decode",
+        tool_name="retrieve_decode",
+        query={},
+        source_refs=("source",),
+        raw_artifact_ref=None,
+        summary="bounded evidence",
+        supports=("goal",),
+        confidence="high",
+        limitations=(),
+        content_sha256=DIGEST,
+        created_at=datetime.now(UTC),
+    )
+    version = ArtifactVersion(
+        1,
+        7,
+        "task",
+        "attempt",
+        1,
+        ArtifactKind.EVIDENCE,
+        DIGEST,
+        ArtifactState.FROZEN,
+        evidence,
+        datetime.now(UTC),
+        datetime.now(UTC),
+    )
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=cast(Any, _Artifacts({"script-build://artifact-versions/1": version})),
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=None,
+    )
+    manifest = await service.resolve_attempt_manifest(context=context)
+    assert manifest.artifact_ref.kind == "evidence"
+    assert manifest.scope_ref == SCOPE
+
+
+@pytest.mark.asyncio
+async def test_paragraph_workspace_uses_positive_branch_and_freezes_on_manifest(
+    database: Any,
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
+    contract = _contract(ScriptTaskKind.PARAGRAPH)
+    (ledger, accepted), context = _scope_fixture(contract)
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=repository,
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=workspaces,
+    )
+    created = await service.create_script_paragraph(
+        payload={"paragraph_index": 1, "name": "opening", "content_range": {}},
+        context=context,
+    )
+    manifest = await service.resolve_attempt_manifest(context=context)
+    version = await repository.get_by_attempt(
+        script_build_id=7, task_id="task", attempt_id="attempt"
+    )
+    assert created["paragraph_id"] > 0
+    assert version.artifact_version_id > 0
+    assert manifest.artifact_ref.digest == version.canonical_sha256
+    assert manifest.artifact_ref.kind == "paragraph"
+
+    with pytest.raises(PhaseTwoCandidateError, match="WRITE_SCOPE_VIOLATION"):
+        await service.create_script_paragraph(
+            payload={
+                "paragraph_index": 2,
+                "name": "forged",
+                "content_range": {},
+                "branch_id": 0,
+            },
+            context=context,
+        )
+
+
+@pytest.mark.asyncio
+async def test_candidate_service_discards_existing_draft_workspace_on_abandoned_attempt(
+    database: Any,
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
+    contract = _contract(ScriptTaskKind.PARAGRAPH)
+    (ledger, accepted), context = _scope_fixture(contract)
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=repository,
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=workspaces,
+    )
+    await service.create_script_paragraph(
+        payload={"paragraph_index": 1, "name": "draft", "content_range": {}},
+        context=context,
+    )
+    await service.discard_attempt(
+        root_trace_id="root",
+        task_id="task",
+        attempt_id="attempt",
+        reason="validator blocked the bounded increment",
+    )
+    write_context = CandidateWriteContext(
+        script_build_id=7,
+        task_id="task",
+        attempt_id="attempt",
+        spec_version=1,
+        objective=contract.objective,
+        input_refs=(),
+        write_scope=contract.write_scope,
+    )
+    assert (await workspaces.require(write_context)).state is ArtifactState.DISCARDED
+    with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"):
+        await service.create_script_paragraph(
+            payload={"paragraph_index": 2, "name": "late", "content_range": {}},
+            context=context,
+        )
+
+
+@pytest.mark.asyncio
+async def test_frozen_image_must_be_accepted_and_digest_and_mime_are_rechecked() -> None:
+    content = b"\x89PNG\r\n\x1a\n" + b"safe"
+    raw_ref = f"script-build://raw-artifacts/sha256/{sha256(content).hexdigest()}"
+    ref = ArtifactRef("script-build://artifact-versions/2", "evidence", "2", DIGEST)
+    item = AcceptedInput("d1", ref, ScriptTaskKind.DECODE_RETRIEVAL, SCOPE, "explicit")
+    bundle = AcceptedInputBundleV1((item,), DIGEST)
+    contract = _contract(ScriptTaskKind.PARAGRAPH)
+    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
+    evidence = EvidenceRecordV1(
+        "e",
+        "decode",
+        "retrieve_decode",
+        {},
+        ("source",),
+        raw_ref,
+        "summary",
+        ("goal",),
+        "high",
+        (),
+        DIGEST,
+        datetime.now(UTC),
+    )
+    version = ArtifactVersion(
+        2,
+        7,
+        "child",
+        "child-attempt",
+        1,
+        ArtifactKind.EVIDENCE,
+        DIGEST,
+        ArtifactState.FROZEN,
+        evidence,
+        datetime.now(UTC),
+        datetime.now(UTC),
+    )
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=cast(Any, _Artifacts({ref.uri: version})),
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=None,
+        raw_artifacts=_RawStore(content),
+    )
+    images = await service.view_frozen_images(raw_artifact_refs=[raw_ref], context=context)
+    assert images[0]["media_type"] == "image/png"
+    assert "url" not in images[0]
+    with pytest.raises(PhaseTwoCandidateError, match="outside the accepted"):
+        await service.view_frozen_images(raw_artifact_refs=[_RAW_OTHER], context=context)
+
+
+@pytest.mark.asyncio
+async def test_active_frontier_write_conflict_is_not_resolved_by_completion_order() -> None:
+    ref1 = ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST)
+    ref2 = ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST)
+    refs = (
+        AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.PARAGRAPH),
+        AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.PARAGRAPH),
+    )
+    contract = _contract(
+        ScriptTaskKind.COMPOSE, refs=refs, adopted=("d1", "d2"), order=("d1", "d2")
+    )
+    bundle = AcceptedInputBundleV1(
+        (
+            AcceptedInput("d1", ref1, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"),
+            AcceptedInput("d2", ref2, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"),
+        ),
+        DIGEST,
+    )
+    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
+    child_contract = _contract(ScriptTaskKind.PARAGRAPH)
+    for decision_id in ("d1", "d2"):
+        task_id = f"task-{decision_id}"
+        ledger.tasks[task_id] = SimpleNamespace(task_id=task_id, attempt_ids=[f"a-{decision_id}"])
+        ledger.attempts[f"a-{decision_id}"] = SimpleNamespace(
+            attempt_id=f"a-{decision_id}", task_id=task_id, spec_version=1
+        )
+        ledger.decisions[decision_id] = SimpleNamespace(
+            task_id=task_id, attempt_id=f"a-{decision_id}"
+        )
+        accepted.contracts[task_id] = child_contract
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=cast(Any, _Artifacts({})),
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=None,
+    )
+    with pytest.raises(PhaseTwoInputError, match="WRITE_SCOPE_CONFLICT"):
+        await service.read_active_frontier(context=context)
+
+
+@pytest.mark.asyncio
+async def test_comparison_candidates_are_contract_derived_and_pass_fairness_precheck() -> None:
+    candidate_refs = (
+        ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST),
+        ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST),
+    )
+    decision_refs = tuple(
+        AcceptedDecisionRef(
+            f"candidate-{index}",
+            ref,
+            SCOPE,
+            ScriptTaskKind.PARAGRAPH,
+        )
+        for index, ref in enumerate(candidate_refs, start=1)
+    )
+    bundle = AcceptedInputBundleV1(
+        tuple(
+            AcceptedInput(
+                decision_ref.decision_id,
+                decision_ref.artifact_ref,
+                ScriptTaskKind.PARAGRAPH,
+                SCOPE,
+                "explicit",
+            )
+            for decision_ref in decision_refs
+        ),
+        DIGEST,
+    )
+    contract = _contract(ScriptTaskKind.COMPARE, comparison_refs=decision_refs)
+    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
+    artifacts = _Artifacts({})
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=cast(Any, artifacts),
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=None,
+    )
+
+    criterion_results = [
+        {
+            "criterion_id": "closed",
+            "candidate_results": [
+                {"artifact_ref": candidate_refs[0].uri, "reason": "uses one visible detail"},
+                {"artifact_ref": candidate_refs[1].uri, "reason": "remains more abstract"},
+            ],
+        }
+    ]
+    frozen = await service.save_comparison_candidate(
+        payload={
+            "criterion_results": criterion_results,
+            "conflicts": ["both candidates write the opening scope"],
+            "recommendation": candidate_refs[0].uri,
+        },
+        context=context,
+    )
+    assert artifacts.frozen is not None
+    comparison = artifacts.frozen.artifact
+    assert isinstance(comparison, ComparisonArtifactV1)
+    assert comparison.candidate_artifact_refs == tuple(ref.uri for ref in candidate_refs)
+    assert comparison.criterion_results == tuple(criterion_results)
+
+    frozen_ref = ArtifactRef.from_dict(frozen["artifact_ref"])
+    service._framework_artifact_store = _SnapshotArtifacts(frozen_ref)
+    rules = await service.deterministic_precheck(
+        context={**context, "snapshot_id": "validation-snapshot"}
+    )
+    assert {str(item["rule_id"]): item["verdict"] for item in rules} == {
+        "business-artifact-owner": "passed",
+        "realized-content": "passed",
+        "candidate-lineage": "passed",
+        "comparison-fairness": "passed",
+    }
+
+    with pytest.raises(PhaseTwoCandidateError, match="derived from the frozen contract"):
+        await service.save_comparison_candidate(
+            payload={
+                "candidate_artifact_refs": [candidate_refs[1].uri],
+                "criterion_results": criterion_results,
+                "recommendation": candidate_refs[1].uri,
+            },
+            context=context,
+        )
+
+
+@pytest.mark.asyncio
+async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract() -> None:
+    ref1 = ArtifactRef("script-build://artifact-versions/1", "structured_script", "1", DIGEST)
+    ref2 = ArtifactRef("script-build://artifact-versions/2", "structured_script", "2", DIGEST)
+    refs = (
+        AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.COMPOSE),
+        AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.COMPOSE),
+    )
+    contract = _contract(
+        ScriptTaskKind.CANDIDATE_PORTFOLIO,
+        refs=refs,
+        adopted=("d1",),
+        held=("d2",),
+        order=("d1",),
+    )
+    bundle = AcceptedInputBundleV1(
+        (
+            AcceptedInput("d1", ref1, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
+            AcceptedInput("d2", ref2, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
+        ),
+        DIGEST,
+    )
+    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
+    paragraph = ScriptParagraphV1(1, 1, 1, None, "opening", {})
+
+    def structured(identifier: int) -> ArtifactVersion:
+        artifact = StructuredScriptArtifactV1(
+            direction_ref="script-build://artifact-versions/10",
+            input_closure_digest=DIGEST,
+            paragraphs=(paragraph,),
+            elements=(),
+            paragraph_element_links=(),
+            source_artifact_refs=(f"script-build://artifact-versions/{20 + identifier}",),
+            evidence_refs=(),
+            acceptance_notes=("accepted",),
+            canonical_sha256=DIGEST,
+        )
+        return ArtifactVersion(
+            identifier,
+            7,
+            f"compose-{identifier}",
+            f"compose-attempt-{identifier}",
+            1,
+            ArtifactKind.STRUCTURED_SCRIPT,
+            DIGEST,
+            ArtifactState.FROZEN,
+            artifact,
+            datetime.now(UTC),
+            datetime.now(UTC),
+        )
+
+    artifacts = _Artifacts({ref1.uri: structured(1), ref2.uri: structured(2)})
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=cast(Any, artifacts),
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=None,
+    )
+    result = await service.save_candidate_portfolio(
+        payload={
+            "unresolved_defects": [
+                {
+                    "severity": "warning",
+                    "defect_code": "MINOR_STYLE",
+                    "criterion_id": "closed",
+                    "scope_ref": SCOPE,
+                    "observed_excerpt": "opening cadence is slightly repetitive",
+                    "evidence_refs": [],
+                    "invalidated_inputs": [],
+                    "recommended_action_class": "revise",
+                }
+            ],
+        },
+        context=context,
+    )
+    assert artifacts.frozen is not None
+    portfolio = artifacts.frozen.artifact
+    assert isinstance(portfolio, CandidatePortfolioArtifactV1)
+    assert portfolio.accepted_decision_ids == ("d1",)
+    assert portfolio.rejected_or_held_decision_ids == ("d2",)
+    assert result["artifact_ref"]["digest"] == canonical_sha256(portfolio.content_payload()).wire
+
+    with pytest.raises(PhaseTwoCandidateError, match="outside the accepted closure"):
+        await service.save_candidate_portfolio(
+            payload={
+                "unresolved_defects": [
+                    {
+                        "severity": "warning",
+                        "defect_code": "UNTRUSTED_EVIDENCE",
+                        "criterion_id": "closed",
+                        "scope_ref": SCOPE,
+                        "observed_excerpt": "unsupported warning",
+                        "evidence_refs": ["script-build://artifact-versions/999"],
+                        "invalidated_inputs": [],
+                        "recommended_action_class": "revise",
+                    }
+                ],
+            },
+            context=context,
+        )
+
+    with pytest.raises(ProtocolViolation, match=r"unsupported enum|must not be blank"):
+        await service.save_candidate_portfolio(
+            payload={
+                "unresolved_defects": [
+                    {"severity": "warning", "defect_code": "INCOMPLETE_WARNING"}
+                ],
+            },
+            context=context,
+        )
+
+    with pytest.raises(PhaseTwoCandidateError, match="derived from the frozen contract"):
+        await service.save_candidate_portfolio(
+            payload={"adopted_structured_script_ref": ref2.uri}, context=context
+        )
+
+
+@pytest.mark.asyncio
+async def test_candidate_portfolio_command_replay_reuses_one_frozen_version(database: Any) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    paragraph = ScriptParagraphV1(1, 1, 1, None, "opening", {})
+
+    async def freeze_structured(identifier: int) -> tuple[ArtifactVersion, ArtifactRef]:
+        return await repository.freeze(
+            script_build_id=7,
+            task_id=f"compose-{identifier}",
+            attempt_id=f"compose-attempt-{identifier}",
+            spec_version=1,
+            artifact=StructuredScriptArtifactV1(
+                direction_ref="script-build://artifact-versions/10",
+                input_closure_digest=DIGEST,
+                paragraphs=(paragraph,),
+                elements=(),
+                paragraph_element_links=(),
+                source_artifact_refs=(f"script-build://artifact-versions/{20 + identifier}",),
+                evidence_refs=(),
+                acceptance_notes=("accepted",),
+            ),
+        )
+
+    (_, stored_ref1), (_, stored_ref2) = await freeze_structured(1), await freeze_structured(2)
+    ref1 = ArtifactRef(
+        stored_ref1.uri,
+        stored_ref1.kind,
+        stored_ref1.version,
+        stored_ref1.digest,
+    )
+    ref2 = ArtifactRef(
+        stored_ref2.uri,
+        stored_ref2.kind,
+        stored_ref2.version,
+        stored_ref2.digest,
+    )
+    refs = (
+        AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.COMPOSE),
+        AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.COMPOSE),
+    )
+    contract = _contract(
+        ScriptTaskKind.CANDIDATE_PORTFOLIO,
+        refs=refs,
+        adopted=("d1",),
+        held=("d2",),
+        order=("d1",),
+    )
+    bundle = AcceptedInputBundleV1(
+        (
+            AcceptedInput("d1", ref1, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
+            AcceptedInput("d2", ref2, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
+        ),
+        DIGEST,
+    )
+    (ledger, accepted), context = _scope_fixture(
+        contract,
+        bundle=bundle,
+        task_id="portfolio",
+        attempt_id="portfolio-attempt",
+    )
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=_TaskStore(ledger),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=repository,
+        accepted_inputs=accepted,
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=None,
+    )
+
+    first = await service.save_candidate_portfolio(
+        payload={"unresolved_defects": []}, context=context
+    )
+    replay = await service.save_candidate_portfolio(
+        payload={"unresolved_defects": []}, context=context
+    )
+    frozen = await repository.get_by_attempt(
+        script_build_id=7,
+        task_id="portfolio",
+        attempt_id="portfolio-attempt",
+    )
+
+    assert first == replay
+    assert first["artifact_ref"]["version"] == str(frozen.artifact_version_id)
+    assert frozen.artifact_type is ArtifactKind.CANDIDATE_PORTFOLIO
+
+
+_RAW_OTHER = "script-build://raw-artifacts/sha256/" + "b" * 64

script_build_host/tests/test_phase_two_contracts.py

@@ -0,0 +1,874 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import replace
+from datetime import UTC, datetime
+from hashlib import sha256
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+from agent import AgentRunner, FileSystemArtifactStore, FileSystemTaskStore, FileSystemTraceStore
+from agent.orchestration import (
+    AgentRole,
+    ArtifactRef,
+    AttemptSubmission,
+    CriterionResult,
+    DecisionAction,
+    OperationStatus,
+    OrchestrationConfig,
+    TaskCoordinator,
+    TaskStatus,
+    ValidationVerdict,
+)
+from agent.orchestration.protocols import ValidatorRunResult, WorkerRunResult
+
+from script_build_host.application.mission_factory import ScriptMissionFactory
+from script_build_host.application.mission_service import ScriptMissionService
+from script_build_host.application.phase_two_planning import PhaseTwoPlanningService
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.records import BuildStatus, MissionBinding, Principal
+from script_build_host.domain.task_contracts import (
+    PhaseTwoLimits,
+    ScriptTaskContractV1,
+    TaskContractError,
+)
+from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
+
+
+def _contract(
+    kind: str,
+    *,
+    scope: str = "script-build://scopes/full",
+    execution_ready: bool = True,
+) -> dict[str, object]:
+    output = {
+        "direction": "script-direction/v1",
+        "candidate-portfolio": "candidate-portfolio/v1",
+        "compose": "structured-script/v1",
+        "paragraph": "paragraph-artifact/v1",
+        "element-set": "element-set-artifact/v1",
+        "structure": "structure-artifact/v1",
+        "compare": "comparison-artifact/v1",
+        "decode-retrieval": "evidence-record/v1",
+    }[kind]
+    payload: dict[str, object] = {
+        "schema_version": "script-task-contract/v1",
+        "task_kind": kind,
+        "scope_ref": scope,
+        "intent_class": (
+            "portfolio"
+            if kind == "candidate-portfolio"
+            else "compose"
+            if kind == "compose"
+            else "explore"
+        ),
+        "objective": f"produce one bounded {kind} increment",
+        "input_decision_refs": [],
+        "base_artifact_ref": None,
+        "write_scope": [scope],
+        "gap_ref": None,
+        "output_schema": output,
+        "criteria": [
+            {
+                "criterion_id": "closed",
+                "description": "the bounded increment is independently verifiable",
+                "hard": True,
+            }
+        ],
+        "budget": {
+            "max_attempts": 4,
+            "max_tokens": 32000,
+            "max_seconds": 900,
+            "max_external_queries": 40,
+            "max_no_improvement": 3,
+        },
+        "supersedes_decision_ids": [],
+        "candidate_closure_decision_refs": [],
+        "adopted_decision_ids": [],
+        "held_or_rejected_decision_ids": [],
+        "compose_order": [],
+        "comparison_decision_refs": [],
+    }
+    if execution_ready and kind in {"compose", "candidate-portfolio"}:
+        ref = {
+            "decision_id": "decision-1",
+            "artifact_ref": {
+                "uri": "script-build://artifact-versions/1",
+                "kind": "structured_script" if kind == "candidate-portfolio" else "paragraph",
+                "version": "1",
+                "digest": "sha256:" + "a" * 64,
+            },
+            "scope_ref": scope,
+            "expected_task_kind": "compose" if kind == "candidate-portfolio" else "paragraph",
+        }
+        payload["candidate_closure_decision_refs"] = [ref]
+        payload["adopted_decision_ids"] = ["decision-1"]
+        payload["compose_order"] = ["decision-1"]
+    return payload
+
+
+class _Bindings:
+    def __init__(self, root: str) -> None:
+        self.binding = MissionBinding(
+            binding_id=1,
+            script_build_id=7,
+            root_trace_id=root,
+            input_snapshot_id=11,
+            active_direction_artifact_version_id=None,
+            accepted_root_artifact_version_id=None,
+            engine_version="test",
+            schema_version="v1",
+            created_at=datetime.now(UTC),
+            updated_at=datetime.now(UTC),
+        )
+
+    async def get_by_root(self, root: str) -> MissionBinding:
+        assert root == self.binding.root_trace_id
+        return self.binding
+
+    async def get_by_build(self, script_build_id: int) -> MissionBinding:
+        assert script_build_id == self.binding.script_build_id
+        return self.binding
+
+
+class _PlaceholderThenReplacementExecutor:
+    """Exercise Coordinator state transitions without bypassing durable operations."""
+
+    def __init__(self, coordinator: TaskCoordinator) -> None:
+        self.coordinator = coordinator
+        self.artifact_version = 100
+
+    async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult:
+        refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
+        kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
+        artifact_kind = {
+            "compose": "structured_script",
+            "structure": "structure",
+            "paragraph": "paragraph",
+            "element-set": "element_set",
+        }[kind]
+        digest = "sha256:" + sha256(context["attempt_id"].encode()).hexdigest()
+        version = str(self.artifact_version)
+        self.artifact_version += 1
+        ref = ArtifactRef(
+            f"script-build://artifact-versions/{version}",
+            artifact_kind,
+            version,
+            digest,
+        )
+        await self.coordinator.submit_attempt(
+            {
+                "role": AgentRole.WORKER.value,
+                "root_trace_id": context["root_trace_id"],
+                "task_id": context["task_id"],
+                "attempt_id": context["attempt_id"],
+                "spec_version": context["spec_version"],
+                "trace_id": context["worker_trace_id"],
+                "tool_call_id": f"submit:{context['attempt_id']}",
+                "operation_id": context["operation_id"],
+                "execution_epoch": context["execution_epoch"],
+            },
+            AttemptSubmission(
+                summary=f"kind={artifact_kind};status=frozen",
+                artifact_refs=[ref],
+            ),
+        )
+        return WorkerRunResult(context["worker_trace_id"], "completed")
+
+    async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
+        refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
+        kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
+        placeholder = kind == "compose" and context["spec_version"] == 1
+        verdict = ValidationVerdict.FAILED if placeholder else ValidationVerdict.PASSED
+        criterion = cast(dict[str, Any], context["task_spec"])["acceptance_criteria"][0]
+        await self.coordinator.submit_validation(
+            {
+                "role": AgentRole.VALIDATOR.value,
+                "root_trace_id": context["root_trace_id"],
+                "task_id": context["task_id"],
+                "attempt_id": context["attempt_id"],
+                "validation_id": context["validation_id"],
+                "snapshot_id": context["snapshot_id"],
+                "trace_id": context["validator_trace_id"],
+                "tool_call_id": f"validate:{context['validation_id']}",
+                "operation_id": context["operation_id"],
+                "execution_epoch": context["execution_epoch"],
+            },
+            verdict,
+            [
+                CriterionResult(
+                    criterion_id=criterion["criterion_id"],
+                    verdict=verdict,
+                    reason=(
+                        "REALIZATION_PLACEHOLDER at artifact.paragraphs[0].description"
+                        if placeholder
+                        else "the replacement is concretely rendered"
+                    ),
+                )
+            ],
+            (
+                "hard_defects=1;defect_codes=REALIZATION_PLACEHOLDER"
+                if placeholder
+                else "hard_defects=0"
+            ),
+            (),
+            ("artifact.paragraphs[0].description",) if placeholder else (),
+            ("REALIZATION_PLACEHOLDER",) if placeholder else (),
+            "split" if placeholder else "accept",
+        )
+        return ValidatorRunResult(context["validator_trace_id"], "completed")
+
+    async def stop(self, trace_id: str) -> bool:
+        del trace_id
+        return True
+
+
+async def _service(tmp_path: Path) -> tuple[PhaseTwoPlanningService, TaskCoordinator, str]:
+    root = "root-contracts"
+    coordinator = TaskCoordinator(
+        FileSystemTaskStore(str(tmp_path / "ledger")),
+        FileSystemArtifactStore(str(tmp_path / "snapshots")),
+        FileSystemTraceStore(str(tmp_path / "traces")),
+    )
+    await coordinator.ensure_ledger(
+        root,
+        {
+            "objective": "build a script",
+            "acceptance_criteria": [
+                {"criterion_id": "ready", "description": "ready", "hard": True}
+            ],
+            "context_refs": ["script-build://inputs/11"],
+        },
+    )
+    service = PhaseTwoPlanningService(
+        coordinator=coordinator,
+        bindings=_Bindings(root),
+        contracts=FileScriptTaskContractStore(tmp_path / "data"),
+    )
+    return service, coordinator, root
+
+
+class _BlockingExecutor:
+    def __init__(self) -> None:
+        self.started = asyncio.Event()
+        self.stopped_trace_ids: list[str] = []
+
+    async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult:
+        self.started.set()
+        await asyncio.Event().wait()
+        raise AssertionError(f"worker unexpectedly resumed: {context['attempt_id']}")
+
+    async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
+        raise AssertionError(f"validator must not start: {context['validation_id']}")
+
+    async def stop(self, trace_id: str) -> bool:
+        self.stopped_trace_ids.append(trace_id)
+        return True
+
+
+class _StopState:
+    def __init__(self) -> None:
+        self.status = BuildStatus.RUNNING
+        self.error_summary: str | None = None
+
+    async def get_status(self, script_build_id: int) -> BuildStatus:
+        assert script_build_id == 7
+        return self.status
+
+    async def set_status(
+        self,
+        script_build_id: int,
+        status: BuildStatus,
+        *,
+        error_summary: str | None = None,
+    ) -> None:
+        assert script_build_id == 7
+        self.status = status
+        self.error_summary = error_summary
+
+
+@pytest.mark.asyncio
+async def test_real_operation_stop_converges_and_forbids_new_dispatch(tmp_path: Path) -> None:
+    root = "root-stop-operation"
+    trace_store = FileSystemTraceStore(str(tmp_path / "stop-traces"))
+    coordinator = TaskCoordinator(
+        FileSystemTaskStore(str(tmp_path / "stop-ledger")),
+        FileSystemArtifactStore(str(tmp_path / "stop-artifacts")),
+        trace_store,
+        config=OrchestrationConfig(stop_grace_seconds=0.01),
+    )
+    await coordinator.ensure_ledger(
+        root,
+        {
+            "objective": "build a script",
+            "acceptance_criteria": [
+                {"criterion_id": "ready", "description": "ready", "hard": True}
+            ],
+            "context_refs": ["script-build://inputs/11"],
+        },
+    )
+    bindings = _Bindings(root)
+    planning = PhaseTwoPlanningService(
+        coordinator=coordinator,
+        bindings=bindings,
+        contracts=FileScriptTaskContractStore(tmp_path / "stop-contracts"),
+    )
+    portfolio = await planning.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "stop:portfolio"},
+    )
+    compose = await planning.plan_script_tasks(
+        contract_payloads=[_contract("compose", execution_ready=False)],
+        parent_task_id=portfolio["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "stop:compose"},
+    )
+    paragraph = await planning.plan_script_tasks(
+        contract_payloads=[_contract("paragraph")],
+        parent_task_id=compose["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "stop:paragraph"},
+    )
+    executor = _BlockingExecutor()
+    coordinator.set_executor(executor)
+
+    async def unused_llm(**_kwargs: Any) -> Any:
+        raise AssertionError("Planner LLM must not run during operation stop")
+
+    runner = AgentRunner(trace_store=trace_store, llm_call=unused_llm)
+    state = _StopState()
+
+    class Authorizer:
+        async def require_access(self, principal: Principal, script_build_id: int) -> None:
+            assert principal.subject == "owner" and script_build_id == 7
+
+    mission = ScriptMissionService(
+        runner=runner,
+        coordinator=coordinator,
+        factory=ScriptMissionFactory(),
+        input_snapshots=SimpleNamespace(),
+        bindings=bindings,
+        legacy_state=state,
+        authorizer=Authorizer(),
+        direction_reconciler=SimpleNamespace(),
+        stop_timeout_seconds=1,
+    )
+    planning.dispatch_guard = mission
+    dispatch = asyncio.create_task(
+        planning.dispatch_script_tasks(
+            task_ids=paragraph["task_ids"],
+            context={"root_trace_id": root, "tool_call_id": "stop:dispatch"},
+        )
+    )
+    await asyncio.wait_for(executor.started.wait(), timeout=1)
+
+    stopped = await mission.stop(7, Principal("owner"))
+    dispatch_result = await asyncio.gather(dispatch, return_exceptions=True)
+
+    assert stopped.status is BuildStatus.STOPPED
+    assert len(stopped.stopped_operation_ids) == 1
+    assert executor.stopped_trace_ids
+    assert not isinstance(dispatch_result[0], asyncio.CancelledError)
+    ledger = await coordinator.task_store.load(root)
+    operation = ledger.operations[stopped.stopped_operation_ids[0]]
+    assert operation.status is OperationStatus.STOPPED
+    assert ledger.tasks[paragraph["task_ids"][0]].status is TaskStatus.NEEDS_REPLAN
+    assert (await mission.stop(7, Principal("owner"))).status is BuildStatus.STOPPED
+
+    with pytest.raises(ProtocolViolation, match="forbidden after stop intent"):
+        await planning.dispatch_script_tasks(
+            task_ids=paragraph["task_ids"],
+            context={"root_trace_id": root, "tool_call_id": "stop:late-dispatch"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_contract_store_is_content_addressed_and_detects_tampering(tmp_path: Path) -> None:
+    store = FileScriptTaskContractStore(tmp_path)
+    from script_build_host.domain.task_contracts import ScriptTaskContractV1
+
+    contract = ScriptTaskContractV1.from_payload(_contract("paragraph"))
+    first = await store.freeze("root-a", contract)
+    second = await store.freeze("root-a", contract)
+    assert first == second
+    path = tmp_path / "script-task-contracts" / "root-a" / "sha256" / first.uri.rsplit("/", 1)[-1]
+    path.write_bytes(b"{}")
+    with pytest.raises(TaskContractError, match="TASK_CONTRACT_DIGEST_MISMATCH"):
+        await store.read("root-a", first.uri)
+    assert not await store.verify("root-a", first.uri, first.digest)
+
+
+@pytest.mark.asyncio
+async def test_orphan_contract_reclamation_uses_only_ledger_references(tmp_path: Path) -> None:
+    service, _, root = await _service(tmp_path)
+    referenced = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "portfolio"},
+    )
+    orphan = await service.contracts.freeze(
+        root, ScriptTaskContractV1.from_payload(_contract("paragraph"))
+    )
+    removed = await service.reclaim_orphan_contracts(root_trace_id=root)
+    assert removed == (orphan.uri,)
+    assert await service.contracts.verify(
+        root,
+        referenced["contracts"][0]["contract_ref"],
+        referenced["contracts"][0]["contract_digest"],
+    )
+    assert not await service.contracts.verify(root, orphan.uri, orphan.digest)
+
+
+@pytest.mark.asyncio
+async def test_failed_operation_discards_workspace_but_recovery_required_preserves_it(
+    tmp_path: Path,
+) -> None:
+    service, _, root = await _service(tmp_path)
+
+    class Lifecycle:
+        def __init__(self) -> None:
+            self.calls: list[dict[str, str]] = []
+
+        async def discard_attempt(self, **values: str) -> None:
+            self.calls.append(values)
+
+    lifecycle = Lifecycle()
+    service.workspace_lifecycle = lifecycle
+    attempts = {
+        "failed-attempt": SimpleNamespace(attempt_id="failed-attempt", task_id="paragraph-task"),
+        "recovery-attempt": SimpleNamespace(
+            attempt_id="recovery-attempt", task_id="paragraph-task"
+        ),
+    }
+    ledger = SimpleNamespace(
+        attempts=attempts,
+        operations={
+            "failed": SimpleNamespace(
+                status=OperationStatus.FAILED,
+                error="worker failed before submit",
+                attempt_ids=["failed-attempt"],
+            ),
+            "recovery": SimpleNamespace(
+                status=OperationStatus.FAILED,
+                error="MISSION_RECOVERY_REQUIRED",
+                attempt_ids=["recovery-attempt"],
+            ),
+        },
+    )
+    await service._reconcile_terminal_workspaces(root, ledger)
+    assert lifecycle.calls == [
+        {
+            "root_trace_id": root,
+            "task_id": "paragraph-task",
+            "attempt_id": "failed-attempt",
+            "reason": "worker failed before submit",
+        }
+    ]
+
+
+def test_comparison_refs_are_kind_specific_and_limits_cannot_be_raised() -> None:
+    candidate_refs = []
+    for index, kind in ((1, "paragraph"), (2, "structure")):
+        candidate_refs.append(
+            {
+                "decision_id": f"decision-{index}",
+                "artifact_ref": {
+                    "uri": f"script-build://artifact-versions/{index}",
+                    "kind": kind,
+                    "version": str(index),
+                    "digest": "sha256:" + str(index) * 64,
+                },
+                "scope_ref": "script-build://scopes/full",
+                "expected_task_kind": kind,
+            }
+        )
+    compare = _contract("compare")
+    compare["comparison_decision_refs"] = candidate_refs
+    assert ScriptTaskContractV1.from_payload(compare).task_kind.value == "compare"
+
+    comparison_ref = {
+        "decision_id": "decision-compare",
+        "artifact_ref": {
+            "uri": "script-build://artifact-versions/3",
+            "kind": "comparison",
+            "version": "3",
+            "digest": "sha256:" + "3" * 64,
+        },
+        "scope_ref": "script-build://scopes/full",
+        "expected_task_kind": "compare",
+    }
+    compose = _contract("compose")
+    compose["comparison_decision_refs"] = [comparison_ref]
+    ScriptTaskContractV1.from_payload(compose)
+    portfolio = _contract("candidate-portfolio")
+    portfolio["comparison_decision_refs"] = [comparison_ref]
+    with pytest.raises(TaskContractError, match="comparison_decision_refs"):
+        ScriptTaskContractV1.from_payload(portfolio)
+    with pytest.raises(TaskContractError, match="TASK_BUDGET_EXCEEDED"):
+        PhaseTwoLimits(max_tasks=65)
+
+
+@pytest.mark.asyncio
+async def test_planning_freezes_contract_and_creates_only_one_portfolio(tmp_path: Path) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    result = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "plan-portfolio"},
+    )
+    task_id = result["task_ids"][0]
+    ledger = await coordinator.task_store.load(root)
+    task = ledger.tasks[task_id]
+    assert task.current_spec.context_refs[0].endswith("candidate-portfolio")
+    assert task.current_spec.context_refs[1].startswith("script-build://task-contracts/sha256/")
+    assert task.current_spec.context_refs[2] == "script-build://inputs/11"
+    frozen = await service.contract_for_task(root, task)
+    assert frozen.contract.scope_ref == "script-build://scopes/full"
+
+    with pytest.raises(TaskContractError, match="one Portfolio"):
+        await service.plan_script_tasks(
+            contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+            parent_task_id=None,
+            context={"root_trace_id": root, "tool_call_id": "plan-portfolio-2"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    invalid = _contract("paragraph")
+    invalid["output_schema"] = "structured-script/v1"
+    with pytest.raises(TaskContractError, match="output_schema"):
+        await service.plan_script_tasks(
+            contract_payloads=[_contract("candidate-portfolio"), invalid],
+            parent_task_id=None,
+            context={"root_trace_id": root},
+        )
+    ledger = await coordinator.task_store.load(root)
+    assert list(ledger.tasks) == [ledger.root_task_id]
+
+
+@pytest.mark.asyncio
+async def test_planning_rejects_unresolved_supersession_before_freeze(tmp_path: Path) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    portfolio = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "portfolio"},
+    )
+    compose = await service.plan_script_tasks(
+        contract_payloads=[_contract("compose", execution_ready=False)],
+        parent_task_id=portfolio["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "compose"},
+    )
+    replacement = _contract("paragraph")
+    replacement["supersedes_decision_ids"] = ["missing-accept"]
+
+    with pytest.raises(TaskContractError, match="import every superseded"):
+        await service.plan_script_tasks(
+            contract_payloads=[replacement],
+            parent_task_id=compose["task_ids"][0],
+            context={"root_trace_id": root, "tool_call_id": "invalid-replacement"},
+        )
+
+    ledger = await coordinator.task_store.load(root)
+    assert set(ledger.tasks) == {
+        ledger.root_task_id,
+        portfolio["task_ids"][0],
+        compose["task_ids"][0],
+    }
+
+
+@pytest.mark.asyncio
+async def test_open_compose_container_cannot_dispatch(tmp_path: Path) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    portfolio = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "p"},
+    )
+    compose = await service.plan_script_tasks(
+        contract_payloads=[_contract("compose", execution_ready=False)],
+        parent_task_id=portfolio["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "c"},
+    )
+    with pytest.raises(TaskContractError, match="not closed"):
+        await service.dispatch_script_tasks(
+            task_ids=compose["task_ids"], context={"root_trace_id": root}
+        )
+    ledger = await coordinator.task_store.load(root)
+    assert not ledger.operations
+
+
+@pytest.mark.asyncio
+async def test_compose_placeholder_split_replacement_then_revised_v2_passes(
+    tmp_path: Path,
+) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
+    portfolio = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "portfolio"},
+    )
+    compose = await service.plan_script_tasks(
+        contract_payloads=[_contract("compose")],
+        parent_task_id=portfolio["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "compose-v1"},
+    )
+    compose_id = compose["task_ids"][0]
+
+    first_cycle = (
+        await service.dispatch_script_tasks(
+            task_ids=(compose_id,),
+            context={"root_trace_id": root, "tool_call_id": "dispatch-compose-v1"},
+        )
+    )[0]
+    assert first_cycle["error"] is None
+    ledger = await coordinator.task_store.load(root)
+    first_validation = ledger.validations[first_cycle["validation_id"]]
+    assert first_validation.verdict is ValidationVerdict.FAILED
+    assert first_validation.unverified_claims == ["artifact.paragraphs[0].description"]
+    assert first_validation.risks == ["REALIZATION_PLACEHOLDER"]
+
+    split = await service.decide_script_task(
+        task_id=compose_id,
+        action=DecisionAction.SPLIT.value,
+        reason="replace only the unrealized opening scope",
+        validation_id=first_validation.validation_id,
+        replacement_contract=None,
+        child_contracts=(_contract("paragraph"),),
+        context={"root_trace_id": root, "tool_call_id": "split-placeholder"},
+    )
+    replacement_id = split["payload"]["child_task_ids"][0]
+    replacement_cycle = (
+        await service.dispatch_script_tasks(
+            task_ids=(replacement_id,),
+            context={"root_trace_id": root, "tool_call_id": "dispatch-replacement"},
+        )
+    )[0]
+    replacement_accept = await service.decide_script_task(
+        task_id=replacement_id,
+        action=DecisionAction.ACCEPT.value,
+        reason="the local replacement passed independent validation",
+        validation_id=replacement_cycle["validation_id"],
+        replacement_contract=None,
+        child_contracts=(),
+        context={"root_trace_id": root, "tool_call_id": "accept-replacement"},
+    )
+
+    ledger = await coordinator.task_store.load(root)
+    assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN
+    replacement_decision_id = replacement_accept["decision_id"]
+    replacement_attempt = ledger.attempts[
+        cast(str, ledger.decisions[replacement_decision_id].attempt_id)
+    ]
+    assert replacement_attempt.submission is not None
+    replacement_ref = replacement_attempt.submission.artifact_refs[0]
+    frozen_replacement = {
+        "decision_id": replacement_decision_id,
+        "artifact_ref": {
+            "uri": replacement_ref.uri,
+            "kind": replacement_ref.kind,
+            "version": replacement_ref.version,
+            "digest": replacement_ref.digest,
+        },
+        "scope_ref": "script-build://scopes/full",
+        "expected_task_kind": "paragraph",
+    }
+    compose_v2 = _contract("compose", execution_ready=False)
+    compose_v2["candidate_closure_decision_refs"] = [frozen_replacement]
+    compose_v2["adopted_decision_ids"] = [replacement_decision_id]
+    compose_v2["compose_order"] = [replacement_decision_id]
+    await service.decide_script_task(
+        task_id=compose_id,
+        action=DecisionAction.REVISE.value,
+        reason="pin the accepted replacement and retry the complete render",
+        validation_id=None,
+        replacement_contract=compose_v2,
+        child_contracts=(),
+        context={"root_trace_id": root, "tool_call_id": "revise-compose-v2"},
+    )
+    second_cycle = (
+        await service.dispatch_script_tasks(
+            task_ids=(compose_id,),
+            context={"root_trace_id": root, "tool_call_id": "dispatch-compose-v2"},
+        )
+    )[0]
+    compose_accept = await service.decide_script_task(
+        task_id=compose_id,
+        action=DecisionAction.ACCEPT.value,
+        reason="the revised full render passed independent validation",
+        validation_id=second_cycle["validation_id"],
+        replacement_contract=None,
+        child_contracts=(),
+        context={"root_trace_id": root, "tool_call_id": "accept-compose-v2"},
+    )
+
+    reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root)
+    compose_task = reloaded.tasks[compose_id]
+    assert compose_task.status is TaskStatus.COMPLETED
+    assert compose_task.current_spec_version == 2
+    assert len(compose_task.attempt_ids) == 2
+    assert [
+        reloaded.validations[validation_id].verdict for validation_id in compose_task.validation_ids
+    ] == [ValidationVerdict.FAILED, ValidationVerdict.PASSED]
+    assert [reloaded.decisions[item].action for item in compose_task.decision_ids] == [
+        DecisionAction.SPLIT,
+        DecisionAction.REVISE,
+        DecisionAction.ACCEPT,
+    ]
+    assert compose_accept["status"] == TaskStatus.COMPLETED.value
+    assert reloaded.tasks[replacement_id].status is TaskStatus.COMPLETED
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "creation_order",
+    [
+        ("element-set", "paragraph", "structure"),
+        ("paragraph", "element-set", "structure"),
+        ("structure", "paragraph", "element-set"),
+    ],
+)
+async def test_exploration_order_and_completion_order_do_not_choose_compose_order(
+    tmp_path: Path,
+    creation_order: tuple[str, str, str],
+) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
+    portfolio = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "portfolio"},
+    )
+    compose = await service.plan_script_tasks(
+        contract_payloads=[_contract("compose", execution_ready=False)],
+        parent_task_id=portfolio["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "compose"},
+    )
+    compose_id = compose["task_ids"][0]
+    task_by_kind: dict[str, str] = {}
+    for kind in creation_order:
+        planned = await service.plan_script_tasks(
+            contract_payloads=[_contract(kind)],
+            parent_task_id=compose_id,
+            context={"root_trace_id": root, "tool_call_id": f"plan:{kind}"},
+        )
+        task_by_kind[kind] = planned["task_ids"][0]
+
+    decision_by_kind: dict[str, str] = {}
+    ref_by_kind: dict[str, ArtifactRef] = {}
+    completion_order = tuple(reversed(creation_order))
+    for kind in completion_order:
+        task_id = task_by_kind[kind]
+        cycle = (
+            await service.dispatch_script_tasks(
+                task_ids=(task_id,),
+                context={"root_trace_id": root, "tool_call_id": f"dispatch:{kind}"},
+            )
+        )[0]
+        accepted = await service.decide_script_task(
+            task_id=task_id,
+            action=DecisionAction.ACCEPT.value,
+            reason=f"the bounded {kind} increment passed validation",
+            validation_id=cycle["validation_id"],
+            replacement_contract=None,
+            child_contracts=(),
+            context={"root_trace_id": root, "tool_call_id": f"accept:{kind}"},
+        )
+        decision_id = accepted["decision_id"]
+        decision_by_kind[kind] = decision_id
+        ledger = await coordinator.task_store.load(root)
+        attempt = ledger.attempts[cast(str, ledger.decisions[decision_id].attempt_id)]
+        assert attempt.submission is not None
+        ref_by_kind[kind] = attempt.submission.artifact_refs[0]
+
+    ledger = await coordinator.task_store.load(root)
+    assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN
+    adoption_kinds = ("paragraph", "structure", "element-set")
+    closure = []
+    for kind in adoption_kinds:
+        ref = ref_by_kind[kind]
+        closure.append(
+            {
+                "decision_id": decision_by_kind[kind],
+                "artifact_ref": {
+                    "uri": ref.uri,
+                    "kind": ref.kind,
+                    "version": ref.version,
+                    "digest": ref.digest,
+                },
+                "scope_ref": "script-build://scopes/full",
+                "expected_task_kind": kind,
+            }
+        )
+    execution_ready = _contract("compose", execution_ready=False)
+    execution_ready["candidate_closure_decision_refs"] = closure
+    execution_ready["adopted_decision_ids"] = [decision_by_kind[kind] for kind in adoption_kinds]
+    execution_ready["compose_order"] = [decision_by_kind[kind] for kind in adoption_kinds]
+    await service.decide_script_task(
+        task_id=compose_id,
+        action=DecisionAction.REVISE.value,
+        reason="formal adoption is based on scope and intent, not timing",
+        validation_id=None,
+        replacement_contract=execution_ready,
+        child_contracts=(),
+        context={"root_trace_id": root, "tool_call_id": "close-compose"},
+    )
+
+    reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root)
+    frozen = await service.contract_for_task(root, reloaded.tasks[compose_id])
+    assert frozen.contract.compose_order == tuple(decision_by_kind[kind] for kind in adoption_kinds)
+    assert [
+        next(kind for kind, task_id in task_by_kind.items() if task_id == child_id)
+        for child_id in reloaded.tasks[compose_id].child_task_ids
+    ] == list(creation_order)
+    assert completion_order == tuple(reversed(creation_order))
+
+
+@pytest.mark.asyncio
+async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path: Path) -> None:
+    service, _, root = await _service(tmp_path)
+    service.limits = PhaseTwoLimits(max_tasks=2)
+    portfolio = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "portfolio"},
+    )
+    compose = await service.plan_script_tasks(
+        contract_payloads=[_contract("compose", execution_ready=False)],
+        parent_task_id=portfolio["task_ids"][0],
+        context={"root_trace_id": root, "tool_call_id": "compose"},
+    )
+    with pytest.raises(TaskContractError, match="Task budget"):
+        await service.plan_script_tasks(
+            contract_payloads=[_contract("decode-retrieval")],
+            parent_task_id=compose["task_ids"][0],
+            context={"root_trace_id": root, "tool_call_id": "retrieval"},
+        )
+
+    service.limits = PhaseTwoLimits(max_depth=1)
+    with pytest.raises(TaskContractError, match="depth"):
+        await service.plan_script_tasks(
+            contract_payloads=[_contract("structure", scope="script-build://scopes/full/local")],
+            parent_task_id=compose["task_ids"][0],
+            context={"root_trace_id": root, "tool_call_id": "too-deep"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_task_contract_cannot_be_silently_changed_in_task_spec(tmp_path: Path) -> None:
+    service, coordinator, root = await _service(tmp_path)
+    result = await service.plan_script_tasks(
+        contract_payloads=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root},
+    )
+    ledger = await coordinator.task_store.load(root)
+    task = ledger.tasks[result["task_ids"][0]]
+    task.specs[-1] = replace(task.specs[-1], objective="tampered")
+    await coordinator.task_store.commit(ledger, expected_revision=ledger.revision)
+    with pytest.raises(TaskContractError, match="DIGEST_MISMATCH"):
+        await service.contract_for_task(root, task)

script_build_host/tests/test_phase_two_inputs_validation.py

@@ -0,0 +1,781 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import replace
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+from agent.orchestration import (
+    AcceptanceCriterion,
+    ArtifactRef,
+    ArtifactSnapshot,
+    AttemptStatus,
+    DecisionAction,
+    PlannerDecision,
+    TaskAttempt,
+    TaskLedger,
+    TaskRecord,
+    TaskSpec,
+    TaskStatus,
+    ValidationReport,
+    ValidationRunStatus,
+    ValidationVerdict,
+)
+
+from script_build_host.agents.validation import (
+    ScriptBuildValidationPolicy,
+    precheck_business_artifact,
+    validate_phase_two_defects,
+    validation_layer,
+)
+from script_build_host.application.phase_two_inputs import (
+    AcceptedInputResolver,
+    ActiveFrontierResolver,
+    PhaseTwoInputError,
+)
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ArtifactVersion,
+    EvidenceRecordV1,
+)
+from script_build_host.domain.phase_two_artifacts import (
+    CandidateLineageV1,
+    ParagraphArtifactV1,
+    ScriptParagraphV1,
+)
+from script_build_host.domain.task_contracts import (
+    AcceptedDecisionRef,
+    AcceptedInput,
+    AcceptedInputBundleV1,
+    ScriptCriterion,
+    ScriptIntentClass,
+    ScriptTaskBudget,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+)
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+
+DIGEST = "sha256:" + "a" * 64
+SNAPSHOT_REF = "script-build://inputs/11"
+SCOPE = "script-build://scopes/opening"
+WRITE = "script-build://writes/paragraphs/1"
+
+
+def _contract(
+    *,
+    kind: ScriptTaskKind = ScriptTaskKind.PARAGRAPH,
+    scope: str = SCOPE,
+    write_scope: tuple[str, ...] = (WRITE,),
+    supersedes: tuple[str, ...] = (),
+) -> ScriptTaskContractV1:
+    schemas = {
+        ScriptTaskKind.PARAGRAPH: "paragraph-artifact/v1",
+        ScriptTaskKind.ELEMENT_SET: "element-set-artifact/v1",
+        ScriptTaskKind.COMPARE: "comparison-artifact/v1",
+        ScriptTaskKind.DECODE_RETRIEVAL: "evidence-record/v1",
+    }
+    return ScriptTaskContractV1(
+        task_kind=kind,
+        scope_ref=scope,
+        intent_class=ScriptIntentClass.EXPLORE,
+        objective="produce one independently verifiable increment",
+        input_decision_refs=(),
+        base_artifact_ref=None,
+        write_scope=write_scope,
+        gap_ref=None,
+        output_schema=schemas[kind],
+        criteria=(ScriptCriterion("closed", "candidate is concretely realized"),),
+        budget=ScriptTaskBudget(),
+        supersedes_decision_ids=supersedes,
+    )
+
+
+def _paragraph_artifact(*, text: str = "specific opening") -> ParagraphArtifactV1:
+    return ParagraphArtifactV1(
+        lineage=CandidateLineageV1(
+            scope_ref=SCOPE,
+            input_snapshot_ref=SNAPSHOT_REF,
+            input_closure_digest=DIGEST,
+            write_scope=(WRITE,),
+        ),
+        paragraphs=(
+            ScriptParagraphV1(
+                paragraph_id=1,
+                paragraph_index=1,
+                level=1,
+                parent_id=None,
+                name="opening",
+                content_range={},
+                description=text,
+            ),
+        ),
+    )
+
+
+def _task(task_id: str, *, parent: str | None, kind: str, status: TaskStatus) -> TaskRecord:
+    return TaskRecord(
+        task_id=task_id,
+        goal_id=None,
+        parent_task_id=parent,
+        display_path=task_id,
+        specs=[
+            TaskSpec(
+                version=1,
+                objective="bounded increment",
+                acceptance_criteria=(AcceptanceCriterion("closed", "closed"),),
+                context_refs=(
+                    f"script-build://task-kinds/{kind}",
+                    SNAPSHOT_REF,
+                ),
+            )
+        ],
+        status=status,
+    )
+
+
+class _TaskStore:
+    def __init__(self, ledger: TaskLedger) -> None:
+        self.ledger = ledger
+
+    async def load(self, root_trace_id: str) -> TaskLedger:
+        assert root_trace_id == self.ledger.root_trace_id
+        return self.ledger
+
+
+class _SnapshotStore:
+    def __init__(self, snapshot: ArtifactSnapshot) -> None:
+        self.snapshot = snapshot
+
+    async def get(self, root_trace_id: str, snapshot_id: str) -> ArtifactSnapshot:
+        assert root_trace_id == "root"
+        assert snapshot_id == self.snapshot.snapshot_id
+        return self.snapshot
+
+
+class _Artifacts:
+    def __init__(self, version: ArtifactVersion) -> None:
+        self.version = version
+
+    async def read_by_ref(self, ref: ArtifactRef, **owners: object) -> ArtifactVersion:
+        assert ref.digest == self.version.canonical_sha256
+        assert owners == {
+            "script_build_id": 7,
+            "task_id": "child",
+            "attempt_id": "attempt-child",
+        }
+        return self.version
+
+
+class _Contracts:
+    def __init__(self, values: dict[str, ScriptTaskContractV1]) -> None:
+        self.values = values
+
+    async def read_for_task(
+        self, *, root_trace_id: str, task: TaskRecord, spec_version: int | None = None
+    ) -> ScriptTaskContractV1:
+        assert root_trace_id == "root"
+        assert spec_version == 1
+        return self.values[task.task_id]
+
+
+def _closed_fixture() -> tuple[AcceptedInputResolver, TaskLedger, TaskAttempt]:
+    root = _task("root-task", parent=None, kind="root", status=TaskStatus.NEEDS_REPLAN)
+    parent = _task("parent", parent="root-task", kind="compose", status=TaskStatus.RUNNING)
+    child = _task("child", parent="parent", kind="paragraph", status=TaskStatus.COMPLETED)
+    parent.child_task_ids.append("child")
+    root.child_task_ids.append("parent")
+    ref = ArtifactRef(
+        uri="script-build://artifact-versions/1",
+        kind=ArtifactKind.PARAGRAPH.value,
+        version="1",
+        digest=DIGEST,
+    )
+    child_attempt = TaskAttempt(
+        attempt_id="attempt-child",
+        task_id="child",
+        spec_version=1,
+        worker_trace_id="worker",
+        worker_preset="script_paragraph_worker",
+        execution_mode="new",
+        accepted_child_decision_ids=(),
+        status=AttemptStatus.SUBMITTED,
+        snapshot_id="artifact-snapshot",
+        submission=SimpleNamespace(artifact_refs=[ref], evidence_refs=[]),
+    )
+    validation = ValidationReport(
+        validation_id="validation-child",
+        task_id="child",
+        attempt_id="attempt-child",
+        spec_version=1,
+        snapshot_id="artifact-snapshot",
+        validator_trace_id="validator",
+        status=ValidationRunStatus.COMPLETED,
+        verdict=ValidationVerdict.PASSED,
+    )
+    decision = PlannerDecision(
+        decision_id="decision-child",
+        task_id="child",
+        action=DecisionAction.ACCEPT,
+        reason="passed",
+        from_status=TaskStatus.AWAITING_DECISION,
+        to_status=TaskStatus.COMPLETED,
+        attempt_id="attempt-child",
+        validation_id="validation-child",
+    )
+    child.attempt_ids.append(child_attempt.attempt_id)
+    child.validation_ids.append(validation.validation_id)
+    child.decision_ids.append(decision.decision_id)
+    parent_attempt = TaskAttempt(
+        attempt_id="attempt-parent",
+        task_id="parent",
+        spec_version=1,
+        worker_trace_id="compose-worker",
+        worker_preset="script_compose_worker",
+        execution_mode="new",
+        accepted_child_decision_ids=(decision.decision_id,),
+    )
+    ledger = TaskLedger(
+        root_trace_id="root",
+        mission="mission",
+        root_task_id="root-task",
+        tasks={item.task_id: item for item in (root, parent, child)},
+        attempts={
+            child_attempt.attempt_id: child_attempt,
+            parent_attempt.attempt_id: parent_attempt,
+        },
+        validations={validation.validation_id: validation},
+        decisions={decision.decision_id: decision},
+    )
+    normalized = {
+        "summary": "bounded",
+        "artifact_refs": [
+            {
+                "uri": ref.uri,
+                "kind": ref.kind,
+                "version": ref.version,
+                "digest": ref.digest,
+                "summary": ref.summary,
+                "metadata": ref.metadata,
+            }
+        ],
+        "evidence_refs": [],
+    }
+    snapshot = ArtifactSnapshot(
+        snapshot_id="artifact-snapshot",
+        attempt_id=child_attempt.attempt_id,
+        normalized_content=normalized,
+        sha256=_snapshot_sha(normalized),
+        artifact_refs=[ref],
+        evidence_refs=[],
+    )
+    artifact = replace(_paragraph_artifact(), canonical_sha256=DIGEST)
+    version = ArtifactVersion(
+        artifact_version_id=1,
+        script_build_id=7,
+        task_id="child",
+        attempt_id=child_attempt.attempt_id,
+        spec_version=1,
+        artifact_type=ArtifactKind.PARAGRAPH,
+        canonical_sha256=DIGEST,
+        state=ArtifactState.FROZEN,
+        artifact=artifact,
+        created_at=datetime.now(UTC),
+        frozen_at=datetime.now(UTC),
+    )
+    resolver = AcceptedInputResolver(
+        task_store=_TaskStore(ledger),
+        artifact_store=_SnapshotStore(snapshot),
+        artifacts=_Artifacts(version),
+        contracts=_Contracts({"child": _contract(), "parent": _contract()}),
+    )
+    return resolver, ledger, parent_attempt
+
+
+@pytest.mark.asyncio
+async def test_accepted_input_closes_decision_validation_snapshot_and_artifact() -> None:
+    resolver, _, attempt = _closed_fixture()
+    bundle = await resolver.resolve(
+        root_trace_id="root",
+        script_build_id=7,
+        task_id="parent",
+        attempt_id=attempt.attempt_id,
+        contract=_contract(),
+        input_snapshot_id="11",
+    )
+    assert [item.decision_id for item in bundle.inputs] == ["decision-child"]
+    assert bundle.inputs[0].source == "direct-child"
+    assert bundle.input_closure_digest.startswith("sha256:")
+
+
+@pytest.mark.asyncio
+async def test_legacy_attempt_without_frozen_child_set_fails_closed() -> None:
+    resolver, ledger, attempt = _closed_fixture()
+    ledger.attempts[attempt.attempt_id] = replace(attempt, accepted_child_decision_ids=None)
+    with pytest.raises(PhaseTwoInputError, match="CHILD_DECISION_INVALID"):
+        await resolver.resolve(
+            root_trace_id="root",
+            script_build_id=7,
+            task_id="parent",
+            attempt_id=attempt.attempt_id,
+            contract=_contract(),
+            input_snapshot_id="11",
+        )
+
+
+@pytest.mark.asyncio
+async def test_mid_retrieval_replacement_freezes_new_evidence_and_old_candidate() -> None:
+    resolver, ledger, parent_attempt = _closed_fixture()
+    old_ref = ledger.attempts["attempt-child"].submission.artifact_refs[0]
+    old_contract = _contract()
+
+    # The old accepted Paragraph remains a completed sibling.  The replacement
+    # consumer gets new Evidence from its own direct Retrieval child and imports
+    # the immutable old Decision explicitly; it never reads a mutable "latest".
+    ledger.tasks["child"].parent_task_id = "root-task"
+    ledger.tasks["root-task"].child_task_ids = ["child", "parent"]
+    replacement_task = _task(
+        "parent", parent="root-task", kind="paragraph", status=TaskStatus.RUNNING
+    )
+    replacement_task.child_task_ids = ["retrieval"]
+    replacement_task.attempt_ids = [parent_attempt.attempt_id]
+    ledger.tasks["parent"] = replacement_task
+
+    retrieval = _task(
+        "retrieval", parent="parent", kind="decode-retrieval", status=TaskStatus.COMPLETED
+    )
+    retrieval_ref = ArtifactRef(
+        "script-build://artifact-versions/2",
+        ArtifactKind.EVIDENCE.value,
+        "2",
+        DIGEST,
+    )
+    retrieval_attempt = TaskAttempt(
+        attempt_id="attempt-retrieval",
+        task_id="retrieval",
+        spec_version=1,
+        worker_trace_id="retrieval-worker",
+        worker_preset="script_decode_retrieval_worker",
+        execution_mode="new",
+        accepted_child_decision_ids=(),
+        status=AttemptStatus.SUBMITTED,
+        snapshot_id="retrieval-snapshot",
+        submission=SimpleNamespace(artifact_refs=[], evidence_refs=[retrieval_ref]),
+    )
+    retrieval_validation = ValidationReport(
+        validation_id="validation-retrieval",
+        task_id="retrieval",
+        attempt_id=retrieval_attempt.attempt_id,
+        spec_version=1,
+        snapshot_id="retrieval-snapshot",
+        validator_trace_id="retrieval-validator",
+        status=ValidationRunStatus.COMPLETED,
+        verdict=ValidationVerdict.PASSED,
+    )
+    retrieval_decision = PlannerDecision(
+        decision_id="decision-retrieval",
+        task_id="retrieval",
+        action=DecisionAction.ACCEPT,
+        reason="new evidence passed",
+        from_status=TaskStatus.AWAITING_DECISION,
+        to_status=TaskStatus.COMPLETED,
+        attempt_id=retrieval_attempt.attempt_id,
+        validation_id=retrieval_validation.validation_id,
+    )
+    retrieval.attempt_ids = [retrieval_attempt.attempt_id]
+    retrieval.validation_ids = [retrieval_validation.validation_id]
+    retrieval.decision_ids = [retrieval_decision.decision_id]
+    ledger.tasks[retrieval.task_id] = retrieval
+    ledger.attempts[retrieval_attempt.attempt_id] = retrieval_attempt
+    ledger.validations[retrieval_validation.validation_id] = retrieval_validation
+    ledger.decisions[retrieval_decision.decision_id] = retrieval_decision
+    ledger.attempts[parent_attempt.attempt_id] = replace(
+        parent_attempt,
+        accepted_child_decision_ids=(retrieval_decision.decision_id,),
+    )
+
+    retrieval_normalized = {
+        "summary": "bounded retrieval evidence",
+        "artifact_refs": [],
+        "evidence_refs": [
+            {
+                "uri": retrieval_ref.uri,
+                "kind": retrieval_ref.kind,
+                "version": retrieval_ref.version,
+                "digest": retrieval_ref.digest,
+                "summary": retrieval_ref.summary,
+                "metadata": retrieval_ref.metadata,
+            }
+        ],
+    }
+    retrieval_snapshot = ArtifactSnapshot(
+        snapshot_id="retrieval-snapshot",
+        attempt_id=retrieval_attempt.attempt_id,
+        normalized_content=retrieval_normalized,
+        sha256=_snapshot_sha(retrieval_normalized),
+        artifact_refs=[],
+        evidence_refs=[retrieval_ref],
+    )
+    evidence = EvidenceRecordV1(
+        evidence_id="mid-retrieval",
+        source_type="decode",
+        tool_name="retrieve_decode",
+        query={"query": "opening detail", "top_k": 3},
+        source_refs=("decode-index://fixture/new-evidence",),
+        raw_artifact_ref=None,
+        summary="One visible detail contradicts the original opening assumption.",
+        supports=(SCOPE,),
+        confidence="high",
+        limitations=(),
+        content_sha256=DIGEST,
+        created_at=datetime.now(UTC),
+    )
+    evidence_version = ArtifactVersion(
+        artifact_version_id=2,
+        script_build_id=7,
+        task_id="retrieval",
+        attempt_id=retrieval_attempt.attempt_id,
+        spec_version=1,
+        artifact_type=ArtifactKind.EVIDENCE,
+        canonical_sha256=DIGEST,
+        state=ArtifactState.FROZEN,
+        artifact=evidence,
+        created_at=datetime.now(UTC),
+        frozen_at=datetime.now(UTC),
+    )
+
+    old_snapshot = resolver._artifact_store.snapshot
+
+    class SnapshotStore:
+        async def get(self, root_trace_id: str, snapshot_id: str) -> ArtifactSnapshot:
+            assert root_trace_id == "root"
+            return {
+                old_snapshot.snapshot_id: old_snapshot,
+                retrieval_snapshot.snapshot_id: retrieval_snapshot,
+            }[snapshot_id]
+
+    old_version = resolver._artifacts.version
+
+    class Artifacts:
+        async def read_by_ref(self, ref: ArtifactRef, **owners: object) -> ArtifactVersion:
+            version = {old_ref.uri: old_version, retrieval_ref.uri: evidence_version}[ref.uri]
+            assert owners == {
+                "script_build_id": 7,
+                "task_id": version.task_id,
+                "attempt_id": version.attempt_id,
+            }
+            return version
+
+    replacement_contract = replace(
+        old_contract,
+        intent_class=ScriptIntentClass.REPLACE,
+        input_decision_refs=(
+            AcceptedDecisionRef(
+                "decision-child",
+                old_ref,
+                SCOPE,
+                ScriptTaskKind.PARAGRAPH,
+            ),
+        ),
+        supersedes_decision_ids=("decision-child",),
+    )
+    resolver._artifact_store = SnapshotStore()
+    resolver._artifacts = Artifacts()
+    resolver._contracts.values = {
+        "child": old_contract,
+        "retrieval": replace(
+            _contract(kind=ScriptTaskKind.DECODE_RETRIEVAL),
+            write_scope=(),
+        ),
+        "parent": replacement_contract,
+    }
+
+    bundle = await resolver.resolve(
+        root_trace_id="root",
+        script_build_id=7,
+        task_id="parent",
+        attempt_id=parent_attempt.attempt_id,
+        contract=replacement_contract,
+        input_snapshot_id="11",
+    )
+    assert [item.decision_id for item in bundle.inputs] == [
+        "decision-retrieval",
+        "decision-child",
+    ]
+    assert [item.source for item in bundle.inputs] == ["direct-child", "explicit"]
+    assert bundle.superseded_decision_ids == ("decision-child",)
+    assert bundle.input_closure_digest.startswith("sha256:")
+
+
+@pytest.mark.asyncio
+async def test_direct_child_is_revalidated_against_consumer_scope_and_kind() -> None:
+    resolver, _, attempt = _closed_fixture()
+    resolver._contracts.values["parent"] = replace(
+        _contract(), scope_ref="script-build://scopes/ending"
+    )
+    with pytest.raises(PhaseTwoInputError, match="INPUT_SCOPE_MISMATCH"):
+        await resolver.resolve(
+            root_trace_id="root",
+            script_build_id=7,
+            task_id="parent",
+            attempt_id=attempt.attempt_id,
+            contract=resolver._contracts.values["parent"],
+            input_snapshot_id="11",
+        )
+
+    resolver, _, attempt = _closed_fixture()
+    consumer = replace(
+        _contract(),
+        task_kind=ScriptTaskKind.CANDIDATE_PORTFOLIO,
+        intent_class=ScriptIntentClass.PORTFOLIO,
+        output_schema="candidate-portfolio/v1",
+    )
+    resolver._contracts.values["parent"] = consumer
+    with pytest.raises(PhaseTwoInputError, match="INPUT_SCOPE_MISMATCH"):
+        await resolver.resolve(
+            root_trace_id="root",
+            script_build_id=7,
+            task_id="parent",
+            attempt_id=attempt.attempt_id,
+            contract=consumer,
+            input_snapshot_id="11",
+        )
+
+
+@pytest.mark.asyncio
+async def test_explicit_input_is_revalidated_against_consumer_scope() -> None:
+    resolver, ledger, attempt = _closed_fixture()
+    child_ref = ledger.attempts["attempt-child"].submission.artifact_refs[0]
+    ledger.tasks["child"].parent_task_id = "root-task"
+    ledger.tasks["parent"].child_task_ids = []
+    ledger.attempts[attempt.attempt_id] = replace(
+        attempt,
+        accepted_child_decision_ids=(),
+    )
+    consumer = replace(
+        _contract(),
+        scope_ref="script-build://scopes/ending",
+        input_decision_refs=(
+            AcceptedDecisionRef(
+                "decision-child",
+                child_ref,
+                SCOPE,
+                ScriptTaskKind.PARAGRAPH,
+            ),
+        ),
+    )
+    resolver._contracts.values["parent"] = consumer
+
+    with pytest.raises(PhaseTwoInputError, match="INPUT_SCOPE_MISMATCH"):
+        await resolver.resolve(
+            root_trace_id="root",
+            script_build_id=7,
+            task_id="parent",
+            attempt_id=attempt.attempt_id,
+            contract=consumer,
+            input_snapshot_id="11",
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mutation", ["verdict", "snapshot", "digest", "not-current"])
+async def test_accepted_input_rejects_broken_causal_edges(mutation: str) -> None:
+    resolver, ledger, attempt = _closed_fixture()
+    if mutation == "verdict":
+        ledger.validations["validation-child"].verdict = ValidationVerdict.FAILED
+    elif mutation == "snapshot":
+        ledger.validations["validation-child"].snapshot_id = "other"
+    elif mutation == "digest":
+        resolver._artifacts.version = replace(
+            resolver._artifacts.version, canonical_sha256="sha256:" + "b" * 64
+        )
+    else:
+        ledger.tasks["child"].decision_ids.append("other-decision")
+    with pytest.raises((PhaseTwoInputError, AssertionError)):
+        await resolver.resolve(
+            root_trace_id="root",
+            script_build_id=7,
+            task_id="parent",
+            attempt_id=attempt.attempt_id,
+            contract=_contract(),
+            input_snapshot_id="11",
+        )
+
+
+def test_active_frontier_replacement_cycle_scope_and_write_conflicts() -> None:
+    ref1 = ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST)
+    ref2 = ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST)
+    bundle = AcceptedInputBundleV1(
+        inputs=(
+            AcceptedInput("d1", ref1, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"),
+            AcceptedInput("d2", ref2, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"),
+        ),
+        input_closure_digest=DIGEST,
+    )
+    frontier = ActiveFrontierResolver()
+    with pytest.raises(PhaseTwoInputError, match="WRITE_SCOPE_CONFLICT"):
+        frontier.resolve(bundle, contracts_by_decision={"d1": _contract(), "d2": _contract()})
+
+    active = frontier.resolve(
+        bundle,
+        contracts_by_decision={
+            "d1": _contract(),
+            "d2": _contract(supersedes=("d1",)),
+        },
+    )
+    assert [item.decision_id for item in active] == ["d2"]
+
+    exact_base_patch = replace(_contract(), base_artifact_ref=ref1)
+    active = frontier.resolve(
+        bundle,
+        contracts_by_decision={"d1": _contract(), "d2": exact_base_patch},
+    )
+    assert [item.decision_id for item in active] == ["d1", "d2"]
+
+    stale_base_patch = replace(
+        _contract(),
+        base_artifact_ref=replace(ref1, digest="sha256:" + "b" * 64),
+    )
+    with pytest.raises(PhaseTwoInputError, match="WRITE_SCOPE_CONFLICT"):
+        frontier.resolve(
+            bundle,
+            contracts_by_decision={"d1": _contract(), "d2": stale_base_patch},
+        )
+
+    with pytest.raises(PhaseTwoInputError, match="SUPERSESSION_CYCLE"):
+        frontier.resolve(
+            bundle,
+            contracts_by_decision={
+                "d1": _contract(supersedes=("d2",)),
+                "d2": _contract(supersedes=("d1",)),
+            },
+        )
+    incompatible = AcceptedInputBundleV1(
+        inputs=(
+            bundle.inputs[0],
+            replace(bundle.inputs[1], scope_ref="script-build://scopes/ending"),
+        ),
+        input_closure_digest=DIGEST,
+    )
+    with pytest.raises(PhaseTwoInputError, match="SUPERSESSION_SCOPE_MISMATCH"):
+        frontier.resolve(
+            incompatible,
+            contracts_by_decision={
+                "d1": _contract(),
+                "d2": _contract(scope="script-build://scopes/ending", supersedes=("d1",)),
+            },
+        )
+
+
+def test_stale_base_revision_fails_closed() -> None:
+    artifact = replace(_paragraph_artifact(), canonical_sha256=DIGEST)
+    version = ArtifactVersion(
+        1,
+        7,
+        "child",
+        "attempt-child",
+        1,
+        ArtifactKind.PARAGRAPH,
+        DIGEST,
+        ArtifactState.FROZEN,
+        artifact,
+        datetime.now(UTC),
+        datetime.now(UTC),
+    )
+    lineage = replace(
+        artifact.lineage,
+        base_artifact_ref="script-build://artifact-versions/1",
+        base_artifact_digest=DIGEST,
+        base_revision=1,
+    )
+    ActiveFrontierResolver.verify_base(lineage=lineage, base=version)
+    with pytest.raises(PhaseTwoInputError, match="STALE_BASE_REVISION"):
+        ActiveFrontierResolver.verify_base(
+            lineage=lineage, base=replace(version, artifact_version_id=2)
+        )
+
+
+def test_four_validation_layers_defects_and_placeholder_preflight() -> None:
+    assert validation_layer("paragraph") == "local"
+    assert validation_layer("compare") == "compare"
+    assert validation_layer("compose") == "global"
+    assert validation_layer("candidate-portfolio") == "governance"
+    assert (
+        ScriptBuildValidationPolicy()
+        .plan(
+            SimpleNamespace(task=_task("t", parent=None, kind="compose", status=TaskStatus.RUNNING))
+        )
+        .validator_preset
+        == "script_candidate_validator"
+    )
+
+    payload = {
+        "defect_code": "REALIZATION_PLACEHOLDER",
+        "criterion_id": "closed",
+        "scope_ref": SCOPE,
+        "observed_excerpt": "TODO",
+        "evidence_refs": ["script-build://artifact-versions/1"],
+        "severity": "hard",
+        "invalidated_inputs": ["decision-child"],
+        "recommended_action_class": "replace",
+    }
+    with pytest.raises(Exception, match="passed validation"):
+        validate_phase_two_defects(
+            [payload],
+            criterion_ids={"closed"},
+            allowed_evidence_refs={"script-build://artifact-versions/1"},
+            passed=True,
+        )
+    defects = validate_phase_two_defects(
+        [payload],
+        criterion_ids={"closed"},
+        allowed_evidence_refs={"script-build://artifact-versions/1"},
+        passed=False,
+    )
+    assert defects[0].blocks_acceptance
+    assert precheck_business_artifact(_paragraph_artifact(text="TODO: later")) == (
+        "artifact.paragraphs[0].description",
+    )
+    assert precheck_business_artifact(_paragraph_artifact(text="此处描述其存在")) == (
+        "artifact.paragraphs[0].description",
+    )
+
+
+def test_defect_rejects_unknown_criterion_evidence_and_unbounded_excerpt() -> None:
+    base = {
+        "defect_code": "X",
+        "criterion_id": "closed",
+        "scope_ref": SCOPE,
+        "observed_excerpt": "x",
+        "evidence_refs": [],
+        "severity": "warning",
+        "invalidated_inputs": [],
+        "recommended_action_class": "repair",
+    }
+    with pytest.raises(Exception, match="outside the frozen contract"):
+        validate_phase_two_defects(
+            [{**base, "criterion_id": "other"}],
+            criterion_ids={"closed"},
+            allowed_evidence_refs=set(),
+            passed=False,
+        )
+    with pytest.raises(Exception, match="500 characters"):
+        validate_phase_two_defects(
+            [{**base, "observed_excerpt": "x" * 501}],
+            criterion_ids={"closed"},
+            allowed_evidence_refs=set(),
+            passed=False,
+        )
+
+
+def test_bundle_digest_changes_when_source_or_order_changes() -> None:
+    first = canonical_sha256(["a", "b"]).wire
+    second = canonical_sha256(["b", "a"]).wire
+    assert first != second
+
+
+def _snapshot_sha(value: dict[str, object]) -> str:
+    canonical = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode()).hexdigest()

script_build_host/tests/test_phase_two_workspace.py

@@ -0,0 +1,628 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import replace
+
+import pytest
+from sqlalchemy import func, select
+
+from script_build_host.domain.artifacts import ArtifactKind, ArtifactState
+from script_build_host.domain.phase_two_artifacts import CandidateLineageV1, ParagraphArtifactV1
+from script_build_host.domain.workspaces import CandidateWriteContext, WorkspaceError
+from script_build_host.infrastructure.legacy_tables import (
+    script_build_element,
+    script_build_paragraph,
+    script_build_paragraph_element,
+    script_build_task_plan_step,
+)
+from script_build_host.infrastructure.tables import artifact_version_table
+from script_build_host.repositories.sqlalchemy import (
+    SqlAlchemyScriptBusinessArtifactRepository,
+)
+from script_build_host.repositories.workspace import (
+    SqlAlchemyCandidateWorkspaceRepository,
+)
+
+
+def _context(attempt: str, *, task: str = "task-1", build: int = 1) -> CandidateWriteContext:
+    return CandidateWriteContext(
+        script_build_id=build,
+        task_id=task,
+        attempt_id=attempt,
+        spec_version=1,
+        objective="produce the opening increment",
+        input_refs=("script-build://artifact-versions/90",),
+        write_scope=("script-build://scopes/opening",),
+    )
+
+
+def _lineage(
+    *,
+    base_ref: str | None = None,
+    base_digest: str | None = None,
+    base_revision: int | None = None,
+) -> CandidateLineageV1:
+    return CandidateLineageV1(
+        scope_ref="script-build://scopes/opening",
+        input_snapshot_ref="script-build://inputs/11",
+        input_closure_digest="sha256:" + "a" * 64,
+        write_scope=("script-build://scopes/opening",),
+        base_artifact_ref=base_ref,
+        base_artifact_digest=base_digest,
+        base_revision=base_revision,
+    )
+
+
+@pytest.mark.asyncio
+async def test_workspace_allocates_positive_branch_and_freezes_full_paragraph(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("attempt-paragraph")
+    workspace = await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH)
+    assert workspace.branch_id == workspace.artifact_version_id > 0
+    assert (
+        await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH) == workspace
+    )
+
+    parent = await repository.create_paragraph(
+        context,
+        paragraph_index=1,
+        name="opening",
+        content_range={"topics": [1, 2]},
+        theme_elements=({"维度": "主题", "原子点": "冲突", "维度类型": "主维度"},),
+        form_elements=({"维度": "形式", "原子点": "对照", "维度类型": "主维度"},),
+    )
+    child = await repository.create_paragraph(
+        context,
+        paragraph_index=2,
+        name="opening detail",
+        content_range={"topics": [1]},
+        level=2,
+        parent_paragraph_id=parent,
+    )
+    await repository.batch_update_paragraphs(
+        context,
+        [
+            {
+                "paragraph_id": parent,
+                "theme": "a concrete conflict",
+                "form": "contrast",
+                "function": "hook",
+                "feeling": "curiosity",
+                "description": "bounded description",
+                "full_description": "fully realized opening",
+            }
+        ],
+    )
+    added = await repository.append_paragraph_atoms(
+        context,
+        paragraph_id=child,
+        column="feeling_elements",
+        atoms=({"维度": "情绪", "原子点": "期待"},),
+    )
+    assert added == 1
+    version, reference = await repository.freeze(context, lineage=_lineage())
+    assert version.state is ArtifactState.FROZEN
+    assert reference.kind == ArtifactKind.PARAGRAPH.value
+    assert isinstance(version.artifact, ParagraphArtifactV1)
+    assert len(version.artifact.paragraphs) == 2
+    assert version.artifact.paragraphs[0].full_description == "fully realized opening"
+
+    with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"):
+        await repository.create_paragraph(
+            context,
+            paragraph_index=3,
+            name="late mutation",
+            content_range={"topics": [2]},
+        )
+    replayed, replay_ref = await repository.freeze(context, lineage=_lineage())
+    assert (replayed.artifact_version_id, replay_ref) == (
+        version.artifact_version_id,
+        reference,
+    )
+    with pytest.raises(WorkspaceError, match="STALE_BASE_REVISION"):
+        await repository.freeze(
+            context,
+            lineage=CandidateLineageV1(
+                scope_ref="script-build://scopes/opening",
+                input_snapshot_ref="script-build://inputs/12",
+                input_closure_digest="sha256:" + "b" * 64,
+                write_scope=("script-build://scopes/opening",),
+            ),
+        )
+    async with sessions() as session:
+        plan = (
+            (
+                await session.execute(
+                    select(script_build_task_plan_step).where(
+                        script_build_task_plan_step.c.script_build_id == 1,
+                        script_build_task_plan_step.c.branch_id == workspace.branch_id,
+                    )
+                )
+            )
+            .mappings()
+            .one()
+        )
+    assert plan["step_key"] == "1"
+    assert plan["round_index"] is None
+    assert plan["status"] == "完成"
+
+
+@pytest.mark.asyncio
+async def test_semantic_write_scope_is_enforced_before_atomic_batch(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    full_context = _context("write-scope")
+    await repository.get_or_create(full_context, artifact_kind=ArtifactKind.PARAGRAPH)
+    first = await repository.create_paragraph(
+        full_context, paragraph_index=1, name="one", content_range={"topics": [1]}
+    )
+    second = await repository.create_paragraph(
+        full_context, paragraph_index=2, name="two", content_range={"topics": [2]}
+    )
+    bounded = replace(full_context, write_scope=(f"script-build://writes/paragraphs/{first}",))
+    with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"):
+        await repository.batch_update_paragraphs(
+            bounded,
+            (
+                {"paragraph_id": first, "description": "authorized but must roll back"},
+                {"paragraph_id": second, "description": "outside scope"},
+            ),
+        )
+    snapshot = await repository.snapshot(full_context)
+    assert [item.description for item in snapshot.paragraphs] == [None, None]
+    with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"):
+        await repository.create_element(
+            bounded,
+            name="not allowed",
+            dimension_primary="实质",
+            dimension_secondary="detail",
+        )
+
+
+@pytest.mark.asyncio
+async def test_content_range_scope_must_cover_every_key_and_value(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = replace(
+        _context("content-range-scope"),
+        write_scope=("script-build://writes/content-ranges/topics/1",),
+    )
+    await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH)
+
+    with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"):
+        await repository.create_paragraph(
+            context,
+            paragraph_index=1,
+            name="partially authorized",
+            content_range={"topics": [1, 2], "sections": ["ending"]},
+        )
+
+    snapshot = await repository.snapshot(context)
+    assert snapshot.paragraphs == ()
+
+
+@pytest.mark.asyncio
+async def test_discard_marks_workspace_and_legacy_plan_blocked(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("discarded-attempt")
+    workspace = await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH)
+    await repository.create_paragraph(
+        context, paragraph_index=1, name="draft", content_range={"topics": [1]}
+    )
+    discarded = await repository.discard(context, reason="validator hard defect")
+    assert discarded.state is ArtifactState.DISCARDED
+    assert (
+        await repository.discard(context, reason="idempotent replay")
+    ).state is ArtifactState.DISCARDED
+    async with sessions() as session:
+        plan = (
+            (
+                await session.execute(
+                    select(script_build_task_plan_step).where(
+                        script_build_task_plan_step.c.script_build_id == context.script_build_id,
+                        script_build_task_plan_step.c.branch_id == workspace.branch_id,
+                    )
+                )
+            )
+            .mappings()
+            .one()
+        )
+    assert plan["status"] == "受阻"
+    assert plan["result_note"] == "validator hard defect"
+    with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"):
+        await repository.create_paragraph(
+            context, paragraph_index=2, name="late", content_range={"topics": [2]}
+        )
+
+
+@pytest.mark.asyncio
+async def test_element_and_link_batch_is_atomic_and_workspace_scoped(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+
+    paragraph_context = _context("base-paragraph")
+    await repository.get_or_create(paragraph_context, artifact_kind=ArtifactKind.PARAGRAPH)
+    paragraph_id = await repository.create_paragraph(
+        paragraph_context,
+        paragraph_index=1,
+        name="body",
+        content_range={"topics": [1]},
+    )
+    base, base_ref = await repository.freeze(paragraph_context, lineage=_lineage())
+
+    element_context = _context("element-attempt", task="element-task")
+    workspace = await repository.get_or_create(
+        element_context,
+        artifact_kind=ArtifactKind.ELEMENT_SET,
+        base_artifact_ref=base_ref,
+    )
+    snapshot = await repository.snapshot(element_context)
+    copied_paragraph = snapshot.paragraphs[0].paragraph_id
+    assert copied_paragraph != paragraph_id
+    assert snapshot.source_identity_map == (
+        {
+            "entity": "paragraph",
+            "source_local_id": paragraph_id,
+            "local_id": copied_paragraph,
+        },
+    )
+    element_id = await repository.create_element(
+        element_context,
+        name="specific observation",
+        dimension_primary="实质",
+        dimension_secondary="detail",
+        commonality_analysis={"shared": "grounded"},
+        topic_support={"topic": 1},
+        weight_score={"score": 0.8},
+        support_elements=({"ref": "source-1"},),
+    )
+    with pytest.raises(WorkspaceError, match="LEGACY_REFERENCE_INVALID"):
+        await repository.batch_link(
+            element_context,
+            [(copied_paragraph, [element_id]), (999999, [element_id])],
+        )
+    async with sessions() as session:
+        assert (
+            await session.scalar(select(func.count()).select_from(script_build_paragraph_element))
+        ) == 0
+    assert await repository.batch_link(element_context, [(copied_paragraph, [element_id])]) == 1
+    with pytest.raises(WorkspaceError, match="STALE_BASE_REVISION"):
+        await repository.freeze(
+            element_context,
+            lineage=_lineage(
+                base_ref=base_ref.uri,
+                base_digest="sha256:" + "f" * 64,
+                base_revision=base.artifact_version_id,
+            ),
+        )
+    version, _ = await repository.freeze(
+        element_context,
+        lineage=_lineage(
+            base_ref=base_ref.uri,
+            base_digest=base_ref.digest,
+            base_revision=base.artifact_version_id,
+        ),
+    )
+    assert version.artifact.lineage.source_lineage[0]["source_local_id"] == paragraph_id
+    assert workspace.branch_id == version.artifact_version_id
+
+
+@pytest.mark.asyncio
+async def test_wrong_adapter_cross_context_and_stale_base_are_rejected(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("element-only")
+    await repository.get_or_create(context, artifact_kind=ArtifactKind.ELEMENT_SET)
+    with pytest.raises(WorkspaceError, match="WRITE_SCOPE_VIOLATION"):
+        await repository.create_paragraph(
+            context,
+            paragraph_index=1,
+            name="forbidden",
+            content_range={"topics": [1]},
+        )
+    with pytest.raises(WorkspaceError, match="LEGACY_REFERENCE_INVALID"):
+        await repository.require(_context("element-only", task="other-task"))
+    with pytest.raises(WorkspaceError, match="STALE_BASE_REVISION"):
+        await repository.freeze(
+            context,
+            lineage=CandidateLineageV1(
+                scope_ref="script-build://scopes/opening",
+                input_snapshot_ref="script-build://inputs/11",
+                input_closure_digest="sha256:" + "a" * 64,
+                write_scope=("script-build://scopes/opening",),
+                base_artifact_ref="script-build://artifact-versions/999",
+                base_artifact_digest="sha256:" + "b" * 64,
+                base_revision=999,
+            ),
+        )
+    async with sessions() as session:
+        counts = {
+            "paragraph": await session.scalar(
+                select(func.count()).select_from(script_build_paragraph)
+            ),
+            "element": await session.scalar(select(func.count()).select_from(script_build_element)),
+        }
+    assert counts == {"paragraph": 0, "element": 0}
+
+
+@pytest.mark.asyncio
+async def test_concurrent_workspace_replay_converges_on_one_attempt_and_branch(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("attempt-concurrent")
+
+    first, second = await asyncio.gather(
+        repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH),
+        repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH),
+    )
+
+    assert first == second
+    assert first.branch_id == first.artifact_version_id > 0
+    async with sessions() as session:
+        rows = (
+            (
+                await session.execute(
+                    select(artifact_version_table).where(
+                        artifact_version_table.c.attempt_id == context.attempt_id
+                    )
+                )
+            )
+            .mappings()
+            .all()
+        )
+    assert len(rows) == 1
+    assert rows[0]["legacy_branch_id"] == rows[0]["id"]
+
+
+@pytest.mark.asyncio
+async def test_batch_update_validates_every_paragraph_before_mutating(database) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("attempt-atomic-update")
+    await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH)
+    first_id = await repository.create_paragraph(
+        context,
+        paragraph_index=1,
+        name="original",
+        content_range={"topics": [1]},
+    )
+    second_id = await repository.create_paragraph(
+        context,
+        paragraph_index=2,
+        name="second",
+        content_range={"topics": [2]},
+    )
+
+    with pytest.raises(WorkspaceError, match="LEGACY_REFERENCE_INVALID"):
+        await repository.batch_update_paragraphs(
+            context,
+            [
+                {"paragraph_id": first_id, "name": "must roll back"},
+                {"paragraph_id": 999_999, "name": "outside workspace"},
+            ],
+        )
+
+    snapshot = await repository.snapshot(context)
+    assert snapshot.paragraphs[0].name == "original"
+
+    with pytest.raises(WorkspaceError, match="paragraph name"):
+        await repository.batch_update_paragraphs(
+            context,
+            [
+                {"paragraph_id": first_id, "description": "must still roll back"},
+                {"paragraph_id": second_id, "name": ""},
+            ],
+        )
+    snapshot = await repository.snapshot(context)
+    assert snapshot.paragraphs[0].description is None
+    assert snapshot.paragraphs[1].name == "second"
+
+
+@pytest.mark.asyncio
+async def test_element_patch_copies_complete_projection_and_freezes_change_manifest(
+    database,
+) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+
+    paragraph_context = _context("paragraph-base")
+    await repository.get_or_create(paragraph_context, artifact_kind=ArtifactKind.PARAGRAPH)
+    source_paragraph_id = await repository.create_paragraph(
+        paragraph_context,
+        paragraph_index=1,
+        name="body",
+        content_range={"topics": [1]},
+    )
+    paragraph_version, paragraph_ref = await repository.freeze(
+        paragraph_context, lineage=_lineage()
+    )
+
+    element_context = _context("element-base", task="element-base-task")
+    await repository.get_or_create(
+        element_context,
+        artifact_kind=ArtifactKind.ELEMENT_SET,
+        base_artifact_ref=paragraph_ref,
+    )
+    copied_paragraph_id = (await repository.snapshot(element_context)).paragraphs[0].paragraph_id
+    source_element_id = await repository.create_element(
+        element_context,
+        name="old detail",
+        dimension_primary="实质",
+        dimension_secondary="detail",
+    )
+    await repository.batch_link(element_context, [(copied_paragraph_id, [source_element_id])])
+    element_version, element_ref = await repository.freeze(
+        element_context,
+        lineage=_lineage(
+            base_ref=paragraph_ref.uri,
+            base_digest=paragraph_ref.digest,
+            base_revision=paragraph_version.artifact_version_id,
+        ),
+    )
+    assert len(element_version.artifact.paragraphs) == 1
+    assert len(element_version.artifact.paragraph_element_links) == 1
+
+    replacement_context = _context("element-replacement", task="element-replacement-task")
+    replacement = await repository.get_or_create(
+        replacement_context,
+        artifact_kind=ArtifactKind.ELEMENT_SET,
+        base_artifact_ref=element_ref,
+    )
+    copied = await repository.snapshot(replacement_context)
+    replacement_paragraph_id = copied.paragraphs[0].paragraph_id
+    replacement_element_id = copied.elements[0].element_id
+    assert replacement_paragraph_id not in {source_paragraph_id, copied_paragraph_id}
+    assert replacement_element_id != source_element_id
+    assert copied.links[0].paragraph_id == replacement_paragraph_id
+    assert copied.links[0].element_id == replacement_element_id
+
+    await repository.update_element(
+        replacement_context,
+        element_id=replacement_element_id,
+        values={"name": "replaced detail", "is_active": False},
+    )
+    new_element_id = await repository.create_element(
+        replacement_context,
+        name="new detail",
+        dimension_primary="实质",
+        dimension_secondary="detail",
+    )
+    await repository.batch_link(replacement_context, [(replacement_paragraph_id, [new_element_id])])
+    replacement_version, _ = await repository.freeze(
+        replacement_context,
+        lineage=_lineage(
+            base_ref=element_ref.uri,
+            base_digest=element_ref.digest,
+            base_revision=element_version.artifact_version_id,
+        ),
+    )
+
+    manifest = replacement_version.artifact.change_manifest
+    assert manifest["created"]["element_ids"] == [new_element_id]
+    assert manifest["created"]["links"] == [
+        {"paragraph_id": replacement_paragraph_id, "element_id": new_element_id}
+    ]
+    assert manifest["updated"]["element_ids"] == [replacement_element_id]
+    assert manifest["deactivated"]["element_ids"] == [replacement_element_id]
+    assert manifest["deleted_links"] == [
+        {"paragraph_id": copied_paragraph_id, "element_id": source_element_id}
+    ]
+    assert replacement.branch_id == replacement_version.artifact_version_id > 0
+
+    for operation in (
+        repository.create_element(
+            replacement_context,
+            name="late",
+            dimension_primary="实质",
+            dimension_secondary="detail",
+        ),
+        repository.update_element(
+            replacement_context,
+            element_id=new_element_id,
+            values={"name": "late"},
+        ),
+        repository.batch_link(replacement_context, [(replacement_paragraph_id, [new_element_id])]),
+        repository.delete_links(
+            replacement_context,
+            pairs=((replacement_paragraph_id, new_element_id),),
+        ),
+    ):
+        with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"):
+            await operation
+
+
+@pytest.mark.asyncio
+async def test_run_454_shape_roundtrips_without_branch_zero(database) -> None:
+    """The legacy 5/11/13 detail shape remains expressible without old Branch/Round rows."""
+
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("run-454-shape", task="compose-454")
+    workspace = await repository.get_or_create(
+        context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT
+    )
+    paragraph_ids = [
+        await repository.create_paragraph(
+            context,
+            paragraph_index=index,
+            name=f"paragraph-{index}",
+            content_range={"topics": [index]},
+        )
+        for index in range(1, 6)
+    ]
+    element_ids = [
+        await repository.create_element(
+            context,
+            name=f"element-{index}",
+            dimension_primary="实质" if index <= 6 else "形式",
+            dimension_secondary=f"dimension-{index}",
+        )
+        for index in range(1, 12)
+    ]
+    link_pairs = [(paragraph_ids[index % 5], [element_ids[index % 11]]) for index in range(11)]
+    link_pairs.extend(
+        [
+            (paragraph_ids[0], [element_ids[1]]),
+            (paragraph_ids[1], [element_ids[2]]),
+        ]
+    )
+    assert await repository.batch_link(context, link_pairs) == 13
+
+    version, reference = await repository.freeze(
+        context,
+        lineage=_lineage(),
+        structured_script={
+            "direction_ref": "script-build://artifact-versions/999",
+            "source_artifact_refs": ["script-build://artifact-versions/888"],
+            "evidence_refs": ["script-build://artifact-versions/777"],
+            "acceptance_notes": ["legacy detail shape preserved"],
+        },
+    )
+    loaded = await artifacts.read_by_ref(reference, script_build_id=1)
+    assert loaded == version
+    assert len(loaded.artifact.paragraphs) == 5
+    assert len(loaded.artifact.elements) == 11
+    assert len(loaded.artifact.paragraph_element_links) == 13
+    assert workspace.branch_id == version.artifact_version_id > 0
+
+    async with sessions() as session:
+        paragraph_branches = set(
+            (
+                await session.execute(
+                    select(script_build_paragraph.c.branch_id).where(
+                        script_build_paragraph.c.script_build_id == 1
+                    )
+                )
+            ).scalars()
+        )
+        element_branches = set(
+            (
+                await session.execute(
+                    select(script_build_element.c.branch_id).where(
+                        script_build_element.c.script_build_id == 1
+                    )
+                )
+            ).scalars()
+        )
+        link_branches = set(
+            (
+                await session.execute(
+                    select(script_build_paragraph_element.c.branch_id).where(
+                        script_build_paragraph_element.c.script_build_id == 1
+                    )
+                )
+            ).scalars()
+        )
+    assert paragraph_branches == element_branches == link_branches == {workspace.branch_id}

script_build_host/tests/test_repositories.py

@@ -5,6 +5,7 @@ from datetime import UTC, datetime
 
 import pytest
 from agent.orchestration import ArtifactRef, EvidenceQuery
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
 
 from script_build_host.agents.validation import ScriptBuildArtifactEvidenceReader
@@ -19,10 +20,13 @@ from script_build_host.domain.errors import (
     ArtifactDigestMismatch,
     ArtifactNotFound,
     ArtifactOwnershipMismatch,
+    InputDigestMismatch,
     InputHashConflict,
 )
 from script_build_host.domain.input_snapshot import ScriptBuildInput
 from script_build_host.domain.records import BuildStatus, PublicationState, PublicationType
+from script_build_host.infrastructure.legacy_tables import script_build_record
+from script_build_host.infrastructure.tables import input_snapshot_table
 from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
 from script_build_host.repositories.sqlalchemy import (
     SqlAlchemyInputSnapshotRepository,
@@ -92,6 +96,21 @@ async def test_concurrent_same_snapshot_freeze_returns_one_version(
     assert first.snapshot_id == second.snapshot_id
 
 
+@pytest.mark.asyncio
+async def test_snapshot_get_recomputes_digest_and_rejects_database_tampering(database) -> None:
+    _, sessions = database
+    repository = SqlAlchemyInputSnapshotRepository(sessions)
+    frozen = await repository.freeze(_input())
+    async with sessions() as session, session.begin():
+        await session.execute(
+            update(input_snapshot_table)
+            .where(input_snapshot_table.c.id == int(frozen.snapshot_id))
+            .values(canonical_json={**frozen.to_input().canonical_payload(), "topic_id": 999})
+        )
+    with pytest.raises(InputDigestMismatch):
+        await repository.get(frozen.snapshot_id, script_build_id=1)
+
+
 @pytest.mark.asyncio
 async def test_binding_create_is_idempotent_and_active_pointer_is_build_scoped(
     database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
@@ -109,6 +128,24 @@ async def test_binding_create_is_idempotent_and_active_pointer_is_build_scoped(
     assert first.binding_id == second.binding_id
     updated = await repository.set_active_direction(script_build_id=1, artifact_version_id=99)
     assert updated.active_direction_artifact_version_id == 99
+    advanced = await repository.compare_and_set_input_snapshot(
+        script_build_id=1,
+        expected_snapshot_id=2,
+        new_snapshot_id=3,
+    )
+    assert advanced.input_snapshot_id == 3
+    replayed = await repository.compare_and_set_input_snapshot(
+        script_build_id=1,
+        expected_snapshot_id=2,
+        new_snapshot_id=3,
+    )
+    assert replayed.input_snapshot_id == 3
+    with pytest.raises(Exception, match="changed concurrently"):
+        await repository.compare_and_set_input_snapshot(
+            script_build_id=1,
+            expected_snapshot_id=2,
+            new_snapshot_id=4,
+        )
 
 
 @pytest.mark.asyncio
@@ -285,5 +322,55 @@ async def test_legacy_status_read_write_and_direction_projection(
     await repository.project_direction(build_id, "# accepted direction")
     await repository.set_status(build_id, BuildStatus.PARTIAL)
     assert await repository.get_status(build_id) is BuildStatus.PARTIAL
+    await repository.set_checkpoint(
+        build_id,
+        checkpoint_code="PHASE_ONE_CAPABILITY_BOUNDARY",
+        summary="direction ready",
+        root_trace_id="root-1",
+    )
+    async with sessions() as session:
+        row = (
+            (
+                await session.execute(
+                    select(script_build_record).where(script_build_record.c.id == build_id)
+                )
+            )
+            .mappings()
+            .one()
+        )
+    assert row["status"] == "partial"
+    assert row["error_message"] == "PHASE_ONE_CAPABILITY_BOUNDARY"
+    assert row["summary"] == "direction ready"
+    assert row["reson_trace_id"] == "root-1"
+    assert row["end_time"] is not None
+    await repository.set_status(build_id, BuildStatus.STOPPING)
+    async with sessions() as session:
+        stopping_row = (
+            (
+                await session.execute(
+                    select(script_build_record).where(script_build_record.c.id == build_id)
+                )
+            )
+            .mappings()
+            .one()
+        )
+    assert stopping_row["status"] == "stopping"
+    assert stopping_row["error_message"] is None
+    assert stopping_row["summary"] is None
+    assert stopping_row["end_time"] is None
+    await repository.begin_phase(build_id)
+    async with sessions() as session:
+        row = (
+            (
+                await session.execute(
+                    select(script_build_record).where(script_build_record.c.id == build_id)
+                )
+            )
+            .mappings()
+            .one()
+        )
+    assert row["status"] == "running"
+    assert row["end_time"] is None
+    assert row["summary"] is None
     with pytest.raises(Exception, match="cannot project script build success"):
         await repository.set_status(build_id, BuildStatus.SUCCESS)

script_build_host/tests/test_trace_verifier.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import pytest
+from agent import FileSystemTraceStore
+
+from script_build_host.domain.errors import InvalidConfiguration
+from script_build_host.infrastructure.trace_verifier import FileSystemTraceStoreVerifier
+
+
+@pytest.mark.asyncio
+async def test_filesystem_trace_verifier_reopens_message_event_and_attachment(tmp_path) -> None:
+    root = tmp_path / "traces"
+    verifier = FileSystemTraceStoreVerifier()
+
+    await verifier.verify(FileSystemTraceStore(str(root)), require_attachments=True)
+
+    probe_directories = list(root.glob("script-build-startup-probe-*"))
+    assert len(probe_directories) == 1
+    assert (probe_directories[0] / "meta.json").is_file()
+    assert list((probe_directories[0] / "attachments").rglob("*.bin"))
+
+
+@pytest.mark.asyncio
+async def test_filesystem_trace_verifier_rejects_an_external_store() -> None:
+    with pytest.raises(InvalidConfiguration, match="another store type"):
+        await FileSystemTraceStoreVerifier().verify(object())