规划：实现内容寻址任务合同、输入闭包与阶段二预算门禁

script_build_host/src/script_build_host/application/phase_two_inputs.py

@@ -0,0 +1,645 @@
+"""Resolve immutable accepted inputs and the active phase-two candidate frontier."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from agent.orchestration import (
+    ArtifactRef,
+    AttemptStatus,
+    DecisionAction,
+    TaskStatus,
+    ValidationRunStatus,
+    ValidationVerdict,
+)
+
+from script_build_host.domain.artifacts import ArtifactState, ArtifactVersion
+from script_build_host.domain.canonical_json import canonical_sha256
+from script_build_host.domain.errors import ScriptBuildError
+from script_build_host.domain.phase_two_artifacts import CandidateLineageV1
+from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
+from script_build_host.domain.ports import ScriptBusinessArtifactRepository
+from script_build_host.domain.task_contracts import (
+    AcceptedDecisionRef,
+    AcceptedInput,
+    AcceptedInputBundleV1,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+    task_kind_from_context_refs,
+)
+
+_CONTRACT_PREFIX = "script-build://task-contracts/sha256/"
+_TERMINAL = {
+    TaskStatus.COMPLETED,
+    TaskStatus.SUPERSEDED,
+    TaskStatus.BLOCKED,
+    TaskStatus.CANCELLED,
+}
+
+
+class PhaseTwoInputError(ScriptBuildError):
+    pass
+
+
+class TaskContractReader(Protocol):
+    async def read_for_task(
+        self, *, root_trace_id: str, task: Any, spec_version: int | None = None
+    ) -> ScriptTaskContractV1: ...
+
+
+@dataclass(frozen=True, slots=True)
+class StoredTaskContractReader:
+    store: ScriptTaskContractStore
+
+    async def read_for_task(
+        self, *, root_trace_id: str, task: Any, spec_version: int | None = None
+    ) -> ScriptTaskContractV1:
+        version = spec_version or task.current_spec_version
+        specs = [item for item in task.specs if item.version == version]
+        if len(specs) != 1:
+            raise PhaseTwoInputError(
+                "TASK_CONTRACT_INVALID", "Task does not contain the requested spec version"
+            )
+        refs = [item for item in specs[0].context_refs if item.startswith(_CONTRACT_PREFIX)]
+        if len(refs) != 1:
+            raise PhaseTwoInputError(
+                "TASK_CONTRACT_INVALID", "TaskSpec must pin exactly one task contract"
+            )
+        return (await self.store.read(root_trace_id, refs[0])).contract
+
+
+@dataclass(frozen=True, slots=True)
+class ResolvedAcceptedDecision:
+    accepted_input: AcceptedInput
+    artifact_version: ArtifactVersion
+    contract: ScriptTaskContractV1
+
+
+class AcceptedInputResolver:
+    """Fail-closed closure over Ledger, Validation snapshot and business artifact."""
+
+    def __init__(
+        self,
+        *,
+        task_store: Any,
+        artifact_store: Any,
+        artifacts: ScriptBusinessArtifactRepository,
+        contracts: TaskContractReader,
+        input_snapshots: Any | None = None,
+    ) -> None:
+        self._task_store = task_store
+        self._artifact_store = artifact_store
+        self._artifacts = artifacts
+        self._contracts = contracts
+        self._input_snapshots = input_snapshots
+
+    async def contract_for_task(
+        self, *, root_trace_id: str, task: Any, spec_version: int
+    ) -> ScriptTaskContractV1:
+        return await self._contracts.read_for_task(
+            root_trace_id=root_trace_id,
+            task=task,
+            spec_version=spec_version,
+        )
+
+    async def resolve(
+        self,
+        *,
+        root_trace_id: str,
+        script_build_id: int,
+        task_id: str,
+        attempt_id: str,
+        contract: ScriptTaskContractV1,
+        input_snapshot_id: str,
+    ) -> AcceptedInputBundleV1:
+        ledger = await self._task_store.load(root_trace_id)
+        if ledger.root_trace_id != root_trace_id:
+            raise PhaseTwoInputError("CHILD_DECISION_INVALID", "Ledger root identity changed")
+        task = ledger.tasks.get(task_id)
+        attempt = ledger.attempts.get(attempt_id)
+        if task is None or attempt is None or attempt.task_id != task_id:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "Attempt is outside the current Task"
+            )
+        if attempt.accepted_child_decision_ids is None:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID",
+                "legacy Attempt has no frozen direct-child decision set",
+            )
+
+        explicit_refs = _contract_decision_refs(contract)
+        explicit_by_id = {item.decision_id: item for item in explicit_refs}
+        ordered: list[tuple[str, str]] = [
+            (decision_id, "direct-child") for decision_id in attempt.accepted_child_decision_ids
+        ]
+        ordered.extend((item.decision_id, "explicit") for item in explicit_refs)
+        deduplicated: list[tuple[str, str]] = []
+        sources: dict[str, set[str]] = {}
+        for decision_id, source in ordered:
+            if decision_id not in sources:
+                deduplicated.append((decision_id, source))
+                sources[decision_id] = set()
+            sources[decision_id].add(source)
+
+        resolved: list[ResolvedAcceptedDecision] = []
+        for decision_id, _ in deduplicated:
+            expected = explicit_by_id.get(decision_id)
+            item = await self.resolve_decision(
+                ledger=ledger,
+                root_trace_id=root_trace_id,
+                script_build_id=script_build_id,
+                decision_id=decision_id,
+                input_snapshot_id=input_snapshot_id,
+                expected=expected,
+                source="+".join(sorted(sources[decision_id])),
+            )
+            if decision_id in attempt.accepted_child_decision_ids:
+                child_task = ledger.tasks[item.artifact_version.task_id]
+                if child_task.parent_task_id != task_id:
+                    raise PhaseTwoInputError(
+                        "CHILD_DECISION_INVALID",
+                        "frozen direct-child decision is not owned by a direct child",
+                    )
+                _validate_consumer_input(contract, item.contract, direct_child=True)
+            else:
+                _validate_consumer_input(contract, item.contract, direct_child=False)
+            resolved.append(item)
+
+        input_values = tuple(item.accepted_input for item in resolved)
+        digest = canonical_sha256(
+            {
+                "schema_version": "accepted-input-bundle/v1",
+                "input_snapshot_ref": _input_snapshot_ref(input_snapshot_id),
+                "inputs": [_accepted_input_payload(item) for item in input_values],
+                "superseded_decision_ids": list(contract.supersedes_decision_ids),
+            }
+        ).wire
+        return AcceptedInputBundleV1(
+            inputs=input_values,
+            input_closure_digest=digest,
+            superseded_decision_ids=contract.supersedes_decision_ids,
+        )
+
+    async def resolve_decision(
+        self,
+        *,
+        ledger: Any,
+        root_trace_id: str,
+        script_build_id: int,
+        decision_id: str,
+        input_snapshot_id: str,
+        expected: AcceptedDecisionRef | None,
+        source: str,
+    ) -> ResolvedAcceptedDecision:
+        decision = ledger.decisions.get(decision_id)
+        if decision is None or decision.action is not DecisionAction.ACCEPT:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "input decision is missing or is not ACCEPT"
+            )
+        task = ledger.tasks.get(decision.task_id)
+        if task is None or task.status is not TaskStatus.COMPLETED:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "accepted input Task is not completed"
+            )
+        if not task.decision_ids or task.decision_ids[-1] != decision_id:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "input is not the Task's current ACCEPT decision"
+            )
+        if decision.attempt_id is None or decision.validation_id is None:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "ACCEPT is missing Attempt or Validation identity"
+            )
+        attempt = ledger.attempts.get(decision.attempt_id)
+        validation = ledger.validations.get(decision.validation_id)
+        if (
+            attempt is None
+            or validation is None
+            or attempt.task_id != task.task_id
+            or validation.task_id != task.task_id
+            or validation.attempt_id != attempt.attempt_id
+            or attempt.status is not AttemptStatus.SUBMITTED
+            or validation.status is not ValidationRunStatus.COMPLETED
+            or validation.verdict is not ValidationVerdict.PASSED
+            or attempt.snapshot_id is None
+            or validation.snapshot_id != attempt.snapshot_id
+            or attempt.submission is None
+            or attempt.spec_version != validation.spec_version
+        ):
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID",
+                "ACCEPT is not closed over one passed Attempt snapshot",
+            )
+        snapshot = await self._artifact_store.get(root_trace_id, attempt.snapshot_id)
+        if (
+            snapshot.attempt_id != attempt.attempt_id
+            or snapshot.sha256 != _snapshot_digest(snapshot.normalized_content)
+            or not _same_ref_sequence(snapshot.artifact_refs, attempt.submission.artifact_refs)
+            or not _same_ref_sequence(snapshot.evidence_refs, attempt.submission.evidence_refs)
+        ):
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "ArtifactSnapshot content or owner does not match Attempt"
+            )
+        refs = [*snapshot.artifact_refs, *snapshot.evidence_refs]
+        if len(refs) != 1:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID", "accepted input must freeze exactly one business ref"
+            )
+        ref = refs[0]
+        if expected is not None and not _same_ref(ref, expected.artifact_ref):
+            raise PhaseTwoInputError(
+                "ACCEPTED_INPUT_CONFLICT", "accepted ref differs from the pinned contract ref"
+            )
+        version = await self._artifacts.read_by_ref(
+            ref,
+            script_build_id=script_build_id,
+            task_id=task.task_id,
+            attempt_id=attempt.attempt_id,
+        )
+        identity_errors = [
+            label
+            for label, invalid in (
+                (
+                    "state",
+                    version.state not in {ArtifactState.FROZEN, ArtifactState.PUBLISHED},
+                ),
+                ("spec_version", version.spec_version != attempt.spec_version),
+                ("kind", version.artifact_type.value != ref.kind),
+                ("digest", version.canonical_sha256 != ref.digest),
+                ("version", str(version.artifact_version_id) != ref.version),
+            )
+            if invalid
+        ]
+        if identity_errors:
+            raise PhaseTwoInputError(
+                "CHILD_DECISION_INVALID",
+                "business Artifact identity is not closed: " + ",".join(identity_errors),
+            )
+        contract = await self.contract_for_task(
+            root_trace_id=root_trace_id,
+            task=task,
+            spec_version=attempt.spec_version,
+        )
+        actual_kind = task_kind_from_context_refs(task.current_spec.context_refs)
+        if actual_kind != contract.task_kind.value:
+            raise PhaseTwoInputError(
+                "TASK_KIND_PRESET_MISMATCH", "Task kind differs from its frozen contract"
+            )
+        attempted_spec = next(item for item in task.specs if item.version == attempt.spec_version)
+        attempted_inputs = {
+            item
+            for item in attempted_spec.context_refs
+            if item.startswith("script-build://inputs/")
+        }
+        current_input_ref = _input_snapshot_ref(input_snapshot_id)
+        if contract.task_kind is not ScriptTaskKind.DIRECTION and attempted_inputs != {
+            current_input_ref
+        }:
+            raise PhaseTwoInputError(
+                "INPUT_SCOPE_MISMATCH", "accepted Task is bound to another InputSnapshot"
+            )
+        if contract.task_kind is ScriptTaskKind.DIRECTION:
+            if len(attempted_inputs) != 1:
+                raise PhaseTwoInputError(
+                    "INPUT_SCOPE_MISMATCH", "Direction must pin one InputSnapshot"
+                )
+            attempted_input_ref = next(iter(attempted_inputs))
+            if attempted_input_ref != current_input_ref:
+                await self._verify_direction_snapshot_lineage(
+                    script_build_id=script_build_id,
+                    current_snapshot_id=input_snapshot_id,
+                    direction_snapshot_id=attempted_input_ref.removeprefix(
+                        "script-build://inputs/"
+                    ),
+                )
+        if expected is not None and (
+            contract.task_kind is not expected.expected_task_kind
+            or not _scope_compatible(expected.scope_ref, contract.scope_ref)
+        ):
+            raise PhaseTwoInputError(
+                "INPUT_SCOPE_MISMATCH", "accepted Task kind or scope differs from the contract"
+            )
+        lineage = getattr(version.artifact, "lineage", None)
+        if isinstance(lineage, CandidateLineageV1) and (
+            lineage.input_snapshot_ref != _input_snapshot_ref(input_snapshot_id)
+        ):
+            raise PhaseTwoInputError(
+                "INPUT_SCOPE_MISMATCH", "accepted Artifact uses a different input snapshot"
+            )
+        return ResolvedAcceptedDecision(
+            accepted_input=AcceptedInput(
+                decision_id=decision_id,
+                artifact_ref=ref,
+                task_kind=contract.task_kind,
+                scope_ref=contract.scope_ref,
+                source=source,
+            ),
+            artifact_version=version,
+            contract=contract,
+        )
+
+    async def _verify_direction_snapshot_lineage(
+        self,
+        *,
+        script_build_id: int,
+        current_snapshot_id: str,
+        direction_snapshot_id: str,
+    ) -> None:
+        if self._input_snapshots is None:
+            raise PhaseTwoInputError(
+                "INPUT_SCOPE_MISMATCH",
+                "Direction snapshot lineage cannot be verified",
+            )
+        current = await self._input_snapshots.get(
+            current_snapshot_id, script_build_id=script_build_id
+        )
+        parent = await self._input_snapshots.get(
+            direction_snapshot_id, script_build_id=script_build_id
+        )
+        parent_business_digest = (
+            parent.business_input_sha256
+            or canonical_sha256(parent.to_input().business_payload()).wire
+        )
+        if (
+            str(current.parent_snapshot_id or "") != direction_snapshot_id
+            or current.parent_snapshot_sha256 != parent.canonical_sha256
+            or not current.business_input_sha256
+            or current.business_input_sha256 != parent_business_digest
+        ):
+            raise PhaseTwoInputError(
+                "INPUT_SCOPE_MISMATCH",
+                "Direction and Phase2 snapshot do not form a valid immutable lineage",
+            )
+
+
+class ActiveFrontierResolver:
+    """Remove explicit replacements, then reject ambiguous overlapping writes."""
+
+    def resolve(
+        self,
+        bundle: AcceptedInputBundleV1,
+        *,
+        contracts_by_decision: dict[str, ScriptTaskContractV1],
+        adopted_decision_ids: Sequence[str] | None = None,
+    ) -> tuple[AcceptedInput, ...]:
+        inputs = {item.decision_id: item for item in bundle.inputs}
+        edges: dict[str, tuple[str, ...]] = {}
+        for decision_id, contract in contracts_by_decision.items():
+            if decision_id not in inputs:
+                raise PhaseTwoInputError(
+                    "ACCEPTED_INPUT_CONFLICT", "frontier contract is outside the input closure"
+                )
+            if contract.scope_ref != inputs[decision_id].scope_ref:
+                raise PhaseTwoInputError(
+                    "INPUT_SCOPE_MISMATCH", "frontier contract scope differs from accepted input"
+                )
+            targets = contract.supersedes_decision_ids
+            for target in targets:
+                replaced = inputs.get(target)
+                if replaced is None:
+                    raise PhaseTwoInputError(
+                        "ACCEPTED_INPUT_CONFLICT", "superseded decision is outside the closure"
+                    )
+                replaced_contract = contracts_by_decision.get(target)
+                replaced_scope = (
+                    replaced_contract.scope_ref
+                    if replaced_contract is not None
+                    else replaced.scope_ref
+                )
+                if not _scope_compatible(contract.scope_ref, replaced_scope):
+                    raise PhaseTwoInputError(
+                        "SUPERSESSION_SCOPE_MISMATCH", "replacement scopes are incompatible"
+                    )
+            edges[decision_id] = targets
+        _reject_cycles(edges)
+        superseded = {item for values in edges.values() for item in values}
+        active = [item for item in bundle.inputs if item.decision_id not in superseded]
+        if adopted_decision_ids is not None:
+            adopted = set(adopted_decision_ids)
+            if not adopted <= {item.decision_id for item in active}:
+                raise PhaseTwoInputError(
+                    "ACCEPTED_INPUT_CONFLICT", "adoption includes superseded or missing inputs"
+                )
+            active = [item for item in active if item.decision_id in adopted]
+        self._reject_write_conflicts(active, contracts_by_decision, edges)
+        return tuple(active)
+
+    @staticmethod
+    def verify_base(
+        *,
+        lineage: CandidateLineageV1,
+        base: ArtifactVersion | None,
+    ) -> None:
+        if lineage.base_artifact_ref is None:
+            if base is not None:
+                raise PhaseTwoInputError("STALE_BASE_REVISION", "unexpected base revision")
+            return
+        if (
+            base is None
+            or lineage.base_revision != base.artifact_version_id
+            or lineage.base_artifact_ref
+            != f"script-build://artifact-versions/{base.artifact_version_id}"
+            or lineage.base_artifact_digest != base.canonical_sha256
+            or base.state is not ArtifactState.FROZEN
+        ):
+            raise PhaseTwoInputError(
+                "STALE_BASE_REVISION", "candidate does not pin the current accepted base"
+            )
+
+    @staticmethod
+    def _reject_write_conflicts(
+        active: Sequence[AcceptedInput],
+        contracts: dict[str, ScriptTaskContractV1],
+        edges: dict[str, tuple[str, ...]],
+    ) -> None:
+        for index, left in enumerate(active):
+            left_contract = contracts[left.decision_id]
+            for right in active[index + 1 :]:
+                right_contract = contracts[right.decision_id]
+                if not _writes_overlap(left_contract.write_scope, right_contract.write_scope):
+                    continue
+                resolved_by_relation = (
+                    right.decision_id in edges.get(left.decision_id, ())
+                    or left.decision_id in edges.get(right.decision_id, ())
+                    or _is_exact_base_patch(left_contract, right)
+                    or _is_exact_base_patch(right_contract, left)
+                )
+                if not resolved_by_relation:
+                    raise PhaseTwoInputError(
+                        "WRITE_SCOPE_CONFLICT",
+                        "active accepted candidates have unresolved overlapping write scopes",
+                    )
+
+
+def _contract_decision_refs(contract: ScriptTaskContractV1) -> tuple[AcceptedDecisionRef, ...]:
+    output: list[AcceptedDecisionRef] = []
+    seen: set[str] = set()
+    for value in (
+        *contract.input_decision_refs,
+        *contract.candidate_closure_decision_refs,
+        *contract.comparison_decision_refs,
+    ):
+        if value.decision_id not in seen:
+            seen.add(value.decision_id)
+            output.append(value)
+    return tuple(output)
+
+
+def _same_ref(left: ArtifactRef, right: ArtifactRef) -> bool:
+    return bool(
+        left.uri == right.uri
+        and left.kind == right.kind
+        and left.version == right.version
+        and left.digest == right.digest
+    )
+
+
+def _is_exact_base_patch(contract: ScriptTaskContractV1, candidate: AcceptedInput) -> bool:
+    base = contract.base_artifact_ref
+    return base is not None and _same_ref(base, candidate.artifact_ref)
+
+
+def _same_ref_sequence(left: Sequence[ArtifactRef], right: Sequence[ArtifactRef]) -> bool:
+    return len(left) == len(right) and all(
+        _same_ref(a, b) for a, b in zip(left, right, strict=True)
+    )
+
+
+def _snapshot_digest(normalized_content: dict[str, Any]) -> str:
+    canonical = json.dumps(
+        normalized_content,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def _accepted_input_payload(value: AcceptedInput) -> dict[str, Any]:
+    return {
+        "decision_id": value.decision_id,
+        "artifact_ref": {
+            "uri": value.artifact_ref.uri,
+            "kind": value.artifact_ref.kind,
+            "version": value.artifact_ref.version,
+            "digest": value.artifact_ref.digest,
+        },
+        "task_kind": value.task_kind.value,
+        "scope_ref": value.scope_ref,
+        "source": value.source,
+    }
+
+
+def _input_snapshot_ref(value: str) -> str:
+    return f"script-build://inputs/{value}"
+
+
+def _scope_compatible(left: str, right: str) -> bool:
+    return left == right or left.startswith(f"{right}/") or right.startswith(f"{left}/")
+
+
+def _validate_consumer_input(
+    consumer: ScriptTaskContractV1,
+    produced: ScriptTaskContractV1,
+    *,
+    direct_child: bool,
+) -> None:
+    cross_scope_element_placement = (
+        not direct_child
+        and consumer.task_kind is ScriptTaskKind.ELEMENT_SET
+        and produced.task_kind is ScriptTaskKind.PARAGRAPH
+    )
+    if not _scope_compatible(consumer.scope_ref, produced.scope_ref) and not (
+        cross_scope_element_placement
+    ):
+        raise PhaseTwoInputError(
+            "INPUT_SCOPE_MISMATCH",
+            "accepted input is outside the consumer scope",
+        )
+    retrieval = {
+        ScriptTaskKind.PATTERN_RETRIEVAL,
+        ScriptTaskKind.DECODE_RETRIEVAL,
+        ScriptTaskKind.EXTERNAL_RETRIEVAL,
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+    }
+    creative = {
+        ScriptTaskKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET,
+        ScriptTaskKind.COMPARE,
+    }
+    direct_allowed = {
+        ScriptTaskKind.DIRECTION: retrieval,
+        ScriptTaskKind.CANDIDATE_PORTFOLIO: {ScriptTaskKind.COMPOSE},
+        ScriptTaskKind.COMPOSE: creative | retrieval,
+        ScriptTaskKind.STRUCTURE: creative | retrieval,
+        ScriptTaskKind.PARAGRAPH: creative | retrieval,
+        ScriptTaskKind.ELEMENT_SET: creative | retrieval,
+        ScriptTaskKind.COMPARE: creative | retrieval,
+    }.get(consumer.task_kind, set())
+    explicit_allowed = {
+        ScriptTaskKind.DIRECTION: retrieval,
+        ScriptTaskKind.CANDIDATE_PORTFOLIO: {
+            ScriptTaskKind.COMPOSE,
+            ScriptTaskKind.COMPARE,
+        },
+        ScriptTaskKind.COMPOSE: creative | retrieval | {ScriptTaskKind.DIRECTION},
+        ScriptTaskKind.STRUCTURE: creative | retrieval | {ScriptTaskKind.DIRECTION},
+        ScriptTaskKind.PARAGRAPH: creative | retrieval | {ScriptTaskKind.DIRECTION},
+        ScriptTaskKind.ELEMENT_SET: creative | retrieval | {ScriptTaskKind.DIRECTION},
+        ScriptTaskKind.COMPARE: creative | retrieval | {ScriptTaskKind.DIRECTION},
+    }.get(consumer.task_kind, set())
+    allowed = direct_allowed if direct_child else explicit_allowed
+    if produced.task_kind not in allowed:
+        raise PhaseTwoInputError(
+            "INPUT_SCOPE_MISMATCH",
+            "accepted Task kind is not valid for the consumer contract and input source",
+        )
+
+
+def _writes_overlap(left: Iterable[str], right: Iterable[str]) -> bool:
+    return any(_scope_compatible(a, b) for a in left for b in right)
+
+
+def _reject_cycles(edges: dict[str, tuple[str, ...]]) -> None:
+    visiting: set[str] = set()
+    visited: set[str] = set()
+
+    def visit(node: str) -> None:
+        if node in visiting:
+            raise PhaseTwoInputError("SUPERSESSION_CYCLE", "supersession graph contains a cycle")
+        if node in visited:
+            return
+        visiting.add(node)
+        for child in edges.get(node, ()):
+            visit(child)
+        visiting.remove(node)
+        visited.add(node)
+
+    for item in edges:
+        visit(item)
+
+
+def all_descendants_terminal(ledger: Any, task_id: str) -> bool:
+    pending = list(ledger.tasks[task_id].child_task_ids)
+    while pending:
+        child_id = pending.pop()
+        child = ledger.tasks.get(child_id)
+        if child is None or child.status not in _TERMINAL:
+            return False
+        pending.extend(child.child_task_ids)
+    return True
+
+
+__all__ = [
+    "AcceptedInputResolver",
+    "ActiveFrontierResolver",
+    "PhaseTwoInputError",
+    "ResolvedAcceptedDecision",
+    "StoredTaskContractReader",
+    "TaskContractReader",
+    "all_descendants_terminal",
+]

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -0,0 +1,1065 @@
+"""Business-governed planning on top of the framework's single-parent Task tree.
+
+The framework remains the only scheduler and state source.  This service only
+freezes ScriptTaskContract values, translates them into TaskSpec drafts, and
+checks script-build invariants before calling Coordinator operations.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Protocol, cast
+
+from agent.orchestration import (
+    DecisionAction,
+    OperationStatus,
+    TaskStatus,
+)
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
+from script_build_host.domain.ports import MissionBindingRepository
+from script_build_host.domain.task_contracts import (
+    FrozenTaskContract,
+    PhaseTwoLimits,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+    TaskContractError,
+    task_kind_from_context_refs,
+)
+
+TASK_KIND_REF_PREFIX = "script-build://task-kinds/"
+TASK_CONTRACT_REF_PREFIX = "script-build://task-contracts/sha256/"
+INPUT_SNAPSHOT_REF_PREFIX = "script-build://inputs/"
+PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
+PHASE_ONE_CAPABILITY_BOUNDARY = "PHASE_ONE_CAPABILITY_BOUNDARY"
+
+TASK_PRESET_BY_KIND: dict[ScriptTaskKind, str] = {
+    ScriptTaskKind.DIRECTION: "script_direction_worker",
+    ScriptTaskKind.PATTERN_RETRIEVAL: "script_pattern_retrieval_worker",
+    ScriptTaskKind.DECODE_RETRIEVAL: "script_decode_retrieval_worker",
+    ScriptTaskKind.EXTERNAL_RETRIEVAL: "script_external_retrieval_worker",
+    ScriptTaskKind.KNOWLEDGE_RETRIEVAL: "script_knowledge_retrieval_worker",
+    ScriptTaskKind.STRUCTURE: "script_structure_worker",
+    ScriptTaskKind.PARAGRAPH: "script_paragraph_worker",
+    ScriptTaskKind.ELEMENT_SET: "script_element_set_worker",
+    ScriptTaskKind.COMPARE: "script_candidate_compare_worker",
+    ScriptTaskKind.COMPOSE: "script_compose_worker",
+    ScriptTaskKind.CANDIDATE_PORTFOLIO: "script_candidate_portfolio_worker",
+}
+
+_PHASE_TWO_KINDS = frozenset(
+    {
+        ScriptTaskKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET,
+        ScriptTaskKind.COMPARE,
+        ScriptTaskKind.COMPOSE,
+        ScriptTaskKind.CANDIDATE_PORTFOLIO,
+        ScriptTaskKind.PATTERN_RETRIEVAL,
+        ScriptTaskKind.DECODE_RETRIEVAL,
+        ScriptTaskKind.EXTERNAL_RETRIEVAL,
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+    }
+)
+_CANDIDATE_KINDS = frozenset(
+    {
+        ScriptTaskKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET,
+        ScriptTaskKind.COMPOSE,
+    }
+)
+RETRIEVAL_KINDS = frozenset(
+    {
+        ScriptTaskKind.PATTERN_RETRIEVAL.value,
+        ScriptTaskKind.DECODE_RETRIEVAL.value,
+        ScriptTaskKind.EXTERNAL_RETRIEVAL.value,
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL.value,
+    }
+)
+_TERMINAL = frozenset(
+    {
+        TaskStatus.COMPLETED,
+        TaskStatus.SUPERSEDED,
+        TaskStatus.BLOCKED,
+        TaskStatus.CANCELLED,
+    }
+)
+_ACTIVE_OPERATION = frozenset(
+    {
+        OperationStatus.PENDING,
+        OperationStatus.RUNNING,
+        OperationStatus.STOP_REQUESTED,
+    }
+)
+
+
+class DispatchGuard(Protocol):
+    async def ensure_dispatch_allowed(self, script_build_id: int) -> None: ...
+
+
+class PhaseTwoClosureGate(Protocol):
+    async def verify_checkpoint(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None: ...
+
+
+class AttemptWorkspaceLifecycle(Protocol):
+    async def discard_attempt(
+        self,
+        *,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: str,
+        reason: str,
+    ) -> None: ...
+
+
+@dataclass(frozen=True, slots=True)
+class PlannedContract:
+    frozen: FrozenTaskContract
+    draft: dict[str, Any]
+
+
+class PhaseTwoPlanningService:
+    """Controlled Planner surface for Phase1 and Phase2 Tasks."""
+
+    def __init__(
+        self,
+        *,
+        coordinator: Any,
+        bindings: MissionBindingRepository,
+        contracts: ScriptTaskContractStore,
+        dispatch_guard: DispatchGuard | None = None,
+        closure_gate: PhaseTwoClosureGate | None = None,
+        workspace_lifecycle: AttemptWorkspaceLifecycle | None = None,
+        limits: PhaseTwoLimits | None = None,
+    ) -> None:
+        self.coordinator = coordinator
+        self.bindings = bindings
+        self.contracts = contracts
+        self.dispatch_guard = dispatch_guard
+        self.closure_gate = closure_gate
+        self.workspace_lifecycle = workspace_lifecycle
+        self.limits = limits or PhaseTwoLimits()
+
+    async def plan_script_tasks(
+        self,
+        *,
+        contracts: Sequence[Mapping[str, Any]] | None = None,
+        contract_payloads: Sequence[Mapping[str, Any]] | None = None,
+        parent_task_id: str | None,
+        after_task_id: str | None = None,
+        focus: bool = False,
+        context: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        root_trace_id = _required(context, "root_trace_id")
+        binding = await self.bindings.get_by_root(root_trace_id)
+        if self.dispatch_guard is not None:
+            await self.dispatch_guard.ensure_dispatch_allowed(binding.script_build_id)
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        await self._reconcile_terminal_workspaces(root_trace_id, ledger)
+        await self._guard_runtime_budget(ledger)
+        parent_id = parent_task_id or ledger.root_task_id
+        parent = ledger.tasks.get(parent_id)
+        if parent is None:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "parent Task is outside this Root")
+        supplied = contracts if contracts is not None else contract_payloads
+        if not supplied:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "at least one contract is required")
+
+        # Parse every value before writing any content-addressed blob.
+        parsed_contracts = tuple(ScriptTaskContractV1.from_payload(value) for value in supplied)
+        await self._guard_task_budget(root_trace_id, ledger, parsed_contracts, parent)
+        await self._guard_hierarchy(root_trace_id, ledger, parent, parsed_contracts)
+        await self._guard_supersessions(root_trace_id, ledger, parsed_contracts)
+        planned = await self._freeze_all(
+            root_trace_id, str(binding.input_snapshot_id), parsed_contracts
+        )
+        result = await self.coordinator.create_tasks(
+            root_trace_id,
+            [item.draft for item in planned],
+            parent_task_id=parent_id,
+            placement={"after_task_id": after_task_id, "focus": focus},
+            idempotency_key=_optional(context, "tool_call_id"),
+        )
+        created_task_ids = [str(item["task_id"]) for item in result.get("tasks", [])]
+        return {
+            **result,
+            "task_ids": created_task_ids,
+            "contracts": [
+                {
+                    "task_kind": item.frozen.contract.task_kind.value,
+                    "scope_ref": item.frozen.contract.scope_ref,
+                    "contract_ref": item.frozen.uri,
+                    "contract_digest": item.frozen.digest,
+                }
+                for item in planned
+            ],
+        }
+
+    async def decide_script_task(
+        self,
+        *,
+        task_id: str,
+        action: str,
+        reason: str,
+        validation_id: str | None,
+        replacement_contract: Mapping[str, Any] | None,
+        child_contracts: Sequence[Mapping[str, Any]],
+        context: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        root_trace_id = _required(context, "root_trace_id")
+        binding = await self.bindings.get_by_root(root_trace_id)
+        if self.dispatch_guard is not None:
+            await self.dispatch_guard.ensure_dispatch_allowed(binding.script_build_id)
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        await self._reconcile_terminal_workspaces(root_trace_id, ledger)
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "Task is outside this Root")
+        is_root = task_id == ledger.root_task_id
+        try:
+            decision_action = DecisionAction(action)
+        except ValueError as exc:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "unsupported decision action") from exc
+        if decision_action is not DecisionAction.BLOCK:
+            await self._guard_runtime_budget(ledger)
+        if decision_action is DecisionAction.SUPERSEDE:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                "accepted history is replaced by a new Task with supersedes_decision_ids",
+            )
+        if not is_root:
+            await self._guard_repair_and_improvement(
+                root_trace_id, ledger, task, decision_action, replacement_contract
+            )
+
+        payload: dict[str, Any] = {"reason": _bounded(reason, 1_000, "reason")}
+        if decision_action is DecisionAction.REVISE and is_root:
+            if replacement_contract is None or child_contracts:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID",
+                    "Root REVISE requires one complete bounded Root specification",
+                )
+            payload.update(_root_replacement(task, replacement_contract))
+        elif decision_action is DecisionAction.REVISE:
+            if replacement_contract is None or child_contracts:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "REVISE requires one complete replacement contract"
+                )
+            contract = ScriptTaskContractV1.from_payload(replacement_contract)
+            current = await self.contract_for_task(root_trace_id, task)
+            if contract.task_kind is not current.contract.task_kind:
+                raise TaskContractError(
+                    "TASK_KIND_PRESET_MISMATCH", "REVISE cannot change the Task kind"
+                )
+            await self._guard_supersessions(root_trace_id, ledger, (contract,))
+            planned_replacement = (
+                await self._freeze_all(root_trace_id, str(binding.input_snapshot_id), (contract,))
+            )[0]
+            payload.update(planned_replacement.draft)
+        elif decision_action is DecisionAction.SPLIT:
+            if replacement_contract is not None or not child_contracts:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "SPLIT requires complete child contracts"
+                )
+            contracts = tuple(ScriptTaskContractV1.from_payload(value) for value in child_contracts)
+            await self._guard_task_budget(root_trace_id, ledger, contracts, task)
+            await self._guard_hierarchy(root_trace_id, ledger, task, contracts)
+            await self._guard_supersessions(root_trace_id, ledger, contracts)
+            planned_children = await self._freeze_all(
+                root_trace_id, str(binding.input_snapshot_id), contracts
+            )
+            payload["tasks"] = [item.draft for item in planned_children]
+        elif replacement_contract is not None or child_contracts:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "this action does not accept Task contracts"
+            )
+
+        if decision_action is DecisionAction.BLOCK and is_root:
+            if reason == PHASE_ONE_CAPABILITY_BOUNDARY:
+                pass
+            elif reason != PHASE_TWO_CANDIDATE_PORTFOLIO_READY:
+                raise TaskContractError(
+                    "PHASE_TWO_BOUNDARY_NOT_READY",
+                    "Root may only block at an explicit phase capability boundary",
+                )
+            elif self.closure_gate is None:
+                raise TaskContractError(
+                    "PHASE_TWO_BOUNDARY_NOT_READY", "portfolio closure gate is not configured"
+                )
+            else:
+                await self.closure_gate.verify_checkpoint(
+                    script_build_id=binding.script_build_id,
+                    root_trace_id=root_trace_id,
+                    input_snapshot_id=str(binding.input_snapshot_id),
+                )
+        result = cast(
+            dict[str, Any],
+            await self.coordinator.decide_task(
+                root_trace_id,
+                task_id,
+                validation_id,
+                decision_action,
+                payload,
+                _optional(context, "tool_call_id"),
+            ),
+        )
+        if not is_root and decision_action in {
+            DecisionAction.REPAIR,
+            DecisionAction.RETRY,
+            DecisionAction.REVISE,
+            DecisionAction.SPLIT,
+            DecisionAction.BLOCK,
+            DecisionAction.CANCEL,
+        }:
+            attempt_id = _decision_attempt_id(ledger, task, validation_id)
+            if attempt_id is not None and self.workspace_lifecycle is not None:
+                await self.workspace_lifecycle.discard_attempt(
+                    root_trace_id=root_trace_id,
+                    task_id=task_id,
+                    attempt_id=attempt_id,
+                    reason=payload["reason"],
+                )
+        if (
+            is_root
+            and decision_action is DecisionAction.BLOCK
+            and reason in {PHASE_ONE_CAPABILITY_BOUNDARY, PHASE_TWO_CANDIDATE_PORTFOLIO_READY}
+            and result.get("status") == "blocked"
+        ):
+            result["terminal_boundary"] = True
+            result["phase_boundary"] = reason
+        return result
+
+    async def dispatch_script_tasks(
+        self, *, task_ids: Sequence[str], context: Mapping[str, Any]
+    ) -> list[dict[str, Any]]:
+        if not task_ids or len(set(task_ids)) != len(task_ids):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "dispatch requires unique Task identities"
+            )
+        root_trace_id = _required(context, "root_trace_id")
+        binding = await self.bindings.get_by_root(root_trace_id)
+        if self.dispatch_guard is not None:
+            await self.dispatch_guard.ensure_dispatch_allowed(binding.script_build_id)
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        await self._guard_runtime_budget(ledger)
+        active_count = sum(item.status in _ACTIVE_OPERATION for item in ledger.operations.values())
+        if active_count + len(task_ids) > self.limits.max_concurrent_operations:
+            raise TaskContractError(
+                "TASK_BUDGET_EXCEEDED",
+                "Phase2 concurrent Operation budget is exhausted",
+            )
+        presets: list[str] = []
+        for task_id in task_ids:
+            task = ledger.tasks.get(task_id)
+            if task is None:
+                raise TaskContractError("TASK_CONTRACT_INVALID", "Task is outside this Root")
+            frozen = await self.contract_for_task(root_trace_id, task)
+            contract = frozen.contract
+            if len(task.attempt_ids) >= min(
+                contract.budget.max_attempts, self.limits.max_attempts_per_task
+            ):
+                raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task attempt budget is exhausted")
+            self._guard_task_usage(ledger, task, contract)
+            await self._guard_supersessions(root_trace_id, ledger, (contract,))
+            contract.require_execution_ready()
+            self._guard_retrieval_parent(ledger, task, contract.task_kind)
+            preset = TASK_PRESET_BY_KIND[contract.task_kind]
+            presets.append(preset)
+        operations = await self.coordinator.dispatch_tasks(
+            root_trace_id,
+            list(task_ids),
+            worker_presets=presets,
+            idempotency_key=_optional(context, "tool_call_id"),
+        )
+        await self._reconcile_terminal_workspaces(
+            root_trace_id, await self.coordinator.task_store.load(root_trace_id)
+        )
+        return [item.to_dict() for item in operations]
+
+    async def ensure_tool_budget(self, *, root_trace_id: str, task_id: str, entry: str) -> None:
+        """Enforce Phase2 budgets at Retrieval and Attempt submission boundaries."""
+
+        if entry not in {"retrieval", "submit"}:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "unknown budget entry")
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "Task is outside this Root")
+        if not self._belongs_to_phase_two(ledger, task):
+            return
+        contract = (await self.contract_for_task(root_trace_id, task)).contract
+        await self._guard_runtime_budget(
+            ledger,
+            external_query_mode="reserved" if entry == "retrieval" else "skip",
+        )
+        self._guard_task_usage(
+            ledger,
+            task,
+            contract,
+            check_external_queries=entry == "retrieval",
+            allow_reserved_retrieval=entry == "retrieval",
+        )
+
+    async def _reconcile_terminal_workspaces(self, root_trace_id: str, ledger: Any) -> None:
+        if self.workspace_lifecycle is None:
+            return
+        for operation in ledger.operations.values():
+            if operation.status not in {OperationStatus.FAILED, OperationStatus.STOPPED}:
+                continue
+            if "MISSION_RECOVERY_REQUIRED" in str(operation.error or ""):
+                continue
+            reason = _bounded(
+                str(operation.error or f"operation {operation.status.value}"),
+                500,
+                "workspace discard reason",
+            )
+            for attempt_id in operation.attempt_ids:
+                attempt = ledger.attempts.get(attempt_id)
+                if attempt is None:
+                    continue
+                await self.workspace_lifecycle.discard_attempt(
+                    root_trace_id=root_trace_id,
+                    task_id=attempt.task_id,
+                    attempt_id=attempt.attempt_id,
+                    reason=reason,
+                )
+
+    async def reclaim_orphan_contracts(self, *, root_trace_id: str) -> tuple[str, ...]:
+        """Prune blobs left by a failed Task creation using Ledger refs as the sole authority."""
+
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        referenced = {
+            ref.removeprefix(TASK_CONTRACT_REF_PREFIX)
+            for task in ledger.tasks.values()
+            for spec in task.specs
+            for ref in spec.context_refs
+            if ref.startswith(TASK_CONTRACT_REF_PREFIX)
+        }
+        return await self.contracts.prune_unreferenced(root_trace_id, referenced)
+
+    async def inspect_script_plan(
+        self, *, task_id: str | None = None, context: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        root_trace_id = _required(context, "root_trace_id")
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        tasks: list[dict[str, Any]] = []
+        for task in sorted(ledger.tasks.values(), key=lambda item: item.display_path):
+            if task_id is not None and task.task_id != task_id:
+                continue
+            if task.task_id == ledger.root_task_id:
+                kind_value = "root"
+                contract_ref = None
+                scope_ref = None
+            else:
+                frozen = await self.contract_for_task(root_trace_id, task)
+                kind_value = frozen.contract.task_kind.value
+                contract_ref = frozen.uri
+                scope_ref = frozen.contract.scope_ref
+            tasks.append(
+                {
+                    "task_id": task.task_id,
+                    "parent_task_id": task.parent_task_id,
+                    "display_path": task.display_path,
+                    "status": task.status.value,
+                    "task_kind": kind_value,
+                    "scope_ref": scope_ref,
+                    "contract_ref": contract_ref,
+                    "current_spec_version": task.current_spec_version,
+                    "attempt_count": len(task.attempt_ids),
+                    "child_task_ids": list(task.child_task_ids),
+                    "blocked_reason": task.blocked_reason,
+                }
+            )
+        return {"root_trace_id": root_trace_id, "revision": ledger.revision, "tasks": tasks}
+
+    async def contract_for_task(self, root_trace_id: str, task: Any) -> FrozenTaskContract:
+        refs = [
+            value
+            for value in task.current_spec.context_refs
+            if value.startswith(TASK_CONTRACT_REF_PREFIX)
+        ]
+        if len(refs) != 1:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "TaskSpec must pin exactly one Task contract"
+            )
+        frozen = await self.contracts.read(root_trace_id, refs[0])
+        expected_kind = task_kind_from_context_refs(task.current_spec.context_refs)
+        if expected_kind != frozen.contract.task_kind.value:
+            raise TaskContractError(
+                "TASK_KIND_PRESET_MISMATCH", "Task kind and contract do not match"
+            )
+        if task.current_spec.objective != frozen.contract.objective:
+            raise TaskContractError(
+                "TASK_CONTRACT_DIGEST_MISMATCH", "Task objective differs from frozen contract"
+            )
+        expected_criteria = [
+            (item.criterion_id, item.description, item.hard) for item in frozen.contract.criteria
+        ]
+        actual_criteria = [
+            (item.criterion_id, item.description, item.hard)
+            for item in task.current_spec.acceptance_criteria
+        ]
+        if actual_criteria != expected_criteria:
+            raise TaskContractError(
+                "TASK_CONTRACT_DIGEST_MISMATCH", "Task criteria differ from frozen contract"
+            )
+        return frozen
+
+    async def _freeze_all(
+        self,
+        root_trace_id: str,
+        snapshot_id: str,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> tuple[PlannedContract, ...]:
+        output: list[PlannedContract] = []
+        for contract in contracts:
+            frozen = await self.contracts.freeze(root_trace_id, contract)
+            output.append(
+                PlannedContract(
+                    frozen=frozen,
+                    draft={
+                        "objective": contract.objective,
+                        "acceptance_criteria": [
+                            {
+                                "criterion_id": item.criterion_id,
+                                "description": item.description,
+                                "hard": item.hard,
+                            }
+                            for item in contract.criteria
+                        ],
+                        "context_refs": [
+                            f"{TASK_KIND_REF_PREFIX}{contract.task_kind.value}",
+                            frozen.uri,
+                            f"{INPUT_SNAPSHOT_REF_PREFIX}{snapshot_id}",
+                        ],
+                    },
+                )
+            )
+        return tuple(output)
+
+    async def _guard_task_budget(
+        self,
+        root_trace_id: str,
+        ledger: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+        parent: Any,
+    ) -> None:
+        existing = [
+            item for item in ledger.tasks.values() if self._belongs_to_phase_two(ledger, item)
+        ]
+        parent_is_phase_two = self._belongs_to_phase_two(ledger, parent)
+        new_phase_two = [
+            item
+            for item in contracts
+            if parent_is_phase_two or item.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO
+        ]
+        if len(existing) + len(new_phase_two) > self.limits.max_tasks:
+            raise TaskContractError("TASK_BUDGET_EXCEEDED", "Phase2 Task budget is exhausted")
+        depth = _task_depth(ledger, parent) + 1
+        if new_phase_two and depth > self.limits.max_depth:
+            raise TaskContractError("TASK_BUDGET_EXCEEDED", "Phase2 Task depth is exhausted")
+
+        existing_by_scope: dict[str, int] = {}
+        for task in existing:
+            existing_kind_value = task_kind_from_context_refs(task.current_spec.context_refs)
+            if existing_kind_value is None:
+                raise TaskContractError(
+                    "TASK_KIND_PRESET_MISMATCH", "Phase2 Task has no controlled kind"
+                )
+            existing_kind = ScriptTaskKind(existing_kind_value)
+            if existing_kind not in _CANDIDATE_KINDS:
+                continue
+            frozen = await self.contract_for_task(root_trace_id, task)
+            key = frozen.contract.scope_ref
+            existing_by_scope[key] = existing_by_scope.get(key, 0) + 1
+        for item in contracts:
+            if item.task_kind not in _CANDIDATE_KINDS:
+                continue
+            key = item.scope_ref
+            existing_by_scope[key] = existing_by_scope.get(key, 0) + 1
+        for _key, count in existing_by_scope.items():
+            if count > self.limits.max_candidates_per_scope:
+                raise TaskContractError(
+                    "TASK_BUDGET_EXCEEDED", "candidate budget for one scope is exhausted"
+                )
+
+    async def _guard_supersessions(
+        self,
+        root_trace_id: str,
+        ledger: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> None:
+        existing_edges: dict[str, tuple[str, ...]] = {}
+        contract_cache: dict[str, ScriptTaskContractV1] = {}
+        for decision_id, decision in ledger.decisions.items():
+            if decision.action is not DecisionAction.ACCEPT or decision.attempt_id is None:
+                continue
+            task = ledger.tasks.get(decision.task_id)
+            attempt = ledger.attempts.get(decision.attempt_id)
+            if (
+                task is None
+                or attempt is None
+                or not task.decision_ids
+                or task.decision_ids[-1] != decision_id
+            ):
+                continue
+            producer = await self.contract_for_task(root_trace_id, task)
+            contract_cache[decision_id] = producer.contract
+            existing_edges[decision_id] = producer.contract.supersedes_decision_ids
+        _reject_supersession_cycles(existing_edges)
+
+        for contract in contracts:
+            imported_ids = {
+                item.decision_id
+                for item in (
+                    *contract.input_decision_refs,
+                    *contract.candidate_closure_decision_refs,
+                    *contract.comparison_decision_refs,
+                )
+            }
+            if not set(contract.supersedes_decision_ids) <= imported_ids:
+                raise TaskContractError(
+                    "CHILD_DECISION_INVALID",
+                    "a replacement must import every superseded ACCEPT decision",
+                )
+            for target in contract.supersedes_decision_ids:
+                decision = ledger.decisions.get(target)
+                target_contract = contract_cache.get(target)
+                if (
+                    decision is None
+                    or decision.action is not DecisionAction.ACCEPT
+                    or target_contract is None
+                ):
+                    raise TaskContractError(
+                        "CHILD_DECISION_INVALID",
+                        "supersession target is not the current ACCEPT decision in this Root",
+                    )
+                if not _scope_contains(contract.scope_ref, target_contract.scope_ref) and not (
+                    _scope_contains(target_contract.scope_ref, contract.scope_ref)
+                ):
+                    raise TaskContractError(
+                        "SUPERSESSION_SCOPE_MISMATCH",
+                        "replacement and superseded scopes are incompatible",
+                    )
+
+    async def _guard_runtime_budget(
+        self, ledger: Any, *, external_query_mode: str = "strict"
+    ) -> None:
+        phase_two_ids = {
+            task.task_id
+            for task in ledger.tasks.values()
+            if self._belongs_to_phase_two(ledger, task)
+        }
+        if not phase_two_ids:
+            return
+        total_tokens = sum(
+            int(stats.total_tokens or 0)
+            for value in (*ledger.attempts.values(), *ledger.validations.values())
+            if value.task_id in phase_two_ids and (stats := value.execution_stats) is not None
+        )
+        if total_tokens >= self.limits.max_total_tokens:
+            raise TaskContractError("TASK_BUDGET_EXCEEDED", "Phase2 token budget is exhausted")
+        started = [
+            _parse_time(ledger.tasks[task_id].created_at)
+            for task_id in phase_two_ids
+            if ledger.tasks[task_id].created_at
+        ]
+        if (
+            started
+            and (datetime.now(UTC) - min(started)).total_seconds() >= self.limits.max_total_seconds
+        ):
+            raise TaskContractError("TASK_BUDGET_EXCEEDED", "Phase2 time budget is exhausted")
+        retrieval_attempts = sum(
+            1
+            for attempt in ledger.attempts.values()
+            if attempt.task_id in phase_two_ids
+            and task_kind_from_context_refs(ledger.tasks[attempt.task_id].current_spec.context_refs)
+            in RETRIEVAL_KINDS
+        )
+        validation_queries = sum(
+            int(value.evidence_queries_used)
+            for value in ledger.validations.values()
+            if value.task_id in phase_two_ids
+        )
+        used_queries = retrieval_attempts + validation_queries
+        exhausted = (
+            used_queries > self.limits.max_external_queries
+            if external_query_mode == "reserved"
+            else external_query_mode == "strict"
+            and used_queries >= self.limits.max_external_queries
+        )
+        if exhausted:
+            raise TaskContractError(
+                "TASK_BUDGET_EXCEEDED", "Phase2 external query budget is exhausted"
+            )
+
+    def _guard_task_usage(
+        self,
+        ledger: Any,
+        task: Any,
+        contract: ScriptTaskContractV1,
+        *,
+        check_external_queries: bool = True,
+        allow_reserved_retrieval: bool = False,
+    ) -> None:
+        attempts = [ledger.attempts[item] for item in task.attempt_ids if item in ledger.attempts]
+        validations = [
+            ledger.validations[item] for item in task.validation_ids if item in ledger.validations
+        ]
+        tokens = sum(
+            int(value.execution_stats.total_tokens or 0)
+            for value in (*attempts, *validations)
+            if value.execution_stats is not None
+        )
+        if tokens >= min(contract.budget.max_tokens, self.limits.max_task_tokens):
+            raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task token budget is exhausted")
+        started = [_parse_time(item.started_at or item.created_at) for item in attempts]
+        if started and (datetime.now(UTC) - min(started)).total_seconds() >= min(
+            contract.budget.max_seconds, self.limits.max_task_seconds
+        ):
+            raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task time budget is exhausted")
+        retrieval_attempts = len(attempts) if contract.task_kind.value in RETRIEVAL_KINDS else 0
+        evidence_queries = sum(int(value.evidence_queries_used) for value in validations)
+        used_queries = retrieval_attempts + evidence_queries
+        query_limit = min(contract.budget.max_external_queries, self.limits.max_external_queries)
+        if check_external_queries and (
+            used_queries > query_limit if allow_reserved_retrieval else used_queries >= query_limit
+        ):
+            raise TaskContractError(
+                "TASK_BUDGET_EXCEEDED", "Task external query budget is exhausted"
+            )
+
+    async def _guard_repair_and_improvement(
+        self,
+        root_trace_id: str,
+        ledger: Any,
+        task: Any,
+        action: DecisionAction,
+        replacement_contract: Mapping[str, Any] | None,
+    ) -> None:
+        if action is DecisionAction.REPAIR:
+            repair_count = sum(
+                ledger.attempts[item].execution_mode == "repair"
+                for item in task.attempt_ids
+                if item in ledger.attempts
+            )
+            if repair_count >= self.limits.max_repairs_per_trace:
+                raise TaskContractError(
+                    "TASK_BUDGET_EXCEEDED", "Worker Trace repair continuation is exhausted"
+                )
+        if action not in {DecisionAction.REPAIR, DecisionAction.RETRY, DecisionAction.REVISE}:
+            return
+        validations = [
+            ledger.validations[item]
+            for item in task.validation_ids
+            if item in ledger.validations and ledger.validations[item].verdict is not None
+        ]
+        current_contract = await self.contract_for_task(root_trace_id, task)
+        no_improvement_limit = min(
+            current_contract.contract.budget.max_no_improvement,
+            self.limits.max_no_improvement,
+        )
+        if _non_improvement_streak(validations) < no_improvement_limit:
+            return
+        if action is DecisionAction.REVISE and replacement_contract is not None:
+            new_contract = ScriptTaskContractV1.from_payload(replacement_contract)
+            old_contract = current_contract.contract
+            if (
+                new_contract.intent_class != old_contract.intent_class
+                or (
+                    _scope_contains(old_contract.scope_ref, new_contract.scope_ref)
+                    and new_contract.scope_ref != old_contract.scope_ref
+                )
+                or set(new_contract.write_scope) < set(old_contract.write_scope)
+            ):
+                return
+        raise TaskContractError(
+            "TASK_BUDGET_EXCEEDED",
+            "three validations made no improvement; change method, narrow scope, split or block",
+        )
+
+    @staticmethod
+    def _belongs_to_phase_two(ledger: Any, task: Any) -> bool:
+        current = task
+        visited: set[str] = set()
+        while current is not None and current.task_id not in visited:
+            visited.add(current.task_id)
+            kind_value = task_kind_from_context_refs(current.current_spec.context_refs)
+            if kind_value == ScriptTaskKind.CANDIDATE_PORTFOLIO.value:
+                return True
+            current = ledger.tasks.get(current.parent_task_id or "")
+        return False
+
+    async def _guard_hierarchy(
+        self,
+        root_trace_id: str,
+        ledger: Any,
+        parent: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> None:
+        parent_kind = (
+            None
+            if parent.task_id == ledger.root_task_id
+            else task_kind_from_context_refs(parent.current_spec.context_refs)
+        )
+        child_kinds = {item.task_kind.value for item in contracts}
+        if parent_kind is None:
+            allowed = {ScriptTaskKind.DIRECTION.value, ScriptTaskKind.CANDIDATE_PORTFOLIO.value}
+            if not child_kinds <= allowed:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "Root accepts only Direction or CandidatePortfolio"
+                )
+            if ScriptTaskKind.CANDIDATE_PORTFOLIO.value in child_kinds:
+                existing = [
+                    item
+                    for item in ledger.tasks.values()
+                    if task_kind_from_context_refs(item.current_spec.context_refs)
+                    == ScriptTaskKind.CANDIDATE_PORTFOLIO.value
+                ]
+                proposed = sum(
+                    item.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO for item in contracts
+                )
+                if existing or proposed != 1:
+                    raise TaskContractError(
+                        "TASK_CONTRACT_INVALID", "Phase2 requires one Portfolio container"
+                    )
+            return
+        parent_contract = (await self.contract_for_task(root_trace_id, parent)).contract
+        if any(
+            not _scope_contains(parent_contract.scope_ref, item.scope_ref) for item in contracts
+        ):
+            raise TaskContractError(
+                "INPUT_SCOPE_MISMATCH", "child Task scope must stay within its parent scope"
+            )
+        if parent_contract.write_scope and any(
+            any(
+                not any(
+                    _scope_contains(parent_write, child_write)
+                    for parent_write in parent_contract.write_scope
+                )
+                for child_write in item.write_scope
+            )
+            for item in contracts
+        ):
+            raise TaskContractError(
+                "WRITE_SCOPE_VIOLATION",
+                "child Task write scope must stay within its parent write scope",
+            )
+        if parent_kind == ScriptTaskKind.DIRECTION.value:
+            if not child_kinds <= RETRIEVAL_KINDS:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "Direction children must be Retrieval Tasks"
+                )
+            return
+        if parent_kind == ScriptTaskKind.CANDIDATE_PORTFOLIO.value:
+            if not child_kinds <= {ScriptTaskKind.COMPOSE.value}:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "Portfolio children must be Compose candidates"
+                )
+            return
+        if parent_kind == ScriptTaskKind.COMPOSE.value:
+            allowed = {
+                ScriptTaskKind.STRUCTURE.value,
+                ScriptTaskKind.PARAGRAPH.value,
+                ScriptTaskKind.ELEMENT_SET.value,
+                ScriptTaskKind.COMPARE.value,
+                *RETRIEVAL_KINDS,
+            }
+            if not child_kinds <= allowed:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID", "Compose contains only local creative increments"
+                )
+            return
+        if parent_kind in {
+            ScriptTaskKind.STRUCTURE.value,
+            ScriptTaskKind.PARAGRAPH.value,
+            ScriptTaskKind.ELEMENT_SET.value,
+            ScriptTaskKind.COMPARE.value,
+        }:
+            allowed = {
+                ScriptTaskKind.STRUCTURE.value,
+                ScriptTaskKind.PARAGRAPH.value,
+                ScriptTaskKind.ELEMENT_SET.value,
+                ScriptTaskKind.COMPARE.value,
+                *RETRIEVAL_KINDS,
+            }
+            if not child_kinds <= allowed:
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID",
+                    "a local creative Task may split only bounded local or Retrieval increments",
+                )
+            return
+        if child_kinds:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "this Task kind cannot own additional children"
+            )
+
+    @staticmethod
+    def _guard_retrieval_parent(ledger: Any, task: Any, kind: ScriptTaskKind) -> None:
+        if kind.value not in RETRIEVAL_KINDS:
+            return
+        parent = ledger.tasks.get(task.parent_task_id or "")
+        if parent is None:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "Retrieval Task has no consumer parent"
+            )
+        parent_kind = task_kind_from_context_refs(parent.current_spec.context_refs)
+        if parent_kind == ScriptTaskKind.DIRECTION.value:
+            return
+        if parent.status in _TERMINAL:
+            raise TaskContractError(
+                "CHILD_DECISION_INVALID",
+                "a completed consumer requires a new replacement Task for new Evidence",
+            )
+
+
+def _required(context: Mapping[str, Any], key: str) -> str:
+    value = context.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise ProtocolViolation(f"protected {key} is required")
+    return value
+
+
+def _optional(context: Mapping[str, Any], key: str) -> str | None:
+    value = context.get(key)
+    return value if isinstance(value, str) and value else None
+
+
+def _bounded(value: str, maximum: int, label: str) -> str:
+    normalized = " ".join(str(value).split())
+    if not normalized or len(normalized) > maximum:
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID", f"{label} must contain at most {maximum} characters"
+        )
+    return normalized
+
+
+def _scope_contains(parent: str, child: str) -> bool:
+    return child == parent or child.startswith(f"{parent.rstrip('/')}/")
+
+
+def _task_depth(ledger: Any, task: Any) -> int:
+    depth = 0
+    current = task
+    visited: set[str] = set()
+    while current.parent_task_id is not None:
+        if current.task_id in visited:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "Task hierarchy contains a cycle")
+        visited.add(current.task_id)
+        parent = ledger.tasks.get(current.parent_task_id)
+        if parent is None:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "Task hierarchy is not closed")
+        depth += 1
+        current = parent
+    return depth
+
+
+def _parse_time(value: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except (TypeError, ValueError) as exc:
+        raise TaskContractError("TASK_BUDGET_EXCEEDED", "durable Task time is invalid") from exc
+    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
+
+
+def _validation_vector(value: Any) -> tuple[int, int, int]:
+    failed = sum(item.verdict.value == "failed" for item in value.criterion_results)
+    inconclusive = sum(item.verdict.value == "inconclusive" for item in value.criterion_results)
+    return failed, inconclusive, len(value.risks) + len(value.unverified_claims)
+
+
+def _non_improvement_streak(values: Sequence[Any]) -> int:
+    if len(values) < 2:
+        return 0
+    streak = 0
+    previous = _validation_vector(values[0])
+    for value in values[1:]:
+        current = _validation_vector(value)
+        streak = streak + 1 if current >= previous else 0
+        previous = current
+    return streak
+
+
+def _reject_supersession_cycles(edges: Mapping[str, Sequence[str]]) -> None:
+    visiting: set[str] = set()
+    visited: set[str] = set()
+
+    def visit(decision_id: str) -> None:
+        if decision_id in visiting:
+            raise TaskContractError(
+                "SUPERSESSION_CYCLE", "accepted supersession graph contains a cycle"
+            )
+        if decision_id in visited:
+            return
+        visiting.add(decision_id)
+        for target in edges.get(decision_id, ()):
+            if target in edges:
+                visit(target)
+        visiting.remove(decision_id)
+        visited.add(decision_id)
+
+    for decision_id in edges:
+        visit(decision_id)
+
+
+def _decision_attempt_id(ledger: Any, task: Any, validation_id: str | None) -> str | None:
+    if validation_id is not None:
+        validation = ledger.validations.get(validation_id)
+        if validation is not None and validation.task_id == task.task_id:
+            return str(validation.attempt_id)
+    return str(task.attempt_ids[-1]) if task.attempt_ids else None
+
+
+def _root_replacement(task: Any, value: Mapping[str, Any]) -> dict[str, Any]:
+    """Allow a safe Root objective refinement without inventing a business contract."""
+
+    if set(value) != {"objective", "acceptance_criteria", "context_refs"}:
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID",
+            "Root replacement must contain objective, acceptance_criteria and context_refs",
+        )
+    objective = _bounded(str(value["objective"]), 2_000, "Root objective")
+    criteria = value["acceptance_criteria"]
+    refs = value["context_refs"]
+    if not isinstance(criteria, Sequence) or isinstance(criteria, (str, bytes)):
+        raise TaskContractError("TASK_CONTRACT_INVALID", "Root criteria must be an array")
+    if not isinstance(refs, Sequence) or isinstance(refs, (str, bytes)):
+        raise TaskContractError("TASK_CONTRACT_INVALID", "Root context_refs must be an array")
+    expected_criteria = [
+        {
+            "criterion_id": item.criterion_id,
+            "description": item.description,
+            "hard": item.hard,
+        }
+        for item in task.current_spec.acceptance_criteria
+    ]
+    expected_refs = list(task.current_spec.context_refs)
+    if list(criteria) != expected_criteria or list(refs) != expected_refs:
+        raise TaskContractError(
+            "TASK_CONTRACT_INVALID",
+            "Root REVISE may refine only objective and cannot expand its contract",
+        )
+    return {
+        "objective": objective,
+        "acceptance_criteria": expected_criteria,
+        "context_refs": expected_refs,
+    }
+
+
+__all__ = [
+    "INPUT_SNAPSHOT_REF_PREFIX",
+    "PHASE_ONE_CAPABILITY_BOUNDARY",
+    "PHASE_TWO_CANDIDATE_PORTFOLIO_READY",
+    "TASK_CONTRACT_REF_PREFIX",
+    "TASK_KIND_REF_PREFIX",
+    "TASK_PRESET_BY_KIND",
+    "AttemptWorkspaceLifecycle",
+    "PhaseTwoClosureGate",
+    "PhaseTwoPlanningService",
+]

