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

active-runtime/app/aios_agent_task_protocol.py

@@ -0,0 +1,829 @@
+#!/usr/bin/env python3
+"""Durable online task protocol for external AIOS agents.
+
+The store is deliberately independent from Git/Gitea mutation. It provides a
+small durable state machine for publishing work, atomic claim/lease ownership,
+heartbeat, idempotent result submission, explicit server-side verification,
+lease recovery, and audit events.
+
+No plaintext agent credential or task lease token is persisted.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+import hashlib
+import hmac
+import json
+from pathlib import Path
+import secrets
+import sqlite3
+from typing import Any, Callable, Iterable, Mapping
+import uuid
+
+from aios_capability_contract import has_capabilities
+
+
+TERMINAL_STATES = frozenset({"SUCCEEDED", "FAILED", "EXPIRED", "CANCELLED"})
+ACTIVE_LEASE_STATES = frozenset({"CLAIMED", "RUNNING"})
+TASK_STATES = frozenset({
+    "QUEUED", "CLAIMED", "RUNNING", "VERIFYING",
+    "SUCCEEDED", "FAILED", "EXPIRED", "CANCELLED",
+})
+VERIFY_DECISIONS = frozenset({"SUCCEEDED", "NEEDS_REWORK", "FAILED"})
+
+
+class TaskProtocolError(RuntimeError):
+    def __init__(self, code: str, http_status: int = 409) -> None:
+        super().__init__(code)
+        self.code = code
+        self.http_status = http_status
+
+
+@dataclass(frozen=True, slots=True)
+class Principal:
+    kind: str
+    identifier: str
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def iso(value: datetime) -> str:
+    return value.astimezone(timezone.utc).isoformat()
+
+
+def parse_time(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    parsed = datetime.fromisoformat(value)
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def canonical_json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def digest_json(value: Any) -> str:
+    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def _string_list(value: Iterable[object] | None, field: str, *, max_items: int = 128) -> list[str]:
+    if value is None:
+        return []
+    if isinstance(value, (str, bytes)) or not isinstance(value, Iterable):
+        raise TaskProtocolError(field.upper() + "_INVALID", 400)
+    result: list[str] = []
+    for item in value:
+        text = str(item).strip()
+        if not text or len(text) > 160:
+            raise TaskProtocolError(field.upper() + "_INVALID", 400)
+        result.append(text)
+        if len(result) > max_items:
+            raise TaskProtocolError(field.upper() + "_TOO_LARGE", 400)
+    return sorted(set(result))
+
+
+def _json_object(value: object | None, field: str, *, max_bytes: int = 128 * 1024) -> dict[str, Any]:
+    if value is None:
+        return {}
+    if not isinstance(value, Mapping):
+        raise TaskProtocolError(field.upper() + "_INVALID", 400)
+    result = dict(value)
+    if len(canonical_json(result).encode("utf-8")) > max_bytes:
+        raise TaskProtocolError(field.upper() + "_TOO_LARGE", 413)
+    return result
+
+
+def _criteria(value: object | None) -> list[Any]:
+    if value is None:
+        return []
+    if isinstance(value, (str, bytes)) or not isinstance(value, list):
+        raise TaskProtocolError("SUCCESS_CRITERIA_INVALID", 400)
+    if len(value) > 128:
+        raise TaskProtocolError("SUCCESS_CRITERIA_TOO_LARGE", 413)
+    if len(canonical_json(value).encode("utf-8")) > 128 * 1024:
+        raise TaskProtocolError("SUCCESS_CRITERIA_TOO_LARGE", 413)
+    return list(value)
+
+
+class TaskStore:
+    """SQLite-backed durable task store.
+
+    SQLite BEGIN IMMEDIATE serializes claims so two agents cannot successfully
+    claim the same queued task. WAL mode allows concurrent readers.
+    """
+
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        now: Callable[[], datetime] = utcnow,
+        token_pepper: str = "",
+    ) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.now = now
+        self.token_pepper = token_pepper
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path, timeout=10, isolation_level=None)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA synchronous=FULL")
+        connection.execute("PRAGMA busy_timeout=10000")
+        return connection
+
+    def _initialize(self) -> None:
+        with self._connect() as connection:
+            connection.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS agents (
+                    agent_id TEXT PRIMARY KEY,
+                    agent_type TEXT NOT NULL,
+                    agent_version TEXT NOT NULL,
+                    capabilities_json TEXT NOT NULL,
+                    metadata_json TEXT NOT NULL,
+                    token_hash TEXT NOT NULL UNIQUE,
+                    status TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    last_seen_at TEXT NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS tasks (
+                    task_id TEXT PRIMARY KEY,
+                    goal TEXT NOT NULL,
+                    instructions TEXT NOT NULL,
+                    success_criteria_json TEXT NOT NULL,
+                    required_capabilities_json TEXT NOT NULL,
+                    constraints_json TEXT NOT NULL,
+                    metadata_json TEXT NOT NULL,
+                    deadline TEXT,
+                    priority INTEGER NOT NULL,
+                    max_attempts INTEGER NOT NULL,
+                    attempt INTEGER NOT NULL DEFAULT 0,
+                    status TEXT NOT NULL,
+                    claimed_by TEXT,
+                    lease_token_hash TEXT,
+                    lease_expires_at TEXT,
+                    result_json TEXT,
+                    result_digest TEXT,
+                    result_idempotency_key TEXT,
+                    verification_json TEXT,
+                    revision INTEGER NOT NULL DEFAULT 1,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL,
+                    FOREIGN KEY(claimed_by) REFERENCES agents(agent_id)
+                );
+
+                CREATE TABLE IF NOT EXISTS idempotency (
+                    scope TEXT NOT NULL,
+                    key TEXT NOT NULL,
+                    payload_digest TEXT NOT NULL,
+                    response_json TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    PRIMARY KEY(scope, key)
+                );
+
+                CREATE TABLE IF NOT EXISTS events (
+                    event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    task_id TEXT,
+                    agent_id TEXT,
+                    event_type TEXT NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    created_at TEXT NOT NULL
+                );
+
+                CREATE INDEX IF NOT EXISTS idx_tasks_queue
+                ON tasks(status, priority DESC, created_at);
+
+                CREATE INDEX IF NOT EXISTS idx_tasks_lease
+                ON tasks(status, lease_expires_at);
+
+                CREATE INDEX IF NOT EXISTS idx_events_task
+                ON events(task_id, event_id);
+                """
+            )
+
+    def _token_hash(self, token: str) -> str:
+        return hashlib.sha256((self.token_pepper + "\0" + token).encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _row_task(row: sqlite3.Row, *, include_result: bool = True) -> dict[str, Any]:
+        value: dict[str, Any] = {
+            "task_id": row["task_id"],
+            "goal": row["goal"],
+            "instructions": row["instructions"],
+            "success_criteria": json.loads(row["success_criteria_json"]),
+            "required_capabilities": json.loads(row["required_capabilities_json"]),
+            "constraints": json.loads(row["constraints_json"]),
+            "metadata": json.loads(row["metadata_json"]),
+            "deadline": row["deadline"],
+            "priority": row["priority"],
+            "max_attempts": row["max_attempts"],
+            "attempt": row["attempt"],
+            "status": row["status"],
+            "claimed_by": row["claimed_by"],
+            "lease_expires_at": row["lease_expires_at"],
+            "revision": row["revision"],
+            "created_at": row["created_at"],
+            "updated_at": row["updated_at"],
+        }
+        if include_result:
+            value["result"] = json.loads(row["result_json"]) if row["result_json"] else None
+            value["verification"] = (
+                json.loads(row["verification_json"]) if row["verification_json"] else None
+            )
+        return value
+
+    def _event(
+        self,
+        connection: sqlite3.Connection,
+        event_type: str,
+        *,
+        task_id: str | None = None,
+        agent_id: str | None = None,
+        payload: Mapping[str, Any] | None = None,
+    ) -> None:
+        connection.execute(
+            "INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at)"
+            " VALUES(?,?,?,?,?)",
+            (task_id, agent_id, event_type, canonical_json(dict(payload or {})), iso(self.now())),
+        )
+
+    def register_agent(
+        self,
+        *,
+        agent_id: str,
+        agent_type: str,
+        agent_version: str,
+        capabilities: Iterable[object],
+        metadata: Mapping[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        agent_id = agent_id.strip()
+        agent_type = agent_type.strip()
+        agent_version = agent_version.strip()
+        if not agent_id or len(agent_id) > 160 or not agent_type or not agent_version:
+            raise TaskProtocolError("AGENT_IDENTITY_INVALID", 400)
+        caps = _string_list(capabilities, "capabilities")
+        meta = _json_object(metadata, "metadata")
+        token = "agt_" + secrets.token_urlsafe(32)
+        now = iso(self.now())
+        try:
+            with self._connect() as connection:
+                connection.execute("BEGIN IMMEDIATE")
+                connection.execute(
+                    "INSERT INTO agents(agent_id,agent_type,agent_version,capabilities_json,"
+                    "metadata_json,token_hash,status,created_at,last_seen_at)"
+                    " VALUES(?,?,?,?,?,?,?,?,?)",
+                    (
+                        agent_id,
+                        agent_type,
+                        agent_version,
+                        canonical_json(caps),
+                        canonical_json(meta),
+                        self._token_hash(token),
+                        "ACTIVE",
+                        now,
+                        now,
+                    ),
+                )
+                self._event(
+                    connection,
+                    "AGENT_REGISTERED",
+                    agent_id=agent_id,
+                    payload={"agent_type": agent_type, "agent_version": agent_version},
+                )
+                connection.commit()
+        except sqlite3.IntegrityError as exc:
+            raise TaskProtocolError("AGENT_ALREADY_EXISTS", 409) from exc
+        return {
+            "agent_id": agent_id,
+            "agent_type": agent_type,
+            "agent_version": agent_version,
+            "capabilities": caps,
+            "agent_token": token,
+            "credential_returned_once": True,
+        }
+
+    def authenticate_agent(self, token: str) -> dict[str, Any]:
+        if not token:
+            raise TaskProtocolError("UNAUTHORIZED", 401)
+        token_hash = self._token_hash(token)
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT * FROM agents WHERE token_hash=? AND status='ACTIVE'",
+                (token_hash,),
+            ).fetchone()
+            if row is None:
+                raise TaskProtocolError("UNAUTHORIZED", 401)
+            connection.execute(
+                "UPDATE agents SET last_seen_at=? WHERE agent_id=?",
+                (iso(self.now()), row["agent_id"]),
+            )
+            return {
+                "agent_id": row["agent_id"],
+                "agent_type": row["agent_type"],
+                "agent_version": row["agent_version"],
+                "capabilities": json.loads(row["capabilities_json"]),
+                "metadata": json.loads(row["metadata_json"]),
+            }
+
+    def _idempotent_lookup(
+        self,
+        connection: sqlite3.Connection,
+        *,
+        scope: str,
+        key: str,
+        payload_digest: str,
+    ) -> dict[str, Any] | None:
+        row = connection.execute(
+            "SELECT payload_digest,response_json FROM idempotency WHERE scope=? AND key=?",
+            (scope, key),
+        ).fetchone()
+        if row is None:
+            return None
+        if not hmac.compare_digest(row["payload_digest"], payload_digest):
+            raise TaskProtocolError("IDEMPOTENCY_CONFLICT", 409)
+        return json.loads(row["response_json"])
+
+    def _idempotent_store(
+        self,
+        connection: sqlite3.Connection,
+        *,
+        scope: str,
+        key: str,
+        payload_digest: str,
+        response: Mapping[str, Any],
+    ) -> None:
+        connection.execute(
+            "INSERT INTO idempotency(scope,key,payload_digest,response_json,created_at)"
+            " VALUES(?,?,?,?,?)",
+            (scope, key, payload_digest, canonical_json(dict(response)), iso(self.now())),
+        )
+
+    def create_task(
+        self,
+        *,
+        task_id: str | None,
+        goal: str,
+        instructions: str,
+        success_criteria: object | None,
+        required_capabilities: Iterable[object] | None = None,
+        constraints: Mapping[str, Any] | None = None,
+        metadata: Mapping[str, Any] | None = None,
+        deadline: str | None = None,
+        priority: int = 0,
+        max_attempts: int = 3,
+        idempotency_key: str,
+    ) -> dict[str, Any]:
+        task_id = (task_id or ("task-" + uuid.uuid4().hex)).strip()
+        goal = goal.strip()
+        instructions = instructions.strip()
+        idempotency_key = idempotency_key.strip()
+        if not task_id or len(task_id) > 180 or not goal or not instructions:
+            raise TaskProtocolError("TASK_BINDING_INVALID", 400)
+        if not idempotency_key or len(idempotency_key) > 160:
+            raise TaskProtocolError("IDEMPOTENCY_KEY_INVALID", 400)
+        if not isinstance(priority, int) or isinstance(priority, bool) or not -100 <= priority <= 100:
+            raise TaskProtocolError("PRIORITY_INVALID", 400)
+        if not isinstance(max_attempts, int) or isinstance(max_attempts, bool) or not 1 <= max_attempts <= 20:
+            raise TaskProtocolError("MAX_ATTEMPTS_INVALID", 400)
+        parsed_deadline = parse_time(deadline)
+        if parsed_deadline and parsed_deadline <= self.now():
+            raise TaskProtocolError("DEADLINE_INVALID", 400)
+        criteria = _criteria(success_criteria)
+        required = _string_list(required_capabilities, "required_capabilities")
+        constraints_value = _json_object(constraints, "constraints")
+        metadata_value = _json_object(metadata, "metadata")
+        payload = {
+            "task_id": task_id,
+            "goal": goal,
+            "instructions": instructions,
+            "success_criteria": criteria,
+            "required_capabilities": required,
+            "constraints": constraints_value,
+            "metadata": metadata_value,
+            "deadline": iso(parsed_deadline) if parsed_deadline else None,
+            "priority": priority,
+            "max_attempts": max_attempts,
+        }
+        payload_digest = digest_json(payload)
+        now = iso(self.now())
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            replay = self._idempotent_lookup(
+                connection,
+                scope="create_task",
+                key=idempotency_key,
+                payload_digest=payload_digest,
+            )
+            if replay is not None:
+                connection.commit()
+                return {**replay, "idempotent": True}
+            try:
+                connection.execute(
+                    "INSERT INTO tasks(task_id,goal,instructions,success_criteria_json,"
+                    "required_capabilities_json,constraints_json,metadata_json,deadline,priority,"
+                    "max_attempts,attempt,status,revision,created_at,updated_at)"
+                    " VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        task_id,
+                        goal,
+                        instructions,
+                        canonical_json(criteria),
+                        canonical_json(required),
+                        canonical_json(constraints_value),
+                        canonical_json(metadata_value),
+                        payload["deadline"],
+                        priority,
+                        max_attempts,
+                        0,
+                        "QUEUED",
+                        1,
+                        now,
+                        now,
+                    ),
+                )
+            except sqlite3.IntegrityError as exc:
+                raise TaskProtocolError("TASK_ID_CONFLICT", 409) from exc
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            result = self._row_task(row)
+            self._idempotent_store(
+                connection,
+                scope="create_task",
+                key=idempotency_key,
+                payload_digest=payload_digest,
+                response=result,
+            )
+            self._event(connection, "TASK_CREATED", task_id=task_id, payload={"priority": priority})
+            connection.commit()
+            return result
+
+    def _expire_locked(self, connection: sqlite3.Connection) -> None:
+        now_dt = self.now()
+        now = iso(now_dt)
+        rows = connection.execute(
+            "SELECT * FROM tasks WHERE status IN ('QUEUED','CLAIMED','RUNNING')"
+        ).fetchall()
+        for row in rows:
+            deadline = parse_time(row["deadline"])
+            if deadline is not None and deadline <= now_dt:
+                connection.execute(
+                    "UPDATE tasks SET status='EXPIRED',claimed_by=NULL,lease_token_hash=NULL,"
+                    "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                    (now, row["task_id"]),
+                )
+                self._event(connection, "TASK_EXPIRED", task_id=row["task_id"])
+                continue
+            if row["status"] in ACTIVE_LEASE_STATES:
+                lease_expires = parse_time(row["lease_expires_at"])
+                if lease_expires is not None and lease_expires <= now_dt:
+                    if row["attempt"] >= row["max_attempts"]:
+                        next_state = "FAILED"
+                    else:
+                        next_state = "QUEUED"
+                    if next_state == "FAILED":
+                        connection.execute(
+                            "UPDATE tasks SET status=?,lease_token_hash=NULL,"
+                            "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                            (next_state, now, row["task_id"]),
+                        )
+                    else:
+                        connection.execute(
+                            "UPDATE tasks SET status=?,claimed_by=NULL,lease_token_hash=NULL,"
+                            "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                            (next_state, now, row["task_id"]),
+                        )
+                    self._event(
+                        connection,
+                        "LEASE_EXPIRED",
+                        task_id=row["task_id"],
+                        agent_id=row["claimed_by"],
+                        payload={"next_state": next_state, "attempt": row["attempt"]},
+                    )
+
+    def expire_leases(self) -> int:
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            before = connection.total_changes
+            self._expire_locked(connection)
+            changed = connection.total_changes - before
+            connection.commit()
+            return changed
+
+    def poll_tasks(self, *, agent_id: str, limit: int = 20) -> list[dict[str, Any]]:
+        if not 1 <= int(limit) <= 100:
+            raise TaskProtocolError("LIMIT_INVALID", 400)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._expire_locked(connection)
+            agent = connection.execute(
+                "SELECT capabilities_json FROM agents WHERE agent_id=? AND status='ACTIVE'",
+                (agent_id,),
+            ).fetchone()
+            if agent is None:
+                raise TaskProtocolError("AGENT_NOT_FOUND", 404)
+            capabilities = json.loads(agent["capabilities_json"])
+            rows = connection.execute(
+                "SELECT * FROM tasks WHERE status='QUEUED' ORDER BY priority DESC,created_at ASC"
+            ).fetchall()
+            result = []
+            for row in rows:
+                required = json.loads(row["required_capabilities_json"])
+                if has_capabilities(capabilities, required):
+                    result.append(self._row_task(row, include_result=False))
+                    if len(result) >= int(limit):
+                        break
+            connection.commit()
+            return result
+
+    def _agent_capabilities(self, connection: sqlite3.Connection, agent_id: str) -> set[str]:
+        row = connection.execute(
+            "SELECT capabilities_json FROM agents WHERE agent_id=? AND status='ACTIVE'",
+            (agent_id,),
+        ).fetchone()
+        if row is None:
+            raise TaskProtocolError("AGENT_NOT_FOUND", 404)
+        return set(json.loads(row["capabilities_json"]))
+
+    def claim_task(
+        self,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_seconds: int = 300,
+    ) -> dict[str, Any]:
+        if not isinstance(lease_seconds, int) or isinstance(lease_seconds, bool) or not 30 <= lease_seconds <= 3600:
+            raise TaskProtocolError("LEASE_SECONDS_INVALID", 400)
+        lease_token = "lease_" + secrets.token_urlsafe(32)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._expire_locked(connection)
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["status"] != "QUEUED":
+                raise TaskProtocolError("TASK_NOT_CLAIMABLE", 409)
+            capabilities = self._agent_capabilities(connection, agent_id)
+            required = json.loads(row["required_capabilities_json"])
+            if not has_capabilities(capabilities, required):
+                raise TaskProtocolError("AGENT_CAPABILITY_MISMATCH", 403)
+            attempt = int(row["attempt"]) + 1
+            if attempt > int(row["max_attempts"]):
+                raise TaskProtocolError("TASK_ATTEMPTS_EXHAUSTED", 409)
+            expires = self.now() + timedelta(seconds=lease_seconds)
+            now = iso(self.now())
+            updated = connection.execute(
+                "UPDATE tasks SET status='CLAIMED',claimed_by=?,lease_token_hash=?,"
+                "lease_expires_at=?,attempt=?,revision=revision+1,updated_at=? "
+                "WHERE task_id=? AND status='QUEUED'",
+                (
+                    agent_id,
+                    self._token_hash(lease_token),
+                    iso(expires),
+                    attempt,
+                    now,
+                    task_id,
+                ),
+            )
+            if updated.rowcount != 1:
+                raise TaskProtocolError("TASK_CLAIM_CONFLICT", 409)
+            self._event(
+                connection,
+                "TASK_CLAIMED",
+                task_id=task_id,
+                agent_id=agent_id,
+                payload={"attempt": attempt, "lease_expires_at": iso(expires)},
+            )
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return {
+                **self._row_task(row, include_result=False),
+                "lease_token": lease_token,
+                "lease_token_returned_once": True,
+            }
+
+    def _validate_lease(
+        self,
+        connection: sqlite3.Connection,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_token: str,
+        allowed_states: frozenset[str] = ACTIVE_LEASE_STATES,
+    ) -> sqlite3.Row:
+        self._expire_locked(connection)
+        row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+        if row is None:
+            raise TaskProtocolError("TASK_NOT_FOUND", 404)
+        if row["status"] not in allowed_states:
+            raise TaskProtocolError("TASK_LEASE_NOT_ACTIVE", 409)
+        if row["claimed_by"] != agent_id:
+            raise TaskProtocolError("TASK_LEASE_OWNER_MISMATCH", 403)
+        if not row["lease_token_hash"] or not hmac.compare_digest(
+            row["lease_token_hash"], self._token_hash(lease_token)
+        ):
+            raise TaskProtocolError("TASK_LEASE_TOKEN_INVALID", 403)
+        expires = parse_time(row["lease_expires_at"])
+        if expires is None or expires <= self.now():
+            raise TaskProtocolError("TASK_LEASE_EXPIRED", 409)
+        return row
+
+    def heartbeat(
+        self,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_token: str,
+        progress: Mapping[str, Any] | None = None,
+        lease_seconds: int = 300,
+    ) -> dict[str, Any]:
+        if not 30 <= int(lease_seconds) <= 3600:
+            raise TaskProtocolError("LEASE_SECONDS_INVALID", 400)
+        progress_value = _json_object(progress, "progress", max_bytes=64 * 1024)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._validate_lease(
+                connection,
+                task_id=task_id,
+                agent_id=agent_id,
+                lease_token=lease_token,
+            )
+            expires = self.now() + timedelta(seconds=int(lease_seconds))
+            now = iso(self.now())
+            connection.execute(
+                "UPDATE tasks SET status='RUNNING',lease_expires_at=?,revision=revision+1,"
+                "updated_at=? WHERE task_id=?",
+                (iso(expires), now, task_id),
+            )
+            self._event(
+                connection,
+                "TASK_HEARTBEAT",
+                task_id=task_id,
+                agent_id=agent_id,
+                payload={"progress": progress_value, "lease_expires_at": iso(expires)},
+            )
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(row, include_result=False)
+
+    def submit_result(
+        self,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_token: str,
+        idempotency_key: str,
+        result: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        idempotency_key = idempotency_key.strip()
+        if not idempotency_key or len(idempotency_key) > 160:
+            raise TaskProtocolError("IDEMPOTENCY_KEY_INVALID", 400)
+        result_value = _json_object(result, "result", max_bytes=1024 * 1024)
+        result_digest = digest_json(result_value)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["result_idempotency_key"] == idempotency_key:
+                if not row["result_digest"] or not hmac.compare_digest(row["result_digest"], result_digest):
+                    raise TaskProtocolError("IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {**self._row_task(row), "idempotent": True}
+            self._validate_lease(
+                connection,
+                task_id=task_id,
+                agent_id=agent_id,
+                lease_token=lease_token,
+            )
+            now = iso(self.now())
+            connection.execute(
+                "UPDATE tasks SET status='VERIFYING',result_json=?,result_digest=?,"
+                "result_idempotency_key=?,lease_token_hash=NULL,lease_expires_at=NULL,"
+                "revision=revision+1,updated_at=? WHERE task_id=?",
+                (
+                    canonical_json(result_value),
+                    result_digest,
+                    idempotency_key,
+                    now,
+                    task_id,
+                ),
+            )
+            self._event(
+                connection,
+                "RESULT_SUBMITTED",
+                task_id=task_id,
+                agent_id=agent_id,
+                payload={"result_digest": result_digest, "idempotency_key": idempotency_key},
+            )
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(row)
+
+    def verify_task(
+        self,
+        *,
+        task_id: str,
+        decision: str,
+        verification: Mapping[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        decision = decision.strip().upper()
+        if decision not in VERIFY_DECISIONS:
+            raise TaskProtocolError("VERIFICATION_DECISION_INVALID", 400)
+        verification_value = _json_object(verification, "verification")
+        verification_value = {"decision": decision, **verification_value}
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["status"] != "VERIFYING":
+                raise TaskProtocolError("TASK_NOT_VERIFYING", 409)
+            if decision == "SUCCEEDED":
+                next_state = "SUCCEEDED"
+            elif decision == "FAILED":
+                next_state = "FAILED"
+            elif int(row["attempt"]) >= int(row["max_attempts"]):
+                next_state = "FAILED"
+                verification_value["reason"] = verification_value.get(
+                    "reason", "MAX_ATTEMPTS_EXHAUSTED"
+                )
+            else:
+                next_state = "QUEUED"
+            now = iso(self.now())
+            if next_state in {"SUCCEEDED", "FAILED"}:
+                connection.execute(
+                    "UPDATE tasks SET status=?,claimed_by=NULL,lease_token_hash=NULL,"
+                    "lease_expires_at=NULL,verification_json=?,revision=revision+1,updated_at=? "
+                    "WHERE task_id=?",
+                    (next_state, canonical_json(verification_value), now, task_id),
+                )
+            else:
+                connection.execute(
+                    "UPDATE tasks SET status=?,claimed_by=NULL,lease_token_hash=NULL,"
+                    "lease_expires_at=NULL,verification_json=?,revision=revision+1,updated_at=? "
+                    "WHERE task_id=?",
+                    (next_state, canonical_json(verification_value), now, task_id),
+                )
+            self._event(
+                connection,
+                "TASK_VERIFIED",
+                task_id=task_id,
+                agent_id=row["claimed_by"],
+                payload={"decision": decision, "next_state": next_state},
+            )
+            updated = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(updated)
+
+    def cancel_task(self, *, task_id: str, reason: str = "") -> dict[str, Any]:
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["status"] in TERMINAL_STATES:
+                connection.commit()
+                return self._row_task(row)
+            connection.execute(
+                "UPDATE tasks SET status='CANCELLED',lease_token_hash=NULL,"
+                "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                (iso(self.now()), task_id),
+            )
+            self._event(connection, "TASK_CANCELLED", task_id=task_id, payload={"reason": reason})
+            updated = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(updated)
+
+    def get_task(self, task_id: str) -> dict[str, Any]:
+        self.expire_leases()
+        with self._connect() as connection:
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            return self._row_task(row)
+
+    def get_events(self, task_id: str, *, limit: int = 200) -> list[dict[str, Any]]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                "SELECT * FROM events WHERE task_id=? ORDER BY event_id DESC LIMIT ?",
+                (task_id, max(1, min(int(limit), 1000))),
+            ).fetchall()
+            return [
+                {
+                    "event_id": row["event_id"],
+                    "task_id": row["task_id"],
+                    "agent_id": row["agent_id"],
+                    "event_type": row["event_type"],
+                    "payload": json.loads(row["payload_json"]),
+                    "created_at": row["created_at"],
+                }
+                for row in reversed(rows)
+            ]

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()