feat(Mission恢复): 支持 Phase1 与 Phase2 沿原 Trace 续跑

接管 OwnerLease 后先归一化持久化 Operation，再依据任务树判断当前阶段。恢复时验证 Phase2 策略迁移与冻结输入，向同一 Planner Trace 追加 recovery 消息，保留既有完成 Attempt；同时提供调试检查点暂停能力。

script_build_host/src/script_build_host/application/mission_service.py

@@ -7,6 +7,7 @@ from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from typing import Any, Protocol
 from uuid import uuid4
 
@@ -20,6 +21,7 @@ from script_build_host.domain.artifacts import (
 from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import (
     DirectionProjectionConflict,
+    MissionAlreadyOwned,
     MissionRecoveryRequired,
     PhaseTwoBoundaryNotReady,
     PhaseTwoPromptsNotFrozen,
@@ -42,6 +44,7 @@ from script_build_host.domain.records import (
     PublicationType,
 )
 from script_build_host.domain.redaction import redact
+from script_build_host.domain.task_contracts import ScriptTaskKind
 from script_build_host.domain.task_refs import task_kind
 
 from .input_snapshot_service import AssembleInputRequest, ScriptInputSnapshotService
@@ -50,6 +53,9 @@ from .mission_factory import ScriptMissionFactory
 PHASE_ONE_CAPABILITY_BOUNDARY = "PHASE_ONE_CAPABILITY_BOUNDARY"
 PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
 _FENCED_BUILD_ID: ContextVar[int | None] = ContextVar("script_build_fenced_build", default=None)
+_ACTIVE_OPERATION_STATUSES = frozenset(
+    {OperationStatus.PENDING, OperationStatus.RUNNING, OperationStatus.STOP_REQUESTED}
+)
 
 
 class BuildTransitionGate:
@@ -1115,6 +1121,148 @@ class ScriptMissionService:
             return
         await begin(script_build_id)
 
+    async def resume_planner(self, script_build_id: int) -> int:
+        """Resume Phase1/2 on the existing Trace after durable operation recovery."""
+
+        lease = None
+        if self.owner_lease_factory is not None:
+            lease = self.owner_lease_factory(script_build_id)
+            token = await lease.acquire(script_build_id)
+            self.register_owner_token(token)
+        try:
+            return await self._resume_planner_owned(script_build_id)
+        finally:
+            if lease is not None:
+                token = self._owner_tokens.get(script_build_id)
+                if token is not None:
+                    self.unregister_owner_token(token)
+                await lease.release()
+
+    async def _resume_planner_owned(self, script_build_id: int) -> int:
+        """Resume after the caller has acquired the mission owner epoch."""
+
+        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}:
+            raise MissionRecoveryRequired("a stopped build cannot resume")
+        snapshot = await self.input_snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=script_build_id
+        )
+        await self.coordinator.recover_operations(binding.root_trace_id)
+        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+        if any(item.status in _ACTIVE_OPERATION_STATUSES for item in ledger.operations.values()):
+            raise MissionRecoveryRequired("active Operations did not normalize")
+        phase = (
+            2
+            if any(
+                task_kind(item.current_spec.context_refs)
+                == ScriptTaskKind.CANDIDATE_PORTFOLIO.value
+                for item in ledger.tasks.values()
+            )
+            else 1
+        )
+        if phase == 2:
+            policy = self.factory.build_phase_two_policy(snapshot)
+            continuation = self.factory.build_phase_two_message(
+                binding,
+                snapshot,
+                direction_artifact_version_id=(
+                    binding.active_direction_artifact_version_id
+                    or await self.direction_reconciler.reconcile(
+                        script_build_id, binding.root_trace_id
+                    )
+                ),
+            )
+            migrations = [
+                item
+                for item in await self.runner.trace_store.get_events(binding.root_trace_id, 0)
+                if item.get("event") == "planner_policy_migrated"
+                or item.get("event_type") == "planner_policy_migrated"
+            ]
+            if len(migrations) != 1:
+                raise PlannerPolicyMigrationRequired(
+                    "phase-two recovery 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,
+            )
+            config = self.factory.build_phase_two_run_config(binding, snapshot)
+        else:
+            config = self.factory.build_phase_one_resume_run_config(binding, snapshot)
+        await self._begin_phase(script_build_id)
+        await self.runner.run_result(
+            messages=[
+                {
+                    "role": "user",
+                    "content": self.factory.build_recovery_message(
+                        phase=phase, script_build_id=script_build_id
+                    ),
+                }
+            ],
+            config=config,
+        )
+        completion = await self.coordinator.root_completion(binding.root_trace_id)
+        expected = (
+            PHASE_TWO_CANDIDATE_PORTFOLIO_READY if phase == 2 else PHASE_ONE_CAPABILITY_BOUNDARY
+        )
+        if (
+            completion.get("status") != TaskStatus.BLOCKED.value
+            or completion.get("blocked_reason") != expected
+        ):
+            raise ProtocolViolation("recovered Planner did not stop at its phase boundary")
+        await self._set_checkpoint(
+            script_build_id,
+            checkpoint_code=expected,
+            summary=f"Phase {phase} recovered from its durable ledger.",
+            root_trace_id=binding.root_trace_id,
+        )
+        return phase
+
+    async def pause_at_debug_checkpoint(
+        self, script_build_id: int, *, checkpoint_code: str
+    ) -> None:
+        """Pause an internal test run without creating terminal stop intent."""
+
+        binding = await self.bindings.get_by_build(script_build_id)
+        ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+        for operation in ledger.operations.values():
+            if operation.status in _ACTIVE_OPERATION_STATUSES:
+                await self.coordinator.stop_operation(
+                    binding.root_trace_id,
+                    operation.operation_id,
+                    idempotency_key=f"debug-pause:{script_build_id}:{operation.operation_id}",
+                )
+        await self.runner.stop(binding.root_trace_id)
+        task = self._runs.get(script_build_id)
+        if task is not None and not task.done():
+            try:
+                await asyncio.wait_for(asyncio.shield(task), timeout=1.0)
+            except TimeoutError:
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+        await self.coordinator.recover_operations(binding.root_trace_id)
+        await self.runner.trace_store.update_trace(
+            binding.root_trace_id, status="stopped", completed_at=datetime.now(UTC)
+        )
+        await self._set_checkpoint(
+            script_build_id,
+            checkpoint_code=checkpoint_code,
+            summary="Internal debug checkpoint; safe to resume with a new OwnerLease.",
+            root_trace_id=binding.root_trace_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):
@@ -1156,6 +1304,12 @@ class ScriptMissionService:
             )
         await self.runner.stop(binding.root_trace_id)
         run_task = self._runs.get(script_build_id)
