v2r2c4 S1 successor: land autonomous resume controller/adapter/integration + tests

active-runtime/app/task_plane/autonomous_resume_aios_adapter.py

@@ -0,0 +1,972 @@
+"""AIOS-native binding layer for the unattended resume controller.
+
+This module intentionally contains no HTTP client, Unix-socket client, database,
+systemd integration, scheduler loop, secret handling, or privileged identity
+mutation.
+
+It converts existing AIOS authorities into the generic callbacks defined by
+``autonomous_resume_integration``. Field code supplies the current authoritative
+implementations for:
+
+- durable Owner authorization/grant projection;
+- canonical governed-worker activation/status;
+- governed task admission;
+- task/claim/TaskLease/TaskPrincipal/WriterLease/CAS ownership;
+- bounded source-lane execution and integration;
+- Gate0;
+- ReleaseReady/deployability;
+- PromotionController;
+- final production acceptance;
+- mission resume lease/checkpoint/effect reconciliation.
+
+The adapter is deterministic and fail-closed. It never fabricates any credential,
+principal, lease, fence, task, commit, receipt, approval, or deployment result.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+import hashlib
+import json
+from typing import Mapping, Protocol, Sequence
+
+from .autonomous_resume_controller import (
+    EffectState,
+    MissionCheckpoint,
+    MissionObservation,
+    MissionPhase,
+    ResumeLease,
+    StepResult,
+    StepStatus,
+)
+from .autonomous_resume_integration import (
+    AuthorityCallbacks,
+    CallableAutonomousExecutionPort,
+    PhaseCallbacks,
+)
+
+
+def _canonical(value) -> bytes:
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def _digest(value) -> str:
+    return hashlib.sha256(_canonical(value)).hexdigest()
+
+
+class WorkerState(str, Enum):
+    ACTIVE = "ACTIVE"
+    STARTING = "STARTING"
+    INACTIVE = "INACTIVE"
+    BLOCKED = "BLOCKED"
+    UNKNOWN = "UNKNOWN"
+
+
+class WorkState(str, Enum):
+    PENDING = "PENDING"
+    RUNNING = "RUNNING"
+    COMPLETE = "COMPLETE"
+    BLOCKED = "BLOCKED"
+    UNKNOWN = "UNKNOWN"
+
+
+class GateState(str, Enum):
+    PASS = "PASS"
+    HOLD = "HOLD"
+    UNKNOWN = "UNKNOWN"
+
+
+class ReadyState(str, Enum):
+    READY = "READY"
+    NOT_READY = "NOT_READY"
+    UNKNOWN = "UNKNOWN"
+
+
+class PromotionState(str, Enum):
+    IN_PROGRESS = "IN_PROGRESS"
+    VERIFIED = "VERIFIED"
+    ROLLED_BACK = "ROLLED_BACK"
+    BLOCKED = "BLOCKED"
+    UNKNOWN = "UNKNOWN"
+
+
+class AcceptanceState(str, Enum):
+    PASS = "PASS"
+    FAIL_ROLLED_BACK = "FAIL_ROLLED_BACK"
+    BLOCKED = "BLOCKED"
+    UNKNOWN = "UNKNOWN"
+
+
+@dataclass(frozen=True)
+class DurableGrantView:
+    grant_id: str
+    active: bool
+    repository: str
+    allowed_target_prefixes: tuple[str, ...]
+    max_task_risk: str
+    expires_at: float | None = None
+    owner_action_required: bool = False
+    evidence_refs: tuple[str, ...] = ()
+
+    def permits(self, *, repository: str, branch: str, risk: str) -> bool:
+        if not self.active:
+            return False
+        if self.repository != repository:
+            return False
+        if not any(branch.startswith(prefix) for prefix in self.allowed_target_prefixes):
+            return False
+        rank = {"low": 0, "medium": 1, "engineering": 2, "high": 3}
+        if risk not in rank or self.max_task_risk not in rank:
+            return False
+        return rank[risk] <= rank[self.max_task_risk]
+
+
+@dataclass(frozen=True)
+class TaskLaneSpec:
+    lane: str
+    task_id: str
+    repository: str
+    target_branch: str
+    risk: str
+    description: str
+    write_set: tuple[str, ...]
+
+    def __post_init__(self) -> None:
+        if not self.lane or not self.task_id or not self.repository:
+            raise ValueError("lane_task_repository_required")
+        if not self.target_branch.startswith("agent/"):
+            raise ValueError("target_branch_must_use_agent_prefix")
+        if not self.write_set:
+            raise ValueError("write_set_required")
+
+
+@dataclass(frozen=True)
+class TaskAdmissionRequest:
+    task_id: str
+    idempotency_key: str
+    repository: str
+    base_oid: str
+    target_branch: str
+    expected_branch_oid: str
+    risk: str
+    description: str
+
+
+@dataclass(frozen=True)
+class TaskAdmissionReceipt:
+    task_id: str
+    generation: int
+    target_branch: str
+    persisted: bool
+    claimable: bool
+    receipt_ref: str
+
+    def __post_init__(self) -> None:
+        if self.generation < 1:
+            raise ValueError("task_generation_must_be_positive")
+        if not self.receipt_ref:
+            raise ValueError("task_admission_receipt_required")
+
+
+@dataclass(frozen=True)
+class WorkerActivationReceipt:
+    state: WorkerState
+    receipt_ref: str | None = None
+    retry_after_seconds: int | None = None
+    owner_action_required: bool = False
+    code: str = "WORKER_STATE"
+
+
+@dataclass(frozen=True)
+class LaneExecutionView:
+    lane: str
+    state: WorkState
+    condition_fingerprint: str
+    receipt_refs: tuple[str, ...] = ()
+    retry_after_seconds: int | None = None
+    code: str = "LANE_STATE"
+
+
+@dataclass(frozen=True)
+class IntegrationView:
+    state: WorkState
+    condition_fingerprint: str
+    receipt_refs: tuple[str, ...] = ()
+    retry_after_seconds: int | None = None
+    code: str = "INTEGRATION_STATE"
+
+
+@dataclass(frozen=True)
+class GateView:
+    state: GateState
+    condition_fingerprint: str
+    receipt_refs: tuple[str, ...] = ()
+    retry_after_seconds: int | None = None
+    code: str = "GATE_STATE"
+
+
+@dataclass(frozen=True)
+class ReleaseReadyView:
+    state: ReadyState
+    condition_fingerprint: str
+    material_digest: str | None
+    receipt_refs: tuple[str, ...] = ()
+    retry_after_seconds: int | None = None
+    code: str = "READY_STATE"
+
+
+@dataclass(frozen=True)
+class PromotionView:
+    state: PromotionState
+    condition_fingerprint: str
+    material_digest: str | None
+    receipt_refs: tuple[str, ...] = ()
+    retry_after_seconds: int | None = None
+    code: str = "PROMOTION_STATE"
+
+
+@dataclass(frozen=True)
+class AcceptanceView:
+    state: AcceptanceState
+    condition_fingerprint: str
+    receipt_refs: tuple[str, ...] = ()
+    code: str = "ACCEPTANCE_STATE"
+
+
+@dataclass(frozen=True)
+class MissionEffectView:
+    effect_state: EffectState
+    condition_fingerprint: str
+    requires_reconcile: bool
+    receipt_refs: tuple[str, ...] = ()
+
+
+class AiosMissionAuthorities(Protocol):
+    """Existing AIOS authority surface consumed by this adapter."""
+
+    # Live truth.
+    def read_head_oid(self, repository: str) -> str: ...
+    def read_material_digest(self, mission_id: str) -> str: ...
+    def read_generation(self, mission_id: str) -> int: ...
+    def read_effect_view(self, mission_id: str) -> MissionEffectView: ...
+    def owner_action_resolved(self, mission_id: str) -> bool: ...
+
+    # Durable mission state / one-winner resume authority.
+    def load_checkpoint(self, mission_id: str) -> MissionCheckpoint | None: ...
+    def acquire_resume_lease(
+        self,
+        mission_id: str,
+        generation: int,
+        idempotency_key: str,
+        lease_seconds: int,
+    ) -> ResumeLease | None: ...
+    def persist_checkpoint(
+        self,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> None: ...
+    def persist_owner_action_required(
+        self,
+        payload: Mapping[str, object],
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> str: ...
+    def release_resume_lease(
+        self,
+        lease: ResumeLease,
+        outcome_code: str,
+    ) -> None: ...
+    def reconcile_effects(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult: ...
+
+    # Admission.
+    def read_durable_grant(self, mission_id: str) -> DurableGrantView: ...
+    def ensure_canonical_worker(
+        self,
+        mission_id: str,
+        generation: int,
+        idempotency_key: str,
+    ) -> WorkerActivationReceipt: ...
+    def create_or_resume_task(
+        self,
+        request: TaskAdmissionRequest,
+    ) -> TaskAdmissionReceipt: ...
+    def read_task_admission(
+        self,
+        lane: TaskLaneSpec,
+    ) -> TaskAdmissionReceipt | None: ...
+
+    # Phase execution.
+    def read_lane_execution(
+        self,
+        lane: TaskLaneSpec,
+        task: TaskAdmissionReceipt,
+    ) -> LaneExecutionView: ...
+    def drive_integration(
+        self,
+        lanes: Sequence[TaskLaneSpec],
+        tasks: Sequence[TaskAdmissionReceipt],
+    ) -> IntegrationView: ...
+    def run_gate0(self, mission_id: str) -> GateView: ...
+    def build_release_ready(self, mission_id: str) -> ReleaseReadyView: ...
+    def run_promotion(self, mission_id: str) -> PromotionView: ...
+    def run_final_acceptance(self, mission_id: str) -> AcceptanceView: ...
+
+
+@dataclass(frozen=True)
+class AiosUnattendedMissionConfig:
+    mission_id: str
+    repository: str
+    lanes: tuple[TaskLaneSpec, ...]
+    admission_risk: str = "medium"
+
+    def __post_init__(self) -> None:
+        if not self.mission_id or not self.repository:
+            raise ValueError("mission_and_repository_required")
+        if not self.lanes:
+            raise ValueError("at_least_one_lane_required")
+        names = [lane.lane for lane in self.lanes]
+        task_ids = [lane.task_id for lane in self.lanes]
+        branches = [lane.target_branch for lane in self.lanes]
+        if len(names) != len(set(names)):
+            raise ValueError("duplicate_lane")
+        if len(task_ids) != len(set(task_ids)):
+            raise ValueError("duplicate_task_id")
+        if len(branches) != len(set(branches)):
+            raise ValueError("duplicate_target_branch")
+        owners: dict[str, str] = {}
+        for lane in self.lanes:
+            if lane.repository != self.repository:
+                raise ValueError("lane_repository_mismatch")
+            for resource in lane.write_set:
+                if resource in owners:
+                    raise ValueError(
+                        f"overlapping_lane_resource:{resource}:"
+                        f"{owners[resource]}:{lane.lane}"
+                    )
+                owners[resource] = lane.lane
+
+
+class AiosUnattendedCallbackFactory:
+    """Construct generic controller callbacks from current AIOS authorities."""
+
+    def __init__(
+        self,
+        authority: AiosMissionAuthorities,
+        config: AiosUnattendedMissionConfig,
+    ) -> None:
+        self.authority = authority
+        self.config = config
+        self._admissions: dict[str, TaskAdmissionReceipt] = {}
+
+    def build_port(self) -> CallableAutonomousExecutionPort:
+        return CallableAutonomousExecutionPort(
+            AuthorityCallbacks(
+                read_observation=self.read_observation,
+                load_checkpoint=self.authority.load_checkpoint,
+                acquire_resume_lease=self.authority.acquire_resume_lease,
+                persist_checkpoint=self.authority.persist_checkpoint,
+                persist_owner_action_required=self.authority.persist_owner_action_required,
+                release_resume_lease=self.authority.release_resume_lease,
+                reconcile=self.authority.reconcile_effects,
+            ),
+            PhaseCallbacks(
+                admission=self.admission,
+                source_execution=self.source_execution,
+                gate0=self.gate0,
+                release_ready=self.release_ready,
+                promotion=self.promotion,
+                final_acceptance=self.final_acceptance,
+            ),
+        )
+
+    def read_observation(self, mission_id: str) -> MissionObservation:
+        if mission_id != self.config.mission_id:
+            raise ValueError("mission_id_mismatch")
+        generation = self.authority.read_generation(mission_id)
+        head_oid = self.authority.read_head_oid(self.config.repository)
+        material = self.authority.read_material_digest(mission_id)
+        effect = self.authority.read_effect_view(mission_id)
+        fingerprint = _digest({
+            "schema": "aios.unattended-condition.v1",
+            "mission_id": mission_id,
+            "generation": generation,
+            "head_oid": head_oid,
+            "material_digest": material,
+            "effect_state": effect.effect_state.value,
+            "effect_condition": effect.condition_fingerprint,
+            "requires_reconcile": effect.requires_reconcile,
+        })
+        return MissionObservation(
+            mission_id=mission_id,
+            generation=generation,
+            head_oid=head_oid,
+            material_digest=material,
+            condition_fingerprint=fingerprint,
+            effect_state=effect.effect_state,
+            requires_reconcile=effect.requires_reconcile,
+            owner_action_resolved=self.authority.owner_action_resolved(mission_id),
+            evidence_refs=effect.receipt_refs,
+        )
+
+    def admission(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        grant = self.authority.read_durable_grant(observation.mission_id)
+        grant_refs = tuple(grant.evidence_refs)
+
+        if not grant.active:
+            return StepResult(
+                StepStatus.BLOCKED_OWNER if grant.owner_action_required
+                else StepStatus.BLOCKED_DETERMINISTIC,
+                "DURABLE_GRANT_NOT_ACTIVE",
+                self._phase_fingerprint(
+                    "admission", observation, {"grant_id": grant.grant_id, "active": False}
+                ),
+                receipt_refs=grant_refs,
+                owner_action_payload=(
+                    {
+                        "boundary": "DURABLE_GRANT_NOT_ACTIVE",
+                        "grant_id": grant.grant_id,
+                    }
+                    if grant.owner_action_required else None
+                ),
+            )
+
+        for lane in self.config.lanes:
+            if not grant.permits(
+                repository=lane.repository,
+                branch=lane.target_branch,
+                risk=lane.risk,
+            ):
+                return StepResult(
+                    StepStatus.BLOCKED_OWNER if grant.owner_action_required
+                    else StepStatus.BLOCKED_DETERMINISTIC,
+                    "DURABLE_GRANT_SCOPE_MISMATCH",
+                    self._phase_fingerprint(
+                        "admission",
+                        observation,
+                        {
+                            "grant_id": grant.grant_id,
+                            "lane": lane.lane,
+                            "branch": lane.target_branch,
+                            "risk": lane.risk,
+                        },
+                    ),
+                    receipt_refs=grant_refs,
+                    owner_action_payload=(
+                        {
+                            "boundary": "DURABLE_GRANT_SCOPE_MISMATCH",
+                            "grant_id": grant.grant_id,
+                            "lane": lane.lane,
+                        }
+                        if grant.owner_action_required else None
+                    ),
+                )
+
+        worker = self.authority.ensure_canonical_worker(
+            observation.mission_id,
+            observation.generation,
+            self._idempotency(
+                "worker",
+                observation.mission_id,
+                observation.generation,
+            ),
+        )
+        worker_refs = self._refs(grant_refs, (worker.receipt_ref or "",))
+
+        if worker.state is WorkerState.UNKNOWN:
+            return StepResult(
+                StepStatus.RECONCILE,
+                worker.code,
+                self._phase_fingerprint(
+                    "worker", observation, {"state": worker.state.value}
+                ),
+                receipt_refs=worker_refs,
+            )
+        if worker.state is WorkerState.BLOCKED:
+            return StepResult(
+                StepStatus.BLOCKED_OWNER if worker.owner_action_required
+                else StepStatus.BLOCKED_DETERMINISTIC,
+                worker.code,
+                self._phase_fingerprint(
+                    "worker", observation, {"state": worker.state.value}
+                ),
+                receipt_refs=worker_refs,
+                owner_action_payload=(
+                    {"boundary": worker.code}
+                    if worker.owner_action_required else None
+                ),
+            )
+        if worker.state in {WorkerState.STARTING, WorkerState.INACTIVE}:
+            return StepResult(
+                StepStatus.IN_PROGRESS,
+                worker.code,
+                self._phase_fingerprint(
+                    "worker", observation, {"state": worker.state.value}
+                ),
+                receipt_refs=worker_refs,
+                retry_after_seconds=worker.retry_after_seconds,
+            )
+        if worker.state is WorkerState.ACTIVE and not worker.receipt_ref:
+            return StepResult(
+                StepStatus.RECONCILE,
+                "WORKER_ACTIVE_WITHOUT_RECEIPT",
+                self._phase_fingerprint(
+                    "worker", observation, {"state": worker.state.value}
+                ),
+                receipt_refs=worker_refs,
+            )
+
+        task_refs: list[str] = list(worker_refs)
+        admissions: dict[str, TaskAdmissionReceipt] = {}
+        for lane in self.config.lanes:
+            request = TaskAdmissionRequest(
+                task_id=lane.task_id,
+                idempotency_key=self._idempotency(
+                    "task",
+                    observation.generation,
+                    lane.task_id,
+                    observation.head_oid,
+                ),
+                repository=lane.repository,
+                base_oid=observation.head_oid,
+                target_branch=lane.target_branch,
+                expected_branch_oid=observation.head_oid,
+                risk=lane.risk,
+                description=lane.description,
+            )
+            receipt = self.authority.create_or_resume_task(request)
+            task_refs.append(receipt.receipt_ref)
+            receipt_error = self._admission_receipt_error(
+                lane,
+                receipt,
+                observation=observation,
+            )
+            if receipt_error is not None:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    receipt_error,
+                    self._phase_fingerprint(
+                        "task",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(task_refs),
+                )
+            if not receipt.persisted:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "TASK_PERSISTENCE_UNKNOWN",
+                    self._phase_fingerprint(
+                        "task",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(task_refs),
+                )
+            if not receipt.claimable:
+                return StepResult(
+                    StepStatus.IN_PROGRESS,
+                    "TASK_NOT_CLAIMABLE_YET",
+                    self._phase_fingerprint(
+                        "task",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(task_refs),
+                )
+            admissions[lane.lane] = receipt
+
+        self._admissions = admissions
+        return StepResult(
+            StepStatus.SUCCEEDED,
+            "GOVERNED_TASKS_ADMITTED",
+            self._phase_fingerprint(
+                "admission",
+                observation,
+                {"tasks": sorted(r.task_id for r in admissions.values())},
+            ),
+            receipt_refs=tuple(task_refs),
+            next_phase=MissionPhase.S2_EXECUTION,
+        )
+
+    def source_execution(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        recovery_refs: list[str] = []
+        for lane in self.config.lanes:
+            if lane.lane in self._admissions:
+                continue
+            receipt = self.authority.read_task_admission(lane)
+            if receipt is None:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "TASK_ADMISSION_NOT_RESOLVED",
+                    self._phase_fingerprint(
+                        "source",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(recovery_refs),
+                )
+            recovery_refs.append(receipt.receipt_ref)
+            receipt_error = self._admission_receipt_error(
+                lane,
+                receipt,
+                observation=observation,
+            )
+            if receipt_error is not None:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    receipt_error,
+                    self._phase_fingerprint(
+                        "source",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(recovery_refs),
+                )
+            if not receipt.persisted:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "TASK_PERSISTENCE_UNKNOWN",
+                    self._phase_fingerprint(
+                        "source",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(recovery_refs),
+                )
+            if not receipt.claimable:
+                return StepResult(
+                    StepStatus.IN_PROGRESS,
+                    "TASK_NOT_CLAIMABLE_YET",
+                    self._phase_fingerprint(
+                        "source",
+                        observation,
+                        {"lane": lane.lane, "task_id": lane.task_id},
+                    ),
+                    receipt_refs=tuple(recovery_refs),
+                )
+            self._admissions[lane.lane] = receipt
+
+        views: list[LaneExecutionView] = []
+        refs: list[str] = list(recovery_refs)
+        for lane in self.config.lanes:
+            view = self.authority.read_lane_execution(
+                lane,
+                self._admissions[lane.lane],
+            )
+            if view.lane != lane.lane:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "LANE_IDENTITY_MISMATCH",
+                    self._phase_fingerprint(
+                        "source",
+                        observation,
+                        {"expected": lane.lane, "actual": view.lane},
+                    ),
+                    receipt_refs=tuple(refs),
+                )
+            if view.state is WorkState.COMPLETE and not view.receipt_refs:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "LANE_COMPLETE_WITHOUT_RECEIPT",
+                    view.condition_fingerprint,
+                    receipt_refs=tuple(refs),
+                )
+            views.append(view)
+            refs.extend(view.receipt_refs)
+
+        unknown = [v for v in views if v.state is WorkState.UNKNOWN]
+        if unknown:
+            return StepResult(
+                StepStatus.RECONCILE,
+                unknown[0].code,
+                self._phase_fingerprint(
+                    "source",
+                    observation,
+                    {"states": [(v.lane, v.state.value) for v in views]},
+                ),
+                receipt_refs=tuple(refs),
+            )
+
+        blocked = [v for v in views if v.state is WorkState.BLOCKED]
+        if blocked:
+            return StepResult(
+                StepStatus.BLOCKED_DETERMINISTIC,
+                blocked[0].code,
+                self._phase_fingerprint(
+                    "source",
+                    observation,
+                    {"states": [(v.lane, v.state.value) for v in views]},
+                ),
+                receipt_refs=tuple(refs),
+            )
+
+        pending = [
+            v for v in views
+            if v.state in {WorkState.PENDING, WorkState.RUNNING}
+        ]
+        if pending:
+            retry_values = [
+                v.retry_after_seconds
+                for v in pending
+                if v.retry_after_seconds is not None
+            ]
+            return StepResult(
+                StepStatus.IN_PROGRESS,
+                "SOURCE_LANES_IN_PROGRESS",
+                self._phase_fingerprint(
+                    "source",
+                    observation,
+                    {"states": [(v.lane, v.state.value) for v in views]},
+                ),
+                receipt_refs=tuple(refs),
+                retry_after_seconds=min(retry_values) if retry_values else None,
+            )
+
+        integration = self.authority.drive_integration(
+            self.config.lanes,
+            tuple(self._admissions[lane.lane] for lane in self.config.lanes),
+        )
+        refs.extend(integration.receipt_refs)
+        if integration.state is WorkState.UNKNOWN:
+            return StepResult(
+                StepStatus.RECONCILE,
+                integration.code,
+                integration.condition_fingerprint,
+                receipt_refs=tuple(refs),
+            )
+        if integration.state is WorkState.BLOCKED:
+            return StepResult(
+                StepStatus.BLOCKED_DETERMINISTIC,
+                integration.code,
+                integration.condition_fingerprint,
+                receipt_refs=tuple(refs),
+            )
+        if integration.state in {WorkState.PENDING, WorkState.RUNNING}:
+            return StepResult(
+                StepStatus.IN_PROGRESS,
+                integration.code,
+                integration.condition_fingerprint,
+                receipt_refs=tuple(refs),
+                retry_after_seconds=integration.retry_after_seconds,
+            )
+        if integration.state is WorkState.COMPLETE and not integration.receipt_refs:
+            return StepResult(
+                StepStatus.RECONCILE,
+                "INTEGRATION_COMPLETE_WITHOUT_RECEIPT",
+                integration.condition_fingerprint,
+                receipt_refs=tuple(refs),
+            )
+        return StepResult(
+            StepStatus.SUCCEEDED,
+            "SOURCE_INTEGRATION_COMPLETE",
+            integration.condition_fingerprint,
+            receipt_refs=tuple(refs),
+            next_phase=MissionPhase.GATE0,
+        )
+
+    def gate0(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        view = self.authority.run_gate0(observation.mission_id)
+        if view.state is GateState.UNKNOWN:
+            status = StepStatus.RECONCILE
+        elif view.state is GateState.HOLD:
+            status = StepStatus.BLOCKED_DETERMINISTIC
+        else:
+            if not view.receipt_refs:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "GATE0_PASS_WITHOUT_RECEIPT",
+                    view.condition_fingerprint,
+                )
+            status = StepStatus.SUCCEEDED
+        return StepResult(
+            status,
+            view.code,
+            view.condition_fingerprint,
+            receipt_refs=view.receipt_refs,
+            retry_after_seconds=view.retry_after_seconds,
+            next_phase=(
+                MissionPhase.RELEASE_READY
+                if status is StepStatus.SUCCEEDED else None
+            ),
+        )
+
+    def release_ready(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        view = self.authority.build_release_ready(observation.mission_id)
+        if view.state is ReadyState.UNKNOWN:
+            status = StepStatus.RECONCILE
+        elif view.state is ReadyState.NOT_READY:
+            status = StepStatus.BLOCKED_DETERMINISTIC
+        else:
+            if not view.material_digest:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "READY_WITHOUT_MATERIAL_DIGEST",
+                    view.condition_fingerprint,
+                    receipt_refs=view.receipt_refs,
+                )
+            if not view.receipt_refs:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "READY_WITHOUT_RECEIPT",
+                    view.condition_fingerprint,
+                )
+            status = StepStatus.SUCCEEDED
+        return StepResult(
+            status,
+            view.code,
+            view.condition_fingerprint,
+            receipt_refs=view.receipt_refs,
+            retry_after_seconds=view.retry_after_seconds,
+            next_phase=(
+                MissionPhase.PROMOTION
+                if status is StepStatus.SUCCEEDED else None
+            ),
+            material_digest=view.material_digest,
+        )
+
+    def promotion(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        view = self.authority.run_promotion(observation.mission_id)
+        if view.state is PromotionState.UNKNOWN:
+            status = StepStatus.RECONCILE
+        elif view.state is PromotionState.IN_PROGRESS:
+            status = StepStatus.IN_PROGRESS
+        elif view.state is PromotionState.VERIFIED:
+            if not view.receipt_refs:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "PROMOTION_VERIFIED_WITHOUT_RECEIPT",
+                    view.condition_fingerprint,
+                    material_digest=view.material_digest,
+                )
+            status = StepStatus.SUCCEEDED
+        elif view.state is PromotionState.ROLLED_BACK:
+            status = StepStatus.BLOCKED_DETERMINISTIC
+        else:
+            status = StepStatus.BLOCKED_DETERMINISTIC
+        return StepResult(
+            status,
+            view.code,
+            view.condition_fingerprint,
+            receipt_refs=view.receipt_refs,
+            retry_after_seconds=view.retry_after_seconds,
+            next_phase=(
+                MissionPhase.FINAL_ACCEPTANCE
+                if status is StepStatus.SUCCEEDED else None
+            ),
+            material_digest=view.material_digest,
+        )
+
+    def final_acceptance(
+        self,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        view = self.authority.run_final_acceptance(observation.mission_id)
+        if view.state is AcceptanceState.UNKNOWN:
+            status = StepStatus.RECONCILE
+        elif view.state is AcceptanceState.PASS:
+            if not view.receipt_refs:
+                return StepResult(
+                    StepStatus.RECONCILE,
+                    "FINAL_ACCEPTANCE_PASS_WITHOUT_RECEIPT",
+                    view.condition_fingerprint,
+                )
+            status = StepStatus.SUCCEEDED
+        else:
+            status = StepStatus.BLOCKED_DETERMINISTIC
+        return StepResult(
+            status,
+            view.code,
+            view.condition_fingerprint,
+            receipt_refs=view.receipt_refs,
+            next_phase=(
+                MissionPhase.COMPLETE
+                if status is StepStatus.SUCCEEDED else None
+            ),
+        )
+
+    @staticmethod
+    def _admission_receipt_error(
+        lane: TaskLaneSpec,
+        receipt: TaskAdmissionReceipt,
+        *,
+        observation: MissionObservation,
+    ) -> str | None:
+        if receipt.task_id != lane.task_id:
+            return "TASK_IDENTITY_MISMATCH"
+        if receipt.target_branch != lane.target_branch:
+            return "TASK_BRANCH_MISMATCH"
+        if receipt.generation != observation.generation:
+            return "TASK_GENERATION_MISMATCH"
+        return None
+
+    @staticmethod
+    def _refs(*groups: Sequence[str]) -> tuple[str, ...]:
+        seen: set[str] = set()
+        values: list[str] = []
+        for group in groups:
+            for item in group:
+                if item and item not in seen:
+                    seen.add(item)
+                    values.append(item)
+        return tuple(values)
+
+    @staticmethod
+    def _idempotency(*parts: object) -> str:
+        return "aios:auto:" + _digest(
+            {"schema": "aios.unattended-idempotency.v1", "parts": list(parts)}
+        )
+
+    @staticmethod
+    def _phase_fingerprint(
+        phase: str,
+        observation: MissionObservation,
+        details: Mapping[str, object],
+    ) -> str:
+        return _digest({
+            "schema": "aios.unattended-phase-condition.v1",
+            "phase": phase,
+            "mission_id": observation.mission_id,
+            "generation": observation.generation,
+            "head_oid": observation.head_oid,
+            "material_digest": observation.material_digest,
+            "details": details,
+        })

active-runtime/app/task_plane/autonomous_resume_controller.py

@@ -0,0 +1,877 @@
+"""Deterministic unattended mission-resume controller for AIOS.
+
+The controller is deliberately small and does not create a second scheduler,
+database, identity system, or release control plane.
+
+It executes exactly one reconciliation/progress cycle per ``run_once`` call.
+An existing AIOS scheduler/timer/worker loop should invoke it periodically.
+
+All consequential effects are delegated through ``AutonomousExecutionPort`` and
+must use the existing AIOS authorities: durable Owner authorization, task
+admission, claim/TaskLease/TaskPrincipal, WriterLease/CAS/fencing, Gate0,
+ReleaseReady, PromotionController and governed rollback.
+
+The controller's responsibilities are limited to:
+- one-winner resume lease acquisition;
+- durable checkpoint progression;
+- bounded retry/backoff;
+- unchanged deterministic blocker suppression;
+- UNKNOWN/ambiguous-effect reconciliation before retry;
+- non-interactive owner-boundary recording;
+- crash-safe phase resumption.
+
+It never manufactures credentials, group membership, principals, leases,
+signatures, branch OIDs, or release receipts.
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field, replace
+from enum import Enum
+import hashlib
+import json
+import time
+from typing import Any, Callable, Mapping, Protocol, Sequence
+
+
+class MissionPhase(str, Enum):
+    S2_ADMISSION = "S2_ADMISSION"
+    S2_EXECUTION = "S2_EXECUTION"
+    GATE0 = "GATE0"
+    RELEASE_READY = "RELEASE_READY"
+    PROMOTION = "PROMOTION"
+    FINAL_ACCEPTANCE = "FINAL_ACCEPTANCE"
+    OWNER_ACTION_REQUIRED = "OWNER_ACTION_REQUIRED"
+    COMPLETE = "COMPLETE"
+
+
+class EffectState(str, Enum):
+    ABSENT = "ABSENT"
+    ACTIVE = "ACTIVE"
+    COMPLETED = "COMPLETED"
+    UNKNOWN = "UNKNOWN"
+
+
+class StepStatus(str, Enum):
+    SUCCEEDED = "SUCCEEDED"
+    IN_PROGRESS = "IN_PROGRESS"
+    RETRYABLE = "RETRYABLE"
+    RECONCILE = "RECONCILE"
+    BLOCKED_DETERMINISTIC = "BLOCKED_DETERMINISTIC"
+    BLOCKED_OWNER = "BLOCKED_OWNER"
+
+
+class OutcomeStatus(str, Enum):
+    COMPLETE = "COMPLETE"
+    PROGRESSED = "PROGRESSED"
+    IN_PROGRESS = "IN_PROGRESS"
+    RETRY_SCHEDULED = "RETRY_SCHEDULED"
+    WAIT_CONDITION_CHANGE = "WAIT_CONDITION_CHANGE"
+    ACTIVE_ELSEWHERE = "ACTIVE_ELSEWHERE"
+    OWNER_ACTION_REQUIRED = "OWNER_ACTION_REQUIRED"
+
+
+_PHASE_NEXT: dict[MissionPhase, MissionPhase] = {
+    MissionPhase.S2_ADMISSION: MissionPhase.S2_EXECUTION,
+    MissionPhase.S2_EXECUTION: MissionPhase.GATE0,
+    MissionPhase.GATE0: MissionPhase.RELEASE_READY,
+    MissionPhase.RELEASE_READY: MissionPhase.PROMOTION,
+    MissionPhase.PROMOTION: MissionPhase.FINAL_ACCEPTANCE,
+    MissionPhase.FINAL_ACCEPTANCE: MissionPhase.COMPLETE,
+}
+
+
+def _canonical(value: Any) -> bytes:
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def _sha256(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+@dataclass(frozen=True)
+class ResumePolicy:
+    resume_lease_seconds: int = 120
+    default_poll_seconds: int = 30
+    deterministic_block_poll_seconds: int = 300
+    transient_max_attempts: int = 3
+    transient_base_delay_seconds: int = 2
+    transient_max_delay_seconds: int = 60
+
+    def __post_init__(self) -> None:
+        for name in (
+            "resume_lease_seconds",
+            "default_poll_seconds",
+            "deterministic_block_poll_seconds",
+            "transient_max_attempts",
+            "transient_base_delay_seconds",
+            "transient_max_delay_seconds",
+        ):
+            if getattr(self, name) < 1:
+                raise ValueError(f"{name}_must_be_positive")
+        if self.transient_max_delay_seconds < self.transient_base_delay_seconds:
+            raise ValueError("transient_max_delay_seconds_too_small")
+
+
+@dataclass(frozen=True)
+class MissionObservation:
+    mission_id: str
+    generation: int
+    head_oid: str
+    material_digest: str
+    condition_fingerprint: str
+    effect_state: EffectState = EffectState.ABSENT
+    requires_reconcile: bool = False
+    owner_action_resolved: bool = False
+    evidence_refs: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.mission_id:
+            raise ValueError("mission_id_required")
+        if self.generation < 1:
+            raise ValueError("generation_must_be_positive")
+        if not self.head_oid:
+            raise ValueError("head_oid_required")
+        if not self.material_digest:
+            raise ValueError("material_digest_required")
+        if not self.condition_fingerprint:
+            raise ValueError("condition_fingerprint_required")
+
+
+@dataclass(frozen=True)
+class ResumeLease:
+    lease_id: str
+    mission_id: str
+    generation: int
+    fencing_token: str
+    expires_at: float
+
+    def __post_init__(self) -> None:
+        if not self.lease_id:
+            raise ValueError("lease_id_required")
+        if not self.mission_id:
+            raise ValueError("mission_id_required")
+        if self.generation < 1:
+            raise ValueError("generation_must_be_positive")
+        if not self.fencing_token:
+            raise ValueError("fencing_token_required")
+
+
+@dataclass(frozen=True)
+class StepResult:
+    status: StepStatus
+    code: str
+    condition_fingerprint: str
+    receipt_refs: tuple[str, ...] = ()
+    retry_after_seconds: int | None = None
+    next_phase: MissionPhase | None = None
+    material_digest: str | None = None
+    owner_action_payload: Mapping[str, Any] | None = None
+
+    def __post_init__(self) -> None:
+        if not self.code:
+            raise ValueError("step_code_required")
+        if not self.condition_fingerprint:
+            raise ValueError("condition_fingerprint_required")
+        if self.retry_after_seconds is not None and self.retry_after_seconds < 0:
+            raise ValueError("retry_after_seconds_must_be_non_negative")
+
+
+@dataclass(frozen=True)
+class MissionCheckpoint:
+    schema: str
+    mission_id: str
+    generation: int
+    phase: MissionPhase
+    head_oid: str
+    material_digest: str
+    last_condition_fingerprint: str
+    attempt_count: int
+    blocked_condition_fingerprint: str | None
+    reconcile_required: bool
+    frozen_material_digest: str | None
+    resume_phase_after_owner: MissionPhase | None
+    receipt_refs: tuple[str, ...]
+    updated_at: float
+    revision: int = 1
+
+    @classmethod
+    def initial(
+        cls,
+        observation: MissionObservation,
+        *,
+        observed_at: float,
+    ) -> "MissionCheckpoint":
+        return cls(
+            schema="aios.autonomous-resume-checkpoint.v1",
+            mission_id=observation.mission_id,
+            generation=observation.generation,
+            phase=MissionPhase.S2_ADMISSION,
+            head_oid=observation.head_oid,
+            material_digest=observation.material_digest,
+            last_condition_fingerprint=observation.condition_fingerprint,
+            attempt_count=0,
+            blocked_condition_fingerprint=None,
+            reconcile_required=False,
+            frozen_material_digest=None,
+            resume_phase_after_owner=None,
+            receipt_refs=tuple(observation.evidence_refs),
+            updated_at=observed_at,
+            revision=1,
+        )
+
+    def __post_init__(self) -> None:
+        if self.schema != "aios.autonomous-resume-checkpoint.v1":
+            raise ValueError("unsupported_checkpoint_schema")
+        if not self.mission_id:
+            raise ValueError("mission_id_required")
+        if self.generation < 1:
+            raise ValueError("generation_must_be_positive")
+        if self.attempt_count < 0:
+            raise ValueError("attempt_count_must_be_non_negative")
+        if self.revision < 1:
+            raise ValueError("revision_must_be_positive")
+
+    @property
+    def digest(self) -> str:
+        payload = asdict(self)
+        payload["phase"] = self.phase.value
+        payload["resume_phase_after_owner"] = (
+            None
+            if self.resume_phase_after_owner is None
+            else self.resume_phase_after_owner.value
+        )
+        payload["receipt_refs"] = list(self.receipt_refs)
+        return _sha256(_canonical(payload))
+
+
+@dataclass(frozen=True)
+class ControllerOutcome:
+    status: OutcomeStatus
+    mission_id: str
+    phase: MissionPhase
+    code: str
+    next_wake_after_seconds: int | None
+    checkpoint_digest: str | None
+    receipt_refs: tuple[str, ...] = ()
+
+
+class AutonomousExecutionPort(Protocol):
+    """Adapter boundary to existing AIOS authorities and executors."""
+
+    def read_observation(self, mission_id: str) -> MissionObservation: ...
+
+    def load_checkpoint(self, mission_id: str) -> MissionCheckpoint | None: ...
+
+    def acquire_resume_lease(
+        self,
+        *,
+        mission_id: str,
+        generation: int,
+        idempotency_key: str,
+        lease_seconds: int,
+    ) -> ResumeLease | None: ...
+
+    def step(
+        self,
+        *,
+        phase: MissionPhase,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult: ...
+
+    def reconcile(
+        self,
+        *,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult: ...
+
+    def persist_checkpoint(
+        self,
+        checkpoint: MissionCheckpoint,
+        *,
+        lease: ResumeLease,
+    ) -> None: ...
+
+    def persist_owner_action_required(
+        self,
+        payload: Mapping[str, Any],
+        *,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> str: ...
+
+    def release_resume_lease(
+        self,
+        lease: ResumeLease,
+        *,
+        outcome_code: str,
+    ) -> None: ...
+
+
+class AutonomousResumeController:
+    """One-cycle deterministic unattended mission controller."""
+
+    def __init__(
+        self,
+        port: AutonomousExecutionPort,
+        *,
+        policy: ResumePolicy | None = None,
+        clock: Callable[[], float] | None = None,
+    ) -> None:
+        self.port = port
+        self.policy = policy or ResumePolicy()
+        self.clock = clock or time.time
+
+    def run_once(self, mission_id: str) -> ControllerOutcome:
+        if not mission_id:
+            raise ValueError("mission_id_required")
+
+        now = float(self.clock())
+        observation = self.port.read_observation(mission_id)
+        checkpoint = self.port.load_checkpoint(mission_id)
+        if checkpoint is None:
+            checkpoint = MissionCheckpoint.initial(observation, observed_at=now)
+            forced_reconcile = False
+        else:
+            checkpoint, forced_reconcile = self._validate_and_refresh_checkpoint(
+                checkpoint,
+                observation,
+                now=now,
+            )
+
+        if checkpoint.phase is MissionPhase.COMPLETE:
+            return self._outcome(
+                OutcomeStatus.COMPLETE,
+                checkpoint,
+                "MISSION_ALREADY_COMPLETE",
+                next_wake=None,
+            )
+
+        if checkpoint.phase is MissionPhase.OWNER_ACTION_REQUIRED:
+            if not observation.owner_action_resolved:
+                return self._outcome(
+                    OutcomeStatus.OWNER_ACTION_REQUIRED,
+                    checkpoint,
+                    "OWNER_ACTION_STILL_REQUIRED",
+                    next_wake=None,
+                )
+            resume_phase = (
+                checkpoint.resume_phase_after_owner
+                or MissionPhase.S2_ADMISSION
+            )
+            checkpoint = replace(
+                checkpoint,
+                phase=resume_phase,
+                resume_phase_after_owner=None,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=False,
+                updated_at=now,
+                revision=checkpoint.revision + 1,
+            )
+
+        if (
+            checkpoint.blocked_condition_fingerprint is not None
+            and checkpoint.blocked_condition_fingerprint
+            == observation.condition_fingerprint
+        ):
+            return self._outcome(
+                OutcomeStatus.WAIT_CONDITION_CHANGE,
+                checkpoint,
+                "UNCHANGED_DETERMINISTIC_BLOCKER",
+                next_wake=self.policy.deterministic_block_poll_seconds,
+            )
+
+        if (
+            checkpoint.blocked_condition_fingerprint is not None
+            and checkpoint.blocked_condition_fingerprint
+            != observation.condition_fingerprint
+        ):
+            checkpoint = replace(
+                checkpoint,
+                blocked_condition_fingerprint=None,
+                attempt_count=0,
+                updated_at=now,
+                revision=checkpoint.revision + 1,
+            )
+
+        lease = self.port.acquire_resume_lease(
+            mission_id=observation.mission_id,
+            generation=observation.generation,
+            idempotency_key=self._resume_idempotency_key(observation),
+            lease_seconds=self.policy.resume_lease_seconds,
+        )
+        if lease is None:
+            return self._outcome(
+                OutcomeStatus.ACTIVE_ELSEWHERE,
+                checkpoint,
+                "RESUME_LEASE_HELD_ELSEWHERE",
+                next_wake=self.policy.default_poll_seconds,
+            )
+
+        try:
+            self._validate_resume_lease(
+                lease,
+                observation=observation,
+                now=now,
+            )
+        except Exception:
+            self.port.release_resume_lease(
+                lease,
+                outcome_code="INVALID_RESUME_LEASE",
+            )
+            raise
+
+        outcome_code = "UNSET"
+        try:
+            reconciling = (
+                forced_reconcile
+                or checkpoint.reconcile_required
+                or observation.requires_reconcile
+                or observation.effect_state is EffectState.UNKNOWN
+            )
+            if reconciling:
+                result = self.port.reconcile(
+                    observation=observation,
+                    checkpoint=checkpoint,
+                    lease=lease,
+                )
+            else:
+                result = self.port.step(
+                    phase=checkpoint.phase,
+                    observation=observation,
+                    checkpoint=checkpoint,
+                    lease=lease,
+                )
+
+            outcome, new_checkpoint = self._apply_result(
+                result=result,
+                observation=observation,
+                checkpoint=checkpoint,
+                now=now,
+                lease=lease,
+                reconciling=reconciling,
+            )
+            outcome_code = outcome.code
+            self.port.persist_checkpoint(new_checkpoint, lease=lease)
+            return replace(
+                outcome,
+                checkpoint_digest=new_checkpoint.digest,
+            )
+        finally:
+            self.port.release_resume_lease(
+                lease,
+                outcome_code=outcome_code,
+            )
+
+    def _validate_and_refresh_checkpoint(
+        self,
+        checkpoint: MissionCheckpoint,
+        observation: MissionObservation,
+        *,
+        now: float,
+    ) -> tuple[MissionCheckpoint, bool]:
+        if checkpoint.mission_id != observation.mission_id:
+            raise ValueError("checkpoint_mission_mismatch")
+        if checkpoint.generation > observation.generation:
+            raise ValueError("checkpoint_generation_ahead_of_authority")
+
+        refreshed = checkpoint
+        forced_reconcile = False
+        if checkpoint.generation < observation.generation:
+            refreshed = replace(
+                checkpoint,
+                generation=observation.generation,
+                phase=MissionPhase.S2_ADMISSION,
+                head_oid=observation.head_oid,
+                material_digest=observation.material_digest,
+                last_condition_fingerprint=observation.condition_fingerprint,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=False,
+                frozen_material_digest=None,
+                resume_phase_after_owner=None,
+                receipt_refs=self._merge_refs(
+                    checkpoint.receipt_refs,
+                    observation.evidence_refs,
+                ),
+                updated_at=now,
+                revision=checkpoint.revision + 1,
+            )
+
+        if (
+            refreshed.phase in {
+                MissionPhase.RELEASE_READY,
+                MissionPhase.PROMOTION,
+                MissionPhase.FINAL_ACCEPTANCE,
+            }
+            and refreshed.frozen_material_digest is not None
+            and refreshed.frozen_material_digest
+            != observation.material_digest
+        ):
+            # Material changed after READY. Promotion/final may already have
+            # observable side effects, so reconcile before rebuilding PREPARE.
+            previous_phase = refreshed.phase
+            force_for_effect = (
+                previous_phase in {
+                    MissionPhase.PROMOTION,
+                    MissionPhase.FINAL_ACCEPTANCE,
+                }
+                and observation.effect_state in {
+                    EffectState.ACTIVE,
+                    EffectState.UNKNOWN,
+                }
+            )
+            refreshed = replace(
+                refreshed,
+                phase=MissionPhase.RELEASE_READY,
+                material_digest=observation.material_digest,
+                head_oid=observation.head_oid,
+                frozen_material_digest=None,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=force_for_effect,
+                last_condition_fingerprint=observation.condition_fingerprint,
+                receipt_refs=self._merge_refs(
+                    refreshed.receipt_refs,
+                    observation.evidence_refs,
+                ),
+                updated_at=now,
+                revision=refreshed.revision + 1,
+            )
+            forced_reconcile = forced_reconcile or force_for_effect
+        elif (
+            refreshed.phase in {
+                MissionPhase.S2_ADMISSION,
+                MissionPhase.S2_EXECUTION,
+                MissionPhase.GATE0,
+            }
+            and refreshed.head_oid != observation.head_oid
+        ):
+            # Source moved before READY. Re-enter S2 admission/replan.
+            refreshed = replace(
+                refreshed,
+                phase=MissionPhase.S2_ADMISSION,
+                head_oid=observation.head_oid,
+                material_digest=observation.material_digest,
+                frozen_material_digest=None,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=False,
+                last_condition_fingerprint=observation.condition_fingerprint,
+                receipt_refs=self._merge_refs(
+                    refreshed.receipt_refs,
+                    observation.evidence_refs,
+                ),
+                updated_at=now,
+                revision=refreshed.revision + 1,
+            )
+
+        return refreshed, forced_reconcile
+
+    def _apply_result(
+        self,
+        *,
+        result: StepResult,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        now: float,
+        lease: ResumeLease,
+        reconciling: bool,
+    ) -> tuple[ControllerOutcome, MissionCheckpoint]:
+        refs = self._merge_refs(
+            checkpoint.receipt_refs,
+            observation.evidence_refs,
+            result.receipt_refs,
+        )
+        base = replace(
+            checkpoint,
+            head_oid=observation.head_oid,
+            material_digest=(
+                result.material_digest
+                or observation.material_digest
+            ),
+            last_condition_fingerprint=result.condition_fingerprint,
+            receipt_refs=refs,
+            updated_at=now,
+            revision=checkpoint.revision + 1,
+        )
+
+        if (
+            result.status is StepStatus.SUCCEEDED
+            and checkpoint.phase in {
+                MissionPhase.PROMOTION,
+                MissionPhase.FINAL_ACCEPTANCE,
+            }
+            and checkpoint.frozen_material_digest is not None
+            and result.material_digest is not None
+            and result.material_digest != checkpoint.frozen_material_digest
+        ):
+            new_checkpoint = replace(
+                base,
+                phase=MissionPhase.RELEASE_READY,
+                material_digest=result.material_digest,
+                frozen_material_digest=None,
+                reconcile_required=True,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+            )
+            return (
+                self._outcome(
+                    OutcomeStatus.IN_PROGRESS,
+                    new_checkpoint,
+                    "RECONCILE_REQUIRED:MATERIAL_CHANGED_DURING_PROMOTION",
+                    next_wake=0,
+                    refs=result.receipt_refs,
+                ),
+                new_checkpoint,
+            )
+
+        if result.status is StepStatus.SUCCEEDED:
+            next_phase = (
+                result.next_phase
+                or _PHASE_NEXT.get(checkpoint.phase, checkpoint.phase)
+            )
+            frozen = base.frozen_material_digest
+            if checkpoint.phase is MissionPhase.RELEASE_READY:
+                frozen = result.material_digest or observation.material_digest
+
+            new_checkpoint = replace(
+                base,
+                phase=next_phase,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=False,
+                frozen_material_digest=frozen,
+            )
+            status = (
+                OutcomeStatus.COMPLETE
+                if next_phase is MissionPhase.COMPLETE
+                else OutcomeStatus.PROGRESSED
+            )
+            code = (
+                "MISSION_COMPLETE"
+                if next_phase is MissionPhase.COMPLETE
+                else result.code
+            )
+            return (
+                self._outcome(
+                    status,
+                    new_checkpoint,
+                    code,
+                    next_wake=(
+                        None
+                        if status is OutcomeStatus.COMPLETE
+                        else self.policy.default_poll_seconds
+                    ),
+                    refs=result.receipt_refs,
+                ),
+                new_checkpoint,
+            )
+
+        if result.status is StepStatus.IN_PROGRESS:
+            new_checkpoint = replace(
+                base,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=reconciling,
+            )
+            return (
+                self._outcome(
+                    OutcomeStatus.IN_PROGRESS,
+                    new_checkpoint,
+                    result.code,
+                    next_wake=(
+                        result.retry_after_seconds
+                        if result.retry_after_seconds is not None
+                        else self.policy.default_poll_seconds
+                    ),
+                    refs=result.receipt_refs,
+                ),
+                new_checkpoint,
+            )
+
+        if result.status is StepStatus.RETRYABLE:
+            attempts = checkpoint.attempt_count + 1
+            if attempts > self.policy.transient_max_attempts:
+                new_checkpoint = replace(
+                    base,
+                    attempt_count=attempts,
+                    blocked_condition_fingerprint=result.condition_fingerprint,
+                    reconcile_required=reconciling,
+                )
+                return (
+                    self._outcome(
+                        OutcomeStatus.WAIT_CONDITION_CHANGE,
+                        new_checkpoint,
+                        "RETRY_BUDGET_EXHAUSTED",
+                        next_wake=self.policy.deterministic_block_poll_seconds,
+                        refs=result.receipt_refs,
+                    ),
+                    new_checkpoint,
+                )
+
+            delay = (
+                result.retry_after_seconds
+                if result.retry_after_seconds is not None
+                else min(
+                    self.policy.transient_base_delay_seconds
+                    * (2 ** (attempts - 1)),
+                    self.policy.transient_max_delay_seconds,
+                )
+            )
+            new_checkpoint = replace(
+                base,
+                attempt_count=attempts,
+                blocked_condition_fingerprint=None,
+                reconcile_required=reconciling,
+            )
+            return (
+                self._outcome(
+                    OutcomeStatus.RETRY_SCHEDULED,
+                    new_checkpoint,
+                    result.code,
+                    next_wake=delay,
+                    refs=result.receipt_refs,
+                ),
+                new_checkpoint,
+            )
+
+        if result.status is StepStatus.BLOCKED_DETERMINISTIC:
+            new_checkpoint = replace(
+                base,
+                attempt_count=checkpoint.attempt_count,
+                blocked_condition_fingerprint=result.condition_fingerprint,
+                reconcile_required=reconciling,
+            )
+            return (
+                self._outcome(
+                    OutcomeStatus.WAIT_CONDITION_CHANGE,
+                    new_checkpoint,
+                    result.code,
+                    next_wake=self.policy.deterministic_block_poll_seconds,
+                    refs=result.receipt_refs,
+                ),
+                new_checkpoint,
+            )
+
+        if result.status is StepStatus.BLOCKED_OWNER:
+            payload = dict(result.owner_action_payload or {})
+            payload.setdefault("mission_id", observation.mission_id)
+            payload.setdefault("generation", observation.generation)
+            payload.setdefault("phase", checkpoint.phase.value)
+            payload.setdefault("code", result.code)
+            record_ref = self.port.persist_owner_action_required(
+                payload,
+                observation=observation,
+                checkpoint=checkpoint,
+                lease=lease,
+            )
+            if not record_ref:
+                raise ValueError("owner_action_record_ref_required")
+            new_checkpoint = replace(
+                base,
+                phase=MissionPhase.OWNER_ACTION_REQUIRED,
+                resume_phase_after_owner=checkpoint.phase,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=False,
+                receipt_refs=self._merge_refs(refs, (record_ref,)),
+            )
+            return (
+                self._outcome(
+                    OutcomeStatus.OWNER_ACTION_REQUIRED,
+                    new_checkpoint,
+                    result.code,
+                    next_wake=None,
+                    refs=(record_ref,),
+                ),
+                new_checkpoint,
+            )
+
+        if result.status is StepStatus.RECONCILE:
+            # The next scheduler cycle must reconcile before any retry.
+            new_checkpoint = replace(
+                base,
+                attempt_count=0,
+                blocked_condition_fingerprint=None,
+                reconcile_required=True,
+            )
+            return (
+                self._outcome(
+                    OutcomeStatus.IN_PROGRESS,
+                    new_checkpoint,
+                    "RECONCILE_REQUIRED:" + result.code,
+                    next_wake=0,
+                    refs=result.receipt_refs,
+                ),
+                new_checkpoint,
+            )
+
+        raise AssertionError(f"unsupported_step_status:{result.status}")
+
+    @staticmethod
+    def _validate_resume_lease(
+        lease: ResumeLease,
+        *,
+        observation: MissionObservation,
+        now: float,
+    ) -> None:
+        if lease.mission_id != observation.mission_id:
+            raise ValueError("resume_lease_mission_mismatch")
+        if lease.generation != observation.generation:
+            raise ValueError("resume_lease_generation_mismatch")
+        if lease.expires_at <= now:
+            raise ValueError("resume_lease_expired")
+
+    def _resume_idempotency_key(
+        self,
+        observation: MissionObservation,
+    ) -> str:
+        payload = {
+            "schema": "aios.resume-effect.v1",
+            "mission_id": observation.mission_id,
+            "generation": observation.generation,
+        }
+        return "resume:sha256:" + _sha256(_canonical(payload))
+
+    @staticmethod
+    def _merge_refs(*groups: Sequence[str]) -> tuple[str, ...]:
+        merged: list[str] = []
+        seen: set[str] = set()
+        for group in groups:
+            for item in group:
+                value = str(item)
+                if value and value not in seen:
+                    seen.add(value)
+                    merged.append(value)
+        return tuple(merged)
+
+    @staticmethod
+    def _outcome(
+        status: OutcomeStatus,
+        checkpoint: MissionCheckpoint,
+        code: str,
+        *,
+        next_wake: int | None,
+        refs: Sequence[str] = (),
+    ) -> ControllerOutcome:
+        return ControllerOutcome(
+            status=status,
+            mission_id=checkpoint.mission_id,
+            phase=checkpoint.phase,
+            code=code,
+            next_wake_after_seconds=next_wake,
+            checkpoint_digest=None,
+            receipt_refs=tuple(refs),
+        )

active-runtime/app/task_plane/autonomous_resume_integration.py

@@ -0,0 +1,209 @@
+"""Thin callable adapter for :mod:`autonomous_resume_controller`.
+
+This module contains no transport, daemon, database or scheduler loop.  It only
+binds current AIOS runtime functions to the stable ``AutonomousExecutionPort``
+contract and exposes one scheduler tick.
+
+A field integrator should:
+- resolve the live current APIs;
+- supply callables that use existing governed authorities;
+- register ``AutonomousSchedulerTick.run`` with the existing scheduler/timer.
+
+No additional control plane is created here.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable
+
+from .autonomous_resume_controller import (
+    AutonomousExecutionPort,
+    AutonomousResumeController,
+    ControllerOutcome,
+    MissionCheckpoint,
+    MissionObservation,
+    MissionPhase,
+    ResumeLease,
+    StepResult,
+)
+
+
+PhaseHandler = Callable[
+    [MissionObservation, MissionCheckpoint, ResumeLease],
+    StepResult,
+]
+ReconcileHandler = PhaseHandler
+
+
+@dataclass(frozen=True)
+class PhaseCallbacks:
+    admission: PhaseHandler
+    source_execution: PhaseHandler
+    gate0: PhaseHandler
+    release_ready: PhaseHandler
+    promotion: PhaseHandler
+    final_acceptance: PhaseHandler
+
+    def for_phase(self, phase: MissionPhase) -> PhaseHandler:
+        mapping = {
+            MissionPhase.S2_ADMISSION: self.admission,
+            MissionPhase.S2_EXECUTION: self.source_execution,
+            MissionPhase.GATE0: self.gate0,
+            MissionPhase.RELEASE_READY: self.release_ready,
+            MissionPhase.PROMOTION: self.promotion,
+            MissionPhase.FINAL_ACCEPTANCE: self.final_acceptance,
+        }
+        try:
+            return mapping[phase]
+        except KeyError as exc:
+            raise ValueError(f"phase_has_no_executor:{phase.value}") from exc
+
+
+@dataclass(frozen=True)
+class AuthorityCallbacks:
+    read_observation: Callable[[str], MissionObservation]
+    load_checkpoint: Callable[[str], MissionCheckpoint | None]
+    acquire_resume_lease: Callable[[str, int, str, int], ResumeLease | None]
+    persist_checkpoint: Callable[[MissionCheckpoint, ResumeLease], None]
+    persist_owner_action_required: Callable[
+        [dict, MissionObservation, MissionCheckpoint, ResumeLease],
+        str,
+    ]
+    release_resume_lease: Callable[[ResumeLease, str], None]
+    reconcile: ReconcileHandler
+
+
+class CallableAutonomousExecutionPort(AutonomousExecutionPort):
+    """Runtime adapter composed from already-governed AIOS callables."""
+
+    def __init__(
+        self,
+        authority: AuthorityCallbacks,
+        phases: PhaseCallbacks,
+    ) -> None:
+        self.authority = authority
+        self.phases = phases
+
+    def read_observation(self, mission_id: str) -> MissionObservation:
+        value = self.authority.read_observation(mission_id)
+        if not isinstance(value, MissionObservation):
+            raise TypeError("read_observation_must_return_MissionObservation")
+        return value
+
+    def load_checkpoint(self, mission_id: str) -> MissionCheckpoint | None:
+        value = self.authority.load_checkpoint(mission_id)
+        if value is not None and not isinstance(value, MissionCheckpoint):
+            raise TypeError("load_checkpoint_must_return_checkpoint_or_none")
+        return value
+
+    def acquire_resume_lease(
+        self,
+        *,
+        mission_id: str,
+        generation: int,
+        idempotency_key: str,
+        lease_seconds: int,
+    ) -> ResumeLease | None:
+        value = self.authority.acquire_resume_lease(
+            mission_id,
+            generation,
+            idempotency_key,
+            lease_seconds,
+        )
+        if value is not None and not isinstance(value, ResumeLease):
+            raise TypeError("acquire_resume_lease_must_return_lease_or_none")
+        return value
+
+    def step(
+        self,
+        *,
+        phase: MissionPhase,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        handler = self.phases.for_phase(phase)
+        value = handler(observation, checkpoint, lease)
+        if not isinstance(value, StepResult):
+            raise TypeError("phase_handler_must_return_StepResult")
+        return value
+
+    def reconcile(
+        self,
+        *,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> StepResult:
+        value = self.authority.reconcile(
+            observation,
+            checkpoint,
+            lease,
+        )
+        if not isinstance(value, StepResult):
+            raise TypeError("reconcile_must_return_StepResult")
+        return value
+
+    def persist_checkpoint(
+        self,
+        checkpoint: MissionCheckpoint,
+        *,
+        lease: ResumeLease,
+    ) -> None:
+        self.authority.persist_checkpoint(checkpoint, lease)
+
+    def persist_owner_action_required(
+        self,
+        payload,
+        *,
+        observation: MissionObservation,
+        checkpoint: MissionCheckpoint,
+        lease: ResumeLease,
+    ) -> str:
+        value = self.authority.persist_owner_action_required(
+            dict(payload),
+            observation,
+            checkpoint,
+            lease,
+        )
+        if not isinstance(value, str) or not value:
+            raise ValueError("owner_action_record_ref_required")
+        return value
+
+    def release_resume_lease(
+        self,
+        lease: ResumeLease,
+        *,
+        outcome_code: str,
+    ) -> None:
+        self.authority.release_resume_lease(lease, outcome_code)
+
+
+@dataclass(frozen=True)
+class SchedulerTickResult:
+    outcome: ControllerOutcome
+    next_wake_after_seconds: int | None
+
+
+class AutonomousSchedulerTick:
+    """One invocation for an existing scheduler/timer.
+
+    This object never sleeps or loops. The host scheduler owns cadence.
+    """
+
+    def __init__(
+        self,
+        controller: AutonomousResumeController,
+        mission_id: str,
+    ) -> None:
+        if not mission_id:
+            raise ValueError("mission_id_required")
+        self.controller = controller
+        self.mission_id = mission_id
+
+    def run(self) -> SchedulerTickResult:
+        outcome = self.controller.run_once(self.mission_id)
+        return SchedulerTickResult(
+            outcome=outcome,
+            next_wake_after_seconds=outcome.next_wake_after_seconds,
+        )

tests/task_plane/test_autonomous_resume_aios_adapter.py

@@ -0,0 +1,458 @@
+from __future__ import annotations
+
+import unittest
+from dataclasses import replace
+
+from task_plane.autonomous_resume_aios_adapter import (
+    AcceptanceState,
+    AcceptanceView,
+    AiosUnattendedCallbackFactory,
+    AiosUnattendedMissionConfig,
+    DurableGrantView,
+    GateState,
+    GateView,
+    IntegrationView,
+    LaneExecutionView,
+    MissionEffectView,
+    PromotionState,
+    PromotionView,
+    ReadyState,
+    ReleaseReadyView,
+    TaskAdmissionReceipt,
+    TaskLaneSpec,
+    WorkerActivationReceipt,
+    WorkerState,
+    WorkState,
+)
+from task_plane.autonomous_resume_controller import (
+    AutonomousResumeController,
+    EffectState,
+    MissionPhase,
+    OutcomeStatus,
+    ResumeLease,
+    StepResult,
+    StepStatus,
+)
+
+
+def lane(name, path):
+    return TaskLaneSpec(
+        lane=name,
+        task_id=f"task-{name}",
+        repository="aios/aios",
+        target_branch=f"agent/{name}",
+        risk="medium",
+        description=name,
+        write_set=(path,),
+    )
+
+
+class FakeAuthorities:
+    def __init__(self):
+        self.head="a"*40
+        self.material="mat-1"
+        self.generation=1
+        self.effect=MissionEffectView(
+            EffectState.ABSENT,"effect-1",False,()
+        )
+        self.grant=DurableGrantView(
+            grant_id="grant-1",
+            active=True,
+            repository="aios/aios",
+            allowed_target_prefixes=("agent/",),
+            max_task_risk="engineering",
+            evidence_refs=("grant:1",),
+        )
+        self.worker=WorkerActivationReceipt(
+            WorkerState.ACTIVE,
+            receipt_ref="worker:1",
+            code="WORKER_ACTIVE",
+        )
+        self.admission_overrides={}
+        self.lane_views={}
+        self.integration=IntegrationView(
+            WorkState.COMPLETE,"integration-1",("integration:1",),
+            code="INTEGRATION_COMPLETE",
+        )
+        self.gate=GateView(
+            GateState.PASS,"gate-1",("gate:1",),code="GATE0_PASS"
+        )
+        self.ready=ReleaseReadyView(
+            ReadyState.READY,"ready-1","mat-1",("ready:1",),
+            code="RELEASE_READY",
+        )
+        self.promotion=PromotionView(
+            PromotionState.VERIFIED,"promotion-1","mat-1",("promotion:1",),
+            code="PROMOTION_VERIFIED",
+        )
+        self.acceptance=AcceptanceView(
+            AcceptanceState.PASS,"accept-1",("accept:1",),
+            code="FINAL_ACCEPTANCE_PASS",
+        )
+        self.checkpoint=None
+        self.owner_resolved=False
+        self.requests=[]
+        self.persisted_admissions={}
+        self.released=[]
+        self.reconcile_calls=0
+
+    def read_head_oid(self, repository): return self.head
+    def read_material_digest(self, mission_id): return self.material
+    def read_generation(self, mission_id): return self.generation
+    def read_effect_view(self, mission_id): return self.effect
+    def owner_action_resolved(self, mission_id): return self.owner_resolved
+    def load_checkpoint(self, mission_id): return self.checkpoint
+    def acquire_resume_lease(self, mission_id,generation,key,seconds):
+        return ResumeLease("lease-1",mission_id,generation,"fence-1",9999)
+    def persist_checkpoint(self, checkpoint, lease): self.checkpoint=checkpoint
+    def persist_owner_action_required(self,payload,observation,checkpoint,lease):
+        return "owner:1"
+    def release_resume_lease(self, lease, outcome_code):
+        self.released.append((lease.lease_id,outcome_code))
+    def reconcile_effects(self,observation,checkpoint,lease):
+        self.reconcile_calls+=1
+        return StepResult(
+            StepStatus.SUCCEEDED,
+            "RECONCILED",
+            "reconcile-1",
+            next_phase=checkpoint.phase,
+        )
+    def read_durable_grant(self,mission_id): return self.grant
+    def ensure_canonical_worker(self,mission_id,generation,idempotency_key):
+        return self.worker
+    def create_or_resume_task(self,request):
+        self.requests.append(request)
+        receipt=self.admission_overrides.get(
+            request.task_id,
+            TaskAdmissionReceipt(
+                request.task_id,1,request.target_branch,True,True,
+                "task:"+request.task_id,
+            )
+        )
+        if receipt.persisted:
+            self.persisted_admissions[request.task_id]=receipt
+        return receipt
+    def read_task_admission(self,lane):
+        return self.persisted_admissions.get(lane.task_id)
+    def read_lane_execution(self,lane,task):
+        return self.lane_views.get(
+            lane.lane,
+            LaneExecutionView(
+                lane.lane,WorkState.COMPLETE,"lane:"+lane.lane,
+                ("lane-receipt:"+lane.lane,),code="LANE_COMPLETE",
+            )
+        )
+    def drive_integration(self,lanes,tasks): return self.integration
+    def run_gate0(self,mission_id): return self.gate
+    def build_release_ready(self,mission_id): return self.ready
+    def run_promotion(self,mission_id): return self.promotion
+    def run_final_acceptance(self,mission_id): return self.acceptance
+
+
+class AiosAdapterTests(unittest.TestCase):
+    def setUp(self):
+        self.auth=FakeAuthorities()
+        self.lanes=(
+            lane("A","a.py"),
+            lane("B","b.py"),
+            lane("C","c.py"),
+        )
+        self.config=AiosUnattendedMissionConfig(
+            mission_id="v2r2c4",
+            repository="aios/aios",
+            lanes=self.lanes,
+        )
+        self.factory=AiosUnattendedCallbackFactory(
+            self.auth,self.config
+        )
+
+    def observation(self):
+        return self.factory.read_observation("v2r2c4")
+
+    def test_config_rejects_write_set_overlap(self):
+        with self.assertRaisesRegex(ValueError,"overlapping_lane_resource"):
+            AiosUnattendedMissionConfig(
+                "v2r2c4","aios/aios",
+                (lane("A","same.py"),lane("B","same.py")),
+            )
+
+    def test_config_rejects_non_agent_branch(self):
+        with self.assertRaisesRegex(ValueError,"agent_prefix"):
+            TaskLaneSpec(
+                "A","task-a","aios/aios","feature/a","medium","x",
+                ("a.py",),
+            )
+
+    def test_observation_fingerprint_changes_with_head(self):
+        first=self.observation().condition_fingerprint
+        self.auth.head="b"*40
+        second=self.observation().condition_fingerprint
+        self.assertNotEqual(first,second)
+
+    def test_inactive_grant_blocks_without_task_creation(self):
+        self.auth.grant=replace(self.auth.grant,active=False)
+        result=self.factory.admission(
+            self.observation(),None,None
+        )
+        self.assertEqual(result.status,StepStatus.BLOCKED_DETERMINISTIC)
+        self.assertEqual(self.auth.requests,[])
+
+    def test_owner_only_grant_boundary_is_typed_owner_blocker(self):
+        self.auth.grant=replace(
+            self.auth.grant,active=False,owner_action_required=True
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.BLOCKED_OWNER)
+        self.assertIsNotNone(result.owner_action_payload)
+
+    def test_grant_scope_mismatch_blocks_before_worker(self):
+        self.auth.grant=replace(
+            self.auth.grant,allowed_target_prefixes=("other/",)
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.BLOCKED_DETERMINISTIC)
+        self.assertEqual(self.auth.requests,[])
+
+    def test_worker_starting_is_in_progress(self):
+        self.auth.worker=WorkerActivationReceipt(
+            WorkerState.STARTING,"worker:start",7,False,"WORKER_STARTING"
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.IN_PROGRESS)
+        self.assertEqual(result.retry_after_seconds,7)
+        self.assertEqual(self.auth.requests,[])
+
+    def test_worker_unknown_reconciles(self):
+        self.auth.worker=WorkerActivationReceipt(
+            WorkerState.UNKNOWN,None,None,False,"WORKER_UNKNOWN"
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_active_worker_creates_all_tasks_deterministically(self):
+        obs=self.observation()
+        result=self.factory.admission(obs,None,None)
+        self.assertEqual(result.status,StepStatus.SUCCEEDED)
+        self.assertEqual(len(self.auth.requests),3)
+        self.assertTrue(all(r.base_oid==obs.head_oid for r in self.auth.requests))
+        keys=[r.idempotency_key for r in self.auth.requests]
+        self.assertEqual(len(keys),len(set(keys)))
+
+    def test_unpersisted_task_requires_reconcile(self):
+        self.auth.admission_overrides["task-B"]=TaskAdmissionReceipt(
+            "task-B",1,"agent/B",False,False,"task:B"
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_unclaimable_task_waits(self):
+        self.auth.admission_overrides["task-B"]=TaskAdmissionReceipt(
+            "task-B",1,"agent/B",True,False,"task:B"
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.IN_PROGRESS)
+
+    def test_source_execution_unknown_lane_reconciles(self):
+        self.factory.admission(self.observation(),None,None)
+        self.auth.lane_views["B"]=LaneExecutionView(
+            "B",WorkState.UNKNOWN,"lane:B",code="LANE_B_UNKNOWN"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_source_execution_running_lane_waits(self):
+        self.factory.admission(self.observation(),None,None)
+        self.auth.lane_views["B"]=LaneExecutionView(
+            "B",WorkState.RUNNING,"lane:B",retry_after_seconds=9,
+            code="LANE_B_RUNNING"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.IN_PROGRESS)
+        self.assertEqual(result.retry_after_seconds,9)
+
+    def test_source_execution_blocked_lane_is_deterministic_block(self):
+        self.factory.admission(self.observation(),None,None)
+        self.auth.lane_views["B"]=LaneExecutionView(
+            "B",WorkState.BLOCKED,"lane:B",code="LANE_B_BLOCKED"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.BLOCKED_DETERMINISTIC)
+
+    def test_source_execution_complete_advances_to_gate0(self):
+        self.factory.admission(self.observation(),None,None)
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.SUCCEEDED)
+        self.assertEqual(result.next_phase,MissionPhase.GATE0)
+
+    def test_missing_admission_receipts_reconciles_if_authority_has_none(self):
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+        self.assertEqual(result.code,"TASK_ADMISSION_NOT_RESOLVED")
+
+    def test_restart_rebuilds_admission_receipts_from_authority(self):
+        self.factory.admission(self.observation(),None,None)
+        restarted=AiosUnattendedCallbackFactory(self.auth,self.config)
+        result=restarted.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.SUCCEEDED)
+        self.assertEqual(result.next_phase,MissionPhase.GATE0)
+
+    def test_stale_generation_admission_receipt_reconciles(self):
+        self.auth.persisted_admissions["task-A"]=TaskAdmissionReceipt(
+            "task-A",2,"agent/A",True,True,"task:A:gen2"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+        self.assertEqual(result.code,"TASK_GENERATION_MISMATCH")
+
+    def test_wrong_branch_admission_receipt_reconciles(self):
+        self.auth.persisted_admissions["task-A"]=TaskAdmissionReceipt(
+            "task-A",1,"agent/wrong",True,True,"task:A:wrong"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+        self.assertEqual(result.code,"TASK_BRANCH_MISMATCH")
+
+    def test_gate_hold_blocks(self):
+        self.auth.gate=GateView(
+            GateState.HOLD,"gate-hold",code="GATE0_HOLD"
+        )
+        result=self.factory.gate0(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.BLOCKED_DETERMINISTIC)
+
+    def test_gate_unknown_reconciles(self):
+        self.auth.gate=GateView(
+            GateState.UNKNOWN,"gate-unknown",code="GATE0_UNKNOWN"
+        )
+        result=self.factory.gate0(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_ready_without_material_digest_reconciles(self):
+        self.auth.ready=ReleaseReadyView(
+            ReadyState.READY,"ready",None,code="READY"
+        )
+        result=self.factory.release_ready(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_ready_freezes_material_for_controller(self):
+        result=self.factory.release_ready(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.SUCCEEDED)
+        self.assertEqual(result.material_digest,"mat-1")
+        self.assertEqual(result.next_phase,MissionPhase.PROMOTION)
+
+    def test_promotion_unknown_reconciles(self):
+        self.auth.promotion=PromotionView(
+            PromotionState.UNKNOWN,"p-unknown","mat-1",code="PROMOTION_UNKNOWN"
+        )
+        result=self.factory.promotion(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_promotion_in_progress_waits(self):
+        self.auth.promotion=PromotionView(
+            PromotionState.IN_PROGRESS,"p-run","mat-1",
+            retry_after_seconds=5,code="PROMOTION_RUNNING"
+        )
+        result=self.factory.promotion(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.IN_PROGRESS)
+        self.assertEqual(result.retry_after_seconds,5)
+
+    def test_promotion_verified_advances(self):
+        result=self.factory.promotion(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.SUCCEEDED)
+        self.assertEqual(result.next_phase,MissionPhase.FINAL_ACCEPTANCE)
+
+    def test_rolled_back_promotion_blocks_condition(self):
+        self.auth.promotion=PromotionView(
+            PromotionState.ROLLED_BACK,"p-rb","mat-1",code="ROLLED_BACK"
+        )
+        result=self.factory.promotion(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.BLOCKED_DETERMINISTIC)
+
+    def test_final_acceptance_pass_completes(self):
+        result=self.factory.final_acceptance(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.SUCCEEDED)
+        self.assertEqual(result.next_phase,MissionPhase.COMPLETE)
+
+    def test_final_acceptance_unknown_reconciles(self):
+        self.auth.acceptance=AcceptanceView(
+            AcceptanceState.UNKNOWN,"a-unknown",code="ACCEPT_UNKNOWN"
+        )
+        result=self.factory.final_acceptance(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+
+    def test_active_worker_without_receipt_reconciles(self):
+        self.auth.worker=WorkerActivationReceipt(
+            WorkerState.ACTIVE,None,None,False,"WORKER_ACTIVE"
+        )
+        result=self.factory.admission(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+        self.assertEqual(result.code,"WORKER_ACTIVE_WITHOUT_RECEIPT")
+
+    def test_complete_lane_without_receipt_reconciles(self):
+        self.factory.admission(self.observation(),None,None)
+        self.auth.lane_views["B"]=LaneExecutionView(
+            "B",WorkState.COMPLETE,"lane:B",(),code="LANE_COMPLETE"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+        self.assertEqual(result.code,"LANE_COMPLETE_WITHOUT_RECEIPT")
+
+    def test_complete_integration_without_receipt_reconciles(self):
+        self.factory.admission(self.observation(),None,None)
+        self.auth.integration=IntegrationView(
+            WorkState.COMPLETE,"integration-empty",(),code="COMPLETE"
+        )
+        result=self.factory.source_execution(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+        self.assertEqual(result.code,"INTEGRATION_COMPLETE_WITHOUT_RECEIPT")
+
+    def test_gate_pass_without_receipt_reconciles(self):
+        self.auth.gate=GateView(GateState.PASS,"gate-empty",(),code="PASS")
+        result=self.factory.gate0(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_ready_without_receipt_reconciles(self):
+        self.auth.ready=ReleaseReadyView(
+            ReadyState.READY,"ready-empty","mat-1",(),code="READY"
+        )
+        result=self.factory.release_ready(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_promotion_verified_without_receipt_reconciles(self):
+        self.auth.promotion=PromotionView(
+            PromotionState.VERIFIED,"p-empty","mat-1",(),code="VERIFIED"
+        )
+        result=self.factory.promotion(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_final_pass_without_receipt_reconciles(self):
+        self.auth.acceptance=AcceptanceView(
+            AcceptanceState.PASS,"a-empty",(),code="PASS"
+        )
+        result=self.factory.final_acceptance(self.observation(),None,None)
+        self.assertEqual(result.status,StepStatus.RECONCILE)
+
+    def test_controller_port_can_execute_admission_cycle(self):
+        controller=AutonomousResumeController(
+            self.factory.build_port(),
+            clock=lambda:100,
+        )
+        out=controller.run_once("v2r2c4")
+        self.assertEqual(out.status,OutcomeStatus.PROGRESSED)
+        self.assertEqual(self.auth.checkpoint.phase,MissionPhase.S2_EXECUTION)
+
+    def test_unknown_effect_routes_controller_to_reconcile(self):
+        self.auth.effect=MissionEffectView(
+            EffectState.UNKNOWN,"unknown-effect",True,()
+        )
+        controller=AutonomousResumeController(
+            self.factory.build_port(),
+            clock=lambda:100,
+        )
+        out=controller.run_once("v2r2c4")
+        self.assertEqual(self.auth.reconcile_calls,1)
+        self.assertEqual(out.status,OutcomeStatus.PROGRESSED)
+
+
+if __name__=="__main__":
+    unittest.main()

tests/task_plane/test_autonomous_resume_controller.py

@@ -0,0 +1,491 @@
+from __future__ import annotations
+
+import unittest
+from dataclasses import replace
+
+from task_plane.autonomous_resume_controller import (
+    AutonomousResumeController,
+    EffectState,
+    MissionCheckpoint,
+    MissionObservation,
+    MissionPhase,
+    OutcomeStatus,
+    ResumeLease,
+    ResumePolicy,
+    StepResult,
+    StepStatus,
+)
+
+
+class FakePort:
+    def __init__(self):
+        self.observation = MissionObservation(
+            mission_id="v2r2c4",
+            generation=1,
+            head_oid="a" * 40,
+            material_digest="mat-1",
+            condition_fingerprint="cond-1",
+        )
+        self.checkpoint = None
+        self.lease_available = True
+        self.step_results = []
+        self.reconcile_results = []
+        self.persisted = []
+        self.owner_records = []
+        self.released = []
+        self.lease_override = None
+        self.persist_failure = None
+        self.step_calls = 0
+        self.reconcile_calls = 0
+
+    def read_observation(self, mission_id):
+        self.assert_mission(mission_id)
+        return self.observation
+
+    def load_checkpoint(self, mission_id):
+        self.assert_mission(mission_id)
+        return self.checkpoint
+
+    def acquire_resume_lease(self, *, mission_id, generation, idempotency_key, lease_seconds):
+        self.assert_mission(mission_id)
+        if not self.lease_available:
+            return None
+        if self.lease_override is not None:
+            return self.lease_override
+        return ResumeLease(
+            lease_id="lease-1",
+            mission_id=mission_id,
+            generation=generation,
+            fencing_token="fence-1",
+            expires_at=999999,
+        )
+
+    def step(self, *, phase, observation, checkpoint, lease):
+        self.step_calls += 1
+        if not self.step_results:
+            raise AssertionError("missing step result")
+        return self.step_results.pop(0)
+
+    def reconcile(self, *, observation, checkpoint, lease):
+        self.reconcile_calls += 1
+        if not self.reconcile_results:
+            raise AssertionError("missing reconcile result")
+        return self.reconcile_results.pop(0)
+
+    def persist_checkpoint(self, checkpoint, *, lease):
+        if self.persist_failure is not None:
+            raise self.persist_failure
+        self.checkpoint = checkpoint
+        self.persisted.append(checkpoint)
+
+    def persist_owner_action_required(self, payload, *, observation, checkpoint, lease):
+        self.owner_records.append(dict(payload))
+        return "evidence:owner-action-1"
+
+    def release_resume_lease(self, lease, *, outcome_code):
+        self.released.append((lease.lease_id, outcome_code))
+
+    @staticmethod
+    def assert_mission(mission_id):
+        if mission_id != "v2r2c4":
+            raise AssertionError("unexpected mission")
+
+
+def success(code="OK", *, next_phase=None, material_digest=None):
+    return StepResult(
+        StepStatus.SUCCEEDED,
+        code,
+        "cond-step",
+        receipt_refs=("receipt:" + code,),
+        next_phase=next_phase,
+        material_digest=material_digest,
+    )
+
+
+class AutonomousResumeControllerTests(unittest.TestCase):
+    def setUp(self):
+        self.port = FakePort()
+        self.now = 100.0
+        self.controller = AutonomousResumeController(
+            self.port,
+            policy=ResumePolicy(
+                resume_lease_seconds=120,
+                default_poll_seconds=30,
+                deterministic_block_poll_seconds=300,
+                transient_max_attempts=3,
+                transient_base_delay_seconds=2,
+                transient_max_delay_seconds=60,
+            ),
+            clock=lambda: self.now,
+        )
+
+    def cycle(self):
+        return self.controller.run_once("v2r2c4")
+
+    def test_initial_success_advances_from_admission_to_execution(self):
+        self.port.step_results=[success("ADMITTED")]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.PROGRESSED)
+        self.assertEqual(self.port.checkpoint.phase, MissionPhase.S2_EXECUTION)
+        self.assertEqual(self.port.step_calls, 1)
+        self.assertEqual(self.port.reconcile_calls, 0)
+        self.assertTrue(self.port.released)
+
+    def test_resume_lease_denied_means_active_elsewhere_without_mutation(self):
+        self.port.lease_available=False
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.ACTIVE_ELSEWHERE)
+        self.assertEqual(self.port.step_calls, 0)
+        self.assertEqual(self.port.persisted, [])
+
+    def test_unknown_effect_reconciles_before_phase_step(self):
+        self.port.observation=replace(
+            self.port.observation,
+            effect_state=EffectState.UNKNOWN,
+        )
+        self.port.reconcile_results=[success("RECONCILED", next_phase=MissionPhase.S2_ADMISSION)]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.PROGRESSED)
+        self.assertEqual(self.port.reconcile_calls, 1)
+        self.assertEqual(self.port.step_calls, 0)
+
+    def test_requires_reconcile_uses_reconcile_port(self):
+        self.port.observation=replace(
+            self.port.observation,
+            requires_reconcile=True,
+        )
+        self.port.reconcile_results=[success("RECONCILED", next_phase=MissionPhase.S2_ADMISSION)]
+        self.cycle()
+        self.assertEqual(self.port.reconcile_calls, 1)
+        self.assertEqual(self.port.step_calls, 0)
+
+    def test_transient_retry_uses_exponential_backoff(self):
+        self.port.step_results=[
+            StepResult(StepStatus.RETRYABLE,"TIMEOUT","transient"),
+        ]
+        first=self.cycle()
+        self.assertEqual(first.status, OutcomeStatus.RETRY_SCHEDULED)
+        self.assertEqual(first.next_wake_after_seconds,2)
+        self.assertEqual(self.port.checkpoint.attempt_count,1)
+
+        self.port.step_results=[
+            StepResult(StepStatus.RETRYABLE,"TIMEOUT","transient"),
+        ]
+        second=self.cycle()
+        self.assertEqual(second.next_wake_after_seconds,4)
+        self.assertEqual(self.port.checkpoint.attempt_count,2)
+
+    def test_retry_budget_exhaustion_blocks_until_condition_change(self):
+        for _ in range(3):
+            self.port.step_results=[
+                StepResult(StepStatus.RETRYABLE,"TIMEOUT","cond-1"),
+            ]
+            self.cycle()
+        self.port.step_results=[
+            StepResult(StepStatus.RETRYABLE,"TIMEOUT","cond-1"),
+        ]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.WAIT_CONDITION_CHANGE)
+        self.assertEqual(self.port.checkpoint.blocked_condition_fingerprint,"cond-1")
+
+        # Same observed condition suppresses another consequential step.
+        step_calls=self.port.step_calls
+        self.port.observation=replace(
+            self.port.observation,
+            condition_fingerprint="cond-1",
+        )
+        out2=self.cycle()
+        self.assertEqual(out2.status, OutcomeStatus.WAIT_CONDITION_CHANGE)
+        self.assertEqual(self.port.step_calls,step_calls)
+
+    def test_condition_change_clears_deterministic_block(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            blocked_condition_fingerprint="old",
+            attempt_count=4,
+        )
+        self.port.observation=replace(
+            self.port.observation,
+            condition_fingerprint="new",
+        )
+        self.port.step_results=[success("RECOVERED")]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.PROGRESSED)
+        self.assertIsNone(self.port.checkpoint.blocked_condition_fingerprint)
+        self.assertEqual(self.port.checkpoint.attempt_count,0)
+
+    def test_deterministic_blocker_is_persisted_without_retry_storm(self):
+        self.port.step_results=[
+            StepResult(
+                StepStatus.BLOCKED_DETERMINISTIC,
+                "WORKER_IDENTITY_UNAVAILABLE",
+                "identity-v1",
+            )
+        ]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.WAIT_CONDITION_CHANGE)
+        self.assertEqual(out.next_wake_after_seconds,300)
+
+    def test_owner_blocker_persists_record_and_stops_only_phase(self):
+        self.port.step_results=[
+            StepResult(
+                StepStatus.BLOCKED_OWNER,
+                "HARDWARE_SIGNATURE_REQUIRED",
+                "owner-boundary-1",
+                owner_action_payload={"envelope_sha256":"abc"},
+            )
+        ]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.OWNER_ACTION_REQUIRED)
+        self.assertEqual(self.port.checkpoint.phase, MissionPhase.OWNER_ACTION_REQUIRED)
+        self.assertEqual(self.port.checkpoint.resume_phase_after_owner, MissionPhase.S2_ADMISSION)
+        self.assertEqual(len(self.port.owner_records),1)
+        self.assertIn("evidence:owner-action-1",self.port.checkpoint.receipt_refs)
+
+    def test_owner_action_resolved_resumes_previous_phase(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            phase=MissionPhase.OWNER_ACTION_REQUIRED,
+            resume_phase_after_owner=MissionPhase.GATE0,
+        )
+        self.port.observation=replace(
+            self.port.observation,
+            owner_action_resolved=True,
+        )
+        self.port.step_results=[
+            success("GATE0_PASS", next_phase=MissionPhase.RELEASE_READY),
+        ]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.PROGRESSED)
+        self.assertEqual(self.port.checkpoint.phase, MissionPhase.RELEASE_READY)
+
+    def test_owner_action_unresolved_returns_without_acquiring_lease(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            phase=MissionPhase.OWNER_ACTION_REQUIRED,
+            resume_phase_after_owner=MissionPhase.GATE0,
+        )
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.OWNER_ACTION_REQUIRED)
+        self.assertEqual(self.port.released,[])
+
+    def test_generation_increase_rewinds_to_admission(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            phase=MissionPhase.GATE0,
+            generation=1,
+        )
+        self.port.observation=replace(
+            self.port.observation,
+            generation=2,
+            head_oid="b"*40,
+            material_digest="mat-2",
+        )
+        self.port.step_results=[success("READMITTED")]
+        self.cycle()
+        self.assertEqual(self.port.checkpoint.phase, MissionPhase.S2_EXECUTION)
+        self.assertEqual(self.port.checkpoint.generation,2)
+
+    def test_checkpoint_generation_ahead_is_rejected(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(cp,generation=2)
+        with self.assertRaisesRegex(ValueError,"checkpoint_generation_ahead"):
+            self.cycle()
+
+    def test_head_change_before_ready_rewinds_to_admission(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(cp,phase=MissionPhase.GATE0)
+        self.port.observation=replace(
+            self.port.observation,
+            head_oid="b"*40,
+            material_digest="mat-2",
+        )
+        self.port.step_results=[success("READMITTED")]
+        self.cycle()
+        self.assertEqual(self.port.checkpoint.phase, MissionPhase.S2_EXECUTION)
+        self.assertEqual(self.port.checkpoint.head_oid,"b"*40)
+
+    def test_release_ready_success_freezes_material_digest(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(cp,phase=MissionPhase.RELEASE_READY)
+        self.port.step_results=[
+            success("READY",material_digest="frozen-1"),
+        ]
+        self.cycle()
+        self.assertEqual(self.port.checkpoint.phase,MissionPhase.PROMOTION)
+        self.assertEqual(self.port.checkpoint.frozen_material_digest,"frozen-1")
+
+    def test_material_change_after_ready_rewinds_to_release_ready(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            phase=MissionPhase.PROMOTION,
+            frozen_material_digest="frozen-1",
+            material_digest="frozen-1",
+        )
+        self.port.observation=replace(
+            self.port.observation,
+            material_digest="changed",
+        )
+        self.port.step_results=[success("READY_AGAIN",material_digest="changed")]
+        self.cycle()
+        self.assertEqual(self.port.checkpoint.phase,MissionPhase.PROMOTION)
+        self.assertEqual(self.port.checkpoint.frozen_material_digest,"changed")
+
+    def test_complete_checkpoint_suppresses_all_effects(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(cp,phase=MissionPhase.COMPLETE)
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.COMPLETE)
+        self.assertEqual(self.port.step_calls,0)
+        self.assertEqual(self.port.released,[])
+
+    def test_final_acceptance_success_completes_mission(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(cp,phase=MissionPhase.FINAL_ACCEPTANCE)
+        self.port.step_results=[success("FINAL_OK")]
+        out=self.cycle()
+        self.assertEqual(out.status, OutcomeStatus.COMPLETE)
+        self.assertEqual(self.port.checkpoint.phase,MissionPhase.COMPLETE)
+
+    def test_reconcile_status_schedules_immediate_reconcile_cycle(self):
+        self.port.step_results=[
+            StepResult(StepStatus.RECONCILE,"CAS_UNKNOWN","cas-unknown")
+        ]
+        out=self.cycle()
+        self.assertEqual(out.status,OutcomeStatus.IN_PROGRESS)
+        self.assertEqual(out.next_wake_after_seconds,0)
+        self.assertTrue(out.code.startswith("RECONCILE_REQUIRED:"))
+
+    def test_step_receipts_are_deduplicated_and_persisted(self):
+        self.port.observation=replace(
+            self.port.observation,
+            evidence_refs=("e:obs",),
+        )
+        self.port.step_results=[
+            StepResult(
+                StepStatus.SUCCEEDED,
+                "OK",
+                "cond-step",
+                receipt_refs=("e:obs","e:step"),
+            )
+        ]
+        self.cycle()
+        self.assertEqual(
+            self.port.checkpoint.receipt_refs,
+            ("e:obs","e:step"),
+        )
+
+
+
+    def test_stale_resume_lease_is_rejected_before_effect(self):
+        self.port.lease_override=ResumeLease(
+            lease_id="stale",
+            mission_id="v2r2c4",
+            generation=1,
+            fencing_token="fence-stale",
+            expires_at=99,
+        )
+        with self.assertRaisesRegex(ValueError,"resume_lease_expired"):
+            self.cycle()
+        self.assertEqual(self.port.step_calls,0)
+        self.assertEqual(self.port.released,[("stale","INVALID_RESUME_LEASE")])
+
+    def test_wrong_generation_resume_lease_is_rejected_before_effect(self):
+        self.port.lease_override=ResumeLease(
+            lease_id="wrong-gen",
+            mission_id="v2r2c4",
+            generation=2,
+            fencing_token="fence-2",
+            expires_at=999,
+        )
+        with self.assertRaisesRegex(ValueError,"resume_lease_generation_mismatch"):
+            self.cycle()
+        self.assertEqual(self.port.step_calls,0)
+        self.assertEqual(self.port.released,[("wrong-gen","INVALID_RESUME_LEASE")])
+
+    def test_checkpoint_persist_failure_still_releases_resume_lease(self):
+        self.port.step_results=[success("ADMITTED")]
+        self.port.persist_failure=RuntimeError("persist failed")
+        with self.assertRaisesRegex(RuntimeError,"persist failed"):
+            self.cycle()
+        self.assertEqual(self.port.released,[("lease-1","ADMITTED")])
+
+    def test_promotion_step_material_mismatch_rewinds_and_reconciles(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            phase=MissionPhase.PROMOTION,
+            frozen_material_digest="frozen-1",
+            material_digest="frozen-1",
+        )
+        self.port.observation=replace(
+            self.port.observation,
+            material_digest="frozen-1",
+        )
+        self.port.step_results=[
+            success("PROMOTED",material_digest="late-change")
+        ]
+        out=self.cycle()
+        self.assertEqual(out.status,OutcomeStatus.IN_PROGRESS)
+        self.assertEqual(self.port.checkpoint.phase,MissionPhase.RELEASE_READY)
+        self.assertTrue(self.port.checkpoint.reconcile_required)
+        self.assertIsNone(self.port.checkpoint.frozen_material_digest)
+        self.assertEqual(out.next_wake_after_seconds,0)
+
+    def test_step_reconcile_flag_is_persisted_until_reconcile_succeeds(self):
+        self.port.step_results=[
+            StepResult(StepStatus.RECONCILE,"CAS_UNKNOWN","cas-unknown")
+        ]
+        first=self.cycle()
+        self.assertTrue(self.port.checkpoint.reconcile_required)
+        self.assertEqual(first.next_wake_after_seconds,0)
+
+        # Next observation is ordinary, but durable checkpoint forces reconcile.
+        self.port.reconcile_results=[
+            success("RECONCILED",next_phase=MissionPhase.S2_ADMISSION)
+        ]
+        second=self.cycle()
+        self.assertEqual(self.port.reconcile_calls,1)
+        self.assertFalse(self.port.checkpoint.reconcile_required)
+        self.assertEqual(second.status,OutcomeStatus.PROGRESSED)
+
+    def test_material_change_during_active_promotion_reconciles_before_prepare(self):
+        cp=MissionCheckpoint.initial(self.port.observation,observed_at=1)
+        self.port.checkpoint=replace(
+            cp,
+            phase=MissionPhase.PROMOTION,
+            frozen_material_digest="frozen-1",
+            material_digest="frozen-1",
+        )
+        self.port.observation=replace(
+            self.port.observation,
+            material_digest="changed",
+            effect_state=EffectState.ACTIVE,
+        )
+        self.port.reconcile_results=[
+            success("PROMOTION_RECONCILED",next_phase=MissionPhase.RELEASE_READY)
+        ]
+        out=self.cycle()
+        self.assertEqual(self.port.reconcile_calls,1)
+        self.assertEqual(self.port.step_calls,0)
+        self.assertEqual(self.port.checkpoint.phase,MissionPhase.RELEASE_READY)
+        self.assertEqual(out.status,OutcomeStatus.PROGRESSED)
+
+    def test_resume_idempotency_is_stable_per_mission_generation(self):
+        key1=self.controller._resume_idempotency_key(self.port.observation)
+        key2=self.controller._resume_idempotency_key(self.port.observation)
+        self.assertEqual(key1,key2)
+        other=replace(self.port.observation,generation=2)
+        self.assertNotEqual(
+            key1,
+            self.controller._resume_idempotency_key(other),
+        )
+
+
+if __name__=="__main__":
+    unittest.main()

tests/task_plane/test_autonomous_resume_integration.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import unittest
+
+from task_plane.autonomous_resume_controller import (
+    AutonomousResumeController,
+    MissionCheckpoint,
+    MissionObservation,
+    MissionPhase,
+    OutcomeStatus,
+    ResumeLease,
+    StepResult,
+    StepStatus,
+)
+from task_plane.autonomous_resume_integration import (
+    AuthorityCallbacks,
+    AutonomousSchedulerTick,
+    CallableAutonomousExecutionPort,
+    PhaseCallbacks,
+)
+
+
+class IntegrationAdapterTests(unittest.TestCase):
+    def setUp(self):
+        self.observation=MissionObservation(
+            mission_id="v2r2c4",
+            generation=1,
+            head_oid="a"*40,
+            material_digest="mat-1",
+            condition_fingerprint="cond-1",
+        )
+        self.checkpoint=None
+        self.phase_seen=[]
+        self.released=[]
+
+        def phase(name):
+            def _handler(obs, cp, lease):
+                self.phase_seen.append(name)
+                return StepResult(
+                    StepStatus.SUCCEEDED,
+                    name.upper(),
+                    "cond-step",
+                )
+            return _handler
+
+        self.phases=PhaseCallbacks(
+            admission=phase("admission"),
+            source_execution=phase("source"),
+            gate0=phase("gate0"),
+            release_ready=phase("ready"),
+            promotion=phase("promotion"),
+            final_acceptance=phase("final"),
+        )
+
+        self.authority=AuthorityCallbacks(
+            read_observation=lambda mission_id:self.observation,
+            load_checkpoint=lambda mission_id:self.checkpoint,
+            acquire_resume_lease=lambda mission_id,generation,key,seconds:ResumeLease(
+                lease_id="lease-1",
+                mission_id=mission_id,
+                generation=generation,
+                fencing_token="fence-1",
+                expires_at=999,
+            ),
+            persist_checkpoint=self._persist,
+            persist_owner_action_required=lambda payload,obs,cp,lease:"owner:1",
+            release_resume_lease=lambda lease,code:self.released.append((lease.lease_id,code)),
+            reconcile=lambda obs,cp,lease:StepResult(
+                StepStatus.SUCCEEDED,
+                "RECONCILED",
+                "cond-r",
+                next_phase=cp.phase,
+            ),
+        )
+        self.port=CallableAutonomousExecutionPort(
+            self.authority,
+            self.phases,
+        )
+
+    def _persist(self, checkpoint, lease):
+        self.checkpoint=checkpoint
+
+    def test_phase_dispatch_maps_all_executable_phases(self):
+        expected={
+            MissionPhase.S2_ADMISSION:"admission",
+            MissionPhase.S2_EXECUTION:"source",
+            MissionPhase.GATE0:"gate0",
+            MissionPhase.RELEASE_READY:"ready",
+            MissionPhase.PROMOTION:"promotion",
+            MissionPhase.FINAL_ACCEPTANCE:"final",
+        }
+        lease=ResumeLease("l","v2r2c4",1,"f",999)
+        cp=MissionCheckpoint.initial(self.observation,observed_at=1)
+        for phase,name in expected.items():
+            self.phases.for_phase(phase)(self.observation,cp,lease)
+            self.assertEqual(self.phase_seen[-1],name)
+
+    def test_non_executable_phase_fails_closed(self):
+        with self.assertRaisesRegex(ValueError,"phase_has_no_executor"):
+            self.phases.for_phase(MissionPhase.COMPLETE)
+
+    def test_callable_port_runs_controller(self):
+        controller=AutonomousResumeController(
+            self.port,
+            clock=lambda:100,
+        )
+        out=controller.run_once("v2r2c4")
+        self.assertEqual(out.status,OutcomeStatus.PROGRESSED)
+        self.assertEqual(self.phase_seen,["admission"])
+        self.assertEqual(self.checkpoint.phase,MissionPhase.S2_EXECUTION)
+
+    def test_scheduler_tick_does_not_loop(self):
+        controller=AutonomousResumeController(
+            self.port,
+            clock=lambda:100,
+        )
+        tick=AutonomousSchedulerTick(controller,"v2r2c4")
+        result=tick.run()
+        self.assertEqual(result.outcome.status,OutcomeStatus.PROGRESSED)
+        self.assertEqual(result.next_wake_after_seconds,30)
+        self.assertEqual(len(self.phase_seen),1)
+
+    def test_wrong_read_observation_type_rejected(self):
+        authority=AuthorityCallbacks(
+            read_observation=lambda mission_id:{"bad":True},
+            load_checkpoint=self.authority.load_checkpoint,
+            acquire_resume_lease=self.authority.acquire_resume_lease,
+            persist_checkpoint=self.authority.persist_checkpoint,
+            persist_owner_action_required=self.authority.persist_owner_action_required,
+            release_resume_lease=self.authority.release_resume_lease,
+            reconcile=self.authority.reconcile,
+        )
+        port=CallableAutonomousExecutionPort(authority,self.phases)
+        with self.assertRaisesRegex(TypeError,"MissionObservation"):
+            port.read_observation("v2r2c4")
+
+    def test_wrong_lease_type_rejected(self):
+        authority=AuthorityCallbacks(
+            read_observation=self.authority.read_observation,
+            load_checkpoint=self.authority.load_checkpoint,
+            acquire_resume_lease=lambda mission_id,generation,key,seconds:"bad",
+            persist_checkpoint=self.authority.persist_checkpoint,
+            persist_owner_action_required=self.authority.persist_owner_action_required,
+            release_resume_lease=self.authority.release_resume_lease,
+            reconcile=self.authority.reconcile,
+        )
+        port=CallableAutonomousExecutionPort(authority,self.phases)
+        with self.assertRaisesRegex(TypeError,"lease_or_none"):
+            port.acquire_resume_lease(
+                mission_id="v2r2c4",
+                generation=1,
+                idempotency_key="k",
+                lease_seconds=120,
+            )
+
+    def test_empty_owner_action_record_rejected(self):
+        authority=AuthorityCallbacks(
+            read_observation=self.authority.read_observation,
+            load_checkpoint=self.authority.load_checkpoint,
+            acquire_resume_lease=self.authority.acquire_resume_lease,
+            persist_checkpoint=self.authority.persist_checkpoint,
+            persist_owner_action_required=lambda payload,obs,cp,lease:"",
+            release_resume_lease=self.authority.release_resume_lease,
+            reconcile=self.authority.reconcile,
+        )
+        port=CallableAutonomousExecutionPort(authority,self.phases)
+        cp=MissionCheckpoint.initial(self.observation,observed_at=1)
+        lease=ResumeLease("l","v2r2c4",1,"f",999)
+        with self.assertRaisesRegex(ValueError,"owner_action_record_ref_required"):
+            port.persist_owner_action_required(
+                {},
+                observation=self.observation,
+                checkpoint=cp,
+                lease=lease,
+            )
+
+
+if __name__=="__main__":
+    unittest.main()