生命周期：实现阶段一续跑、作品集边界与停止状态矩阵

script_build_host/src/script_build_host/application/mission_service.py

@@ -6,14 +6,25 @@ import asyncio
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Protocol
 from uuid import uuid4
 
 from agent.orchestration import DecisionAction, OperationStatus, TaskStatus, ValidationVerdict
 
-from script_build_host.agents.validation import task_kind
-from script_build_host.domain.artifacts import ArtifactKind, ScriptDirectionArtifactV1
-from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ScriptDirectionArtifactV1,
+)
+from script_build_host.domain.canonical_json import canonical_sha256
+from script_build_host.domain.errors import (
+    DirectionProjectionConflict,
+    MissionRecoveryRequired,
+    PhaseTwoBoundaryNotReady,
+    PhaseTwoPromptsNotFrozen,
+    PlannerPolicyMigrationRequired,
+    ProtocolViolation,
+)
 from script_build_host.domain.ports import (
     BuildAuthorizer,
     LegacyBuildStateRepository,
@@ -29,12 +40,14 @@ from script_build_host.domain.records import (
     PublicationState,
     PublicationType,
 )
-from script_build_host.infrastructure.redaction import redact
+from script_build_host.domain.redaction import redact
+from script_build_host.domain.task_refs import task_kind
 
 from .input_snapshot_service import AssembleInputRequest, ScriptInputSnapshotService
 from .mission_factory import ScriptMissionFactory
 
 PHASE_ONE_CAPABILITY_BOUNDARY = "PHASE_ONE_CAPABILITY_BOUNDARY"
+PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
 
 
 class BuildTransitionGate:
@@ -82,6 +95,26 @@ class StopResult:
     stopped_operation_ids: tuple[str, ...] = ()
 
 
+@dataclass(frozen=True, slots=True)
+class PhaseAdvanceResult:
+    script_build_id: int
+    status: BuildStatus
+    root_trace_id: str
+    input_snapshot_id: str
+
+
+class PhaseTwoBoundaryVerifier(Protocol):
+    """Host extension point owned by the phase-two TaskContract subsystem."""
+
+    async def verify_before_advance(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None: ...
+
+    async def verify_checkpoint(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None: ...
+
+
 class DirectionReconciler:
     """Idempotently project one passed and accepted Direction artifact."""
 
@@ -155,6 +188,8 @@ class DirectionReconciler:
         )
         if not isinstance(version.artifact, ScriptDirectionArtifactV1):
             raise ProtocolViolation("accepted direction reference has the wrong artifact type")
+        if version.state not in {ArtifactState.FROZEN, ArtifactState.PUBLISHED}:
+            raise ProtocolViolation("accepted direction artifact is not frozen")
         publication = await self.publications.prepare(
             script_build_id=script_build_id,
             publication_type=PublicationType.DIRECTION,
@@ -162,20 +197,29 @@ class DirectionReconciler:
             artifact_version_id=version.artifact_version_id,
             expected_sha256=version.canonical_sha256,
         )
-        if publication.state is PublicationState.PUBLISHED:
-            return version.artifact_version_id
         try:
             status = await self.legacy_state.get_status(script_build_id)
             if status in {BuildStatus.STOPPING, BuildStatus.STOPPED}:
                 raise ProtocolViolation("direction publication is forbidden while stopping")
-            await self.legacy_state.project_direction(
-                script_build_id, version.artifact.legacy_markdown
-            )
-            await self.bindings.set_active_direction(
-                script_build_id=script_build_id,
-                artifact_version_id=version.artifact_version_id,
-            )
-            await self.publications.mark_published(publication.publication_id)
+            get_direction = getattr(self.legacy_state, "get_direction", None)
+            projected = await get_direction(script_build_id) if get_direction is not None else None
+            if projected is None or not projected.strip():
+                await self.legacy_state.project_direction(
+                    script_build_id, version.artifact.legacy_markdown
+                )
+            elif projected != version.artifact.legacy_markdown:
+                raise DirectionProjectionConflict()
+            binding = await self.bindings.get_by_build(script_build_id)
+            active = binding.active_direction_artifact_version_id
+            if active is None:
+                await self.bindings.set_active_direction(
+                    script_build_id=script_build_id,
+                    artifact_version_id=version.artifact_version_id,
+                )
+            elif active != version.artifact_version_id:
+                raise DirectionProjectionConflict()
+            if publication.state is not PublicationState.PUBLISHED:
+                await self.publications.mark_published(publication.publication_id)
         except Exception as exc:
             await self.publications.mark_failed(
                 publication.publication_id,
@@ -205,6 +249,11 @@ class ScriptMissionService:
         stop_timeout_seconds: float = 5.0,
         transition_gate: BuildTransitionGate | None = None,
         runtime_manifest_provider: RuntimeManifestProvider | None = None,
+        enabled_phase: int = 1,
+        phase_two_prompt_requests: tuple[PromptRequest, ...] = (),
+        phase_two_required_presets: tuple[str, ...] = (),
+        phase_two_boundary_verifier: PhaseTwoBoundaryVerifier | None = None,
+        phase_two_model_manifest: dict[str, Any] | None = None,
     ) -> None:
         self.runner = runner
         self.coordinator = coordinator
@@ -224,7 +273,15 @@ class ScriptMissionService:
             else BuildTransitionGate()
         )
         self.runtime_manifest_provider = runtime_manifest_provider
+        if enabled_phase not in {1, 2}:
+            raise ValueError("enabled_phase must be 1 or 2")
+        self.enabled_phase = enabled_phase
+        self.phase_two_prompt_requests = phase_two_prompt_requests
+        self.phase_two_required_presets = frozenset(phase_two_required_presets)
+        self.phase_two_boundary_verifier = phase_two_boundary_verifier
+        self.phase_two_model_manifest = dict(phase_two_model_manifest or {})
         self._runs: dict[int, asyncio.Task[None]] = {}
+        self._phase_advance_locks: dict[int, asyncio.Lock] = {}
 
     async def start(self, command: StartScriptBuildCommand) -> ScriptMissionStartResult:
         await self.authorizer.require_source_access(
@@ -283,6 +340,8 @@ class ScriptMissionService:
                 )
             )
             snapshot = await self.input_snapshots.freeze(assembled)