script_build_host/src/script_build_host/infrastructure/config.py

@@ -7,6 +7,20 @@ from pydantic import Field, SecretStr, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from sqlalchemy.engine import make_url
 
+from script_build_host.domain.task_contracts import (
+    MAX_ATTEMPTS_PER_TASK,
+    MAX_CANDIDATES_PER_SCOPE,
+    MAX_EXTERNAL_QUERIES,
+    MAX_NO_IMPROVEMENT,
+    MAX_PHASE_TWO_SECONDS,
+    MAX_PHASE_TWO_TASKS,
+    MAX_PHASE_TWO_TOKENS,
+    MAX_TASK_DEPTH,
+    MAX_TASK_SECONDS,
+    MAX_TASK_TOKENS,
+    PhaseTwoLimits,
+)
+
 
 class ScriptBuildSettings(BaseSettings):
     model_config = SettingsConfigDict(
@@ -42,9 +56,35 @@ class ScriptBuildSettings(BaseSettings):
     max_images: int = Field(default=12, ge=0, le=100)
     max_image_bytes: int = Field(default=10_000_000, gt=0)
     max_total_image_bytes: int = Field(default=50_000_000, gt=0)
+    enable_image_understanding: bool = False
+    enabled_phase: int = Field(default=2, ge=1, le=2)
     max_upload_bytes: int = Field(default=5_000_000, gt=0)
     max_upload_points: int = Field(default=500, gt=0, le=10_000)
     max_upload_items: int = Field(default=5_000, gt=0, le=100_000)
+    phase_two_max_tasks: int = Field(default=MAX_PHASE_TWO_TASKS, ge=1, le=MAX_PHASE_TWO_TASKS)
+    phase_two_max_depth: int = Field(default=MAX_TASK_DEPTH, ge=1, le=MAX_TASK_DEPTH)
+    phase_two_max_candidates_per_scope: int = Field(
+        default=MAX_CANDIDATES_PER_SCOPE, ge=1, le=MAX_CANDIDATES_PER_SCOPE
+    )
+    phase_two_max_attempts_per_task: int = Field(
+        default=MAX_ATTEMPTS_PER_TASK, ge=1, le=MAX_ATTEMPTS_PER_TASK
+    )
+    phase_two_max_concurrent_operations: int = Field(default=4, ge=1, le=4)
+    phase_two_max_task_tokens: int = Field(default=MAX_TASK_TOKENS, ge=1, le=MAX_TASK_TOKENS)
+    phase_two_max_total_tokens: int = Field(
+        default=MAX_PHASE_TWO_TOKENS, ge=1, le=MAX_PHASE_TWO_TOKENS
+    )
+    phase_two_max_task_seconds: int = Field(default=MAX_TASK_SECONDS, ge=1, le=MAX_TASK_SECONDS)
+    phase_two_max_total_seconds: int = Field(
+        default=MAX_PHASE_TWO_SECONDS, ge=1, le=MAX_PHASE_TWO_SECONDS
+    )
+    phase_two_max_external_queries: int = Field(
+        default=MAX_EXTERNAL_QUERIES, ge=1, le=MAX_EXTERNAL_QUERIES
+    )
+    phase_two_max_no_improvement: int = Field(
+        default=MAX_NO_IMPROVEMENT, ge=1, le=MAX_NO_IMPROVEMENT
+    )
+    phase_two_max_repairs_per_trace: int = Field(default=1, ge=1, le=1)
 
     @model_validator(mode="after")
     def validate_database_separation(self) -> Self:
@@ -64,3 +104,19 @@ class ScriptBuildSettings(BaseSettings):
 
     def write_dsn(self) -> str:
         return self.write_database_url.get_secret_value()
+
+    def phase_two_limits(self) -> PhaseTwoLimits:
+        return PhaseTwoLimits(
+            max_tasks=self.phase_two_max_tasks,
+            max_depth=self.phase_two_max_depth,
+            max_candidates_per_scope=self.phase_two_max_candidates_per_scope,
+            max_attempts_per_task=self.phase_two_max_attempts_per_task,
+            max_concurrent_operations=self.phase_two_max_concurrent_operations,
+            max_task_tokens=self.phase_two_max_task_tokens,
+            max_total_tokens=self.phase_two_max_total_tokens,
+            max_task_seconds=self.phase_two_max_task_seconds,
+            max_total_seconds=self.phase_two_max_total_seconds,
+            max_external_queries=self.phase_two_max_external_queries,
+            max_no_improvement=self.phase_two_max_no_improvement,
+            max_repairs_per_trace=self.phase_two_max_repairs_per_trace,
+        )