feat(online-agent): add durable checkpoint and retry-safe resume

active-runtime/app/aios_agent_task_retry_extension.py

@@ -0,0 +1,427 @@
+"""Owner-controlled, CAS-bound retry extension for exhausted Online Agent Tasks."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+from typing import Any, Mapping
+
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,159}$")
+MAX_BODY = 64 * 1024
+AUTHORIZE_EXTENSION_ID_ENV = "AIOS_AGENT_TASK_AUTHORIZATION_ID"
+
+
+class RetryExtensionError(RuntimeError):
+    def __init__(self, code: str, status: int = 409) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+def _canonical(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def _sha(value: object) -> str:
+    return hashlib.sha256(_canonical(value)).hexdigest()
+
+
+def _required_authorization_id() -> str:
+    value = os.environ.get(AUTHORIZE_EXTENSION_ID_ENV, "").strip()
+    if not value:
+        raise RuntimeError("TASK_RETRY_AUTHORIZATION_ID_MISSING")
+    if KEY_RE.fullmatch(value) is None:
+        raise RuntimeError("TASK_RETRY_AUTHORIZATION_ID_INVALID")
+    return value
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _require_string(value: object, code: str, pattern: re.Pattern[str] | None = None) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise RetryExtensionError(code, 400)
+    value = value.strip()
+    if pattern is not None and pattern.fullmatch(value) is None:
+        raise RetryExtensionError(code, 400)
+    return value
+
+
+@dataclass(frozen=True)
+class RetryExtensionRequest:
+    expected_status: str
+    expected_revision: int
+    expected_attempt: int
+    expected_max_attempts: int
+    next_max_attempts: int
+    previous_verification_sha256: str
+    authorization_id: str
+    idempotency_key: str
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "RetryExtensionRequest":
+        allowed = {
+            "expected_status", "expected_revision", "expected_attempt", "expected_max_attempts",
+            "next_max_attempts", "previous_verification_sha256", "authorization_id", "idempotency_key",
+        }
+        if set(value) != allowed:
+            raise RetryExtensionError("RETRY_EXTENSION_FIELDS_INVALID", 400)
+        status = _require_string(value["expected_status"], "EXPECTED_STATUS_INVALID")
+        if status != "FAILED":
+            raise RetryExtensionError("EXPECTED_STATUS_INVALID", 400)
+        ints: dict[str, int] = {}
+        for field in ("expected_revision", "expected_attempt", "expected_max_attempts", "next_max_attempts"):
+            item = value[field]
+            if not isinstance(item, int) or isinstance(item, bool) or item < 0:
+                raise RetryExtensionError("RETRY_EXTENSION_INTEGER_INVALID", 400)
+            ints[field] = item
+        if ints["expected_attempt"] != ints["expected_max_attempts"]:
+            raise RetryExtensionError("RETRY_EXTENSION_NOT_EXHAUSTED", 409)
+        if ints["next_max_attempts"] != ints["expected_max_attempts"] + 1:
+            raise RetryExtensionError("RETRY_EXTENSION_MUST_BE_BOUNDED_ONE_STEP", 400)
+        if ints["next_max_attempts"] > 50:
+            raise RetryExtensionError("RETRY_EXTENSION_LIMIT_EXCEEDED", 400)
+        verification = _require_string(value["previous_verification_sha256"], "VERIFICATION_SHA256_INVALID")
+        if re.fullmatch(r"[0-9a-f]{64}", verification) is None:
+            raise RetryExtensionError("VERIFICATION_SHA256_INVALID", 400)
+        authorization = _require_string(value["authorization_id"], "AUTHORIZATION_ID_INVALID", KEY_RE)
+        key = _require_string(value["idempotency_key"], "IDEMPOTENCY_KEY_INVALID", KEY_RE)
+        return cls(
+            status,
+            ints["expected_revision"],
+            ints["expected_attempt"],
+            ints["expected_max_attempts"],
+            ints["next_max_attempts"],
+            verification,
+            authorization,
+            key,
+        )
+
+    def public(self) -> dict[str, Any]:
+        return {
+            "expected_status": self.expected_status,
+            "expected_revision": self.expected_revision,
+            "expected_attempt": self.expected_attempt,
+            "expected_max_attempts": self.expected_max_attempts,
+            "next_max_attempts": self.next_max_attempts,
+            "previous_verification_sha256": self.previous_verification_sha256,
+            "authorization_id": self.authorization_id,
+            "idempotency_key": self.idempotency_key,
+        }
+
+
+class RetryExtensionStore:
+    """Owns the official mutation. Direct/manual DB edits remain forbidden."""
+
+    def __init__(self, db_path: Path) -> None:
+        self.db_path = Path(db_path)
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.db_path, timeout=30.0)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA busy_timeout=30000")
+        return connection
+
+    @staticmethod
+    def _columns(connection: sqlite3.Connection, table: str) -> dict[str, sqlite3.Row]:
+        return {str(row["name"]): row for row in connection.execute(f'PRAGMA table_info("{table}")')}
+
+    @staticmethod
+    def _event_table(connection: sqlite3.Connection) -> tuple[str, dict[str, sqlite3.Row]]:
+        tables = {
+            row["name"]
+            for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'")
+        }
+        for name in ("task_events", "events"):
+            if name not in tables:
+                continue
+            columns = RetryExtensionStore._columns(connection, name)
+            if "task_id" in columns and ("event_type" in columns or "type" in columns):
+                return name, columns
+        raise RetryExtensionError("TASK_EVENT_TABLE_UNSUPPORTED", 503)
+
+    @staticmethod
+    def _verification_hash(raw: object) -> str:
+        if raw in (None, ""):
+            value: object = {}
+        elif isinstance(raw, str):
+            try:
+                value = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise RetryExtensionError("TASK_VERIFICATION_INVALID", 409) from exc
+        else:
+            value = raw
+        return _sha(value)
+
+    @staticmethod
+    def _ensure_extension_table(connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS task_retry_extensions (
+              task_id TEXT NOT NULL,
+              idempotency_key TEXT NOT NULL,
+              request_sha256 TEXT NOT NULL,
+              authorization_id TEXT NOT NULL,
+              previous_verification_sha256 TEXT NOT NULL,
+              previous_max_attempts INTEGER NOT NULL,
+              next_max_attempts INTEGER NOT NULL,
+              expected_revision INTEGER NOT NULL,
+              applied_revision INTEGER NOT NULL,
+              created_at TEXT NOT NULL,
+              PRIMARY KEY(task_id, idempotency_key)
+            )
+            """
+        )
+
+    @staticmethod
+    def _insert_event(
+        connection: sqlite3.Connection,
+        task_id: str,
+        payload: Mapping[str, Any],
+    ) -> None:
+        table, columns = RetryExtensionStore._event_table(connection)
+        values: dict[str, Any] = {"task_id": task_id}
+        if "event_type" in columns:
+            values["event_type"] = "TASK_RETRY_EXTENSION_AUTHORIZED"
+        else:
+            values["type"] = "TASK_RETRY_EXTENSION_AUTHORIZED"
+        payload_text = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        if "payload_json" in columns:
+            values["payload_json"] = payload_text
+        elif "payload" in columns:
+            values["payload"] = payload_text
+        else:
+            raise RetryExtensionError("TASK_EVENT_PAYLOAD_UNSUPPORTED", 503)
+        if "created_at" in columns:
+            values["created_at"] = _now()
+        if "agent_id" in columns:
+            values["agent_id"] = "owner-control"
+        if "actor_id" in columns:
+            values["actor_id"] = "owner-control"
+        names = list(values)
+        sql = f'INSERT INTO "{table}" ({",".join(f"""\"{name}\"""" for name in names)}) VALUES ({",".join("?" for _ in names)})'
+        connection.execute(sql, tuple(values[name] for name in names))
+
+    def extend(self, task_id: str, request: RetryExtensionRequest) -> dict[str, Any]:
+        task_id = _require_string(task_id, "TASK_ID_INVALID", TASK_RE)
+        required_authorization_id = _required_authorization_id()
+        if request.authorization_id != required_authorization_id:
+            raise RetryExtensionError("RETRY_EXTENSION_AUTHORIZATION_ID_MISMATCH", 403)
+        request_payload = request.public()
+        request_sha = _sha(request_payload)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._ensure_extension_table(connection)
+            replay = connection.execute(
+                "SELECT * FROM task_retry_extensions WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise RetryExtensionError("RETRY_EXTENSION_IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {
+                    "task_id": task_id,
+                    "status": "QUEUED",
+                    "attempt": request.expected_attempt,
+                    "max_attempts": int(replay["next_max_attempts"]),
+                    "revision": int(replay["applied_revision"]),
+                    "idempotent": True,
+                    "authorization_id": replay["authorization_id"],
+                    "previous_verification_sha256": replay["previous_verification_sha256"],
+                }
+
+            columns = self._columns(connection, "tasks")
+            required = {
+                "task_id", "status", "attempt", "max_attempts", "revision", "claimed_by",
+                "lease_token_hash", "lease_expires_at", "verification_json",
+            }
+            if not required.issubset(columns):
+                raise RetryExtensionError("TASK_SCHEMA_RETRY_EXTENSION_UNSUPPORTED", 503)
+            task = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if task is None:
+                raise RetryExtensionError("TASK_NOT_FOUND", 404)
+            actual_verification_sha = self._verification_hash(task["verification_json"])
+            if actual_verification_sha != request.previous_verification_sha256:
+                raise RetryExtensionError("RETRY_EXTENSION_VERIFICATION_DRIFT", 409)
+            if (
+                task["status"] != request.expected_status
+                or int(task["revision"]) != request.expected_revision
+                or int(task["attempt"]) != request.expected_attempt
+                or int(task["max_attempts"]) != request.expected_max_attempts
+            ):
+                raise RetryExtensionError("RETRY_EXTENSION_CAS_MISMATCH", 409)
+            for lease_field in ("lease_token_hash", "lease_expires_at"):
+                if lease_field in columns and task[lease_field] not in (None, ""):
+                    raise RetryExtensionError("RETRY_EXTENSION_ACTIVE_LEASE_DENY", 409)
+
+            previous_claimed_by = task["claimed_by"]
+            applied_revision = request.expected_revision + 1
+            cursor = connection.execute(
+                """
+                UPDATE tasks
+                   SET status='QUEUED', max_attempts=?, revision=?, claimed_by=NULL,
+                       lease_token_hash=NULL, lease_expires_at=NULL
+                 WHERE task_id=? AND status=? AND revision=? AND attempt=? AND max_attempts=?
+                   AND claimed_by IS ?
+                   AND lease_token_hash IS NULL AND lease_expires_at IS NULL
+                """,
+                (
+                    request.next_max_attempts,
+                    applied_revision,
+                    task_id,
+                    request.expected_status,
+                    request.expected_revision,
+                    request.expected_attempt,
+                    request.expected_max_attempts,
+                    previous_claimed_by,
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise RetryExtensionError("RETRY_EXTENSION_CAS_MISMATCH", 409)
+
+            event_payload = {
+                "schema": "aios.task-retry-extension.event.v1",
+                "task_id": task_id,
+                "authorization_id": request.authorization_id,
+                "idempotency_key": request.idempotency_key,
+                "previous_verification_sha256": request.previous_verification_sha256,
+                "previous_max_attempts": request.expected_max_attempts,
+                "next_max_attempts": request.next_max_attempts,
+                "previous_revision": request.expected_revision,
+                "applied_revision": applied_revision,
+                "request_sha256": request_sha,
+                "previous_claimed_by": previous_claimed_by,
+                "stale_terminal_claim_normalized": previous_claimed_by not in (None, ""),
+            }
+            self._insert_event(connection, task_id, event_payload)
+            connection.execute(
+                """
+                INSERT INTO task_retry_extensions(
+                  task_id,idempotency_key,request_sha256,authorization_id,previous_verification_sha256,
+                  previous_max_attempts,next_max_attempts,expected_revision,applied_revision,created_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    task_id,
+                    request.idempotency_key,
+                    request_sha,
+                    request.authorization_id,
+                    request.previous_verification_sha256,
+                    request.expected_max_attempts,
+                    request.next_max_attempts,
+                    request.expected_revision,
+                    applied_revision,
+                    _now(),
+                ),
+            )
+            connection.commit()
+            return {
+                "task_id": task_id,
+                "status": "QUEUED",
+                "attempt": request.expected_attempt,
+                "max_attempts": request.next_max_attempts,
+                "revision": applied_revision,
+                "idempotent": False,
+                "authorization_id": request.authorization_id,
+                "previous_verification_sha256": request.previous_verification_sha256,
+            }
+
+
+class RetryExtensionHandler(BaseHTTPRequestHandler):
+    store: RetryExtensionStore | None = None
+    control_token: str = ""
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def _json(self, value: Mapping[str, Any], status: int) -> None:
+        data = json.dumps(dict(value), sort_keys=True, separators=(",", ":")).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def _authorized(self) -> bool:
+        header = self.headers.get("Authorization", "")
+        return (
+            header.startswith("Bearer ")
+            and bool(self.control_token)
+            and hmac.compare_digest(header[7:], self.control_token)
+        )
+
+    def do_GET(self) -> None:
+        if self.path == "/health":
+            self._json({"status": "ok", "service": "aios-agent-task-retry-extension"}, 200)
+        else:
+            self._json({"error": "NOT_FOUND"}, 404)
+
+    def do_POST(self) -> None:
+        match = re.fullmatch(r"/v1/tasks/([^/]+)/retry-extension", self.path)
+        if match is None:
+            self._json({"error": "NOT_FOUND"}, 404)
+            return
+        if not self._authorized():
+            self._json({"error": "UNAUTHORIZED"}, 401)
+            return
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            self._json({"error": "CONTENT_LENGTH_INVALID"}, 400)
+            return
+        if length < 1 or length > MAX_BODY:
+            self._json({"error": "REQUEST_SIZE_INVALID"}, 413)
+            return
+        try:
+            value = json.loads(self.rfile.read(length).decode("utf-8"))
+            if not isinstance(value, Mapping):
+                raise ValueError
+            request = RetryExtensionRequest.from_mapping(value)
+            if self.store is None:
+                raise RetryExtensionError("RUNTIME_UNAVAILABLE", 503)
+            result = self.store.extend(match.group(1), request)
+            self._json(result, 200 if result["idempotent"] else 201)
+        except RetryExtensionError as exc:
+            self._json({"error": exc.code}, exc.status)
+        except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
+            self._json({"error": "REQUEST_INVALID"}, 400)
+
+
+def load_control_token(path: Path) -> str:
+    if path.is_symlink() or not path.is_file():
+        raise RuntimeError("TASK_RETRY_CONTROL_CREDENTIAL_UNAVAILABLE")
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 32:
+        raise RuntimeError("TASK_RETRY_CONTROL_CREDENTIAL_INVALID")
+    return value
+
+
+def serve() -> None:
+    db = Path(os.environ.get("AIOS_AGENT_TASK_DB", "/var/lib/aios-agent-task/state.sqlite3"))
+    _required_authorization_id()
+    token_path = Path(
+        os.environ.get(
+            "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+            str(Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials")) / "agent_task_control_token"),
+        )
+    )
+    host = os.environ.get("AIOS_TASK_RETRY_HOST", "127.0.0.1")
+    port = int(os.environ.get("AIOS_TASK_RETRY_PORT", "18941"))
+    RetryExtensionHandler.store = RetryExtensionStore(db)
+    RetryExtensionHandler.control_token = load_control_token(token_path)
+    ThreadingHTTPServer((host, port), RetryExtensionHandler).serve_forever()
+
+
+if __name__ == "__main__":
+    serve()

active-runtime/app/aios_isolated_executor.py

@@ -0,0 +1,306 @@
+"""Uncredentialed deterministic executor for governed Online Agent changes.
+
+This process never receives Agent/Owner/Bridge credentials and never writes Git refs.
+It operates only on an exchange bundle and returns hash-bound changed files + test evidence.
+"""
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+import errno
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import socket
+import subprocess
+import tarfile
+import tempfile
+import time
+from typing import Any, Mapping, Sequence
+
+ALLOWED_PROGRAMS = {"/usr/bin/python3", "/usr/bin/git", "/bin/true", "/usr/bin/true"}
+MAX_ARCHIVE_BYTES = 64 * 1024 * 1024
+MAX_FILE_BYTES = 4 * 1024 * 1024
+SECRET_ENV_TOKENS = ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "AUTHORIZATION", "PRIVATE_KEY")
+REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,254}$")
+SENSITIVE_VALUE_RE = re.compile(
+    r"(?i)(authorization\s*[:=]\s*bearer\s+\S+|authorization\s+bearer\s+\S+|"
+    r"(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
+)
+
+
+class IsolatedExecutorError(RuntimeError):
+    pass
+
+
+def sha256(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def safe_relative(value: object) -> str:
+    if not isinstance(value, str) or not value or value.startswith("/"):
+        raise IsolatedExecutorError("PATH_INVALID")
+    path = Path(value)
+    if ".." in path.parts or "." in path.parts or any(part in {"", ".git"} for part in path.parts):
+        raise IsolatedExecutorError("PATH_INVALID")
+    return path.as_posix()
+
+
+def resolve_under(root: Path, relative: str) -> Path:
+    root = root.resolve()
+    path = (root / safe_relative(relative)).resolve(strict=False)
+    if path == root or root not in path.parents:
+        raise IsolatedExecutorError("PATH_ESCAPE")
+    return path
+
+
+def _credential_exposure(directory: Path) -> str | None:
+    """Return an exposure code; EACCES is the intended credential isolation proof."""
+    try:
+        if not directory.exists():
+            return None
+        if directory.is_symlink():
+            return "CREDENTIAL_MOUNT_SYMLINK"
+        if os.access(directory, os.R_OK):
+            return "CREDENTIAL_MOUNT_ACCESSIBLE"
+        if directory.is_dir():
+            for child in directory.iterdir():
+                if child.is_file() and os.access(child, os.R_OK):
+                    return "CREDENTIAL_FILE_ACCESSIBLE"
+    except PermissionError:
+        return None
+    return None
+
+
+def assert_uncredentialed() -> None:
+    credentials = Path("/run/credentials")
+    candidates = {credentials}
+    credentials_directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+    candidates.add(credentials_directory)
+    leaked = sorted(
+        key for key, value in os.environ.items()
+        if value
+        and (any(token in key.upper() for token in SECRET_ENV_TOKENS) or SENSITIVE_VALUE_RE.search(str(value)))
+    )
+    if leaked:
+        raise IsolatedExecutorError("CREDENTIAL_ENV_EXPOSED:" + ",".join(leaked))
+    for directory in candidates:
+        exposure = _credential_exposure(directory)
+        if exposure is not None:
+            raise IsolatedExecutorError(exposure)
+    # In the intended PrivateNetwork namespace host loopback services are unreachable.
+    try:
+        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    except OSError as exc:
+        if exc.errno in {errno.EACCES, errno.EPERM, errno.EAFNOSUPPORT}:
+            return
+        raise IsolatedExecutorError("NETWORK_ISOLATION_PROBE_FAILED") from exc
+    probe.settimeout(0.15)
+    try:
+        if probe.connect_ex(("127.0.0.1", 18892)) == 0:
+            raise IsolatedExecutorError("GOVERNED_BRIDGE_REACHABLE")
+    finally:
+        probe.close()
+
+
+def extract_source(archive: Path, root: Path) -> None:
+    if archive.is_symlink() or not archive.is_file() or archive.stat().st_size > MAX_ARCHIVE_BYTES:
+        raise IsolatedExecutorError("SOURCE_ARCHIVE_INVALID")
+    with tarfile.open(archive, "r:gz") as tf:
+        for member in tf.getmembers():
+            path = Path(member.name)
+            if path.is_absolute() or ".." in path.parts or ".git" in path.parts:
+                raise IsolatedExecutorError("SOURCE_ARCHIVE_PATH_INVALID")
+            if member.issym() or member.islnk():
+                raise IsolatedExecutorError("SOURCE_ARCHIVE_LINK_DENY")
+        tf.extractall(root, filter="data")
+
+
+def run_command(item: Mapping[str, Any], root: Path) -> dict[str, Any]:
+    argv = item.get("argv")
+    if not isinstance(argv, Sequence) or isinstance(argv, (str, bytes)) or not argv:
+        raise IsolatedExecutorError("COMMAND_INVALID")
+    argv = [str(part) for part in argv]
+    if argv[0] not in ALLOWED_PROGRAMS:
+        raise IsolatedExecutorError("COMMAND_PROGRAM_DENY")
+    if argv[0] == "/usr/bin/python3" and "-m" in argv:
+        index = argv.index("-m")
+        if index + 1 >= len(argv) or argv[index + 1] not in {"pytest", "py_compile"}:
+            raise IsolatedExecutorError("PYTHON_MODULE_DENY")
+    if argv[0] == "/usr/bin/git" and any(
+        token in {"push", "update-ref", "commit", "merge", "rebase", "reset", "clean", "branch"}
+        for token in argv[1:]
+    ):
+        raise IsolatedExecutorError("GIT_MUTATION_DENY")
+    timeout = int(item.get("timeout_seconds") or 300)
+    if not 1 <= timeout <= 1800:
+        raise IsolatedExecutorError("COMMAND_TIMEOUT_INVALID")
+    cwd = root if item.get("cwd", ".") == "." else resolve_under(root, str(item.get("cwd")))
+    started = time.monotonic()
+    completed = subprocess.run(
+        argv,
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=timeout,
+        env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8"},
+    )
+    result = {
+        "label": str(item.get("label") or "command"),
+        "exit_code": completed.returncode,
+        "stdout_sha256": sha256(completed.stdout.encode("utf-8", "replace")),
+        "stderr_sha256": sha256(completed.stderr.encode("utf-8", "replace")),
+        "duration_ms": int((time.monotonic() - started) * 1000),
+    }
+    if completed.returncode != 0:
+        raise IsolatedExecutorError(
+            "COMMAND_FAILED:" + result["label"] + ":" + str(completed.returncode)
+        )
+    return result
+
+
+def apply_changes(request: Mapping[str, Any], root: Path) -> list[str]:
+    changes = request.get("change_set")
+    if not isinstance(changes, list) or not changes:
+        raise IsolatedExecutorError("CHANGE_SET_INVALID")
+    changed: list[str] = []
+    for change in changes:
+        if not isinstance(change, Mapping):
+            raise IsolatedExecutorError("CHANGE_INVALID")
+        relative = safe_relative(change.get("path"))
+        path = resolve_under(root, relative)
+        current = path.read_bytes() if path.is_file() else None
+        expected = change.get("expected_sha256")
+        if expected is not None and (current is None or sha256(current) != expected):
+            raise IsolatedExecutorError("SOURCE_DRIFT:" + relative)
+        operation = change.get("operation")
+        if operation == "write_file":
+            content = change.get("content")
+            if not isinstance(content, str):
+                raise IsolatedExecutorError("WRITE_CONTENT_INVALID")
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(content, encoding="utf-8")
+        elif operation == "replace_text":
+            if current is None:
+                raise IsolatedExecutorError("REPLACE_SOURCE_MISSING")
+            old = str(change.get("old"))
+            new = str(change.get("new"))
+            count = int(change.get("count", 1))
+            text = current.decode("utf-8")
+            if text.count(old) != count:
+                raise IsolatedExecutorError("REPLACE_COUNT_INVALID:" + relative)
+            path.write_text(text.replace(old, new), encoding="utf-8")
+        elif operation == "delete_file":
+            if path.exists():
+                path.unlink()
+        else:
+            raise IsolatedExecutorError("CHANGE_OPERATION_INVALID")
+        changed.append(relative)
+    return sorted(set(changed))
+
+
+def changed_bundle(root: Path, changed: list[str]) -> list[dict[str, Any]]:
+    files = []
+    for relative in changed:
+        path = resolve_under(root, relative)
+        if path.exists():
+            data = path.read_bytes()
+            if len(data) > MAX_FILE_BYTES:
+                raise IsolatedExecutorError("CHANGED_FILE_TOO_LARGE:" + relative)
+            files.append({
+                "path": relative,
+                "deleted": False,
+                "sha256": sha256(data),
+                "content_b64": base64.b64encode(data).decode("ascii"),
+            })
+        else:
+            files.append({"path": relative, "deleted": True, "sha256": None, "content_b64": None})
+    return files
+
+
+def execute_request(request_path: Path, result_path: Path) -> None:
+    assert_uncredentialed()
+    request = json.loads(request_path.read_text(encoding="utf-8"))
+    if request.get("schema") != "aios.isolated-execution.request.v1":
+        raise IsolatedExecutorError("REQUEST_SCHEMA_INVALID")
+    request_id = request.get("request_id")
+    if not isinstance(request_id, str) or not REQUEST_ID_RE.fullmatch(request_id):
+        raise IsolatedExecutorError("REQUEST_ID_INVALID")
+    source_archive = Path(str(request.get("source_archive") or ""))
+    if not source_archive.is_absolute():
+        raise IsolatedExecutorError("SOURCE_ARCHIVE_PATH_INVALID")
+    expected_archive_sha = str(request.get("source_archive_sha256") or "")
+    if sha256(source_archive.read_bytes()) != expected_archive_sha:
+        raise IsolatedExecutorError("SOURCE_ARCHIVE_SHA_MISMATCH")
+    with tempfile.TemporaryDirectory(prefix="aios-isolated-executor-") as temporary:
+        root = Path(temporary) / "workspace"
+        root.mkdir()
+        extract_source(source_archive, root)
+        changed = apply_changes(request, root)
+        commands = [
+            run_command(item, root)
+            for item in request.get("verification", [])
+            if isinstance(item, Mapping)
+        ]
+        result = {
+            "schema": "aios.isolated-execution.result.v1",
+            "request_id": request_id,
+            "request_sha256": sha256(canonical(request)),
+            "changed_files": changed_bundle(root, changed),
+            "verification": commands,
+            "credential_isolation": {
+                "credential_env_absent": True,
+                "credential_mount_inaccessible": True,
+                "governed_bridge_unreachable": True,
+            },
+        }
+        result["result_sha256"] = sha256(canonical(result))
+        temporary_result = result_path.with_suffix(".tmp")
+        temporary_result.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")), encoding="utf-8")
+        os.replace(temporary_result, result_path)
+
+
+def serve(root: Path) -> None:
+    assert_uncredentialed()
+    requests = root / "requests"
+    results = root / "results"
+    failed = root / "failed"
+    for path in (requests, results, failed):
+        path.mkdir(parents=True, exist_ok=True)
+    while True:
+        for request in sorted(requests.glob("*.json")):
+            if request.is_symlink():
+                continue
+            result = results / request.name
+            try:
+                execute_request(request, result)
+                request.unlink()
+            except Exception as exc:
+                (failed / request.name).write_text(
+                    json.dumps({"error": str(exc) or type(exc).__name__}, sort_keys=True),
+                    encoding="utf-8",
+                )
+                request.unlink(missing_ok=True)
+        time.sleep(0.5)
+
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--root", default="/var/lib/aios-online-agent-executor")
+    parser.add_argument("--request")
+    parser.add_argument("--result")
+    args = parser.parse_args()
+    if args.request and args.result:
+        execute_request(Path(args.request), Path(args.result))
+    else:
+        serve(Path(args.root))

active-runtime/app/aios_online_agent_checkpoint.py

@@ -0,0 +1,301 @@
+"""Durable checkpoint journal for the generic Online Agent worker.
+
+The JSON task state remains a fast cache.  This SQLite journal is the durable,
+append-only evidence used for restart/resume/replay decisions.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+from pathlib import Path
+import sqlite3
+from typing import Any, Mapping
+
+PHASE_ORDER = {
+    "claimed": 10,
+    "workspace_prepared": 20,
+    "verification_passed": 30,
+    "source_capture_done": 40,
+    "commit_done": 40,
+    "reconciliation_required": 45,
+    "result_pending": 50,
+    "result_submitted": 60,
+}
+
+
+class CheckpointError(RuntimeError):
+    pass
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str).encode("utf-8")
+
+
+def sha256_json(value: object) -> str:
+    return hashlib.sha256(canonical(value)).hexdigest()
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@dataclass(frozen=True)
+class CheckpointBinding:
+    task_id: str
+    attempt: int
+    execution_digest: str
+    job_id: str
+    repository: str
+    base_oid: str
+    expected_branch_oid: str
+    target_branch: str
+    generation: int
+
+    def material(self) -> dict[str, Any]:
+        return {
+            "task_id": self.task_id,
+            "attempt": self.attempt,
+            "execution_digest": self.execution_digest,
+            "job_id": self.job_id,
+            "repository": self.repository,
+            "base_oid": self.base_oid,
+            "expected_branch_oid": self.expected_branch_oid,
+            "target_branch": self.target_branch,
+            "generation": self.generation,
+        }
+
+    @property
+    def binding_sha256(self) -> str:
+        return sha256_json(self.material())
+
+
+class CheckpointStore:
+    def __init__(self, path: Path) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self._connect() as connection:
+            connection.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS checkpoints(
+                  task_id TEXT NOT NULL,
+                  attempt INTEGER NOT NULL,
+                  sequence INTEGER NOT NULL,
+                  phase TEXT NOT NULL,
+                  phase_order INTEGER NOT NULL,
+                  execution_digest TEXT NOT NULL,
+                  job_id TEXT NOT NULL,
+                  repository TEXT NOT NULL,
+                  base_oid TEXT NOT NULL,
+                  expected_branch_oid TEXT NOT NULL,
+                  target_branch TEXT NOT NULL,
+                  generation INTEGER NOT NULL,
+                  binding_sha256 TEXT NOT NULL,
+                  payload_json TEXT NOT NULL,
+                  payload_sha256 TEXT NOT NULL,
+                  created_at TEXT NOT NULL,
+                  PRIMARY KEY(task_id, attempt, sequence)
+                );
+                CREATE INDEX IF NOT EXISTS idx_checkpoints_task_latest
+                  ON checkpoints(task_id, attempt DESC, sequence DESC);
+                CREATE UNIQUE INDEX IF NOT EXISTS idx_checkpoints_exact_replay
+                  ON checkpoints(task_id, attempt, phase, payload_sha256);
+                CREATE TABLE IF NOT EXISTS checkpoint_heads(
+                  task_id TEXT PRIMARY KEY,
+                  attempt INTEGER NOT NULL,
+                  sequence INTEGER NOT NULL,
+                  phase TEXT NOT NULL,
+                  phase_order INTEGER NOT NULL,
+                  execution_digest TEXT NOT NULL,
+                  binding_sha256 TEXT NOT NULL,
+                  payload_sha256 TEXT NOT NULL,
+                  updated_at TEXT NOT NULL
+                );
+                """
+            )
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path, timeout=30.0)
+        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=30000")
+        return connection
+
+    @staticmethod
+    def _phase_order(phase: str) -> int:
+        try:
+            return PHASE_ORDER[phase]
+        except KeyError as exc:
+            raise CheckpointError("CHECKPOINT_PHASE_INVALID") from exc
+
+    def append(
+        self,
+        binding: CheckpointBinding,
+        phase: str,
+        payload: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        order = self._phase_order(phase)
+        payload_value = dict(payload)
+        payload_sha = sha256_json(payload_value)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            replay = connection.execute(
+                """
+                SELECT * FROM checkpoints
+                 WHERE task_id=? AND attempt=? AND phase=? AND payload_sha256=?
+                """,
+                (binding.task_id, binding.attempt, phase, payload_sha),
+            ).fetchone()
+            if replay is not None:
+                connection.commit()
+                return dict(replay)
+
+            head = connection.execute(
+                "SELECT * FROM checkpoint_heads WHERE task_id=?",
+                (binding.task_id,),
+            ).fetchone()
+            if head is not None:
+                head_attempt = int(head["attempt"])
+                head_order = int(head["phase_order"])
+                if binding.attempt < head_attempt:
+                    raise CheckpointError("CHECKPOINT_ATTEMPT_REGRESSION")
+                if binding.attempt == head_attempt:
+                    if head["execution_digest"] != binding.execution_digest:
+                        raise CheckpointError("CHECKPOINT_EXECUTION_DIGEST_CONFLICT")
+                    if head["binding_sha256"] != binding.binding_sha256:
+                        raise CheckpointError("CHECKPOINT_SCOPE_MISMATCH_DENY")
+                    if order < head_order and phase != "reconciliation_required":
+                        raise CheckpointError("CHECKPOINT_PHASE_REGRESSION")
+                    sequence = int(head["sequence"]) + 1
+                else:
+                    sequence = 1
+            else:
+                sequence = 1
+
+            created_at = now()
+            connection.execute(
+                """
+                INSERT INTO checkpoints(
+                  task_id,attempt,sequence,phase,phase_order,execution_digest,job_id,
+                  repository,base_oid,expected_branch_oid,target_branch,generation,
+                  binding_sha256,payload_json,payload_sha256,created_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    binding.task_id,
+                    binding.attempt,
+                    sequence,
+                    phase,
+                    order,
+                    binding.execution_digest,
+                    binding.job_id,
+                    binding.repository,
+                    binding.base_oid,
+                    binding.expected_branch_oid,
+                    binding.target_branch,
+                    binding.generation,
+                    binding.binding_sha256,
+                    canonical(payload_value).decode("utf-8"),
+                    payload_sha,
+                    created_at,
+                ),
+            )
+            connection.execute(
+                """
+                INSERT INTO checkpoint_heads(
+                  task_id,attempt,sequence,phase,phase_order,execution_digest,
+                  binding_sha256,payload_sha256,updated_at
+                ) VALUES (?,?,?,?,?,?,?,?,?)
+                ON CONFLICT(task_id) DO UPDATE SET
+                  attempt=excluded.attempt,
+                  sequence=excluded.sequence,
+                  phase=excluded.phase,
+                  phase_order=excluded.phase_order,
+                  execution_digest=excluded.execution_digest,
+                  binding_sha256=excluded.binding_sha256,
+                  payload_sha256=excluded.payload_sha256,
+                  updated_at=excluded.updated_at
+                """,
+                (
+                    binding.task_id,
+                    binding.attempt,
+                    sequence,
+                    phase,
+                    order,
+                    binding.execution_digest,
+                    binding.binding_sha256,
+                    payload_sha,
+                    created_at,
+                ),
+            )
+            connection.commit()
+            row = connection.execute(
+                "SELECT * FROM checkpoints WHERE task_id=? AND attempt=? AND sequence=?",
+                (binding.task_id, binding.attempt, sequence),
+            ).fetchone()
+            return dict(row)
+
+    def latest(self, task_id: str) -> dict[str, Any] | None:
+        with self._connect() as connection:
+            row = connection.execute(
+                """
+                SELECT * FROM checkpoints
+                 WHERE task_id=?
+                 ORDER BY attempt DESC, sequence DESC
+                 LIMIT 1
+                """,
+                (task_id,),
+            ).fetchone()
+            if row is None:
+                return None
+            value = dict(row)
+            value["payload"] = json.loads(value.pop("payload_json"))
+            return value
+
+    def latest_for_attempt(self, task_id: str, attempt: int) -> dict[str, Any] | None:
+        with self._connect() as connection:
+            row = connection.execute(
+                """
+                SELECT * FROM checkpoints
+                 WHERE task_id=? AND attempt=?
+                 ORDER BY sequence DESC
+                 LIMIT 1
+                """,
+                (task_id, attempt),
+            ).fetchone()
+            if row is None:
+                return None
+            value = dict(row)
+            value["payload"] = json.loads(value.pop("payload_json"))
+            return value
+
+    def verify_latest_binding(self, binding: CheckpointBinding) -> dict[str, Any] | None:
+        value = self.latest(binding.task_id)
+        if value is None:
+            return None
+        if int(value["attempt"]) > binding.attempt:
+            raise CheckpointError("CHECKPOINT_FUTURE_ATTEMPT")
+        # A previous attempt may be reused only when the immutable execution/scope binding matches.
+        if value["execution_digest"] != binding.execution_digest:
+            return None
+        expected = CheckpointBinding(
+            task_id=value["task_id"],
+            attempt=int(value["attempt"]),
+            execution_digest=value["execution_digest"],
+            job_id=value["job_id"],
+            repository=value["repository"],
+            base_oid=value["base_oid"],
+            expected_branch_oid=value["expected_branch_oid"],
+            target_branch=value["target_branch"],
+            generation=int(value["generation"]),
+        )
+        current_scope = binding.material()
+        historical_scope = expected.material()
+        current_scope.pop("attempt", None)
+        historical_scope.pop("attempt", None)
+        if sha256_json(current_scope) != sha256_json(historical_scope):
+            raise CheckpointError("CHECKPOINT_SCOPE_MISMATCH_DENY")
+        return value

active-runtime/app/aios_online_agent_execution.py

@@ -0,0 +1,397 @@
+"""Canonical governed execution projection for Online Agent Tasks.
+
+The durable governed binding is authoritative.  Natural-language task fields and
+worker configuration are never allowed to invent repository, ref, generation, or
+idempotency coordinates.
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import hashlib
+import json
+import re
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+import urllib.error
+import urllib.request
+
+
+EXECUTION_SCHEMA = "aios.online-agent.execution.v2"
+PROJECTION_SCHEMA = "aios.online-agent.task-projection.v1"
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,159}$")
+PROTECTED_PREFIXES = ("main", "master", "refs/", "tags/", "tag/", "release/", "baseline/", "production/")
+MUTATION_ACTIONS = frozenset({"deterministic"})
+SUPPORTED_ACTIONS = frozenset({"deterministic", "source_capture"})
+
+
+class ExecutionProjectionError(RuntimeError):
+    """A canonical fail-closed projection or validation error."""
+
+
+class OnlineTaskProjectionClient:
+    """Official HTTP projection client using only a control credential reference."""
+
+    def __init__(self, api_base: str, control_token_path: Path) -> None:
+        self.api_base = api_base.rstrip("/")
+        self.control_token_path = Path(control_token_path)
+
+    def _token(self) -> str:
+        path = self.control_token_path
+        if path.is_symlink() or not path.is_file():
+            _error("ONLINE_TASK_PROJECTION_CONTROL_UNAVAILABLE")
+        value = path.read_text(encoding="utf-8").strip()
+        if len(value) < 32:
+            _error("ONLINE_TASK_PROJECTION_CONTROL_INVALID")
+        return value
+
+    def create(
+        self,
+        projected_task: Mapping[str, Any],
+        *,
+        binding: "GovernedExecutionBinding",
+    ) -> dict[str, Any]:
+        body = dict(projected_task)
+        payload = _canonical(body).encode("utf-8")
+        request = urllib.request.Request(
+            self.api_base + "/v1/tasks",
+            method="POST",
+            data=payload,
+            headers={
+                "Authorization": "Bearer " + self._token(),
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=15) as response:
+                result = json.loads(response.read().decode("utf-8"))
+                status = response.status
+        except urllib.error.HTTPError as exc:
+            raw = exc.read(8192).decode("utf-8", "replace")
+            try:
+                error = json.loads(raw).get("error")
+            except (AttributeError, json.JSONDecodeError):
+                error = "HTTP_ERROR"
+            raise ExecutionProjectionError(f"ONLINE_TASK_PROJECTION_FAILED:{exc.code}:{error}") from exc
+        except (OSError, urllib.error.URLError, TimeoutError) as exc:
+            raise ExecutionProjectionError("ONLINE_TASK_PROJECTION_UNAVAILABLE") from exc
+        if status not in {200, 201} or not isinstance(result, dict):
+            _error("ONLINE_TASK_PROJECTION_RESPONSE_INVALID")
+        if result.get("task_id") != binding.task_id or result.get("status") not in {"QUEUED", "VERIFYING", "SUCCEEDED"}:
+            _error("ONLINE_TASK_PROJECTION_RESPONSE_INVALID")
+        metadata = result.get("metadata")
+        if not isinstance(metadata, Mapping):
+            _error("ONLINE_TASK_PROJECTION_RESPONSE_INVALID")
+        returned_execution = validate_execution_envelope(
+            _object(metadata.get("execution"), "ONLINE_TASK_PROJECTION_RESPONSE_INVALID"),
+            binding=binding,
+        )
+        submitted_metadata = _object(body.get("metadata"), "ONLINE_TASK_PROJECTION_REQUEST_INVALID")
+        submitted_execution = validate_execution_envelope(
+            _object(submitted_metadata.get("execution"), "ONLINE_TASK_PROJECTION_REQUEST_INVALID"),
+            binding=binding,
+        )
+        if _canonical(returned_execution) != _canonical(submitted_execution):
+            _error("ONLINE_TASK_PROJECTION_RESPONSE_BINDING_INVALID")
+        return result
+
+
+def _error(code: str) -> None:
+    raise ExecutionProjectionError(code)
+
+
+def _object(value: object, code: str) -> dict[str, Any]:
+    if not isinstance(value, Mapping):
+        _error(code)
+    return dict(value)
+
+
+def _string(value: object, code: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        _error(code)
+    return value.strip()
+
+
+def _task_id(value: object) -> str:
+    value = _string(value, "TASK_ID_INVALID")
+    if TASK_RE.fullmatch(value) is None:
+        _error("TASK_ID_INVALID")
+    return value
+
+
+def _oid(value: object, code: str) -> str:
+    value = _string(value, code)
+    if OID_RE.fullmatch(value) is None:
+        _error(code)
+    return value
+
+
+def _branch(value: object) -> str:
+    value = _string(value, "TARGET_BRANCH_INVALID")
+    if BRANCH_RE.fullmatch(value) is None or value.startswith(PROTECTED_PREFIXES):
+        _error("TARGET_BRANCH_INVALID")
+    return value
+
+
+def _generation(value: object) -> int:
+    if not isinstance(value, int) or isinstance(value, bool) or value < 1:
+        _error("GENERATION_INVALID")
+    return value
+
+
+def _canonical(value: object) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _digest(value: object) -> str:
+    return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
+
+
+def _source_signals(value: object) -> tuple[str, ...]:
+    """Normalize untrusted natural-language hints before they enter a signed envelope."""
+    if value is None:
+        return ()
+    if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
+        _error("SOURCE_CAPTURE_SIGNALS_INVALID")
+    result: list[str] = []
+    for item in value:
+        if not isinstance(item, str):
+            _error("SOURCE_CAPTURE_SIGNALS_INVALID")
+        text = " ".join(item.split()).strip()
+        if not text:
+            continue
+        # Projection may accept long task prose, but execution envelopes stay bounded.
+        text = text[:512].rstrip()
+        if text not in result:
+            result.append(text)
+        if len(result) == 16:
+            break
+    return tuple(result)
+
+
+def _capabilities(value: object) -> tuple[str, ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
+        _error("CAPABILITY_SCOPE_INVALID")
+    result = tuple(sorted({_string(item, "CAPABILITY_SCOPE_INVALID") for item in value}))
+    if any(len(item) > 120 for item in result):
+        _error("CAPABILITY_SCOPE_INVALID")
+    return result
+
+
+@dataclass(frozen=True, slots=True)
+class GovernedExecutionBinding:
+    task_id: str
+    job_id: str
+    repository: str
+    base_oid: str
+    expected_branch_oid: str
+    target_branch: str
+    branch_mode: str
+    generation: int
+    idempotency_key: str
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "GovernedExecutionBinding":
+        data = dict(value)
+        task_id = _task_id(data.get("task_id"))
+        job_id = _string(data.get("job_id"), "JOB_ID_REQUIRED")
+        if TASK_RE.fullmatch(job_id) is None:
+            _error("JOB_ID_INVALID")
+        repository = _string(data.get("repository"), "REPOSITORY_INVALID")
+        if repository != "aios/aios":
+            _error("WRONG_REPOSITORY_ENVELOPE_DENY")
+        base_oid = _oid(data.get("base_oid"), "BASE_OID_INVALID")
+        expected = _oid(data.get("expected_branch_oid"), "EXPECTED_BRANCH_OID_INVALID")
+        target = _branch(data.get("target_branch"))
+        mode = _string(data.get("branch_mode"), "BRANCH_MODE_MISSING")
+        if mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}:
+            _error("BRANCH_MODE_INVALID")
+        generation = _generation(data.get("generation"))
+        if mode == "CREATE_NEW" and expected != base_oid:
+            _error("EXPECTED_BASE_CONFLICT")
+        key = _string(data.get("idempotency_key"), "IDEMPOTENCY_KEY_INVALID")
+        if KEY_RE.fullmatch(key) is None:
+            _error("IDEMPOTENCY_KEY_INVALID")
+        return cls(task_id, job_id, repository, base_oid, expected, target, mode, generation, key)
+
+    def public(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def binding_from_bootstrap(value: Mapping[str, Any]) -> GovernedExecutionBinding:
+    return GovernedExecutionBinding.from_mapping(value)
+
+
+def validate_execution_envelope(
+    value: Mapping[str, Any],
+    *,
+    binding: GovernedExecutionBinding | None = None,
+    worker_capabilities: Sequence[str] | None = None,
+) -> dict[str, Any]:
+    envelope = dict(value)
+    if envelope.get("schema") != EXECUTION_SCHEMA:
+        _error("EXECUTION_ENVELOPE_SCHEMA_VALIDATION")
+    candidate = GovernedExecutionBinding.from_mapping(envelope)
+    if binding is not None and candidate != binding:
+        comparisons = (
+            (candidate.task_id, binding.task_id, "TASK_BINDING_CONFLICT"),
+            (candidate.job_id, binding.job_id, "JOB_BINDING_CONFLICT"),
+            (candidate.repository, binding.repository, "WRONG_REPOSITORY_ENVELOPE_DENY"),
+            (candidate.base_oid, binding.base_oid, "WRONG_BASE_ENVELOPE_DENY"),
+            (candidate.expected_branch_oid, binding.expected_branch_oid, "WRONG_EXPECTED_HEAD_ENVELOPE_DENY"),
+            (candidate.target_branch, binding.target_branch, "WRONG_BRANCH_ENVELOPE_DENY"),
+            (candidate.generation, binding.generation, "WRONG_GENERATION_ENVELOPE_DENY"),
+            (candidate.branch_mode, binding.branch_mode, "BRANCH_MODE_BINDING_CONFLICT"),
+            (candidate.idempotency_key, binding.idempotency_key, "IDEMPOTENCY_BINDING_CONFLICT"),
+        )
+        for actual, expected, code in comparisons:
+            if actual != expected:
+                _error(code)
+    action = _string(envelope.get("action"), "EXECUTION_ACTION_MISSING")
+    if action not in SUPPORTED_ACTIONS:
+        _error("EXECUTION_ACTION_UNSUPPORTED")
+    capability_scope = _capabilities(envelope.get("capability_scope"))
+    if worker_capabilities is not None and not set(capability_scope).issubset(set(worker_capabilities)):
+        _error("CAPABILITY_MISMATCH_DENY")
+    workspace = _object(envelope.get("workspace"), "WORKSPACE_POLICY_REQUIRED")
+    if workspace.get("authority_git_dir") != "/data/aios-gitops/repo.git" or workspace.get("read_only_repository") is not True:
+        _error("WORKSPACE_POLICY_DENY")
+    governance = _object(envelope.get("governance"), "GOVERNANCE_POLICY_REQUIRED")
+    governed_coordinates = (
+        (governance.get("task_id"), candidate.task_id, "TASK_BINDING_CONFLICT"),
+        (governance.get("job_id"), candidate.job_id, "JOB_BINDING_CONFLICT"),
+        (governance.get("idempotency_key"), candidate.idempotency_key, "IDEMPOTENCY_BINDING_CONFLICT"),
+    )
+    for actual_coordinate, expected_coordinate, code in governed_coordinates:
+        if actual_coordinate != expected_coordinate:
+            _error(code)
+    commit_key = _string(governance.get("commit_idempotency_key"), "COMMIT_IDEMPOTENCY_KEY_INVALID")
+    if KEY_RE.fullmatch(commit_key) is None:
+        _error("COMMIT_IDEMPOTENCY_KEY_INVALID")
+    if governance.get("writer_lease") is not (action in MUTATION_ACTIONS):
+        _error("WRITER_LEASE_POLICY_DENY")
+    for field in ("main_write", "production_write", "tag_write", "baseline_write", "force_push", "protected_ref_bypass"):
+        if governance.get(field, False) is not False:
+            _error("POLICY_DENY")
+    changes = envelope.get("change_set")
+    verification = envelope.get("verification")
+    if not isinstance(changes, list) or not isinstance(verification, list):
+        _error("EXECUTION_OPERATIONS_INVALID")
+    if action == "source_capture":
+        if changes or verification:
+            _error("SOURCE_CAPTURE_MUTATION_DENY")
+        signals = list(_source_signals(envelope.get("source_signals") or []))
+        if envelope.get("source_signals") != signals:
+            _error("SOURCE_CAPTURE_SIGNALS_INVALID")
+    elif not changes:
+        _error("CHANGE_SET_INVALID")
+    declared = str(envelope.get("execution_digest") or "")
+    material = {key: item for key, item in envelope.items() if key != "execution_digest"}
+    actual = _digest(material)
+    if declared != actual:
+        _error("EXECUTION_DIGEST_INVALID")
+    return {**material, "execution_digest": actual}
+
+
+def project_execution_envelope(
+    binding: GovernedExecutionBinding,
+    execution_request: Mapping[str, Any] | None = None,
+    *,
+    required_capabilities: Sequence[str] = (),
+) -> dict[str, Any]:
+    request = dict(execution_request or {})
+    action = str(request.get("action") or "source_capture")
+    if action not in SUPPORTED_ACTIONS:
+        _error("EXECUTION_ACTION_UNSUPPORTED")
+    capabilities = _capabilities(request.get("capability_scope", required_capabilities))
+    governance = {
+        "task_id": binding.task_id,
+        "job_id": binding.job_id,
+        "idempotency_key": binding.idempotency_key,
+        "commit_idempotency_key": str(request.get("commit_idempotency_key") or f"{binding.task_id}-commit-g{binding.generation}"),
+        "writer_lease": action in MUTATION_ACTIONS,
+        "main_write": False,
+        "production_write": False,
+        "tag_write": False,
+        "baseline_write": False,
+        "force_push": False,
+        "protected_ref_bypass": False,
+    }
+    supplied_governance = request.get("governance")
+    if supplied_governance is not None:
+        governance.update(_object(supplied_governance, "GOVERNANCE_POLICY_REQUIRED"))
+    material: dict[str, Any] = {
+        "schema": EXECUTION_SCHEMA,
+        **binding.public(),
+        "action": action,
+        "adapter": str(request.get("adapter") or ("deterministic" if action == "deterministic" else "source_capture")),
+        "capability_scope": list(capabilities),
+        "change_set": list(request.get("change_set") or []),
+        "verification": list(request.get("verification") or []),
+        "source_signals": list(_source_signals(request.get("source_signals") or [])) if action == "source_capture" else [],
+        "governance": governance,
+        "workspace": {
+            "authority_git_dir": "/data/aios-gitops/repo.git",
+            "read_only_repository": True,
+            "cleanup_policy": "always",
+        },
+    }
+    envelope = {**material, "execution_digest": _digest(material)}
+    return validate_execution_envelope(envelope, binding=binding)
+
+
+def project_online_task(
+    binding: GovernedExecutionBinding,
+    task_request: Mapping[str, Any],
+    execution_request: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    request = dict(task_request)
+    required = _capabilities(request.get("required_capabilities"))
+    execution = project_execution_envelope(binding, execution_request, required_capabilities=required)
+    metadata = dict(request.get("metadata") or {})
+    if "execution" in metadata:
+        validate_execution_envelope(_object(metadata["execution"], "EXECUTION_ENVELOPE_INVALID"), binding=binding)
+        if metadata["execution"] != execution:
+            _error("EXECUTION_PROJECTION_CONFLICT")
+    metadata["execution"] = execution
+    metadata["execution_projection"] = {
+        "schema": PROJECTION_SCHEMA,
+        "task_id": binding.task_id,
+        "job_id": binding.job_id,
+        "execution_digest": execution["execution_digest"],
+    }
+    return {
+        **request,
+        "task_id": binding.task_id,
+        "metadata": metadata,
+        "required_capabilities": list(required),
+        "idempotency_key": str(request.get("idempotency_key") or f"online-task:{binding.idempotency_key}"),
+    }
+
+
+def resolve_task_execution(
+    task: Mapping[str, Any],
+    binding: GovernedExecutionBinding,
+    *,
+    worker_capabilities: Sequence[str] | None = None,
+) -> tuple[dict[str, Any], str]:
+    metadata = task.get("metadata")
+    if isinstance(metadata, Mapping) and "execution" in metadata:
+        value = validate_execution_envelope(
+            _object(metadata["execution"], "EXECUTION_ENVELOPE_INVALID"),
+            binding=binding,
+            worker_capabilities=worker_capabilities,
+        )
+        return value, "persisted"
+    projected = project_execution_envelope(
+        binding,
+        {"action": "source_capture", "capability_scope": task.get("required_capabilities") or []},
+        required_capabilities=task.get("required_capabilities") or [],
+    )
+    validate_execution_envelope(projected, binding=binding, worker_capabilities=worker_capabilities)
+    return projected, "recovered_projection"

active-runtime/app/aios_online_agent_plan_store.py

@@ -0,0 +1,187 @@
+"""Durable, authority-bound deterministic execution plans for Online Agent Tasks."""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from dataclasses import dataclass
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+from typing import Any, Mapping
+
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+
+class ExecutionPlanError(RuntimeError):
+    pass
+
+
+def _canonical(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def _digest(value: object) -> str:
+    return hashlib.sha256(_canonical(value)).hexdigest()
+
+
+def _task_key(task_id: str) -> str:
+    readable = "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in task_id).strip("-")[:72] or "task"
+    return readable + "-" + hashlib.sha256(task_id.encode("utf-8")).hexdigest()[:24]
+
+
+@dataclass(frozen=True)
+class ExecutionPlanRecord:
+    task_id: str
+    job_id: str
+    source_head_oid: str
+    generation: int
+    idempotency_key: str
+    plan_digest: str
+    execution: dict[str, Any]
+
+    def public(self) -> dict[str, Any]:
+        return {
+            "schema": "aios.online-agent.plan-record.v1",
+            "task_id": self.task_id,
+            "job_id": self.job_id,
+            "source_head_oid": self.source_head_oid,
+            "generation": self.generation,
+            "idempotency_key": self.idempotency_key,
+            "plan_digest": self.plan_digest,
+            "execution": dict(self.execution),
+        }
+
+
+class ExecutionPlanStore:
+    def __init__(self, root: Path) -> None:
+        self.root = Path(root)
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, task_id: str) -> Path:
+        return self.root / (_task_key(task_id) + ".json")
+
+    @contextmanager
+    def _lock(self, task_id: str):
+        path = self.root / (_task_key(task_id) + ".lock")
+        with path.open("a+b") as handle:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+            try:
+                yield
+            finally:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+
+    @staticmethod
+    def _source_head(binding: GovernedExecutionBinding) -> str:
+        return binding.expected_branch_oid if binding.branch_mode == "CONTINUE_EXISTING" else binding.base_oid
+
+    def _read(self, task_id: str) -> ExecutionPlanRecord | None:
+        path = self._path(task_id)
+        if not path.is_file() or path.is_symlink():
+            return None
+        value = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(value, Mapping) or value.get("schema") != "aios.online-agent.plan-record.v1":
+            raise ExecutionPlanError("EXECUTION_PLAN_RECORD_INVALID")
+        execution = value.get("execution")
+        if not isinstance(execution, Mapping):
+            raise ExecutionPlanError("EXECUTION_PLAN_RECORD_INVALID")
+        return ExecutionPlanRecord(
+            task_id=str(value.get("task_id") or ""),
+            job_id=str(value.get("job_id") or ""),
+            source_head_oid=str(value.get("source_head_oid") or ""),
+            generation=int(value.get("generation") or 0),
+            idempotency_key=str(value.get("idempotency_key") or ""),
+            plan_digest=str(value.get("plan_digest") or ""),
+            execution=dict(execution),
+        )
+
+    def register(
+        self,
+        binding: GovernedExecutionBinding,
+        request: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        with self._lock(binding.task_id):
+            return self._register_locked(binding, request)
+
+    def _register_locked(
+        self,
+        binding: GovernedExecutionBinding,
+        request: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        if request.get("schema") != "aios.online-agent.plan.v1":
+            raise ExecutionPlanError("EXECUTION_PLAN_SCHEMA_INVALID")
+        if request.get("task_id") != binding.task_id or request.get("job_id") != binding.job_id:
+            raise ExecutionPlanError("EXECUTION_PLAN_BINDING_CONFLICT")
+        if int(request.get("generation") or 0) != binding.generation:
+            raise ExecutionPlanError("EXECUTION_PLAN_GENERATION_CONFLICT")
+        source_head = str(request.get("source_head_oid") or "")
+        if source_head != self._source_head(binding):
+            raise ExecutionPlanError("EXECUTION_PLAN_SOURCE_HEAD_CONFLICT")
+        key = str(request.get("idempotency_key") or "")
+        if len(key) < 8:
+            raise ExecutionPlanError("EXECUTION_PLAN_IDEMPOTENCY_INVALID")
+        execution_request = request.get("execution")
+        if not isinstance(execution_request, Mapping) or execution_request.get("action") != "deterministic":
+            raise ExecutionPlanError("EXECUTION_PLAN_DETERMINISTIC_REQUIRED")
+        envelope = project_execution_envelope(binding, execution_request)
+        material = {
+            "task_id": binding.task_id,
+            "job_id": binding.job_id,
+            "source_head_oid": source_head,
+            "generation": binding.generation,
+            "idempotency_key": key,
+            "execution": dict(execution_request),
+        }
+        plan_digest = _digest(material)
+        existing = self._read(binding.task_id)
+        expected_previous = request.get("expected_previous_plan_digest")
+        if existing is not None:
+            if existing.idempotency_key == key:
+                if existing.plan_digest != plan_digest:
+                    raise ExecutionPlanError("EXECUTION_PLAN_IDEMPOTENCY_CONFLICT")
+                return {**existing.public(), "idempotent": True, "execution_digest": envelope["execution_digest"]}
+            if expected_previous != existing.plan_digest:
+                raise ExecutionPlanError("EXECUTION_PLAN_REPLACEMENT_CAS_MISMATCH")
+            if existing.source_head_oid != source_head:
+                raise ExecutionPlanError("EXECUTION_PLAN_STALE_REPLACEMENT_DENY")
+        elif expected_previous not in (None, ""):
+            raise ExecutionPlanError("EXECUTION_PLAN_REPLACEMENT_CAS_MISMATCH")
+
+        record = ExecutionPlanRecord(
+            task_id=binding.task_id,
+            job_id=binding.job_id,
+            source_head_oid=source_head,
+            generation=binding.generation,
+            idempotency_key=key,
+            plan_digest=plan_digest,
+            execution=dict(execution_request),
+        )
+        path = self._path(binding.task_id)
+        temporary = path.with_suffix(".tmp")
+        temporary.write_text(
+            json.dumps(record.public(), sort_keys=True, separators=(",", ":")),
+            encoding="utf-8",
+        )
+        temporary.replace(path)
+        return {**record.public(), "idempotent": False, "execution_digest": envelope["execution_digest"]}
+
+    def resolve(self, binding: GovernedExecutionBinding) -> dict[str, Any] | None:
+        record = self._read(binding.task_id)
+        if record is None:
+            return None
+        if (
+            record.task_id != binding.task_id
+            or record.job_id != binding.job_id
+            or record.generation != binding.generation
+        ):
+            raise ExecutionPlanError("EXECUTION_PLAN_BINDING_CONFLICT")
+        if record.source_head_oid != self._source_head(binding):
+            raise ExecutionPlanError("EXECUTION_PLAN_SOURCE_HEAD_DRIFT")
+        envelope = project_execution_envelope(binding, record.execution)
+        return {
+            "execution": dict(record.execution),
+            "plan_digest": record.plan_digest,
+            "execution_digest": envelope["execution_digest"],
+        }

deploy/systemd/aios-agent-task-retry-extension.service

@@ -0,0 +1,36 @@
+[Unit]
+Description=AIOS owner-controlled Task retry-extension API
+After=aios-agent-task-api.service
+Wants=aios-agent-task-api.service
+
+[Service]
+Type=simple
+User=aios-agent-task
+Group=aios-agent-task
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+EnvironmentFile=/etc/aios-gitops/agent-task-retry-extension.env
+Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_TASK_RETRY_HOST=127.0.0.1
+Environment=AIOS_TASK_RETRY_PORT=18941
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_retry_extension.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+ReadWritePaths=/var/lib/aios-agent-task
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-online-agent-isolated-executor.service

@@ -0,0 +1,33 @@
+[Unit]
+Description=AIOS uncredentialed deterministic Planner/Executor
+After=local-fs.target
+
+[Service]
+Type=simple
+User=aios-engineering-executor
+Group=aios-executor-exchange
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_isolated_executor.py --root /var/lib/aios-online-agent-executor
+Restart=always
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateNetwork=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=localhost
+InaccessiblePaths=/run/credentials
+ReadWritePaths=/var/lib/aios-online-agent-executor
+UMask=0027
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-online-agent-worker.service

@@ -0,0 +1,46 @@
+[Unit]
+Description=AIOS governed Online Agent Worker
+After=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
+Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
+
+[Service]
+Type=simple
+User=aios-engineering-worker
+Group=aios-engineering-worker
+SupplementaryGroups=gitops aios-executor-exchange
+LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
+Environment=PYTHONUNBUFFERED=1
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_AGENT_ID=online-engineering-worker
+Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
+Environment=AIOS_AGENT_VERSION=2026-08-17.2-durable
+Environment=AIOS_WORKER_CAPABILITIES=git.read,git.test,governed.write.branch,governed.write.file
+Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
+Environment=AIOS_ISOLATED_EXECUTOR_ROOT=/var/lib/aios-online-agent-executor
+Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-worker
+Environment=AIOS_WORKER_CHECKPOINT_DB=/var/lib/aios-online-agent-worker/checkpoints.sqlite3
+Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-worker/workspaces
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_online_agent_worker.py
+Restart=always
+RestartSec=5
+TimeoutStopSec=20
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadWritePaths=/var/lib/aios-online-agent-worker
+UMask=0027
+
+[Install]
+WantedBy=multi-user.target

deploy/sysusers.d/aios-online-agent-executor.conf

@@ -0,0 +1,4 @@
+g aios-executor-exchange -
+u aios-engineering-executor - "AIOS isolated engineering executor" /var/lib/aios-online-agent-executor /usr/sbin/nologin
+m aios-engineering-executor aios-executor-exchange
+m aios-engineering-worker aios-executor-exchange

deploy/tmpfiles.d/aios-online-agent-executor.conf

@@ -0,0 +1,5 @@
+d /var/lib/aios-online-agent-executor 2770 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-online-agent-executor/requests 2770 aios-engineering-worker aios-executor-exchange -
+d /var/lib/aios-online-agent-executor/results 2770 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-online-agent-executor/failed 2770 aios-engineering-executor aios-executor-exchange -
+d /var/lib/aios-online-agent-executor/sources 2770 aios-engineering-worker aios-executor-exchange -

docs/superpowers/specs/2026-08-17-r42-p1-durable-execution.md

@@ -0,0 +1,75 @@
+# R4.2 P1 — Durable Online Agent Execution
+
+Status: candidate implementation. This document does not imply protected-main promotion or Production deployment.
+
+## State model
+
+The generic worker uses two local durable layers:
+
+1. `tasks/<task_id>.json`: atomic fast-state cache for compatibility and quick restart.
+2. `checkpoints.sqlite3`: append-only attempt/phase journal with `synchronous=FULL`.
+
+Checkpoint identity is bound to:
+
+- task_id
+- attempt
+- execution_digest
+- job_id
+- repository
+- base_oid
+- expected_branch_oid
+- target_branch
+- generation
+
+A checkpoint from another repository/branch/generation is never reused.
+
+Phases are monotonic within an attempt:
+
+`claimed → workspace_prepared → verification_passed → source_capture_done|commit_done → result_pending → result_submitted`
+
+`reconciliation_required` is an explicit recovery phase.
+
+## Attempt-aware result idempotency
+
+Result submission idempotency keys include the official Task attempt and the execution digest. The same
+attempt reuses the same key after a timeout/restart; a later official attempt cannot collide with an
+earlier result.
+
+Before submitting a result the worker durably writes `result_pending`, including the result hash and
+idempotency key. After a successful or exact-idempotency-reconciled API response it writes
+`result_submitted`.
+
+A 409 is accepted only if the Task API already contains the exact same result body and the Task is
+VERIFYING/SUCCEEDED. A conflicting result remains a hard failure.
+
+## Resume rules
+
+- `source_capture_done` and `result_pending` may be reused across later official attempts only when the
+  execution/scope binding is identical. This avoids repeating read-only source capture.
+- A deterministic task that crashed before governed commit replays from a new clean detached worktree.
+  This is safe because there is no external side effect yet.
+- `commit_done` is never recommitted. Recovery requires exact Job receipt OID convergence plus exactly
+  one ACTIVE WriterLease whose `current_head_oid` equals the governed commit.
+- delayed Git visibility remains `RECONCILIATION_REQUIRED`, preserving an explicit recovery entry.
+
+## Crash cleanup
+
+On worker startup, stale `task-*` workspaces under the configured workspace root are removed before
+polling begins. The path boundary is verified before deletion.
+
+## Contracts
+
+P1 gates include:
+
+- CHECKPOINT_DURABLE_WRITE
+- RESUME_FROM_CHECKPOINT
+- RETRY_IDEMPOTENT_NO_DUPLICATE_SIDE_EFFECT
+- CRASH_RECOVERY_CONVERGENCE
+- WORKER_RESTART_RESUME
+- CHECKPOINT_SCOPE_ISOLATION
+- RESULT_SUBMISSION_REPLAY_SAFE
+- WORKSPACE_CLEANUP_RECOVERY
+- LEASE_RENEWAL_RECOVERY
+
+WriterLease and governed Job authority remain in the existing Bridge/GitOps durable stores; the worker
+checkpoint DB is not a competing authority for Git state.

tests/test_agent_task_retry_extension.py

@@ -0,0 +1,117 @@
+from pathlib import Path
+import hashlib, json, sqlite3, sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+from aios_agent_task_retry_extension import RetryExtensionRequest, RetryExtensionStore, RetryExtensionError
+
+AUTHORIZATION_ID="owner-scheme-a-20260817"
+
+
+@pytest.fixture(autouse=True)
+def _authorization_id(monkeypatch):
+    monkeypatch.setenv("AIOS_AGENT_TASK_AUTHORIZATION_ID", AUTHORIZATION_ID)
+
+
+def canon(v):
+    return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(",",":")).encode()).hexdigest()
+
+
+def db(tmp_path):
+    p=tmp_path/"state.sqlite3"
+    c=sqlite3.connect(p)
+    c.executescript("""
+    CREATE TABLE tasks(
+      task_id TEXT PRIMARY KEY,status TEXT,attempt INTEGER,max_attempts INTEGER,revision INTEGER,
+      claimed_by TEXT,lease_token_hash TEXT,lease_expires_at TEXT,verification_json TEXT,result_json TEXT
+    );
+    CREATE TABLE task_events(
+      event_id INTEGER PRIMARY KEY AUTOINCREMENT,task_id TEXT,event_type TEXT,agent_id TEXT,
+      payload_json TEXT,created_at TEXT
+    );
+    """)
+    verification={"decision":"NEEDS_REWORK","reason":"BASE_OID_BINDING_INVALID"}
+    c.execute("INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?)",
+              ("task-a","FAILED",10,10,41,None,None,None,json.dumps(verification),json.dumps({"status":"FAILED"})))
+    c.commit(); c.close()
+    return p,verification
+
+
+def request(v):
+    return RetryExtensionRequest.from_mapping({
+      "expected_status":"FAILED","expected_revision":41,"expected_attempt":10,"expected_max_attempts":10,
+      "next_max_attempts":11,"previous_verification_sha256":canon(v),
+      "authorization_id":AUTHORIZATION_ID,"idempotency_key":"retry-task-a-attempt11",
+    })
+
+
+def test_retry_extension_cas_and_audit(tmp_path):
+    p,v=db(tmp_path)
+    out=RetryExtensionStore(p).extend("task-a",request(v))
+    assert out["status"]=="QUEUED" and out["max_attempts"]==11 and out["revision"]==42
+    c=sqlite3.connect(p)
+    assert c.execute("select status,attempt,max_attempts,revision from tasks").fetchone()==("QUEUED",10,11,42)
+    event=c.execute("select event_type,payload_json from task_events").fetchone()
+    assert event[0]=="TASK_RETRY_EXTENSION_AUTHORIZED"
+    assert json.loads(event[1])["authorization_id"]=="owner-scheme-a-20260817"
+
+
+def test_retry_extension_idempotent_replay(tmp_path):
+    p,v=db(tmp_path); store=RetryExtensionStore(p); req=request(v)
+    first=store.extend("task-a",req); second=store.extend("task-a",req)
+    assert first["idempotent"] is False and second["idempotent"] is True
+    c=sqlite3.connect(p)
+    assert c.execute("select count(*) from task_retry_extensions").fetchone()[0]==1
+
+
+def test_retry_extension_normalizes_stale_terminal_claim_without_active_lease(tmp_path):
+    p,v=db(tmp_path)
+    c=sqlite3.connect(p)
+    c.execute("update tasks set claimed_by='previous-worker'")
+    c.commit(); c.close()
+    out=RetryExtensionStore(p).extend("task-a",request(v))
+    assert out["status"]=="QUEUED"
+    c=sqlite3.connect(p)
+    assert c.execute(
+        "select claimed_by,lease_token_hash,lease_expires_at from tasks"
+    ).fetchone()==(None,None,None)
+    payload=json.loads(c.execute("select payload_json from task_events").fetchone()[0])
+    assert payload["previous_claimed_by"]=="previous-worker"
+    assert payload["stale_terminal_claim_normalized"] is True
+    assert c.execute("select count(*) from task_events").fetchone()[0]==1
+
+
+def test_retry_extension_verification_drift_denied(tmp_path):
+    p,v=db(tmp_path)
+    bad=request({"decision":"NEEDS_REWORK","reason":"other"})
+    with pytest.raises(RetryExtensionError,match="RETRY_EXTENSION_VERIFICATION_DRIFT"):
+        RetryExtensionStore(p).extend("task-a",bad)
+
+
+def test_retry_extension_authorization_id_mismatch_rejected(tmp_path, monkeypatch):
+    p,v=db(tmp_path)
+    monkeypatch.setenv("AIOS_AGENT_TASK_AUTHORIZATION_ID", "other-owner-id")
+    with pytest.raises(RetryExtensionError,match="RETRY_EXTENSION_AUTHORIZATION_ID_MISMATCH"):
+        RetryExtensionStore(p).extend("task-a",request(v))
+
+
+def test_retry_extension_bounded_one_step():
+    v={"decision":"NEEDS_REWORK"}
+    data={
+      "expected_status":"FAILED","expected_revision":41,"expected_attempt":10,"expected_max_attempts":10,
+      "next_max_attempts":12,"previous_verification_sha256":canon(v),
+      "authorization_id":"owner-scheme-a-20260817","idempotency_key":"retry-task-a-attempt11",
+    }
+    with pytest.raises(RetryExtensionError,match="RETRY_EXTENSION_MUST_BE_BOUNDED_ONE_STEP"):
+        RetryExtensionRequest.from_mapping(data)
+
+
+def test_retry_extension_unit_is_unprivileged_and_requires_owner_binding():
+    unit=(ROOT/"deploy/systemd/aios-agent-task-retry-extension.service").read_text()
+    assert "User=aios-agent-task" in unit
+    assert "Group=aios-agent-task" in unit
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "EnvironmentFile=/etc/aios-gitops/agent-task-retry-extension.env" in unit
+    assert "User=root" not in unit

tests/test_isolated_executor.py

@@ -0,0 +1,67 @@
+from pathlib import Path
+import hashlib,json,os,tarfile,sys,tempfile
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+import aios_isolated_executor as mod
+
+def make_archive(tmp_path):
+    root=tmp_path/"src"; root.mkdir()
+    (root/"hello.py").write_text("VALUE = 1\n")
+    archive=tmp_path/"source.tgz"
+    with tarfile.open(archive,"w:gz") as tf: tf.add(root/"hello.py",arcname="hello.py")
+    return archive
+
+def test_isolated_executor_applies_hash_bound_change(tmp_path,monkeypatch):
+    archive=make_archive(tmp_path)
+    monkeypatch.delenv("CREDENTIALS_DIRECTORY",raising=False)
+    monkeypatch.setattr(mod,"assert_uncredentialed",lambda:None)
+    request={
+      "schema":"aios.isolated-execution.request.v1",
+      "request_id":"task-req-0000000000000000",
+      "source_archive":str(archive),
+      "source_archive_sha256":mod.sha256(archive.read_bytes()),
+      "change_set":[{
+        "path":"hello.py","operation":"replace_text","old":"VALUE = 1","new":"VALUE = 2","count":1,
+        "expected_sha256":mod.sha256(b"VALUE = 1\n")
+      }],
+      "verification":[{"label":"compile","argv":["/usr/bin/python3","-m","py_compile","hello.py"],"cwd":".","timeout_seconds":30}]
+    }
+    req=tmp_path/"req.json"; res=tmp_path/"res.json"
+    req.write_text(json.dumps(request))
+    mod.execute_request(req,res)
+    out=json.loads(res.read_text())
+    assert out["request_id"]=="task-req-0000000000000000"
+    assert out["credential_isolation"]["credential_env_absent"] is True
+    changed=out["changed_files"][0]
+    assert changed["path"]=="hello.py" and changed["deleted"] is False
+
+def test_isolated_executor_source_drift_denied(tmp_path,monkeypatch):
+    archive=make_archive(tmp_path); monkeypatch.setattr(mod,"assert_uncredentialed",lambda:None)
+    request={
+      "schema":"aios.isolated-execution.request.v1","request_id":"task-req-0000000000000001",
+      "source_archive":str(archive),
+      "source_archive_sha256":mod.sha256(archive.read_bytes()),
+      "change_set":[{"path":"hello.py","operation":"write_file","content":"x\n","expected_sha256":"0"*64}],
+      "verification":[]
+    }
+    req=tmp_path/"req.json"; res=tmp_path/"res.json"; req.write_text(json.dumps(request))
+    with pytest.raises(mod.IsolatedExecutorError,match="SOURCE_DRIFT"):
+        mod.execute_request(req,res)
+
+def test_credential_boundary_eacces_is_accepted(monkeypatch):
+    class DeniedDirectory:
+        def exists(self): return True
+        def is_symlink(self): return False
+        def is_dir(self): return True
+        def iterdir(self): raise PermissionError(13,"permission denied")
+    monkeypatch.setattr(mod.os,"access",lambda *_args,**_kwargs:False)
+    assert mod._credential_exposure(DeniedDirectory()) is None
+
+def test_systemd_unit_has_no_credentials_and_private_network():
+    unit=(ROOT/"deploy/systemd/aios-online-agent-isolated-executor.service").read_text()
+    assert "PrivateNetwork=true" in unit
+    assert "InaccessiblePaths=/run/credentials" in unit
+    assert "LoadCredential" not in unit
+    assert "SupplementaryGroups=gitops" not in unit

tests/test_online_agent_durable_checkpoint.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+from aios_online_agent_checkpoint import (
+    CheckpointBinding,
+    CheckpointError,
+    CheckpointStore,
+)
+import aios_online_agent_worker as worker
+
+
+def binding(attempt: int = 1, *, tenant_head: str = "b"*40):
+    return CheckpointBinding(
+        task_id="task-p1-a",
+        attempt=attempt,
+        execution_digest="d"*64,
+        job_id="job-p1-a",
+        repository="aios/aios",
+        base_oid="a"*40,
+        expected_branch_oid=tenant_head,
+        target_branch="agent/task-p1-a",
+        generation=1,
+    )
+
+
+def test_CHECKPOINT_DURABLE_WRITE(tmp_path):
+    store=CheckpointStore(tmp_path/"checkpoints.sqlite3")
+    first=store.append(binding(), "claimed", {"phase":"claimed","attempt":1})
+    second=store.append(binding(), "workspace_prepared", {"phase":"workspace_prepared","attempt":1})
+    assert first["sequence"] == 1
+    assert second["sequence"] == 2
+    latest=store.latest("task-p1-a")
+    assert latest["phase"]=="workspace_prepared"
+    assert latest["payload"]["attempt"]==1
+
+
+def test_CHECKPOINT_SCOPE_ISOLATION(tmp_path):
+    store=CheckpointStore(tmp_path/"checkpoints.sqlite3")
+    store.append(binding(), "claimed", {"phase":"claimed"})
+    changed=CheckpointBinding(
+        task_id="task-p1-a",attempt=1,execution_digest="d"*64,job_id="job-p1-a",
+        repository="aios/other",base_oid="a"*40,expected_branch_oid="b"*40,
+        target_branch="agent/task-p1-a",generation=1,
+    )
+    with pytest.raises(CheckpointError,match="CHECKPOINT_SCOPE_MISMATCH_DENY"):
+        store.append(changed,"workspace_prepared",{"phase":"workspace_prepared"})
+
+
+def test_RESUME_FROM_CHECKPOINT_across_attempt_same_binding(tmp_path):
+    store=CheckpointStore(tmp_path/"checkpoints.sqlite3")
+    store.append(binding(1), "source_capture_done", {
+        "phase":"source_capture_done",
+        "attempt":1,
+        "result":{"status":"SUCCEEDED","classification":"NEEDS_REWORK_SOURCE_CAPTURE"},
+    })
+    resumed=store.verify_latest_binding(binding(2))
+    assert resumed is not None
+    assert resumed["attempt"]==1
+    assert resumed["phase"]=="source_capture_done"
+
+
+def test_checkpoint_new_attempt_can_start_from_claimed(tmp_path):
+    store=CheckpointStore(tmp_path/"checkpoints.sqlite3")
+    store.append(binding(1),"result_submitted",{"phase":"result_submitted"})
+    cp=store.append(binding(2),"claimed",{"phase":"claimed"})
+    assert cp["attempt"]==2
+    assert cp["sequence"]==1
+
+
+def envelope():
+    return worker.ExecutionEnvelope(
+        task_id="task-p1-a",
+        job_id="job-p1-a",
+        repository="aios/aios",
+        base_oid="a"*40,
+        expected_branch_oid="b"*40,
+        target_branch="agent/task-p1-a",
+        branch_mode="CONTINUE_EXISTING",
+        generation=1,
+        idempotency_key="governed-p1-a",
+        action="source_capture",
+        adapter="source_capture",
+        capability_scope=("git.read","git.test"),
+        source_signals=(),
+        change_set=(),
+        verification=(),
+        governance={"task_id":"task-p1-a","job_id":"job-p1-a"},
+        workspace={},
+        digest="d"*64,
+    )
+
+
+def test_RESULT_SUBMISSION_REPLAY_SAFE_attempt_aware_key():
+    env=envelope()
+    result={"status":"SUCCEEDED","execution_digest":env.digest}
+    one=worker.result_idempotency_key(env.task_id,1,env,result)
+    two=worker.result_idempotency_key(env.task_id,2,env,result)
+    assert one != two
+    assert "-a1-result-" in one
+    assert "-a2-result-" in two
+    assert one == worker.result_idempotency_key(env.task_id,1,env,result)
+
+
+class RetryClient:
+    def __init__(self):
+        self.calls=[]
+    def result(self,task_id,lease_token,key,result):
+        self.calls.append((task_id,key,dict(result)))
+        return (599,{"error":"SERVICE_UNAVAILABLE"}) if len(self.calls)==1 else (201,{"status":"VERIFYING"})
+    def get(self,task_id):
+        return 500,{}
+
+
+def test_RETRY_IDEMPOTENT_NO_DUPLICATE_SIDE_EFFECT_reuses_same_key(monkeypatch):
+    client=RetryClient()
+    monkeypatch.setattr(worker.time,"sleep",lambda _:None)
+    status,response=worker.submit_result_replay_safe(
+        client,"task-p1-a","lease-token-"+"x"*32,"stable-key",{"status":"SUCCEEDED"}
+    )
+    assert status==201
+    assert [item[1] for item in client.calls]==["stable-key","stable-key"]
+
+
+class ConflictClient:
+    def __init__(self,result):
+        self.result_value=result
+    def result(self,*args):
+        return 409,{"error":"IDEMPOTENCY_CONFLICT"}
+    def get(self,task_id):
+        return 200,{"status":"VERIFYING","result":dict(self.result_value)}
+
+
+def test_RESULT_SUBMISSION_REPLAY_SAFE_reconciles_exact_result():
+    result={"status":"SUCCEEDED","value":1}
+    status,response=worker.submit_result_replay_safe(
+        ConflictClient(result),"task-p1-a","lease-token-"+"x"*32,"stable-key",result
+    )
+    assert status==200
+    assert response["idempotent"] is True
+    assert response["reconciled"] is True
+
+
+def test_RESULT_SUBMISSION_REPLAY_SAFE_rejects_conflicting_result():
+    status,response=worker.submit_result_replay_safe(
+        ConflictClient({"status":"SUCCEEDED","value":2}),
+        "task-p1-a","lease-token-"+"x"*32,"stable-key",
+        {"status":"SUCCEEDED","value":1},
+    )
+    assert status==409
+    assert response["error"]=="IDEMPOTENCY_CONFLICT"
+
+
+def test_WORKSPACE_CLEANUP_RECOVERY(tmp_path,monkeypatch):
+    root=tmp_path/"workspaces"; root.mkdir()
+    stale=root/"task-old"; stale.mkdir(); (stale/"x").write_text("x")
+    monkeypatch.setattr(worker,"WORKSPACE_ROOT",root)
+    monkeypatch.setattr(worker,"STATE_ROOT",tmp_path/"state")
+    removed=worker.recover_stale_workspaces()
+    assert removed==1
+    assert not stale.exists()
+
+
+class LeaseBridge:
+    def __init__(self):
+        self.calls=0
+    def call(self,route,body):
+        assert route=="/v2/writer-lease"
+        self.calls+=1
+        if self.calls==1:
+            return 599,{"error":"SERVICE_UNAVAILABLE"}
+        return 200,{
+            "lease_id":"lease-p1-a",
+            "job_id":"job-p1-a",
+            "task_id":"task-p1-a",
+            "repository":"aios/aios",
+            "base_oid":"a"*40,
+            "expected_branch_oid":"b"*40,
+            "branch_mode":"CONTINUE_EXISTING",
+            "generation":1,
+            "target_branch":"agent/task-p1-a",
+            "state":"ACTIVE",
+        }
+
+
+class LeaseHeartbeatStub:
+    def ensure(self):
+        return None
+
+
+def test_LEASE_RENEWAL_RECOVERY_retries_same_binding(monkeypatch):
+    env=envelope()
+    env=worker.ExecutionEnvelope(
+        task_id=env.task_id,job_id=env.job_id,repository=env.repository,
+        base_oid=env.base_oid,expected_branch_oid=env.expected_branch_oid,
+        target_branch=env.target_branch,branch_mode=env.branch_mode,generation=env.generation,
+        idempotency_key=env.idempotency_key,action=env.action,adapter=env.adapter,
+        capability_scope=env.capability_scope,source_signals=env.source_signals,
+        change_set=env.change_set,verification=env.verification,
+        governance={"task_id":env.task_id,"job_id":env.job_id,"lease_seconds":900},
+        workspace=env.workspace,digest=env.digest,
+    )
+    bridge=LeaseBridge()
+    monkeypatch.setattr(worker.time,"sleep",lambda _:None)
+    monkeypatch.setattr(worker,"STATE_ROOT",Path("/tmp")/"aios-p1-test-state")
+    response=worker.acquire_writer_lease_replay_safe(env,bridge,LeaseHeartbeatStub())
+    assert response["lease_id"]=="lease-p1-a"
+    assert bridge.calls==2

tests/test_online_agent_plan_store.py

@@ -0,0 +1,60 @@
+from pathlib import Path
+import sys
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_online_agent_execution import GovernedExecutionBinding
+from aios_online_agent_plan_store import ExecutionPlanStore, ExecutionPlanError
+
+BASE="a642b045775b9154a00ae78543037fbef4fb447b"
+HEAD="71523aef6e48b1e2fa516c456ca189b7a039104b"
+
+def binding(head=HEAD):
+    return GovernedExecutionBinding.from_mapping({
+      "task_id":"task-plan-a","job_id":"job-plan-a","repository":"aios/aios","base_oid":BASE,
+      "expected_branch_oid":head,"target_branch":"agent/task-plan-a","branch_mode":"CONTINUE_EXISTING",
+      "generation":1,"idempotency_key":"governed-plan-a",
+    })
+
+def request():
+    return {
+      "schema":"aios.online-agent.plan.v1","task_id":"task-plan-a","job_id":"job-plan-a",
+      "source_head_oid":HEAD,"generation":1,"idempotency_key":"plan-task-a-v1",
+      "execution":{
+        "action":"deterministic",
+        "change_set":[{"path":"x.py","operation":"write_file","content":"x=1\n"}],
+        "verification":[{"label":"compile","argv":["/usr/bin/python3","-m","py_compile","x.py"],"cwd":".","timeout_seconds":30}],
+        "capability_scope":["git.read","git.test"],
+      }
+    }
+
+def test_plan_register_and_resolve(tmp_path):
+    store=ExecutionPlanStore(tmp_path)
+    created=store.register(binding(),request())
+    assert created["idempotent"] is False
+    resolved=store.resolve(binding())
+    assert resolved["plan_digest"]==created["plan_digest"]
+    assert resolved["execution"]["action"]=="deterministic"
+
+def test_plan_is_idempotent(tmp_path):
+    store=ExecutionPlanStore(tmp_path); req=request()
+    first=store.register(binding(),req); second=store.register(binding(),req)
+    assert second["idempotent"] is True
+    assert first["plan_digest"]==second["plan_digest"]
+
+def test_plan_source_head_drift_denied(tmp_path):
+    store=ExecutionPlanStore(tmp_path); store.register(binding(),request())
+    with pytest.raises(ExecutionPlanError,match="SOURCE_HEAD_DRIFT"):
+        store.resolve(binding("b"*40))
+
+def test_plan_replacement_requires_cas(tmp_path):
+    store=ExecutionPlanStore(tmp_path); first=store.register(binding(),request())
+    changed=request(); changed["idempotency_key"]="plan-task-a-v2"
+    changed["execution"]["change_set"][0]["content"]="x=2\n"
+    with pytest.raises(ExecutionPlanError,match="REPLACEMENT_CAS_MISMATCH"):
+        store.register(binding(),changed)
+    changed["expected_previous_plan_digest"]=first["plan_digest"]
+    second=store.register(binding(),changed)
+    assert second["plan_digest"]!=first["plan_digest"]