+            if self.enabled_phase >= 2:
+                self._require_phase_two_prompts(snapshot)
             root_trace_id = str(uuid4())
             binding = await self.bindings.create(
                 script_build_id=script_build_id,
@@ -312,6 +371,17 @@ class ScriptMissionService:
         )
 
     async def run(self, script_build_id: int) -> None:
+        await self.run_to_enabled_phase(script_build_id)
+
+    async def run_to_enabled_phase(self, script_build_id: int) -> None:
+        await self.run_phase_one(script_build_id)
+        if (
+            self.enabled_phase >= 2
+            and await self.legacy_state.get_status(script_build_id) is BuildStatus.PARTIAL
+        ):
+            await self._run_phase_two_transition(script_build_id)
+
+    async def run_phase_one(self, script_build_id: int) -> None:
         binding = await self.bindings.get_by_build(script_build_id)
         snapshot = await self.input_snapshots.get(
             str(binding.input_snapshot_id), script_build_id=script_build_id
@@ -338,7 +408,12 @@ class ScriptMissionService:
             async with self.transition_gate.hold(script_build_id):
                 if await self.legacy_state.get_status(script_build_id) == BuildStatus.STOPPING:
                     return
-                await self.legacy_state.set_status(script_build_id, BuildStatus.PARTIAL)
+                await self._set_checkpoint(
+                    script_build_id,
+                    checkpoint_code=PHASE_ONE_CAPABILITY_BOUNDARY,
+                    summary="Phase one accepted and published one immutable direction.",
+                    root_trace_id=binding.root_trace_id,
+                )
         except Exception as exc:
             if await self.legacy_state.get_status(script_build_id) != BuildStatus.STOPPING:
                 await self.legacy_state.set_status(
@@ -348,14 +423,682 @@ class ScriptMissionService:
                 )
             raise
 
+    async def advance_to_phase_two(
+        self,
+        script_build_id: int,
+        principal: Principal,
+    ) -> PhaseAdvanceResult:
+        """Idempotently start Phase2 for one authorized Phase1 checkpoint."""
+
+        await self.authorizer.require_access(principal, script_build_id)
+        lock = self._phase_advance_locks.setdefault(script_build_id, asyncio.Lock())
+        async with lock:
+            return await self._advance_to_phase_two_locked(script_build_id)
+
+    async def _advance_to_phase_two_locked(
+        self,
+        script_build_id: int,
+    ) -> PhaseAdvanceResult:
+        binding = await self.bindings.get_by_build(script_build_id)
+        status = await self.legacy_state.get_status(script_build_id)
+        if status in {BuildStatus.STOPPING, BuildStatus.STOPPED, BuildStatus.FAILED}:
+            raise PhaseTwoBoundaryNotReady(f"cannot advance a {status.value} build")
+        if status is BuildStatus.SUCCESS:
+            raise PhaseTwoBoundaryNotReady("a successful build cannot be advanced")
+        active = self._runs.get(script_build_id)
+        if active is not None and not active.done():
+            return PhaseAdvanceResult(
+                script_build_id,
+                BuildStatus.RUNNING,
+                binding.root_trace_id,
+                str(binding.input_snapshot_id),
+            )
+        completion = await self.coordinator.root_completion(binding.root_trace_id)
+        if (
+            status is BuildStatus.PARTIAL
+            and completion.get("status") == TaskStatus.BLOCKED.value
+            and completion.get("blocked_reason") == PHASE_TWO_CANDIDATE_PORTFOLIO_READY
+        ):
+            return PhaseAdvanceResult(
+                script_build_id,
+                BuildStatus.PARTIAL,
+                binding.root_trace_id,
+                str(binding.input_snapshot_id),
+            )
+        if completion.get("status") == TaskStatus.NEEDS_REPLAN.value:
+            binding, snapshot, direction_version_id = await self._prepare_phase_two_reentry(
+                script_build_id
+            )
+        else:
+            binding, snapshot, direction_version_id = await self._prepare_phase_two_transition(
+                script_build_id
+            )
+        task = asyncio.create_task(
+            self.run_phase_two(
+                script_build_id,
+                binding=binding,
+                snapshot=snapshot,
+                direction_artifact_version_id=direction_version_id,
+            ),
+            name=f"script-build:{script_build_id}:phase-two",
+        )
+        self._runs[script_build_id] = task
+        task.add_done_callback(lambda done: self._discard_run(script_build_id, done))
+        return PhaseAdvanceResult(
+            script_build_id,
+            BuildStatus.RUNNING,
+            binding.root_trace_id,
+            snapshot.snapshot_id,
+        )
+
+    async def _run_phase_two_transition(self, script_build_id: int) -> None:
+        binding, snapshot, direction_version_id = await self._prepare_phase_two_transition(
+            script_build_id
+        )
+        await self.run_phase_two(
+            script_build_id,
+            binding=binding,
+            snapshot=snapshot,
+            direction_artifact_version_id=direction_version_id,
+        )
+
+    async def _prepare_phase_two_transition(self, script_build_id: int) -> tuple[Any, Any, int]:
+        binding = await self.bindings.get_by_build(script_build_id)
+        snapshot = await self.input_snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=script_build_id
+        )
+        original_snapshot_id = str(snapshot.snapshot_id)
+        # Validate the immutable Phase1 checkpoint before appending a v2
+        # snapshot.  A rejected continuation must not mutate snapshot lineage.
+        direction_version_id = await self._verify_phase_two_preconditions(
+            script_build_id, binding, snapshot
+        )
+        snapshot, binding = await self._ensure_phase_two_snapshot(snapshot, binding)
+        if str(snapshot.snapshot_id) != str(binding.input_snapshot_id):
+            raise PhaseTwoBoundaryNotReady("binding does not reference the prepared snapshot")
+        if str(snapshot.snapshot_id) != original_snapshot_id:
+            # A v2 prompt-only snapshot has a new identity, so revalidate the
+            # binding and Phase1 closure against the updated lineage.
+            direction_version_id = await self._verify_phase_two_preconditions(
+                script_build_id, binding, snapshot
+            )
+        return binding, snapshot, direction_version_id
+
+    async def _prepare_phase_two_reentry(self, script_build_id: int) -> tuple[Any, Any, int]:
+        """Verify the only Phase2 crash point that is safe before stage-three recovery.
+
+        This path applies after the signed policy is durable and Root was UNBLOCKed,
+        but before any Phase2 model output, Task or Operation was created.
+        """
+
+        status = await self.legacy_state.get_status(script_build_id)
+        if status not in {BuildStatus.PARTIAL, BuildStatus.RUNNING}:
+            raise PhaseTwoBoundaryNotReady("build is not at a resumable phase-two transition")
+        binding = await self.bindings.get_by_build(script_build_id)
+        snapshot = await self.input_snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=script_build_id
+        )
+        if (
+            binding.script_build_id != script_build_id
+            or snapshot.script_build_id != script_build_id
+            or str(binding.input_snapshot_id) != str(snapshot.snapshot_id)
+        ):
+            raise PhaseTwoBoundaryNotReady("binding and InputSnapshot ownership differs")
+        self._require_phase_two_prompts(snapshot)
+        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+        root = ledger.tasks[ledger.root_task_id]
+        if root.status is not TaskStatus.NEEDS_REPLAN:
+            raise PhaseTwoBoundaryNotReady("Root is not at the safe phase-two reentry point")
+        if any(
+            task_kind(item.current_spec.context_refs) == "candidate-portfolio"
+            for item in ledger.tasks.values()
+        ):
+            raise MissionRecoveryRequired()
+        if any(
+            item.status
+            in {OperationStatus.PENDING, OperationStatus.RUNNING, OperationStatus.STOP_REQUESTED}
+            for item in ledger.operations.values()
+        ):
+            raise MissionRecoveryRequired()
+
+        direction_version_id = await self.direction_reconciler.reconcile(
+            script_build_id, binding.root_trace_id
+        )
+        binding = await self.bindings.get_by_build(script_build_id)
+        publication = await self.direction_reconciler.publications.get_by_build(
+            script_build_id, publication_type=PublicationType.DIRECTION
+        )
+        if (
+            publication is None
+            or publication.state is not PublicationState.PUBLISHED
+            or publication.artifact_version_id != direction_version_id
+            or binding.active_direction_artifact_version_id != direction_version_id
+        ):
+            raise PhaseTwoBoundaryNotReady("direction publication is not durably closed")
+        await self._verify_direction_snapshot_lineage(
+            script_build_id=script_build_id,
+            ledger=ledger,
+            current_snapshot=snapshot,
+        )
+
+        policy = self.factory.build_phase_two_policy(snapshot)
+        continuation_message = self.factory.build_phase_two_message(
+            binding,
+            snapshot,
+            direction_artifact_version_id=direction_version_id,
+        )
+        events = await self.runner.trace_store.get_events(binding.root_trace_id, 0)
+        migrations = [
+            item
+            for item in events
+            if item.get("event") == "planner_policy_migrated"
+            or item.get("event_type") == "planner_policy_migrated"
+        ]
+        if len(migrations) != 1:
+            raise PlannerPolicyMigrationRequired(
+                "safe phase-two reentry requires one durable policy migration"
+            )
+        payload = migrations[0].get("data") or migrations[0].get("payload") or migrations[0]
+        await self._verify_policy_migration(
+            root_trace_id=binding.root_trace_id,
+            payload=payload,
+            policy=policy,
+            policy_digest=_canonical_digest(policy),
+            toolset_digest=_canonical_digest(
+                sorted(self.runner.tools.get_tool_names(groups=["script_build"]))
+            ),
+            input_snapshot_id=str(snapshot.snapshot_id),
+            continuation_message=continuation_message,
+        )
+        if await self._has_phase_two_output(
+            binding.root_trace_id, int(payload.get("user_sequence", 0))
+        ):
+            raise MissionRecoveryRequired()
+        if self.phase_two_boundary_verifier is not None:
+            await self.phase_two_boundary_verifier.verify_before_advance(
+                script_build_id=script_build_id,
+                root_trace_id=binding.root_trace_id,
+                input_snapshot_id=snapshot.snapshot_id,
+            )
+        return binding, snapshot, direction_version_id
+
+    async def run_phase_two(
+        self,
+        script_build_id: int,
+        *,
+        binding: Any | None = None,
+        snapshot: Any | None = None,
+        direction_artifact_version_id: int | None = None,
+    ) -> None:
+        binding = binding or await self.bindings.get_by_build(script_build_id)
+        snapshot = snapshot or await self.input_snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=script_build_id
+        )
+        if direction_artifact_version_id is None:
+            direction_artifact_version_id = await self._verify_phase_two_preconditions(
+                script_build_id, binding, snapshot
+            )
+        policy = self.factory.build_phase_two_policy(snapshot)
+        continuation_message = self.factory.build_phase_two_message(
+            binding,
+            snapshot,
+            direction_artifact_version_id=direction_artifact_version_id,
+        )
+        policy_digest = _canonical_digest(policy)
+        toolset_digest = _canonical_digest(
+            sorted(self.runner.tools.get_tool_names(groups=["script_build"]))
+        )
+        trace_store = self.runner.trace_store
+        events = await trace_store.get_events(binding.root_trace_id, 0)
+        migrations = [
+            item
+            for item in events
+            if item.get("event") == "planner_policy_migrated"
+            or item.get("event_type") == "planner_policy_migrated"
+        ]
+        migration = migrations[0] if migrations else None
+        iterator: AsyncIterator[Any] | None = None
+        try:
+            if len(migrations) > 1:
+                raise PlannerPolicyMigrationRequired(
+                    "multiple Planner policy migration events are not permitted"
+                )
+            if migration is None:
+                prepared = await self._find_unrecorded_policy_messages(
+                    root_trace_id=binding.root_trace_id,
+                    policy=policy,
+                    continuation_message=continuation_message,
+                )
+                if prepared is None:
+                    iterator = self.runner.run(
+                        messages=[
+                            {"role": "system", "content": policy},
+                            {
+                                "role": "user",
+                                "content": continuation_message,
+                            },
+                        ],
+                        config=self.factory.build_phase_two_run_config(binding, snapshot),
+                    )
+                    try:
+                        await anext(iterator)  # prepared Trace
+                        system_message = await anext(iterator)
+                        user_message = await anext(iterator)
+                    except Exception as exc:
+                        raise PlannerPolicyMigrationRequired(
+                            "phase-two policy messages could not be durably appended"
+                        ) from exc
+                    if (
+                        getattr(system_message, "role", None) != "system"
+                        or getattr(user_message, "role", None) != "user"
+                    ):
+                        raise PlannerPolicyMigrationRequired(
+                            "phase-two policy messages were not durably prepared"
+                        )
+                    prepared = (system_message.sequence, user_message.sequence)
+                system_sequence, user_sequence = prepared
+                migration_payload = {
+                    "policy_version": "script-build-phase-two/v1",
+                    "policy_digest": policy_digest,
+                    "toolset_digest": toolset_digest,
+                    "input_snapshot_id": snapshot.snapshot_id,
+                    "system_sequence": system_sequence,
+                    "user_sequence": user_sequence,
+                }
+                try:
+                    await trace_store.append_event(
+                        binding.root_trace_id,
+                        "planner_policy_migrated",
+                        migration_payload,
+                    )
+                except Exception as exc:
+                    raise PlannerPolicyMigrationRequired(
+                        "phase-two policy event could not be durably appended"
+                    ) from exc
+                if iterator is None and await self._has_phase_two_output(
+                    binding.root_trace_id, user_sequence
+                ):
+                    raise MissionRecoveryRequired()
+            if migration is not None:
+                payload = migration.get("data") or migration.get("payload") or migration
+                await self._verify_policy_migration(
+                    root_trace_id=binding.root_trace_id,
+                    payload=payload,
+                    policy=policy,
+                    policy_digest=policy_digest,
+                    toolset_digest=toolset_digest,
+                    input_snapshot_id=str(snapshot.snapshot_id),
+                    continuation_message=continuation_message,
+                )
+                if await self._has_phase_two_output(
+                    binding.root_trace_id, int(payload.get("user_sequence", 0))
+                ):
+                    raise MissionRecoveryRequired()
+            ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+            root = ledger.tasks[ledger.root_task_id]
+            if root.status == TaskStatus.BLOCKED:
+                if root.blocked_reason != PHASE_ONE_CAPABILITY_BOUNDARY:
+                    raise PhaseTwoBoundaryNotReady("Root is blocked at another boundary")
+                await self.coordinator.decide_task(
+                    binding.root_trace_id,
+                    root.task_id,
+                    None,
+                    DecisionAction.UNBLOCK,
+                    {},
+                    f"phase-two-unblock:{script_build_id}",
+                )
+            elif root.status != TaskStatus.NEEDS_REPLAN:
+                raise PhaseTwoBoundaryNotReady("Root is not ready for phase-two planning")
+            async with self.transition_gate.hold(script_build_id):
+                status = await self.legacy_state.get_status(script_build_id)
+                if status is BuildStatus.STOPPING:
+                    return
+                if status is BuildStatus.PARTIAL:
+                    await self._begin_phase(script_build_id)
+                elif status is not BuildStatus.RUNNING:
+                    raise PhaseTwoBoundaryNotReady("build cannot enter phase two")
+            if iterator is None:
+                await self.runner.run_result(
+                    messages=[],
+                    config=self.factory.build_phase_two_run_config(binding, snapshot),
+                )
+            else:
+                async for _ in iterator:
+                    pass
+            completion = await self.coordinator.root_completion(binding.root_trace_id)
+            if (
+                completion.get("status") != TaskStatus.BLOCKED.value
+                or completion.get("blocked_reason") != PHASE_TWO_CANDIDATE_PORTFOLIO_READY
+            ):
+                raise ProtocolViolation("mission did not stop at the phase-two boundary")
+            if self.phase_two_boundary_verifier is not None:
+                await self.phase_two_boundary_verifier.verify_checkpoint(
+                    script_build_id=script_build_id,
+                    root_trace_id=binding.root_trace_id,
+                    input_snapshot_id=snapshot.snapshot_id,
+                )
+            async with self.transition_gate.hold(script_build_id):
+                if await self.legacy_state.get_status(script_build_id) == BuildStatus.STOPPING:
+                    return
+                await self._set_checkpoint(
+                    script_build_id,
+                    checkpoint_code=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
+                    summary="Phase two accepted one closed candidate portfolio.",
+                    root_trace_id=binding.root_trace_id,
+                )
+        except MissionRecoveryRequired:
+            # Existing Phase2 output is intentionally left for the stage-three
+            # recovery workflow.  Do not turn a recoverable durable run into a
+            # terminal failure merely because this process cannot resume it.
+            raise
+        except Exception as exc:
+            if await self.legacy_state.get_status(script_build_id) != BuildStatus.STOPPING:
+                await self.legacy_state.set_status(
+                    script_build_id,
+                    BuildStatus.FAILED,
+                    error_summary=getattr(exc, "code", type(exc).__name__),
+                )
+            raise
+
+    async def _verify_phase_two_preconditions(
+        self, script_build_id: int, binding: Any, snapshot: Any
+    ) -> int:
+        if (
+            binding.script_build_id != script_build_id
+            or str(binding.input_snapshot_id) != str(snapshot.snapshot_id)
+            or snapshot.script_build_id != script_build_id
+        ):
+            raise PhaseTwoBoundaryNotReady(
+                "binding, build and current InputSnapshot ownership differs"
+            )
+        status = await self.legacy_state.get_status(script_build_id)
+        if status is not BuildStatus.PARTIAL:
+            raise PhaseTwoBoundaryNotReady("build is not at a partial checkpoint")
+        completion = await self.coordinator.root_completion(binding.root_trace_id)
+        if (
+            completion.get("status") != TaskStatus.BLOCKED.value
+            or completion.get("blocked_reason") != PHASE_ONE_CAPABILITY_BOUNDARY
+        ):
+            raise PhaseTwoBoundaryNotReady("Root is not at the phase-one boundary")
+        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+        root = ledger.tasks[ledger.root_task_id]
+        if not root.decision_ids:
+            raise PhaseTwoBoundaryNotReady("Root has no phase-one boundary decision")
+        last_decision = ledger.decisions[root.decision_ids[-1]]
+        if (
+            last_decision.action is not DecisionAction.BLOCK
+            or last_decision.reason != PHASE_ONE_CAPABILITY_BOUNDARY
+        ):
+            raise PhaseTwoBoundaryNotReady("Root's latest decision is not the phase-one boundary")
+        if any(
+            task_kind(item.current_spec.context_refs) == "candidate-portfolio"
+            for item in ledger.tasks.values()
+        ):
+            raise PhaseTwoBoundaryNotReady("phase-two portfolio work already exists")
+        active = [
+            item
+            for item in ledger.operations.values()
+            if item.status
+            in {OperationStatus.PENDING, OperationStatus.RUNNING, OperationStatus.STOP_REQUESTED}
+        ]
+        if active:
+            raise PhaseTwoBoundaryNotReady("phase one still has active operations")
+        trace = await self.runner.trace_store.get_trace(binding.root_trace_id)
+        if trace is None or trace.agent_type != "script_planner":
+            raise PhaseTwoBoundaryNotReady("Root Trace is not the script Planner")
+        direction_version_id = await self.direction_reconciler.reconcile(
+            script_build_id, binding.root_trace_id
+        )
+        binding = await self.bindings.get_by_build(script_build_id)
+        publication = await self.direction_reconciler.publications.get_by_build(
+            script_build_id, publication_type=PublicationType.DIRECTION
+        )
+        if (
+            publication is None
+            or publication.state is not PublicationState.PUBLISHED
+            or publication.artifact_version_id != direction_version_id
+            or binding.active_direction_artifact_version_id != direction_version_id
+        ):
+            raise PhaseTwoBoundaryNotReady("direction publication is not durably closed")
+        await self._verify_direction_snapshot_lineage(
+            script_build_id=script_build_id,
+            ledger=ledger,
+            current_snapshot=snapshot,
+        )
+        if self.phase_two_boundary_verifier is not None:
+            await self.phase_two_boundary_verifier.verify_before_advance(
+                script_build_id=script_build_id,
+                root_trace_id=binding.root_trace_id,
+                input_snapshot_id=snapshot.snapshot_id,
+            )
+        return direction_version_id
+
+    async def _verify_direction_snapshot_lineage(
+        self,
+        *,
+        script_build_id: int,
+        ledger: Any,
+        current_snapshot: Any,
+    ) -> None:
+        direction_tasks = [
+            item
+            for item in ledger.tasks.values()
+            if task_kind(item.current_spec.context_refs) == "direction"
+            and item.status is TaskStatus.COMPLETED
+        ]
+        if len(direction_tasks) != 1:
+            raise PhaseTwoBoundaryNotReady("one completed Direction Task is required")
+        direction = direction_tasks[0]
+        accepted = [
+            ledger.decisions[item]
+            for item in direction.decision_ids
+            if ledger.decisions[item].action is DecisionAction.ACCEPT
+        ]
+        if len(accepted) != 1 or accepted[0].attempt_id is None:
+            raise PhaseTwoBoundaryNotReady("Direction requires one immutable ACCEPT Attempt")
+        attempt = ledger.attempts.get(accepted[0].attempt_id)
+        if attempt is None:
+            raise PhaseTwoBoundaryNotReady("Direction ACCEPT Attempt is missing")
+        attempted_specs = [item for item in direction.specs if item.version == attempt.spec_version]
+        if len(attempted_specs) != 1:
+            raise PhaseTwoBoundaryNotReady("Direction Attempt TaskSpec is missing")
+        refs = {
+            item
+            for item in attempted_specs[0].context_refs
+            if item.startswith("script-build://inputs/")
+        }
+        if len(refs) != 1:
+            raise PhaseTwoBoundaryNotReady("Direction Attempt must pin one InputSnapshot")
+        direction_snapshot_id = next(iter(refs)).removeprefix("script-build://inputs/")
+        current_snapshot_id = str(current_snapshot.snapshot_id)
+        if direction_snapshot_id == current_snapshot_id:
+            return
+        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
+        )
+        current_business_digest = canonical_sha256(
+            current_snapshot.to_input().business_payload()
+        ).wire
+        if (
+            str(current_snapshot.parent_snapshot_id or "") != direction_snapshot_id
+            or current_snapshot.parent_snapshot_sha256 != parent.canonical_sha256
+            or not current_snapshot.business_input_sha256
+            or current_snapshot.business_input_sha256 != parent_business_digest
+            or current_business_digest != parent_business_digest
+        ):
+            raise PhaseTwoBoundaryNotReady(
+                "Direction and Phase2 InputSnapshots do not form a valid immutable lineage"
+            )
+
+    async def _verify_policy_migration(
+        self,
+        *,
+        root_trace_id: str,
+        payload: Any,
+        policy: str,
+        policy_digest: str,
+        toolset_digest: str,
+        input_snapshot_id: str,
+        continuation_message: str,
+    ) -> None:
+        if not isinstance(payload, dict):
+            raise PlannerPolicyMigrationRequired("Planner policy event payload is invalid")
+        if (
+            payload.get("policy_version") != "script-build-phase-two/v1"
+            or payload.get("policy_digest") != policy_digest
+            or payload.get("toolset_digest") != toolset_digest
+            or str(payload.get("input_snapshot_id")) != input_snapshot_id
+        ):
+            raise PlannerPolicyMigrationRequired(
+                "Planner policy identity, toolset or InputSnapshot changed"
+            )
+        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)
+            or system_sequence <= 0
+            or user_sequence <= system_sequence
+        ):
+            raise PlannerPolicyMigrationRequired("Planner policy message sequence is invalid")
+        messages = await self.runner.trace_store.get_trace_messages(root_trace_id)
+        by_sequence = {item.sequence: item for item in messages}
+        system_message = by_sequence.get(system_sequence)
+        user_message = by_sequence.get(user_sequence)
+        if (
+            system_message is None
+            or system_message.role != "system"
+            or system_message.content != policy
+            or user_message is None
+            or user_message.role != "user"
+            or user_message.content != continuation_message
+            or getattr(user_message, "parent_sequence", None) != system_sequence
+        ):
+            raise PlannerPolicyMigrationRequired(
+                "Planner policy event is not backed by the expected durable messages"
+            )
+
+    async def _find_unrecorded_policy_messages(
+        self,
+        *,
+        root_trace_id: str,
+        policy: str,
+        continuation_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 == continuation_message
+        ]
+        if not systems and not users:
+            return None
+        if len(systems) != 1 or len(users) != 1:
+            raise PlannerPolicyMigrationRequired(
+                "unrecorded Planner policy messages are incomplete or duplicated"
+            )
+        system_message = systems[0]
+        user_message = users[0]
+        if (
+            user_message.sequence <= system_message.sequence
+            or getattr(user_message, "parent_sequence", None) != system_message.sequence
+        ):
+            raise PlannerPolicyMigrationRequired(
+                "unrecorded Planner policy messages are outside one durable parent chain"
+            )
+        return system_message.sequence, user_message.sequence
+
+    async def _ensure_phase_two_snapshot(self, snapshot: Any, binding: Any) -> tuple[Any, Any]:
+        if self._has_phase_two_prompts(snapshot):
+            return snapshot, binding
+        if not self.phase_two_prompt_requests:
+            raise PhaseTwoPromptsNotFrozen()
+        extended = await self.input_snapshots.extend_prompt_lineage(
+            snapshot,
+            prompt_requests=self.phase_two_prompt_requests,
+            model_manifest=_merge_model_manifests(
+                snapshot.model_manifest, self.phase_two_model_manifest
+            ),
+        )
+        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),
+        )
+        self._require_phase_two_prompts(extended)
+        return extended, binding
+
+    def _has_phase_two_prompts(self, snapshot: Any) -> bool:
+        if not self.phase_two_required_presets:
+            return True
+        present = {str(item.get("preset") or "") for item in snapshot.prompt_manifest}
+        return self.phase_two_required_presets.issubset(present)
+
+    def _require_phase_two_prompts(self, snapshot: Any) -> None:
+        if not self._has_phase_two_prompts(snapshot):
+            raise PhaseTwoPromptsNotFrozen()
+
+    async def _has_phase_two_output(self, root_trace_id: str, after_sequence: int) -> bool:
+        messages = await self.runner.trace_store.get_trace_messages(root_trace_id)
+        return any(
+            item.sequence > after_sequence and item.role in {"assistant", "tool"}
+            for item in messages
+        )
+
+    async def _set_checkpoint(
+        self,
+        script_build_id: int,
+        *,
+        checkpoint_code: str,
+        summary: str,
+        root_trace_id: str,
+    ) -> None:
+        setter = getattr(self.legacy_state, "set_checkpoint", None)
+        if setter is None:
+            await self.legacy_state.set_status(script_build_id, BuildStatus.PARTIAL)
+            return
+        await setter(
+            script_build_id,
+            checkpoint_code=checkpoint_code,
+            summary=summary,
+            root_trace_id=root_trace_id,
+        )
+
+    async def _begin_phase(self, script_build_id: int) -> None:
+        begin = getattr(self.legacy_state, "begin_phase", None)
+        if begin is None:
+            await self.legacy_state.set_status(script_build_id, BuildStatus.RUNNING)
+            return
+        await begin(script_build_id)
+
     async def stop(self, script_build_id: int, principal: Principal) -> StopResult:
         await self.authorizer.require_access(principal, script_build_id)
         async with self.transition_gate.hold(script_build_id):
             status = await self.legacy_state.get_status(script_build_id)
             if status == BuildStatus.STOPPED:
                 return StopResult(script_build_id, BuildStatus.STOPPED)