+        if run_task is None and await self._converge_orphaned_stop(script_build_id, binding):
+            return StopResult(
+                script_build_id,
+                BuildStatus.STOPPED,
+                tuple(item.operation_id for item in active),
+            )
         try:
             await asyncio.wait_for(
                 self._wait_stopped(
@@ -1186,6 +1340,45 @@ class ScriptMissionService:
             tuple(item.operation_id for item in active),
         )
 
+    async def _converge_orphaned_stop(self, script_build_id: int, binding: Any) -> bool:
+        """Take over a dead process and close its durable stop state."""
+
+        if self.owner_lease_factory is None or self.fenced_command_gate is None:
+            return False
+        lease = self.owner_lease_factory(script_build_id)
+        try:
+            token = await lease.acquire(script_build_id)
+        except MissionAlreadyOwned:
+            return False
+        try:
+            await self.coordinator.recover_operations(binding.root_trace_id)
+            ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+            if any(
+                item.status in _ACTIVE_OPERATION_STATUSES for item in ledger.operations.values()
+            ):
+                return False
+            await self.fenced_command_gate.verify_stop_owner(token)
+            trace = await self.runner.trace_store.get_trace(binding.root_trace_id)
+            if trace is not None and trace.status not in {"completed", "failed", "stopped"}:
+                await self.runner.trace_store.update_trace(
+                    binding.root_trace_id,
+                    status="stopped",
+                    completed_at=datetime.now(UTC),
+                )
+                await self.runner.trace_store.append_event(
+                    binding.root_trace_id,
+                    "host_stop_converged",
+                    {
+                        "owner_epoch": token.owner_epoch,
+                        "stop_epoch": token.stop_epoch,
+                    },
+                )
+            await self.fenced_command_gate.verify_stop_owner(token)
+            await self.fenced_command_gate.complete_stop(token)
+            return True
+        finally:
+            await lease.release()
+
     async def ensure_dispatch_allowed(self, script_build_id: int) -> None:
         if _FENCED_BUILD_ID.get() == script_build_id:
             return