feat(script-build): continue root delivery through phase three

script_build_host/src/script_build_host/agents/presets.py

@@ -15,6 +15,8 @@ from .prompts import (
     PORTFOLIO_WORKER_PROMPT,
     RETRIEVAL_VALIDATOR_PROMPT,
     RETRIEVAL_WORKER_PROMPT,
+    ROOT_VALIDATOR_PROMPT,
+    ROOT_WORKER_PROMPT,
     SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
     STRUCTURE_WORKER_PROMPT,
 )
@@ -178,6 +180,18 @@ def register_script_presets() -> None:
             prompt=PORTFOLIO_WORKER_PROMPT,
         ),
     )
+    register_preset(
+        "script_root_worker",
+        _worker(
+            [
+                "read_input_snapshot",
+                "read_accepted_portfolio",
+                "legacy_projection_dry_run",
+                "save_root_delivery_manifest",
+            ],
+            prompt=ROOT_WORKER_PROMPT,
+        ),
+    )
     validator_tools = [
         "read_input_snapshot",
         "view_frozen_images",
@@ -219,6 +233,24 @@ def register_script_presets() -> None:
             system_prompt=SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
         ),
     )
+    register_preset(
+        "script_root_validator",
+        AgentPreset(
+            role=AgentRole.VALIDATOR,
+            allowed_tools=[
+                "read_input_snapshot",
+                "read_accepted_portfolio",
+                "query_validation_evidence",
+                "deterministic_precheck",
+                "submit_validation",
+            ],
+            denied_tools=validator_denied,
+            max_iterations=25,
+            temperature=0.0,
+            skills=[],
+            system_prompt=ROOT_VALIDATOR_PROMPT,
+        ),
+    )
 
 
 __all__ = ["register_script_presets"]

script_build_host/src/script_build_host/agents/prompt_catalog.py

@@ -15,6 +15,7 @@ PHASE_TWO_PRESETS = frozenset(
         "script_candidate_portfolio_worker",
     }
 )
+PHASE_THREE_PRESETS = frozenset({"script_root_worker", "script_root_validator"})
 
 
 def script_prompt_requests() -> tuple[PromptRequest, ...]:
@@ -31,4 +32,21 @@ def script_prompt_requests() -> tuple[PromptRequest, ...]:
     )
 
 
-__all__ = ["PHASE_TWO_PRESETS", "script_prompt_requests"]
+def phase_three_prompt_requests() -> tuple[PromptRequest, ...]:
+    return tuple(
+        PromptRequest(
+            biz_type=str(item["biz_type"]),
+            filename=str(item["source_id"]),
+            preset=str(item["preset"]),
+            role=str(item["role"]),
+        )
+        for item in prompt_contracts.phase_three_prompt_manifest()
+    )
+
+
+__all__ = [
+    "PHASE_THREE_PRESETS",
+    "PHASE_TWO_PRESETS",
+    "phase_three_prompt_requests",
+    "script_prompt_requests",
+]

script_build_host/src/script_build_host/agents/prompts/__init__.py

@@ -10,9 +10,12 @@ from .contracts import (
     PORTFOLIO_WORKER_PROMPT,
     RETRIEVAL_VALIDATOR_PROMPT,
     RETRIEVAL_WORKER_PROMPT,
+    ROOT_VALIDATOR_PROMPT,
+    ROOT_WORKER_PROMPT,
     SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
     STRUCTURE_WORKER_PROMPT,
     phase_one_prompt_manifest,
+    phase_three_prompt_manifest,
     script_build_prompt_manifest,
 )
 
@@ -26,8 +29,11 @@ __all__ = [
     "PORTFOLIO_WORKER_PROMPT",
     "RETRIEVAL_VALIDATOR_PROMPT",
     "RETRIEVAL_WORKER_PROMPT",
+    "ROOT_VALIDATOR_PROMPT",
+    "ROOT_WORKER_PROMPT",
     "SCRIPT_CANDIDATE_VALIDATOR_PROMPT",
     "STRUCTURE_WORKER_PROMPT",
     "phase_one_prompt_manifest",
+    "phase_three_prompt_manifest",
     "script_build_prompt_manifest",
 ]

script_build_host/src/script_build_host/agents/prompts/contracts.py

@@ -25,6 +25,8 @@ ELEMENT_SET_WORKER_PROMPT = _read("element_set_worker.md")
 COMPARISON_WORKER_PROMPT = _read("comparison_worker.md")
 COMPOSE_WORKER_PROMPT = _read("compose_worker.md")
 PORTFOLIO_WORKER_PROMPT = _read("portfolio_worker.md")
+ROOT_WORKER_PROMPT = _read("root_worker.md")
+ROOT_VALIDATOR_PROMPT = _read("root_validator.md")
 
 
 def script_build_prompt_manifest() -> tuple[dict[str, object], ...]:
@@ -126,6 +128,31 @@ def script_build_prompt_manifest() -> tuple[dict[str, object], ...]:
     )
 
 
+def phase_three_prompt_manifest() -> tuple[dict[str, object], ...]:
+    values = {
+        "script_root_worker": ("worker", "root_worker.md", ROOT_WORKER_PROMPT, 1),
+        "script_root_validator": (
+            "validator",
+            "root_validator.md",
+            ROOT_VALIDATOR_PROMPT,
+            1,
+        ),
+    }
+    return tuple(
+        {
+            "biz_type": preset,
+            "preset": preset,
+            "role": role,
+            "source": "file",
+            "source_id": filename,
+            "version": version,
+            "content": content,
+            "content_sha256": f"sha256:{sha256(content.encode('utf-8')).hexdigest()}",
+        }
+        for preset, (role, filename, content, version) in values.items()
+    )
+
+
 def phase_one_prompt_manifest() -> tuple[dict[str, object], ...]:
     """Compatibility view used by older callers and Phase1 fixture tests."""
 

script_build_host/src/script_build_host/agents/prompts/root_validator.md

@@ -0,0 +1,8 @@
+# Script Root Delivery Validator
+
+Independently validate the frozen RootDeliveryManifest and the exact accepted closure it pins.
+
+Run deterministic prechecks first. Verify schema/digests, Direction/Portfolio/StructuredScript
+closure, topic and persona coverage, evidence, narrative structure, placeholder/meta-description
+absence, and legacy projection readability. A passed result requires an empty blocking-defect
+set. Report criterion-level evidence and submit Validation; never edit or publish content.

script_build_host/src/script_build_host/agents/prompts/root_worker.md

@@ -0,0 +1,11 @@
+# Script Root Delivery Worker
+
+You close one immutable accepted delivery; you do not rewrite its script.
+
+- Read the frozen InputSnapshot and the accepted Direction/Portfolio/StructuredScript closure.
+- Run the legacy projection dry-run with a concise build summary.
+- Save exactly one RootDeliveryManifest. Report any genuine blocking defect; never hide it.
+- Submit the Attempt after the manifest is frozen.
+
+Do not create or edit Paragraphs, Elements, Links, evidence, or candidate artifacts. Do not
+select a different candidate, scan “latest” state, or invent artifact references.

script_build_host/src/script_build_host/agents/validation.py

@@ -50,6 +50,7 @@ PHASE_TWO_KINDS = frozenset(
         "candidate-portfolio",
     }
 )