-            await self.legacy_state.set_status(script_build_id, BuildStatus.STOPPING)
-        binding = await self.bindings.get_by_build(script_build_id)
+            if status in {BuildStatus.FAILED, BuildStatus.SUCCESS}:
+                return StopResult(script_build_id, status)
+            if status is not BuildStatus.STOPPING:
+                await self.legacy_state.set_status(script_build_id, BuildStatus.STOPPING)
+        try:
+            binding = await self.bindings.get_by_build(script_build_id)
+        except Exception as exc:
+            if status in {BuildStatus.RUNNING, BuildStatus.STOPPING, BuildStatus.PARTIAL}:
+                await self.legacy_state.set_status(
+                    script_build_id,
+                    BuildStatus.FAILED,
+                    error_summary="MISSION_BINDING_MISSING",
+                )
+            raise exc
         ledger = await self.coordinator.task_store.load(binding.root_trace_id)
         active = [
             operation
@@ -385,6 +1128,14 @@ class ScriptMissionService:
                 timeout=self.stop_timeout_seconds,
             )
         except TimeoutError:
+            await self.legacy_state.set_status(
+                script_build_id,
+                BuildStatus.STOPPING,
+                error_summary=(
+                    "STOP_TIMEOUT active_operations="
+                    + ",".join(item.operation_id for item in active)
+                ),
+            )
             return StopResult(
                 script_build_id,
                 BuildStatus.STOPPING,
@@ -445,10 +1196,25 @@ class ScriptMissionService:
             task.exception()
 
 
+def _canonical_digest(value: Any) -> str:
+    return canonical_sha256(value).wire
+
+
+def _merge_model_manifests(current: dict[str, Any], configured: dict[str, Any]) -> dict[str, Any]:
+    current_presets = current.get("presets", current)
+    configured_presets = configured.get("presets", configured)
+    if not isinstance(current_presets, dict) or not isinstance(configured_presets, dict):
+        raise ProtocolViolation("model manifests must contain preset objects")
+    return {"presets": {**configured_presets, **current_presets}}
+
+
 __all__ = [
     "PHASE_ONE_CAPABILITY_BOUNDARY",
+    "PHASE_TWO_CANDIDATE_PORTFOLIO_READY",
     "BuildTransitionGate",
     "DirectionReconciler",
+    "PhaseAdvanceResult",
+    "PhaseTwoBoundaryVerifier",
     "ScriptMissionService",
     "ScriptMissionStartResult",
     "StartScriptBuildCommand",

script_build_host/src/script_build_host/application/phase_two_boundary.py

@@ -0,0 +1,243 @@
+"""Business gate for entering and closing the phase-two capability boundary."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from agent.orchestration import DecisionAction, OperationStatus, TaskStatus
+
+from script_build_host.domain.artifacts import ArtifactKind, ArtifactState
+from script_build_host.domain.errors import PhaseTwoBoundaryNotReady, ScriptBuildError
+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_contracts import ScriptTaskKind
+from script_build_host.domain.task_refs import task_kind
+
+from .phase_two_inputs import AcceptedInputResolver, all_descendants_terminal
+
+_ACTIVE_OPERATIONS = {
+    OperationStatus.PENDING,
+    OperationStatus.RUNNING,
+    OperationStatus.STOP_REQUESTED,
+}
+
+
+class ScriptPhaseTwoBoundaryVerifier:
+    """Prove that one accepted portfolio closes one immutable candidate set."""
+
+    def __init__(
+        self,
+        *,
+        task_store: Any,
+        artifact_store: Any,
+        artifacts: ScriptBusinessArtifactRepository,
+        bindings: MissionBindingRepository,
+        accepted_inputs: AcceptedInputResolver,
+    ) -> None:
+        self._task_store = task_store
+        self._artifact_store = artifact_store
+        self._artifacts = artifacts
+        self._bindings = bindings
+        self._accepted_inputs = accepted_inputs
+
+    async def verify_before_advance(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None:
+        try:
+            await self._verify_before_advance(
+                script_build_id=script_build_id,
+                root_trace_id=root_trace_id,
+                input_snapshot_id=input_snapshot_id,
+            )
+        except PhaseTwoBoundaryNotReady:
+            raise
+        except ScriptBuildError as exc:
+            raise PhaseTwoBoundaryNotReady(exc.summary) from exc
+
+    async def _verify_before_advance(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None:
+        ledger = await self._owned_ledger(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            input_snapshot_id=input_snapshot_id,
+        )
+        if any(
+            task_kind(item.current_spec.context_refs) == ScriptTaskKind.CANDIDATE_PORTFOLIO.value
+            for item in ledger.tasks.values()
+        ):
+            raise PhaseTwoBoundaryNotReady("a phase-two portfolio already exists")
+        if any(item.status in _ACTIVE_OPERATIONS for item in ledger.operations.values()):
+            raise PhaseTwoBoundaryNotReady("active Operations prevent phase-two continuation")
+
+    async def verify_checkpoint(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None:
+        try:
+            await self._verify_checkpoint(
+                script_build_id=script_build_id,
+                root_trace_id=root_trace_id,
+                input_snapshot_id=input_snapshot_id,
+            )
+        except PhaseTwoBoundaryNotReady:
+            raise
+        except ScriptBuildError as exc:
+            raise PhaseTwoBoundaryNotReady(exc.summary) from exc
+
+    async def _verify_checkpoint(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> None:
+        ledger = await self._owned_ledger(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            input_snapshot_id=input_snapshot_id,
+        )
+        root = ledger.tasks.get(ledger.root_task_id or "")
+        if root is None:
+            raise PhaseTwoBoundaryNotReady("Root Task is missing")
+        portfolio_tasks = [
+            item
+            for item in ledger.tasks.values()
+            if task_kind(item.current_spec.context_refs) == ScriptTaskKind.CANDIDATE_PORTFOLIO.value
+        ]
+        if len(portfolio_tasks) != 1:
+            raise PhaseTwoBoundaryNotReady("exactly one CandidatePortfolio Task is required")
+        task = portfolio_tasks[0]
+        if task.parent_task_id != root.task_id or task.status is not TaskStatus.COMPLETED:
+            raise PhaseTwoBoundaryNotReady(
+                "CandidatePortfolio must be one completed direct child of Root"
+            )
+        accepted = [
+            ledger.decisions[item]
+            for item in task.decision_ids
+            if ledger.decisions[item].action is DecisionAction.ACCEPT
+        ]
+        if (
+            len(accepted) != 1
+            or not task.decision_ids
+            or task.decision_ids[-1] != accepted[0].decision_id
+        ):
+            raise PhaseTwoBoundaryNotReady(
+                "CandidatePortfolio requires one current ACCEPT decision"
+            )
+        decision = accepted[0]
+        if decision.attempt_id is None:
+            raise PhaseTwoBoundaryNotReady("CandidatePortfolio ACCEPT is missing its Attempt")
+        attempt = ledger.attempts.get(decision.attempt_id)
+        if attempt is None:
+            raise PhaseTwoBoundaryNotReady("CandidatePortfolio Attempt is missing")
+        if not all_descendants_terminal(ledger, task.task_id):
+            raise PhaseTwoBoundaryNotReady("CandidatePortfolio subtree is not terminal")
+        contract = await self._accepted_inputs.contract_for_task(
+            root_trace_id=root_trace_id,
+            task=task,
+            spec_version=attempt.spec_version,
+        )
+        if contract.task_kind is not ScriptTaskKind.CANDIDATE_PORTFOLIO:
+            raise PhaseTwoBoundaryNotReady("CandidatePortfolio contract kind is invalid")
+        contract.require_execution_ready()
+        bundle = await self._accepted_inputs.resolve(
+            root_trace_id=root_trace_id,
+            script_build_id=script_build_id,
+            task_id=task.task_id,
+            attempt_id=attempt.attempt_id,
+            contract=contract,
+            input_snapshot_id=input_snapshot_id,
+        )
+        resolved_portfolio = await self._accepted_inputs.resolve_decision(
+            ledger=ledger,
+            root_trace_id=root_trace_id,
+            script_build_id=script_build_id,
+            decision_id=decision.decision_id,
+            input_snapshot_id=input_snapshot_id,
+            expected=None,
+            source="boundary",
+        )
+        version = resolved_portfolio.artifact_version
+        if version.artifact_type is not ArtifactKind.CANDIDATE_PORTFOLIO or not isinstance(
+            version.artifact, CandidatePortfolioArtifactV1
+        ):
+            raise PhaseTwoBoundaryNotReady("accepted portfolio Artifact has the wrong kind")
+        portfolio = version.artifact
+        if portfolio.input_closure_digest != bundle.input_closure_digest:
+            raise PhaseTwoBoundaryNotReady("portfolio input closure digest does not match")
+
+        closure_ids = {item.decision_id for item in contract.candidate_closure_decision_refs}
+        dispositions = (
+            set(portfolio.accepted_decision_ids)
+            | set(portfolio.superseded_decision_ids)
+            | set(portfolio.rejected_or_held_decision_ids)
+        )
+        if closure_ids != dispositions:
+            raise PhaseTwoBoundaryNotReady(
+                "portfolio does not dispose its complete candidate closure"
+            )
+        if set(portfolio.accepted_decision_ids) != set(contract.adopted_decision_ids):
+            raise PhaseTwoBoundaryNotReady("portfolio adoption differs from its execution contract")
+        if tuple(portfolio.compose_order) != tuple(contract.compose_order):
+            raise PhaseTwoBoundaryNotReady("portfolio compose order differs from its contract")
+        if any(item.blocks_acceptance for item in portfolio.unresolved_defects):
+            raise PhaseTwoBoundaryNotReady("portfolio retains a blocking defect")
+
+        candidates = [
+            await self._read_structured_script(script_build_id, uri)
+            for uri in portfolio.candidate_structured_script_refs
+        ]
+        adopted = [
+            item
+            for item in candidates
+            if _artifact_uri(item.artifact_version_id) == portfolio.adopted_structured_script_ref
+        ]
+        if len(adopted) != 1:
+            raise PhaseTwoBoundaryNotReady("portfolio must adopt exactly one StructuredScript")
+        adopted_decision_refs = {
+            item.artifact_ref.uri
+            for item in bundle.inputs
+            if item.decision_id in portfolio.accepted_decision_ids
+        }
+        if portfolio.adopted_structured_script_ref not in adopted_decision_refs:
+            raise PhaseTwoBoundaryNotReady(
+                "adopted StructuredScript is not backed by an adopted ACCEPT decision"
+            )
+        if any(item.status in _ACTIVE_OPERATIONS for item in ledger.operations.values()):
+            raise PhaseTwoBoundaryNotReady("active or stopping Operations prevent checkpoint")
+
+    async def _owned_ledger(
+        self, *, script_build_id: int, root_trace_id: str, input_snapshot_id: str
+    ) -> Any:
+        binding = await self._bindings.get_by_build(script_build_id)
+        if binding.root_trace_id != root_trace_id or str(binding.input_snapshot_id) != str(
+            input_snapshot_id
+        ):
+            raise PhaseTwoBoundaryNotReady("binding, Root or InputSnapshot ownership differs")
+        ledger = await self._task_store.load(root_trace_id)
+        if ledger.root_trace_id != root_trace_id:
+            raise PhaseTwoBoundaryNotReady("Ledger belongs to another Root")
+        return ledger
+
+    async def _read_structured_script(self, script_build_id: int, uri: str) -> Any:
+        identifier = uri.removeprefix("script-build://artifact-versions/")
+        if not identifier.isdigit() or _artifact_uri(int(identifier)) != uri:
+            raise PhaseTwoBoundaryNotReady("candidate StructuredScript ref is not immutable")
+        version = await self._artifacts.get_by_id(int(identifier), script_build_id=script_build_id)
+        if (
+            version.state is not ArtifactState.FROZEN
+            or version.artifact_type is not ArtifactKind.STRUCTURED_SCRIPT
+            or not isinstance(version.artifact, StructuredScriptArtifactV1)
+            or version.canonical_sha256 != version.artifact.canonical_sha256
+        ):
+            raise PhaseTwoBoundaryNotReady("candidate StructuredScript is not closed")
+        return version
+
+
+def _artifact_uri(identifier: int) -> str:
+    return f"script-build://artifact-versions/{identifier}"
+
+
+__all__ = ["ScriptPhaseTwoBoundaryVerifier"]

script_build_host/src/script_build_host/repositories/legacy_state.py

@@ -54,6 +54,8 @@ class SqlAlchemyLegacyBuildStateRepository:
         if status is BuildStatus.SUCCESS:
             raise ProtocolViolation("phase one cannot project script build success")
         values: dict[str, Any] = {"status": status.value}
+        if status in {BuildStatus.RUNNING, BuildStatus.STOPPING}:
+            values.update(end_time=None, error_message=None, summary=None)
         if error_summary is not None:
             values["error_message"] = redact_text(error_summary)
         if status in {
@@ -98,3 +100,43 @@ class SqlAlchemyLegacyBuildStateRepository:
             )
             if cast(CursorResult[Any], result).rowcount != 1:
                 raise BuildNotFound()
+
+    async def get_direction(self, script_build_id: int) -> str | None:
+        async with self._sessions() as session:
+            value = await session.scalar(
+                select(script_build_record.c.script_direction).where(
+                    script_build_record.c.id == script_build_id,
+                    script_build_record.c.is_deleted.is_(False),
+                )
+            )
+        return str(value) if value is not None else None
+
+    async def set_checkpoint(
+        self,
+        script_build_id: int,
+        *,
+        checkpoint_code: str,
+        summary: str,
+        root_trace_id: str,
+    ) -> None:
+        values = {
+            "status": BuildStatus.PARTIAL.value,
+            "error_message": redact_text(checkpoint_code)[:255],
+            "summary": redact_text(summary)[:2000],
+            "reson_trace_id": redact_text(root_trace_id)[:200],
+            "end_time": datetime.now(UTC),
+        }
+        async with self._sessions() as session, session.begin():
+            result = await session.execute(
+                update(script_build_record)
+                .where(
+                    script_build_record.c.id == script_build_id,
+                    script_build_record.c.is_deleted.is_(False),
+                )
+                .values(**values)
+            )
+            if cast(CursorResult[Any], result).rowcount != 1:
+                raise BuildNotFound()
+
+    async def begin_phase(self, script_build_id: int) -> None:
+        await self.set_status(script_build_id, BuildStatus.RUNNING)