+ROOT_DELIVERY_KIND = "root-delivery"
 _ARTIFACT_KIND_BY_TASK_KIND = {
     "direction": ArtifactKind.DIRECTION,
     "structure": ArtifactKind.STRUCTURE,
@@ -58,6 +59,7 @@ _ARTIFACT_KIND_BY_TASK_KIND = {
     "compare": ArtifactKind.COMPARISON,
     "compose": ArtifactKind.STRUCTURED_SCRIPT,
     "candidate-portfolio": ArtifactKind.CANDIDATE_PORTFOLIO,
+    ROOT_DELIVERY_KIND: ArtifactKind.ROOT_DELIVERY_MANIFEST,
 }
 _DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
 
@@ -80,6 +82,8 @@ class ScriptBuildValidationPolicy:
         kind = task_kind(context.task.current_spec.context_refs)
         if kind in RETRIEVAL_KINDS:
             preset = "script_retrieval_validator"
+        elif kind == ROOT_DELIVERY_KIND:
+            preset = "script_root_validator"
         elif kind == "direction" or kind in PHASE_TWO_KINDS:
             preset = "script_candidate_validator"
         else:
@@ -141,9 +145,13 @@ def precheck_snapshot(context: ValidationContext) -> list[DeterministicRuleResul
     results.append(
         DeterministicRuleResult(
             rule_id=(
-                "phase-two-artifact-shape"
-                if kind in PHASE_TWO_KINDS
-                else "phase-one-artifact-shape"
+                "root-delivery-artifact-shape"
+                if kind == ROOT_DELIVERY_KIND
+                else (
+                    "phase-two-artifact-shape"
+                    if kind in PHASE_TWO_KINDS
+                    else "phase-one-artifact-shape"
+                )
             ),
             verdict=ValidationVerdict.PASSED if shape_ok else ValidationVerdict.FAILED,
             reason=(
@@ -163,6 +171,8 @@ def validation_layer(kind: str) -> str:
         return "global"
     if kind == "candidate-portfolio":
         return "governance"
+    if kind == ROOT_DELIVERY_KIND:
+        return "root-delivery"
     if kind in PHASE_TWO_KINDS:
         return "local"
     if kind in RETRIEVAL_KINDS or kind == "direction":

script_build_host/src/script_build_host/application/input_snapshot_service.py

@@ -127,6 +127,7 @@ class ScriptInputSnapshotService:
         *,
         prompt_requests: tuple[PromptRequest, ...],
         model_manifest: dict[str, Any] | None = None,
+        version: int = 2,
     ) -> ScriptBuildInputSnapshotV1:
         """Append a v2 snapshot containing phase-two runtime inputs.
 
@@ -163,7 +164,7 @@ class ScriptInputSnapshotService:
             parent_snapshot_sha256=snapshot.canonical_sha256,
             business_input_sha256=business_digest,
         )
-        return await self._snapshots.freeze(extended, version=2)
+        return await self._snapshots.freeze(extended, version=version)
 
 
 def _redacted_rows(

script_build_host/src/script_build_host/application/mission_factory.py

@@ -205,6 +205,80 @@ class ScriptMissionFactory:
             name=f"Script build {binding.script_build_id} phase two",
         )
 
+    def build_phase_three_policy(self, snapshot: ScriptBuildInputSnapshotV1) -> str:
+        frozen = _frozen_prompt(snapshot, "script_planner") or ""
+        policy = json.dumps(
+            {
+                "policy": "script-build-phase-three/v1",
+                "phase": 3,
+                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
+                "input_sha256": snapshot.canonical_sha256,
+                "instruction": (
+                    "Deliver only the accepted Direction and CandidatePortfolio closure. "
+                    "Dispatch the Host-frozen root-delivery contract, require independent "
+                    "validation, then explicitly ACCEPT Root. Do not rewrite the script."
+                ),
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+        return f"{frozen}\n\n{policy}".strip()
+
+    def build_phase_three_message(
+        self,
+        binding: MissionBinding,
+        snapshot: ScriptBuildInputSnapshotV1,
+        *,
+        direction_ref: str,
+        portfolio_ref: str,
+        structured_script_ref: str,
+    ) -> str:
+        return json.dumps(
+            {
+                "mission": "close_root_delivery",
+                "script_build_id": binding.script_build_id,
+                "phase": 3,
+                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
+                "input_sha256": snapshot.canonical_sha256,
+                "accepted_direction_ref": direction_ref,
+                "accepted_candidate_portfolio_ref": portfolio_ref,
+                "adopted_structured_script_ref": structured_script_ref,
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+
+    def build_phase_three_run_config(
+        self,
+        binding: MissionBinding,
+        snapshot: ScriptBuildInputSnapshotV1,
+    ) -> RunConfig:
+        planner_config = _planner_model_config(snapshot.model_manifest)
+        return RunConfig(
+            agent_type="script_planner",
+            model=planner_config["model"],
+            temperature=planner_config["temperature"],
+            max_iterations=planner_config["max_iterations"],
+            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            trace_id=binding.root_trace_id,
+            root_task_spec=None,
+            tool_groups=None,
+            parallel_tool_execution=False,
+            enable_memory=False,
+            enable_research_flow=False,
+            knowledge=KnowledgeConfig(
+                enable_extraction=False,
+                enable_completion_extraction=False,
+                enable_injection=False,
+            ),
+            context={
+                "script_build_id": binding.script_build_id,
+                "input_snapshot_id": snapshot.snapshot_id,
+                "phase": 3,
+            },
+            name=f"Script build {binding.script_build_id} phase three",
+        )
+
 
 def _planner_model_config(manifest: dict[str, object]) -> dict[str, object]:
     raw_presets = manifest.get("presets", manifest)

script_build_host/src/script_build_host/application/phase_three.py

@@ -0,0 +1,583 @@
+"""Idempotent Phase2-to-Phase3 Root continuation."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator, Callable
+from dataclasses import dataclass
+from typing import Any, cast
+
+from agent.orchestration import DecisionAction, OperationStatus, TaskStatus, ValidationVerdict
+
+from script_build_host.application.input_snapshot_service import ScriptInputSnapshotService
+from script_build_host.application.mission_factory import ScriptMissionFactory
+from script_build_host.domain.artifacts import ArtifactState, ScriptDirectionArtifactV1
+from script_build_host.domain.canonical_json import canonical_sha256
+from script_build_host.domain.errors import (
+    MissionRecoveryRequired,
+    PhaseThreeBoundaryNotReady,
+    ProtocolViolation,
+)
+from script_build_host.domain.phase_two_artifacts import CandidatePortfolioArtifactV1
+from script_build_host.domain.ports import (
+    BuildAuthorizer,
+    LegacyBuildStateRepository,
+    MissionBindingRepository,
+    PromptRequest,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.domain.records import BuildStatus, MissionOwnerToken, Principal
+from script_build_host.domain.task_contracts import (
+    AcceptedDecisionRef,
+    ScriptCriterion,
+    ScriptIntentClass,
+    ScriptTaskBudget,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+)
+from script_build_host.domain.task_refs import TASK_KIND_PREFIX, task_kind
+from script_build_host.infrastructure.ownership import OwnerLease
+
+PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
+PHASE_THREE_ROOT_ACCEPTED = "PHASE_THREE_ROOT_ACCEPTED"
+PHASE_THREE_POLICY = "script-build-phase-three/v1"
+
+
+@dataclass(frozen=True, slots=True)
+class PhaseThreeAdvanceResult:
+    script_build_id: int
+    status: BuildStatus
+    root_trace_id: str
+    input_snapshot_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class _AcceptedRootClosure:
+    direction: AcceptedDecisionRef
+    portfolio: AcceptedDecisionRef
+    structured_script_ref: str
+
+
+class PhaseThreeContinuationService:
+    def __init__(
+        self,
+        *,
+        runner: Any,
+        coordinator: Any,
+        factory: ScriptMissionFactory,
+        input_snapshots: ScriptInputSnapshotService,
+        bindings: MissionBindingRepository,
+        artifacts: ScriptBusinessArtifactRepository,
+        legacy_state: LegacyBuildStateRepository,
+        authorizer: BuildAuthorizer,
+        contracts: Any,
+        owner_lease_factory: Callable[[int], OwnerLease],
+        phase_two_boundary_verifier: Any,
+        prompt_requests: tuple[PromptRequest, ...],
+        model_manifest: dict[str, Any] | None = None,
+        fenced_command_gate: Any | None = None,
+        owner_registry: Any | None = None,
+    ) -> None:
+        self.runner = runner
+        self.coordinator = coordinator
+        self.factory = factory
+        self.input_snapshots = input_snapshots
+        self.bindings = bindings
+        self.artifacts = artifacts
+        self.legacy_state = legacy_state
+        self.authorizer = authorizer
+        self.contracts = contracts
+        self.owner_lease_factory = owner_lease_factory
+        self.phase_two_boundary_verifier = phase_two_boundary_verifier
+        self.prompt_requests = prompt_requests
+        self.model_manifest = dict(model_manifest or {})
+        self.fenced_command_gate = fenced_command_gate
+        self.owner_registry = owner_registry
+        self._locks: dict[int, asyncio.Lock] = {}
+        self._runs: dict[int, asyncio.Task[None]] = {}
+
+    async def advance(self, script_build_id: int, principal: Principal) -> PhaseThreeAdvanceResult:
+        await self.authorizer.require_access(principal, script_build_id)
+        lock = self._locks.setdefault(script_build_id, asyncio.Lock())
+        async with lock:
+            binding = await self.bindings.get_by_build(script_build_id)
+            active = self._runs.get(script_build_id)
+            if active is not None and not active.done():
+                return PhaseThreeAdvanceResult(
+                    script_build_id,
+                    BuildStatus.RUNNING,
+                    binding.root_trace_id,
+                    str(binding.input_snapshot_id),
+                )
+            completion = await self.coordinator.root_completion(binding.root_trace_id)
+            status = await self.legacy_state.get_status(script_build_id)
+            if completion.get("status") == TaskStatus.COMPLETED.value:
+                return PhaseThreeAdvanceResult(
+                    script_build_id,
+                    status,
+                    binding.root_trace_id,
+                    str(binding.input_snapshot_id),
+                )
+            if status is not BuildStatus.PARTIAL:
+                raise PhaseThreeBoundaryNotReady("phase-three advance requires a partial build")
+            if (
+                completion.get("status") != TaskStatus.BLOCKED.value
+                or completion.get("blocked_reason") != PHASE_TWO_CANDIDATE_PORTFOLIO_READY
+            ):
+                raise PhaseThreeBoundaryNotReady(
+                    "phase-three advance requires the phase-two portfolio checkpoint"
+                )
+            if self.phase_two_boundary_verifier is None:
+                raise PhaseThreeBoundaryNotReady("phase-two closure verifier is unavailable")
+            await self.phase_two_boundary_verifier.verify_checkpoint(
+                script_build_id=script_build_id,
+                root_trace_id=binding.root_trace_id,
+                input_snapshot_id=str(binding.input_snapshot_id),
+            )
+            ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+            if any(
+                item.status
+                in {
+                    OperationStatus.PENDING,
+                    OperationStatus.RUNNING,
+                    OperationStatus.STOP_REQUESTED,
+                }
+                for item in ledger.operations.values()
+            ):
+                raise PhaseThreeBoundaryNotReady("phase two still has active operations")
+            lease = self.owner_lease_factory(script_build_id)
+            token = await lease.acquire(script_build_id)
+            task = asyncio.create_task(
+                self._run_owned(script_build_id, token, lease),
+                name=f"script-build:{script_build_id}:phase-three",
+            )
+            self._runs[script_build_id] = task
+            task.add_done_callback(lambda done: self._discard(script_build_id, done))
+            return PhaseThreeAdvanceResult(
+                script_build_id,
+                BuildStatus.RUNNING,
+                binding.root_trace_id,
+                str(binding.input_snapshot_id),
+            )
+
+    async def _run_owned(
+        self, script_build_id: int, token: MissionOwnerToken, lease: OwnerLease
+    ) -> None:
+        if self.owner_registry is not None:
+            self.owner_registry.register_owner_token(token)
+        try:
+            await self.run(script_build_id, owner_token=token)
+        finally:
+            if self.owner_registry is not None:
+                self.owner_registry.unregister_owner_token(token)
+            await lease.release()
+
+    async def run(self, script_build_id: int, *, owner_token: MissionOwnerToken) -> None:
+        await self._verify_owner(owner_token)
+        binding = await self.bindings.get_by_build(script_build_id)
+        if binding.root_trace_id != owner_token.root_trace_id:
+            raise PhaseThreeBoundaryNotReady("owner token is bound to another Root")
+        status = await self.legacy_state.get_status(script_build_id)
+        if status is not BuildStatus.PARTIAL:
+            raise PhaseThreeBoundaryNotReady(f"cannot continue a {status.value} build")
+        snapshot = await self.input_snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=script_build_id
+        )
+        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+        root = ledger.tasks[ledger.root_task_id]
+        if root.status is TaskStatus.COMPLETED:
+            return
+        if root.status is TaskStatus.BLOCKED:
+            if root.blocked_reason != PHASE_TWO_CANDIDATE_PORTFOLIO_READY:
+                raise PhaseThreeBoundaryNotReady("Root is not at the phase-two checkpoint")
+            if self.phase_two_boundary_verifier is None:
+                raise PhaseThreeBoundaryNotReady("phase-two closure verifier is unavailable")
+            await self.phase_two_boundary_verifier.verify_checkpoint(
+                script_build_id=script_build_id,
+                root_trace_id=binding.root_trace_id,
+                input_snapshot_id=str(binding.input_snapshot_id),
+            )
+        active_operations = [
+            item
+            for item in ledger.operations.values()
+            if item.status
+            in {OperationStatus.PENDING, OperationStatus.RUNNING, OperationStatus.STOP_REQUESTED}
+        ]
+        if active_operations:
+            raise PhaseThreeBoundaryNotReady("phase two still has active operations")
+        closure = await self._accepted_root_closure(script_build_id, ledger)
+        await self._verify_owner(owner_token)
+        snapshot, binding = await self._ensure_snapshot(snapshot, binding)
+        root_contract = await self._execute_fenced(
+            owner_token,
+            lambda: self._freeze_root_contract(
+                binding.root_trace_id, root, closure, str(snapshot.snapshot_id)
+            ),
+        )
+        policy = self.factory.build_phase_three_policy(snapshot)
+        message = self.factory.build_phase_three_message(
+            binding,
+            snapshot,
+            direction_ref=closure.direction.artifact_ref.uri,
+            portfolio_ref=closure.portfolio.artifact_ref.uri,
+            structured_script_ref=closure.structured_script_ref,
+        )
+        iterator = await self._execute_fenced(
+            owner_token,
+            lambda: self._prepare_policy_messages(binding, snapshot, policy, message),
+        )
+        await self._verify_owner(owner_token)
+        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+        root = ledger.tasks[ledger.root_task_id]
+        if root.status is TaskStatus.BLOCKED:
+            await self._execute_fenced(
+                owner_token,
+                lambda: self.coordinator.decide_task(
+                    binding.root_trace_id,
+                    root.task_id,
+                    None,
+                    DecisionAction.UNBLOCK,
+                    {},
+                    f"phase-three-unblock:{script_build_id}",
+                ),
+            )
+            ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+            root = ledger.tasks[ledger.root_task_id]
+        if task_kind(root.current_spec.context_refs) != ScriptTaskKind.ROOT_DELIVERY.value:
+            if root.status is not TaskStatus.NEEDS_REPLAN:
+                raise PhaseThreeBoundaryNotReady("Root cannot accept its delivery contract")
+            await self._execute_fenced(
+                owner_token,
+                lambda: self.coordinator.decide_task(
+                    binding.root_trace_id,
+                    root.task_id,
+                    None,
+                    DecisionAction.REVISE,
+                    {
+                        "reason": "enter the Host-frozen phase-three delivery contract",
+                        "objective": root.current_spec.objective,
+                        "acceptance_criteria": [
+                            {
+                                "criterion_id": item.criterion_id,
+                                "description": item.description,
+                                "hard": item.hard,
+                            }
+                            for item in root.current_spec.acceptance_criteria
+                        ],
+                        "context_refs": [
+                            f"{TASK_KIND_PREFIX}{ScriptTaskKind.ROOT_DELIVERY.value}",
+                            root_contract.uri,
+                            f"script-build://inputs/{snapshot.snapshot_id}",
+                        ],
+                    },
+                    f"phase-three-contract:{script_build_id}",
+                ),
+            )
+        await self.legacy_state.begin_phase(script_build_id)
+        await self._verify_owner(owner_token)
+        if iterator is None:
+            await self.runner.run_result(
+                messages=[], config=self.factory.build_phase_three_run_config(binding, snapshot)
+            )
+        else:
+            async for _ in iterator:
+                pass
+        completion = await self.coordinator.root_completion(binding.root_trace_id)
+        await self._verify_owner(owner_token)
+        if completion.get("status") != TaskStatus.COMPLETED.value:
+            raise ProtocolViolation("phase-three Root was not explicitly ACCEPTed")
+        await self.legacy_state.set_checkpoint(
+            script_build_id,
+            checkpoint_code=PHASE_THREE_ROOT_ACCEPTED,
+            summary="Root accepted one validated immutable delivery manifest.",
+            root_trace_id=binding.root_trace_id,
+        )
+
+    async def _verify_owner(self, token: MissionOwnerToken) -> None:
+        if self.fenced_command_gate is not None:
+            await self.fenced_command_gate.verify(token)
+
+    async def _execute_fenced(self, token: MissionOwnerToken, mutation: Any) -> Any:
+        if self.owner_registry is not None:
+            return await self.owner_registry.execute_fenced(token.script_build_id, mutation)
+        await self._verify_owner(token)
+        return await mutation()
+
+    async def _ensure_snapshot(self, snapshot: Any, binding: Any) -> tuple[Any, Any]:
+        presets = {str(item.get("preset")) for item in snapshot.prompt_manifest}
+        required = {item.preset for item in self.prompt_requests}
+        if required <= presets:
+            return snapshot, binding
+        extended = await self.input_snapshots.extend_prompt_lineage(
+            snapshot,
+            prompt_requests=self.prompt_requests,
+            model_manifest=self.model_manifest or snapshot.model_manifest,
+            version=3,
+        )
+        binding = await self.bindings.compare_and_set_input_snapshot(
+            script_build_id=binding.script_build_id,
+            expected_snapshot_id=int(snapshot.snapshot_id),
+            new_snapshot_id=int(extended.snapshot_id),
+        )
+        return extended, binding
+
+    async def _accepted_root_closure(
+        self, script_build_id: int, ledger: Any
+    ) -> _AcceptedRootClosure:
+        root = ledger.tasks[ledger.root_task_id]
+        if len(root.child_task_ids) != 2:
+            raise PhaseThreeBoundaryNotReady(
+                "Root must have exactly the accepted Direction and Portfolio children"
+            )
+        accepted: dict[str, AcceptedDecisionRef] = {}
+        for child_id in root.child_task_ids:
+            child = ledger.tasks[child_id]
+            decisions = [
+                ledger.decisions[item]
+                for item in child.decision_ids
+                if ledger.decisions[item].action is DecisionAction.ACCEPT
+            ]
+            if len(decisions) != 1 or not decisions[0].attempt_id or not decisions[0].validation_id:
+                raise PhaseThreeBoundaryNotReady(
+                    "each Root child requires exactly one closed ACCEPT"
+                )
+            attempt = ledger.attempts[decisions[0].attempt_id]
+            validation = ledger.validations[decisions[0].validation_id]
+            if (
+                attempt.submission is None
+                or len(attempt.submission.artifact_refs) != 1
+                or validation.attempt_id != attempt.attempt_id
+                or validation.snapshot_id != attempt.snapshot_id
+                or validation.verdict is not ValidationVerdict.PASSED
+            ):
+                raise PhaseThreeBoundaryNotReady(
+                    "Root child ACCEPT is not closed over one passed snapshot"
+                )
+            kind = task_kind(child.current_spec.context_refs)
+            if (
+                kind
+                not in {
+                    ScriptTaskKind.DIRECTION.value,
+                    ScriptTaskKind.CANDIDATE_PORTFOLIO.value,
+                }
+                or kind in accepted
+            ):
+                raise PhaseThreeBoundaryNotReady(
+                    "Root children must be one Direction and one Portfolio"
+                )
+            artifact_ref = attempt.submission.artifact_refs[0]
+            version = await self.artifacts.read_by_ref(
+                artifact_ref,
+                script_build_id=script_build_id,
+                task_id=child.task_id,
+                attempt_id=attempt.attempt_id,
+            )
+            if version.state not in {ArtifactState.FROZEN, ArtifactState.PUBLISHED}:
+                raise PhaseThreeBoundaryNotReady("Root child Artifact is not frozen")
+            if kind == ScriptTaskKind.DIRECTION.value and not isinstance(
+                version.artifact, ScriptDirectionArtifactV1
+            ):
+                raise PhaseThreeBoundaryNotReady("accepted Direction has the wrong Artifact type")
+            accepted[kind] = AcceptedDecisionRef(
+                decision_id=decisions[0].decision_id,
+                artifact_ref=artifact_ref,
+                scope_ref="script-build://scope/root",
+                expected_task_kind=ScriptTaskKind(kind),
+            )
+        if set(accepted) != {
+            ScriptTaskKind.DIRECTION.value,
+            ScriptTaskKind.CANDIDATE_PORTFOLIO.value,
+        }:
+            raise PhaseThreeBoundaryNotReady("Root requires accepted Direction and Portfolio")
+        portfolio_ref = accepted[ScriptTaskKind.CANDIDATE_PORTFOLIO.value].artifact_ref
+        portfolio_version = await self.artifacts.read_by_ref(
+            portfolio_ref, script_build_id=script_build_id
+        )
+        if not isinstance(portfolio_version.artifact, CandidatePortfolioArtifactV1):
+            raise PhaseThreeBoundaryNotReady("accepted Portfolio has the wrong Artifact type")
+        return _AcceptedRootClosure(
+            accepted[ScriptTaskKind.DIRECTION.value],
+            accepted[ScriptTaskKind.CANDIDATE_PORTFOLIO.value],
+            portfolio_version.artifact.adopted_structured_script_ref,
+        )
+
+    async def _freeze_root_contract(
+        self,
+        root_trace_id: str,
+        root: Any,
+        closure: _AcceptedRootClosure,
+        snapshot_id: str,
+    ) -> Any:
+        contract = ScriptTaskContractV1(
+            task_kind=ScriptTaskKind.ROOT_DELIVERY,
+            scope_ref="script-build://scope/root",
+            intent_class=ScriptIntentClass.DELIVER,
+            objective=root.current_spec.objective,
+            input_decision_refs=(closure.direction, closure.portfolio),
+            base_artifact_ref=None,
+            write_scope=(),
+            gap_ref=None,
+            output_schema="root-delivery-manifest/v1",
+            criteria=tuple(
+                ScriptCriterion(item.criterion_id, item.description, item.hard)
+                for item in root.current_spec.acceptance_criteria
+            ),
+            budget=ScriptTaskBudget(),
+        )
+        frozen = await self.contracts.freeze(root_trace_id, contract)
+        # Snapshot identity is present in TaskSpec; this guards accidental use of
+        # a contract prepared for a different continuation.
+        if not snapshot_id:
+            raise PhaseThreeBoundaryNotReady("phase-three InputSnapshot is missing")
+        return frozen
+
+    async def _prepare_policy_messages(
+        self, binding: Any, snapshot: Any, policy: str, message: str
+    ) -> AsyncIterator[Any] | None:
+        events = await self.runner.trace_store.get_events(binding.root_trace_id, 0)
+        migrations = []
+        for event in events:
+            if (
+                event.get("event") != "planner_policy_migrated"
+                and event.get("event_type") != "planner_policy_migrated"
+            ):
+                continue
+            payload = event.get("data") or event.get("payload") or event
+            if payload.get("policy_version") == PHASE_THREE_POLICY:
+                migrations.append(payload)
+        policy_digest = canonical_sha256(policy).wire
+        toolset_digest = canonical_sha256(
+            sorted(self.runner.tools.get_tool_names(groups=["script_build"]))
+        ).wire
+        if len(migrations) > 1:
+            raise PhaseThreeBoundaryNotReady("multiple phase-three policy migrations exist")
+        if migrations:
+            payload = migrations[0]
+            expected = {
+                "policy_version": PHASE_THREE_POLICY,
+                "policy_digest": policy_digest,
+                "toolset_digest": toolset_digest,
+                "input_snapshot_id": snapshot.snapshot_id,
+            }
+            if any(str(payload.get(key)) != str(value) for key, value in expected.items()):
+                raise PhaseThreeBoundaryNotReady("phase-three policy migration digest differs")
+            system_sequence, user_sequence = await self._verify_policy_messages(
+                binding.root_trace_id, payload, policy, message
+            )
+            if await self._has_output_after(binding.root_trace_id, user_sequence):
+                raise MissionRecoveryRequired()
+            return None
+        prepared = await self._find_policy_messages(binding.root_trace_id, policy, message)
+        if prepared is not None:
+            system_sequence, user_sequence = prepared
+            await self.runner.trace_store.append_event(
+                binding.root_trace_id,
+                "planner_policy_migrated",
+                {
+                    "policy_version": PHASE_THREE_POLICY,
+                    "policy_digest": policy_digest,
+                    "toolset_digest": toolset_digest,
+                    "input_snapshot_id": snapshot.snapshot_id,
+                    "system_sequence": system_sequence,
+                    "user_sequence": user_sequence,
+                },
+            )
+            if await self._has_output_after(binding.root_trace_id, user_sequence):
+                raise MissionRecoveryRequired()
+            return None
+        iterator = cast(
+            AsyncIterator[Any],
+            self.runner.run(
+                messages=[
+                    {"role": "system", "content": policy},
+                    {"role": "user", "content": message},
+                ],
+                config=self.factory.build_phase_three_run_config(binding, snapshot),
+            ),
+        )
+        try:
+            await anext(iterator)
+            system_message = await anext(iterator)
+            user_message = await anext(iterator)
+        except Exception as exc:
+            raise PhaseThreeBoundaryNotReady(
+                "phase-three policy messages were not durable"
+            ) from exc
+        if (
+            getattr(system_message, "role", None) != "system"
+            or getattr(user_message, "role", None) != "user"
+        ):
+            raise PhaseThreeBoundaryNotReady("phase-three policy message order differs")
+        await self.runner.trace_store.append_event(
+            binding.root_trace_id,
+            "planner_policy_migrated",
+            {
+                "policy_version": PHASE_THREE_POLICY,
+                "policy_digest": policy_digest,
+                "toolset_digest": toolset_digest,
+                "input_snapshot_id": snapshot.snapshot_id,
+                "system_sequence": system_message.sequence,
+                "user_sequence": user_message.sequence,
+            },
+        )
+        return iterator
+
+    async def _find_policy_messages(
+        self, root_trace_id: str, policy: str, message: str
+    ) -> tuple[int, int] | None:
+        messages = await self.runner.trace_store.get_trace_messages(root_trace_id)
+        systems = [item for item in messages if item.role == "system" and item.content == policy]
+        users = [item for item in messages if item.role == "user" and item.content == message]
+        if not systems and not users:
+            return None
+        if len(systems) != 1 or len(users) != 1:
+            raise PhaseThreeBoundaryNotReady("phase-three policy messages are duplicated")
+        if getattr(users[0], "parent_sequence", None) != systems[0].sequence:
+            raise PhaseThreeBoundaryNotReady("phase-three policy message parent chain differs")
+        return systems[0].sequence, users[0].sequence
+
+    async def _verify_policy_messages(
+        self,
+        root_trace_id: str,
+        payload: Any,
+        policy: str,
+        message: str,
+    ) -> tuple[int, int]:
+        system_sequence = payload.get("system_sequence")
+        user_sequence = payload.get("user_sequence")
+        if (
+            isinstance(system_sequence, bool)
+            or not isinstance(system_sequence, int)
+            or isinstance(user_sequence, bool)
+            or not isinstance(user_sequence, int)
+        ):
+            raise PhaseThreeBoundaryNotReady("phase-three policy sequence is invalid")
+        observed = await self._find_policy_messages(root_trace_id, policy, message)
+        if observed != (system_sequence, user_sequence):
+            raise PhaseThreeBoundaryNotReady(
+                "phase-three migration is not backed by its durable messages"
+            )
+        return observed
+
+    async def _has_output_after(self, root_trace_id: str, user_sequence: int) -> bool:
+        messages = await self.runner.trace_store.get_trace_messages(root_trace_id)
+        return any(
+            item.sequence > user_sequence and item.role not in {"system", "user"}
+            for item in messages
+        )
+
+    def _discard(self, script_build_id: int, task: asyncio.Task[None]) -> None:
+        if self._runs.get(script_build_id) is task:
+            self._runs.pop(script_build_id, None)
+        try:
+            task.exception()
+        except (asyncio.CancelledError, Exception):
+            pass
+
+
+__all__ = [
+    "PHASE_THREE_POLICY",
+    "PHASE_THREE_ROOT_ACCEPTED",
+    "PhaseThreeAdvanceResult",
+    "PhaseThreeContinuationService",
+]

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -48,6 +48,7 @@ TASK_PRESET_BY_KIND: dict[ScriptTaskKind, str] = {
     ScriptTaskKind.COMPARE: "script_candidate_compare_worker",
     ScriptTaskKind.COMPOSE: "script_compose_worker",
     ScriptTaskKind.CANDIDATE_PORTFOLIO: "script_candidate_portfolio_worker",
+    ScriptTaskKind.ROOT_DELIVERY: "script_root_worker",
 }
 
 _PHASE_TWO_KINDS = frozenset(
@@ -167,6 +168,13 @@ class PhaseTwoPlanningService:
         parent = ledger.tasks.get(parent_id)
         if parent is None:
             raise TaskContractError("TASK_CONTRACT_INVALID", "parent Task is outside this Root")
+        if parent_id == ledger.root_task_id and (
+            task_kind_from_context_refs(parent.current_spec.context_refs)
+            == ScriptTaskKind.ROOT_DELIVERY.value
+        ):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "Root delivery cannot plan additional child Tasks"
+            )
         supplied = contracts if contracts is not None else contract_payloads
         if not supplied:
             raise TaskContractError("TASK_CONTRACT_INVALID", "at least one contract is required")
@@ -226,6 +234,16 @@ class PhaseTwoPlanningService:
             decision_action = DecisionAction(action)
         except ValueError as exc:
             raise TaskContractError("TASK_CONTRACT_INVALID", "unsupported decision action") from exc
+        if (
+            is_root
+            and task_kind_from_context_refs(task.current_spec.context_refs)
+            == ScriptTaskKind.ROOT_DELIVERY.value
+            and decision_action in {DecisionAction.REVISE, DecisionAction.SPLIT}
+        ):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                "the Host-frozen Root delivery contract cannot be replaced or split",
+            )
         if decision_action is not DecisionAction.BLOCK:
             await self._guard_runtime_budget(ledger)
         if decision_action is DecisionAction.SUPERSEDE:

script_build_host/src/script_build_host/application/root_delivery.py

@@ -0,0 +1,345 @@
+"""Protected Root delivery tools over one frozen accepted closure."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict
+from typing import Any
+
+from agent.orchestration import DecisionAction, ValidationVerdict
+
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ArtifactVersion,
+    ScriptDirectionArtifactV1,
+)
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.phase_three_artifacts import (
+    RootDeliveryManifestV1,
+    canonicalize_legacy_projection,
+    root_delivery_input_closure_digest,
+)
+from script_build_host.domain.phase_two_artifacts import (
+    CandidatePortfolioArtifactV1,
+    StructuredScriptArtifactV1,
+)
+from script_build_host.domain.ports import (
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.domain.task_refs import task_kind
+
+
+class RootDeliveryService:
+    def __init__(
+        self,
+        *,
+        bindings: MissionBindingRepository,
+        task_store: Any,
+        artifacts: ScriptBusinessArtifactRepository,
+    ) -> None:
+        self._bindings = bindings
+        self._task_store = task_store
+        self._artifacts = artifacts
+
+    async def read_accepted_portfolio(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        closure = await self._closure(context)
+        return {
+            "direction_ref": _uri(closure.direction),
+            "candidate_portfolio_ref": _uri(closure.portfolio),
+            "structured_script_ref": _uri(closure.structured),
+            "direction": closure.direction.artifact.content_payload(),
+            "candidate_portfolio": closure.portfolio.artifact.content_payload(),
+            "structured_script": closure.structured.artifact.content_payload(),
+        }
+
+    async def legacy_projection_dry_run(
+        self, *, build_summary: str, context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        closure = await self._closure(context)
+        direction = _direction(closure.direction)
+        structured = _structured(closure.structured)
+        canonical = canonicalize_legacy_projection(
+            structured,
+            direction=direction.legacy_markdown,
+            summary=build_summary,
+        )
+        return {
+            "legacy_projection_digest": canonical.canonical_sha256,
+            "paragraph_count": canonical.paragraph_count,
+            "element_count": canonical.element_count,
+            "link_count": canonical.link_count,
+        }
+
+    async def save_root_delivery_manifest(
+        self,
+        *,
+        build_summary: str,
+        blocking_defects: Sequence[str],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        closure = await self._closure(context)
+        binding = await self._bindings.get_by_root(_required(context, "root_trace_id"))
+        ledger = await self._task_store.load(binding.root_trace_id)
+        task = ledger.tasks.get(_required(context, "task_id"))
+        attempt = ledger.attempts.get(_required(context, "attempt_id"))
+        if task is None or attempt is None or attempt.task_id != task.task_id:
+            raise ProtocolViolation("Root delivery Attempt is outside protected context")
+        if (
+            task.task_id != ledger.root_task_id
+            or task_kind(task.current_spec.context_refs) != "root-delivery"
+        ):
+            raise ProtocolViolation("Root delivery manifest can be saved by Root only")
+        if attempt.spec_version != task.current_spec.version:
+            raise ProtocolViolation("Root delivery Attempt does not use the current TaskSpec")
+        direction = _direction(closure.direction)
+        portfolio = _portfolio(closure.portfolio)
+        structured = _structured(closure.structured)
+        canonical = canonicalize_legacy_projection(
+            structured,
+            direction=direction.legacy_markdown,
+            summary=build_summary,
+        )
+        manifest = RootDeliveryManifestV1(
+            direction_ref=_uri(closure.direction),
+            candidate_portfolio_ref=_uri(closure.portfolio),
+            structured_script_ref=_uri(closure.structured),
+            input_closure_digest=_closure_digest(closure),
+            legacy_projection_digest=canonical.canonical_sha256,
+            build_summary=build_summary,
+            blocking_defects=tuple(blocking_defects),
+        )
+        if portfolio.adopted_structured_script_ref != manifest.structured_script_ref:
+            raise ProtocolViolation("accepted Portfolio does not adopt the frozen StructuredScript")
+        version, ref = await self._artifacts.freeze(
+            script_build_id=binding.script_build_id,
+            task_id=task.task_id,
+            attempt_id=attempt.attempt_id,
+            spec_version=attempt.spec_version,
+            artifact=manifest,
+        )
+        return {
+            "artifact_ref": asdict(ref),
+            "artifact_version_id": version.artifact_version_id,
+            "legacy_projection_digest": canonical.canonical_sha256,
+        }
+
+    async def deterministic_precheck(
+        self, *, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]:
+        binding = await self._bindings.get_by_root(_required(context, "root_trace_id"))
+        ledger = await self._task_store.load(binding.root_trace_id)
+        attempt = ledger.attempts.get(_required(context, "attempt_id"))
+        if attempt is None or attempt.submission is None:
+            raise ProtocolViolation("Root validation Attempt has no submission")
+        refs = [
+            item
+            for item in attempt.submission.artifact_refs
+            if item.kind == ArtifactKind.ROOT_DELIVERY_MANIFEST.value
+        ]
+        if len(refs) != 1:
+            return (
+                {
+                    "rule_id": "root-manifest-shape",
+                    "verdict": "failed",
+                    "reason": "Root validation requires exactly one delivery manifest",
+                    "error": None,
+                    "evidence_refs": [],
+                },
+            )
+        version = await self._artifacts.read_by_ref(
+            refs[0],
+            script_build_id=binding.script_build_id,
+            task_id=attempt.task_id,
+            attempt_id=attempt.attempt_id,
+        )
+        manifest = version.artifact
+        closure = await self._closure(context)
+        root_manifest = manifest if isinstance(manifest, RootDeliveryManifestV1) else None
+        manifest_shape_passed = root_manifest is not None
+        defects_passed = root_manifest is not None and not root_manifest.blocking_defects
+        closure_passed = root_manifest is not None and (
+            root_manifest.direction_ref == _uri(closure.direction)
+            and root_manifest.candidate_portfolio_ref == _uri(closure.portfolio)
+            and root_manifest.structured_script_ref == _uri(closure.structured)
+            and root_manifest.input_closure_digest == _closure_digest(closure)
+        )
+        projection_passed = False
+        if root_manifest is not None:
+            canonical = canonicalize_legacy_projection(
+                _structured(closure.structured),
+                direction=_direction(closure.direction).legacy_markdown,
+                summary=root_manifest.build_summary,
+            )
+            projection_passed = canonical.canonical_sha256 == root_manifest.legacy_projection_digest
+        return (
+            {
+                "rule_id": "root-manifest-shape",
+                "verdict": "passed" if manifest_shape_passed else "failed",
+                "reason": (
+                    "Root Attempt owns one frozen delivery manifest"
+                    if manifest_shape_passed
+                    else "Root Attempt submission is not a delivery manifest"
+                ),
+                "error": None,
+                "evidence_refs": [],
+            },
+            {
+                "rule_id": "root-blocking-defects-empty",
+                "verdict": "passed" if defects_passed else "failed",
+                "reason": (
+                    "Root delivery has no blocking defects"
+                    if defects_passed
+                    else "Root delivery retains blocking defects"
+                ),
+                "error": None,
+                "evidence_refs": [],
+            },
+            {
+                "rule_id": "root-input-closure",
+                "verdict": "passed" if closure_passed else "failed",
+                "reason": (
+                    "Root delivery pins the exact accepted immutable closure"
+                    if closure_passed
+                    else "Root delivery immutable closure differs"
+                ),
+                "error": None,
+                "evidence_refs": [],
+            },
+            {
+                "rule_id": "root-legacy-projection",
+                "verdict": "passed" if projection_passed else "failed",
+                "reason": (
+                    "Root delivery legacy projection digest is reproducible"
+                    if projection_passed
+                    else "Root delivery legacy projection digest differs"
+                ),
+                "error": None,
+                "evidence_refs": [],
+            },
+        )
+
+    async def _closure(self, context: Mapping[str, Any]) -> _RootClosure:
+        root_trace_id = _required(context, "root_trace_id")
+        binding = await self._bindings.get_by_root(root_trace_id)
+        ledger = await self._task_store.load(root_trace_id)
+        attempt = ledger.attempts.get(_required(context, "attempt_id"))
+        task = ledger.tasks.get(_required(context, "task_id"))
+        if (
+            attempt is None
+            or task is None
+            or attempt.task_id != task.task_id
+            or task.task_id != ledger.root_task_id
+            or attempt.accepted_child_decision_ids is None
+        ):
+            raise ProtocolViolation("Root Attempt has no frozen accepted child closure")
+        versions: dict[str, ArtifactVersion] = {}
+        for decision_id in attempt.accepted_child_decision_ids:
+            decision = ledger.decisions.get(decision_id)
+            if (
+                decision is None
+                or decision.action is not DecisionAction.ACCEPT
+                or not decision.attempt_id
+                or not decision.validation_id
+            ):
+                raise ProtocolViolation("Root accepted child decision is invalid")
+            child_attempt = ledger.attempts.get(decision.attempt_id)
+            child_validation = ledger.validations.get(decision.validation_id)
+            child_task = ledger.tasks.get(decision.task_id)
+            if (
+                child_attempt is None
+                or child_validation is None
+                or child_task is None
+                or child_task.parent_task_id != task.task_id
+                or child_attempt.submission is None
+                or child_validation.attempt_id != child_attempt.attempt_id
+                or child_validation.snapshot_id != child_attempt.snapshot_id
+                or child_validation.verdict is not ValidationVerdict.PASSED
+            ):
+                raise ProtocolViolation("Root child ACCEPT is not closed over a passed snapshot")
+            refs = child_attempt.submission.artifact_refs
+            if len(refs) != 1:
+                raise ProtocolViolation("Root child must submit exactly one business Artifact")
+            version = await self._artifacts.read_by_ref(
+                refs[0],
+                script_build_id=binding.script_build_id,
+                task_id=child_task.task_id,
+                attempt_id=child_attempt.attempt_id,
+            )
+            if version.state not in {ArtifactState.FROZEN, ArtifactState.PUBLISHED}:
+                raise ProtocolViolation("Root child Artifact is not frozen")
+            versions[task_kind(child_task.current_spec.context_refs) or ""] = version
+        if set(versions) != {"direction", "candidate-portfolio"}:
+            raise ProtocolViolation(
+                "Root requires exactly accepted Direction and CandidatePortfolio"
+            )
+        portfolio = _portfolio(versions["candidate-portfolio"])
+        structured_id = _artifact_id(portfolio.adopted_structured_script_ref)
+        structured = await self._artifacts.get_by_id(
+            structured_id, script_build_id=binding.script_build_id
+        )
+        _structured(structured)
+        return _RootClosure(versions["direction"], versions["candidate-portfolio"], structured)
+
+
+class _RootClosure:
+    def __init__(
+        self,
+        direction: ArtifactVersion,
+        portfolio: ArtifactVersion,
+        structured: ArtifactVersion,
+    ):
+        self.direction = direction
+        self.portfolio = portfolio
+        self.structured = structured
+
+
+def _direction(version: ArtifactVersion) -> ScriptDirectionArtifactV1:
+    if not isinstance(version.artifact, ScriptDirectionArtifactV1):
+        raise ProtocolViolation("Root Direction artifact has the wrong type")
+    return version.artifact
+
+
+def _portfolio(version: ArtifactVersion) -> CandidatePortfolioArtifactV1:
+    if not isinstance(version.artifact, CandidatePortfolioArtifactV1):
+        raise ProtocolViolation("Root Portfolio artifact has the wrong type")
+    return version.artifact
+
+
+def _structured(version: ArtifactVersion) -> StructuredScriptArtifactV1:
+    if not isinstance(version.artifact, StructuredScriptArtifactV1):
+        raise ProtocolViolation("Root StructuredScript artifact has the wrong type")
+    return version.artifact
+
+
+def _uri(version: ArtifactVersion) -> str:
+    return f"script-build://artifact-versions/{version.artifact_version_id}"
+
+
+def _closure_digest(closure: _RootClosure) -> str:
+    return root_delivery_input_closure_digest(
+        direction_ref=_uri(closure.direction),
+        direction_digest=closure.direction.canonical_sha256,
+        candidate_portfolio_ref=_uri(closure.portfolio),
+        candidate_portfolio_digest=closure.portfolio.canonical_sha256,
+        structured_script_ref=_uri(closure.structured),
+        structured_script_digest=closure.structured.canonical_sha256,
+    )
+
+
+def _artifact_id(uri: str) -> int:
+    value = uri.removeprefix("script-build://artifact-versions/")
+    if not value.isdigit():
+        raise ProtocolViolation("Root delivery Artifact URI is invalid")
+    return int(value)
+
+
+def _required(context: Mapping[str, Any], key: str) -> str:
+    value = context.get(key)
+    if not isinstance(value, str) or not value:
+        raise ProtocolViolation(f"protected context is missing {key}")
+    return value
+
+
+__all__ = ["RootDeliveryService"]

script_build_host/src/script_build_host/tools/contracts.py

@@ -124,4 +124,29 @@ class ScriptCandidateToolPort(Protocol):
     async def resolve_attempt_manifest(self, *, context: Mapping[str, Any]) -> AttemptManifest: ...
 
 
-__all__ = ["AttemptManifest", "ScriptCandidateToolPort", "ScriptPlannerToolPort"]
+class RootDeliveryToolPort(Protocol):
+    async def read_accepted_portfolio(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]: ...
+
+    async def legacy_projection_dry_run(
+        self, *, build_summary: str, context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def save_root_delivery_manifest(
+        self,
+        *,
+        build_summary: str,
+        blocking_defects: Sequence[str],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+    async def deterministic_precheck(
+        self, *, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+
+__all__ = [
+    "AttemptManifest",
+    "RootDeliveryToolPort",
+    "ScriptCandidateToolPort",
+    "ScriptPlannerToolPort",
+]

script_build_host/src/script_build_host/tools/gateway.py

@@ -8,7 +8,7 @@ framework's protected context and then checked against durable bindings.
 from __future__ import annotations
 
 import re
-from collections.abc import Mapping, Sequence
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime
 from typing import Any, Protocol, cast
@@ -25,6 +25,7 @@ from agent.orchestration import (
 from script_build_host.agents.validation import (
     PHASE_TWO_KINDS,
     RETRIEVAL_KINDS,
+    ROOT_DELIVERY_KIND,
     precheck_business_artifact,
     precheck_snapshot,
     task_kind,
@@ -45,7 +46,7 @@ from script_build_host.domain.ports import (
 )
 from script_build_host.infrastructure.redaction import redact, redact_text
 
-from .contracts import ScriptCandidateToolPort, ScriptPlannerToolPort
+from .contracts import RootDeliveryToolPort, ScriptCandidateToolPort, ScriptPlannerToolPort
 
 _DEFECT_SEVERITIES = frozenset({"warning", "hard", "critical"})
 _DEFECT_ACTIONS = frozenset(
@@ -82,6 +83,10 @@ class ImageAdapter(Protocol):
 class DispatchGuard(Protocol):
     async def ensure_dispatch_allowed(self, script_build_id: int) -> None: ...
 
+    async def execute_fenced(
+        self, script_build_id: int, mutation: Callable[[], Awaitable[Any]]
+    ) -> Any: ...
+
 
 class PhaseTwoBudgetGuard(Protocol):
     async def ensure_tool_budget(self, *, root_trace_id: str, task_id: str, entry: str) -> None: ...
@@ -103,6 +108,7 @@ class LegacyScriptToolGateway:
         planner_tools: ScriptPlannerToolPort | None = None,
         candidate_tools: ScriptCandidateToolPort | None = None,
         budget_guard: PhaseTwoBudgetGuard | None = None,
+        root_tools: RootDeliveryToolPort | None = None,
     ) -> None:
         self.bindings = bindings
         self.snapshots = snapshots
@@ -114,6 +120,7 @@ class LegacyScriptToolGateway:
         self.planner_tools = planner_tools
         self.candidate_tools = candidate_tools
         self.budget_guard = budget_guard
+        self.root_tools = root_tools
         self._active_retrieval_attempts: set[tuple[str, str, str]] = set()
 
     async def ensure_dispatch_allowed(self, context: Mapping[str, Any]) -> None:
@@ -121,6 +128,14 @@ class LegacyScriptToolGateway:
         if self.dispatch_guard is not None:
             await self.dispatch_guard.ensure_dispatch_allowed(binding.script_build_id)
 
+    async def _execute_fenced(
+        self, context: Mapping[str, Any], mutation: Callable[[], Awaitable[Any]]
+    ) -> Any:
+        binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
+        if self.dispatch_guard is None:
+            return await mutation()
+        return await self.dispatch_guard.execute_fenced(binding.script_build_id, mutation)
+
     async def plan_script_tasks(
         self,
         *,
@@ -128,10 +143,16 @@ class LegacyScriptToolGateway:
         parent_task_id: str | None,
         context: Mapping[str, Any],
     ) -> Mapping[str, Any]:
-        return await self._planner().plan_script_tasks(
-            contract_payloads=contracts,
-            parent_task_id=parent_task_id,
-            context=context,
+        return cast(
+            Mapping[str, Any],
+            await self._execute_fenced(
+                context,
+                lambda: self._planner().plan_script_tasks(
+                    contract_payloads=contracts,
+                    parent_task_id=parent_task_id,
+                    context=context,
+                ),
+            ),
         )
 
     async def decide_script_task(
@@ -145,14 +166,20 @@ class LegacyScriptToolGateway:
         child_contracts: Sequence[Mapping[str, Any]],
         context: Mapping[str, Any],
     ) -> Mapping[str, Any]:
-        return await self._planner().decide_script_task(
-            task_id=task_id,
-            action=action,
-            reason=reason,
-            validation_id=validation_id,
-            replacement_contract=replacement_contract,
-            child_contracts=child_contracts,
-            context=context,
+        return cast(
+            Mapping[str, Any],
+            await self._execute_fenced(
+                context,
+                lambda: self._planner().decide_script_task(
+                    task_id=task_id,
+                    action=action,
+                    reason=reason,
+                    validation_id=validation_id,
+                    replacement_contract=replacement_contract,
+                    child_contracts=child_contracts,
+                    context=context,
+                ),
+            ),
         )
 
     async def inspect_script_plan(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
@@ -161,8 +188,13 @@ class LegacyScriptToolGateway:
     async def dispatch_script_tasks(
         self, *, task_ids: Sequence[str], context: Mapping[str, Any]
     ) -> Sequence[Mapping[str, Any]]:
-        await self.ensure_dispatch_allowed(context)
-        return await self._planner().dispatch_script_tasks(task_ids=task_ids, context=context)
+        return cast(
+            Sequence[Mapping[str, Any]],
+            await self._execute_fenced(
+                context,
+                lambda: self._planner().dispatch_script_tasks(task_ids=task_ids, context=context),
+            ),
+        )
 
     async def read_input_snapshot(self, context: Mapping[str, Any]) -> dict[str, Any]:
         binding, snapshot = await self._scope(context)
@@ -182,6 +214,35 @@ class LegacyScriptToolGateway:
         ]
         return cast(dict[str, Any], _json_values(payload))
 
+    async def read_accepted_portfolio(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        return await self._root_tools().read_accepted_portfolio(context=context)
+
+    async def legacy_projection_dry_run(
+        self, build_summary: str, context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        return await self._root_tools().legacy_projection_dry_run(
+            build_summary=build_summary, context=context
+        )
+
+    async def save_root_delivery_manifest(
+        self,
+        *,
+        build_summary: str,
+        blocking_defects: Sequence[str],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        return cast(
+            Mapping[str, Any],
+            await self._execute_fenced(
+                context,
+                lambda: self._root_tools().save_root_delivery_manifest(
+                    build_summary=build_summary,
+                    blocking_defects=blocking_defects,
+                    context=context,
+                ),
+            ),
+        )
+
     async def load_frozen_strategy(
         self, strategy_ref: str, context: Mapping[str, Any]
     ) -> dict[str, Any]:
@@ -473,8 +534,33 @@ class LegacyScriptToolGateway:
                 task_id=_required(context, "task_id"),
                 entry="submit",
             )
-        manifest = await self._candidates().resolve_attempt_manifest(context=context)
-        ref = manifest.artifact_ref
+        root_trace_id = _required(context, "root_trace_id")
+        task_id = _required(context, "task_id")
+        attempt_id = _required(context, "attempt_id")
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise ProtocolViolation("attempt task is outside the protected mission")
+        if task_kind(task.current_spec.context_refs) == ROOT_DELIVERY_KIND:
+            binding = await self.bindings.get_by_root(root_trace_id)
+            root_version = await self.artifacts.get_by_attempt(
+                script_build_id=binding.script_build_id,
+                task_id=task_id,
+                attempt_id=attempt_id,
+            )
+            if root_version.artifact_type is not ArtifactKind.ROOT_DELIVERY_MANIFEST:
+                raise ProtocolViolation("Root Attempt does not own a delivery manifest")
+            ref = ArtifactRef(
+                uri=(f"script-build://artifact-versions/{root_version.artifact_version_id}"),
+                kind=root_version.artifact_type.value,
+                version=str(root_version.artifact_version_id),
+                digest=root_version.canonical_sha256,
+            )
+            scope_ref = "script-build://scope/root"
+        else:
+            manifest = await self._candidates().resolve_attempt_manifest(context=context)
+            ref = manifest.artifact_ref
+            scope_ref = manifest.scope_ref
         if ref.kind not in _BUSINESS_ARTIFACT_KINDS or not ref.digest:
             raise ProtocolViolation("attempt manifest must contain one frozen business artifact")
         binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
@@ -496,7 +582,7 @@ class LegacyScriptToolGateway:
             version=ref.version,
             digest=ref.digest,
         )
-        summary = _attempt_summary(safe_ref, manifest.scope_ref)
+        summary = _attempt_summary(safe_ref, scope_ref)
         submission = AttemptSubmission(
             summary=summary,
             artifact_refs=([] if safe_ref.kind == ArtifactKind.EVIDENCE.value else [safe_ref]),
@@ -504,7 +590,10 @@ class LegacyScriptToolGateway:
         )
         return cast(
             dict[str, Any],
-            await self.coordinator.submit_attempt(dict(context), submission),
+            await self._execute_fenced(
+                context,
+                lambda: self.coordinator.submit_attempt(dict(context), submission),
+            ),
         )
 
     async def submit_structured_validation(
@@ -602,6 +691,15 @@ class LegacyScriptToolGateway:
                 raise ProtocolViolation(
                     "passed validation conflicts with phase-two deterministic precheck"
                 )
+        if (
+            overall is ValidationVerdict.PASSED
+            and task_kind(task.current_spec.context_refs) == ROOT_DELIVERY_KIND
+        ):
+            root_rules = await self._root_tools().deterministic_precheck(context=context)
+            if any(item.get("verdict") != "passed" for item in root_rules):
+                raise ProtocolViolation(
+                    "passed validation conflicts with Root deterministic precheck"
+                )
         if overall is not ValidationVerdict.PASSED and not normalized_defects:
             raise ProtocolViolation("failed or inconclusive validation must report a defect")
         if any(
@@ -630,15 +728,18 @@ class LegacyScriptToolGateway:
         )
         return cast(
             dict[str, Any],
-            await self.coordinator.submit_validation(
-                dict(context),
-                overall,
-                tuple(result_by_id.values()),
-                summary,
-                evidence_refs,
-                (),
-                codes[:50],
-                bounded_recommendation,
+            await self._execute_fenced(
+                context,
+                lambda: self.coordinator.submit_validation(
+                    dict(context),
+                    overall,
+                    tuple(result_by_id.values()),
+                    summary,
+                    evidence_refs,
+                    (),
+                    codes[:50],
+                    bounded_recommendation,
+                ),
             ),
         )
 
@@ -750,6 +851,8 @@ class LegacyScriptToolGateway:
         kind = task_kind(task.current_spec.context_refs)
         if kind in PHASE_TWO_KINDS:
             rules.extend(await self._candidates().deterministic_precheck(context=context))
+        elif kind == "root-delivery":
+            rules.extend(await self._root_tools().deterministic_precheck(context=context))
         return {"rule_results": rules}
 
     async def _scope(self, context: Mapping[str, Any]) -> tuple[Any, Any]:
@@ -771,6 +874,11 @@ class LegacyScriptToolGateway:
             raise ProtocolViolation("script candidate tools are not configured")
         return self.candidate_tools
 
+    def _root_tools(self) -> RootDeliveryToolPort:
+        if self.root_tools is None:
+            raise ProtocolViolation("script Root delivery tools are not configured")
+        return self.root_tools
+
 
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)

script_build_host/src/script_build_host/tools/registry.py

@@ -24,6 +24,7 @@ TASK_PRESET_BY_KIND = {
     "compare": "script_candidate_compare_worker",
     "compose": "script_compose_worker",
     "candidate-portfolio": "script_candidate_portfolio_worker",
+    "root-delivery": "script_root_worker",
 }
 
 
@@ -558,6 +559,40 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, save_candidate_portfolio, capabilities=["write"])
 
+    async def read_accepted_portfolio(context: dict[str, Any] | None = None) -> str:
+        """Read the Root Attempt's accepted Direction, Portfolio and adopted script."""
+
+        return _json(await gateway.read_accepted_portfolio(context or {}))
+
+    _register(registry, read_accepted_portfolio, capabilities=["read"])
+
+    async def legacy_projection_dry_run(
+        build_summary: str,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Compute the physical-ID-independent legacy projection digest."""
+
+        return _json(await gateway.legacy_projection_dry_run(build_summary, context or {}))
+
+    _register(registry, legacy_projection_dry_run, capabilities=["read"])
+
+    async def save_root_delivery_manifest(
+        build_summary: str,
+        blocking_defects: list[str] | None = None,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Freeze the unique Root delivery manifest without rewriting script content."""
+
+        return _json(
+            await gateway.save_root_delivery_manifest(
+                build_summary=build_summary,
+                blocking_defects=blocking_defects or [],
+                context=context or {},
+            )
+        )
+
+    _register(registry, save_root_delivery_manifest, capabilities=["write"])
+
     async def query_validation_evidence(
         query: str, limit: int = 10, context: dict[str, Any] | None = None
     ) -> str: