fix(gitops): repair continuation normalization and durable head

active-runtime/app/aios_agent_task_api.py

@@ -1,315 +0,0 @@
-#!/usr/bin/env python3
-"""HTTP API for AIOS external-agent online task distribution.
-
-The service is intended to bind to loopback and be published through the
-existing TLS reverse proxy. Control operations use a server-managed control
-Bearer token. Agents receive individual Bearer tokens at registration; only
-token hashes are persisted.
-"""
-from __future__ import annotations
-
-from dataclasses import dataclass
-import hmac
-import json
-import os
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
-import re
-from typing import Any, Mapping
-from urllib.parse import urlparse
-
-from aios_agent_task_protocol import TaskProtocolError, TaskStore
-
-
-MAX_BODY = 1024 * 1024
-TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
-
-
-@dataclass(frozen=True, slots=True)
-class APIConfig:
-    db_path: Path
-    control_token_path: Path
-    host: str = "127.0.0.1"
-    port: int = 18940
-    token_pepper: str = ""
-
-    @classmethod
-    def from_env(cls) -> "APIConfig":
-        return cls(
-            db_path=Path(os.getenv(
-                "AIOS_AGENT_TASK_DB",
-                "/var/lib/aios-agent-task/state.sqlite3",
-            )),
-            control_token_path=Path(os.getenv(
-                "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
-                "/run/credentials/aios-agent-task-api.service/control_token",
-            )),
-            host=os.getenv("AIOS_AGENT_TASK_HOST", "127.0.0.1"),
-            port=int(os.getenv("AIOS_AGENT_TASK_PORT", "18940")),
-            token_pepper=os.getenv("AIOS_AGENT_TASK_TOKEN_PEPPER", ""),
-        )
-
-
-class AgentTaskApplication:
-    def __init__(self, config: APIConfig) -> None:
-        self.config = config
-        self.store = TaskStore(
-            config.db_path,
-            token_pepper=config.token_pepper,
-        )
-
-    def _control_token(self) -> str:
-        path = self.config.control_token_path
-        if path.is_symlink() or not path.is_file():
-            raise TaskProtocolError("CONTROL_CREDENTIAL_UNAVAILABLE", 503)
-        value = path.read_text(encoding="utf-8").strip()
-        if len(value) < 32:
-            raise TaskProtocolError("CONTROL_CREDENTIAL_INVALID", 503)
-        return value
-
-    @staticmethod
-    def _bearer(headers: Mapping[str, str]) -> str:
-        value = str(headers.get("Authorization") or "")
-        if not value.startswith("Bearer "):
-            return ""
-        return value[7:].strip()
-
-    def authenticate_control(self, headers: Mapping[str, str]) -> None:
-        supplied = self._bearer(headers)
-        expected = self._control_token()
-        if not supplied or not hmac.compare_digest(supplied, expected):
-            raise TaskProtocolError("UNAUTHORIZED", 401)
-
-    def authenticate_agent(self, headers: Mapping[str, str]) -> dict[str, Any]:
-        return self.store.authenticate_agent(self._bearer(headers))
-
-    def principal(self, headers: Mapping[str, str]) -> tuple[str, dict[str, Any] | None]:
-        supplied = self._bearer(headers)
-        if not supplied:
-            raise TaskProtocolError("UNAUTHORIZED", 401)
-        try:
-            expected = self._control_token()
-        except TaskProtocolError:
-            expected = ""
-        if expected and hmac.compare_digest(supplied, expected):
-            return "control", None
-        return "agent", self.store.authenticate_agent(supplied)
-
-    def health(self) -> dict[str, Any]:
-        return {
-            "status": "ok",
-            "schema": "aios.agent-task.v1",
-            "task_states": [
-                "QUEUED", "CLAIMED", "RUNNING", "VERIFYING",
-                "SUCCEEDED", "FAILED", "EXPIRED", "CANCELLED",
-            ],
-            "lease_recovery": True,
-            "atomic_claim": True,
-            "idempotent_result_submission": True,
-            "server_side_verification_required": True,
-            "plaintext_credentials_persisted": False,
-        }
-
-    def register_agent(self, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        self.authenticate_control(headers)
-        return self.store.register_agent(
-            agent_id=str(body.get("agent_id") or ""),
-            agent_type=str(body.get("agent_type") or ""),
-            agent_version=str(body.get("agent_version") or ""),
-            capabilities=body.get("capabilities") or [],
-            metadata=body.get("metadata") if isinstance(body.get("metadata"), Mapping) else {},
-        )
-
-    def create_task(self, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        self.authenticate_control(headers)
-        return self.store.create_task(
-            task_id=str(body.get("task_id") or "") or None,
-            goal=str(body.get("goal") or ""),
-            instructions=str(body.get("instructions") or ""),
-            success_criteria=body.get("success_criteria"),
-            required_capabilities=body.get("required_capabilities") or [],
-            constraints=body.get("constraints") if isinstance(body.get("constraints"), Mapping) else {},
-            metadata=body.get("metadata") if isinstance(body.get("metadata"), Mapping) else {},
-            deadline=str(body.get("deadline") or "") or None,
-            priority=body.get("priority", 0),
-            max_attempts=body.get("max_attempts", 3),
-            idempotency_key=str(body.get("idempotency_key") or ""),
-        )
-
-    def poll(self, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        agent = self.authenticate_agent(headers)
-        tasks = self.store.poll_tasks(
-            agent_id=agent["agent_id"],
-            limit=int(body.get("limit", 20)),
-        )
-        return {"agent_id": agent["agent_id"], "tasks": tasks}
-
-    def claim(self, task_id: str, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        agent = self.authenticate_agent(headers)
-        return self.store.claim_task(
-            task_id=task_id,
-            agent_id=agent["agent_id"],
-            lease_seconds=int(body.get("lease_seconds", 300)),
-        )
-
-    def heartbeat(self, task_id: str, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        agent = self.authenticate_agent(headers)
-        return self.store.heartbeat(
-            task_id=task_id,
-            agent_id=agent["agent_id"],
-            lease_token=str(body.get("lease_token") or ""),
-            progress=body.get("progress") if isinstance(body.get("progress"), Mapping) else {},
-            lease_seconds=int(body.get("lease_seconds", 300)),
-        )
-
-    def result(self, task_id: str, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        agent = self.authenticate_agent(headers)
-        result = body.get("result")
-        if not isinstance(result, Mapping):
-            raise TaskProtocolError("RESULT_INVALID", 400)
-        return self.store.submit_result(
-            task_id=task_id,
-            agent_id=agent["agent_id"],
-            lease_token=str(body.get("lease_token") or ""),
-            idempotency_key=str(body.get("idempotency_key") or ""),
-            result=result,
-        )
-
-    def verify(self, task_id: str, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        self.authenticate_control(headers)
-        return self.store.verify_task(
-            task_id=task_id,
-            decision=str(body.get("decision") or ""),
-            verification=body.get("verification") if isinstance(body.get("verification"), Mapping) else {},
-        )
-
-    def cancel(self, task_id: str, headers: Mapping[str, str], body: Mapping[str, Any]) -> dict[str, Any]:
-        self.authenticate_control(headers)
-        return self.store.cancel_task(task_id=task_id, reason=str(body.get("reason") or ""))
-
-    def get_task(self, task_id: str, headers: Mapping[str, str]) -> dict[str, Any]:
-        kind, agent = self.principal(headers)
-        task = self.store.get_task(task_id)
-        if kind == "agent" and task.get("claimed_by") != agent["agent_id"]:
-            raise TaskProtocolError("FORBIDDEN", 403)
-        return task
-
-    def events(self, task_id: str, headers: Mapping[str, str]) -> dict[str, Any]:
-        self.authenticate_control(headers)
-        return {"task_id": task_id, "events": self.store.get_events(task_id)}
-
-
-class Handler(BaseHTTPRequestHandler):
-    app: AgentTaskApplication | None = None
-    server_version = "AIOSAgentTaskAPI/1"
-
-    def log_message(self, _format: str, *_args: Any) -> None:
-        return
-
-    def _headers(self) -> dict[str, str]:
-        return {key: value for key, value in self.headers.items()}
-
-    def _json(self, value: Any, status: int = 200) -> None:
-        payload = json.dumps(value, ensure_ascii=False, sort_keys=True).encode("utf-8")
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json; charset=utf-8")
-        self.send_header("Cache-Control", "no-store")
-        self.send_header("Content-Length", str(len(payload)))
-        self.end_headers()
-        self.wfile.write(payload)
-
-    def _body(self) -> dict[str, Any]:
-        try:
-            size = int(self.headers.get("Content-Length", "0"))
-        except ValueError as exc:
-            raise TaskProtocolError("CONTENT_LENGTH_INVALID", 400) from exc
-        if size < 0 or size > MAX_BODY:
-            raise TaskProtocolError("REQUEST_TOO_LARGE", 413)
-        raw = self.rfile.read(size) if size else b"{}"
-        try:
-            value = json.loads(raw.decode("utf-8"))
-        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
-            raise TaskProtocolError("JSON_INVALID", 400) from exc
-        if not isinstance(value, dict):
-            raise TaskProtocolError("JSON_OBJECT_REQUIRED", 400)
-        return value
-
-    @staticmethod
-    def _task_route(path: str) -> tuple[str, str | None]:
-        match = re.fullmatch(
-            r"/v1/tasks/([A-Za-z0-9][A-Za-z0-9._:-]{2,179})(?:/(claim|heartbeat|result|verify|cancel|events))?",
-            path,
-        )
-        if match:
-            return match.group(1), match.group(2)
-        return "", None
-
-    def _run(self, callback) -> None:
-        try:
-            callback()
-        except TaskProtocolError as exc:
-            self._json({"error": exc.code}, exc.http_status)
-        except Exception as exc:
-            self._json({"error": "INTERNAL_ERROR", "detail": type(exc).__name__}, 500)
-
-    def do_GET(self) -> None:
-        def action() -> None:
-            if self.app is None:
-                raise TaskProtocolError("SERVICE_UNAVAILABLE", 503)
-            path = urlparse(self.path).path.rstrip("/") or "/"
-            if path == "/health":
-                self._json(self.app.health())
-                return
-            task_id, suffix = self._task_route(path)
-            if task_id and suffix is None:
-                self._json(self.app.get_task(task_id, self._headers()))
-                return
-            if task_id and suffix == "events":
-                self._json(self.app.events(task_id, self._headers()))
-                return
-            self._json({"error": "NOT_FOUND"}, 404)
-        self._run(action)
-
-    def do_POST(self) -> None:
-        def action() -> None:
-            if self.app is None:
-                raise TaskProtocolError("SERVICE_UNAVAILABLE", 503)
-            path = urlparse(self.path).path.rstrip("/") or "/"
-            body = self._body()
-            if path == "/v1/agents/register":
-                self._json(self.app.register_agent(self._headers(), body), 201)
-                return
-            if path == "/v1/tasks":
-                self._json(self.app.create_task(self._headers(), body), 201)
-                return
-            if path == "/v1/tasks/poll":
-                self._json(self.app.poll(self._headers(), body))
-                return
-            task_id, suffix = self._task_route(path)
-            if not task_id:
-                self._json({"error": "NOT_FOUND"}, 404)
-                return
-            if suffix == "claim":
-                self._json(self.app.claim(task_id, self._headers(), body))
-            elif suffix == "heartbeat":
-                self._json(self.app.heartbeat(task_id, self._headers(), body))
-            elif suffix == "result":
-                self._json(self.app.result(task_id, self._headers(), body))
-            elif suffix == "verify":
-                self._json(self.app.verify(task_id, self._headers(), body))
-            elif suffix == "cancel":
-                self._json(self.app.cancel(task_id, self._headers(), body))
-            else:
-                self._json({"error": "NOT_FOUND"}, 404)
-        self._run(action)
-
-
-def serve(config: APIConfig | None = None) -> None:
-    config = config or APIConfig.from_env()
-    application = AgentTaskApplication(config)
-    Handler.app = application
-    ThreadingHTTPServer((config.host, config.port), Handler).serve_forever()
-
-
-if __name__ == "__main__":
-    serve()

active-runtime/app/aios_agent_task_protocol.py

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

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -13,6 +13,8 @@ BRIDGE_OPERATIONS = (
     "getGovernedWriteCapabilities",
     "createGovernedTask",
     "acquireWriterLease",
+    "bootstrapWriterLease",
+    "renewWriterLease",
     "commitFileGoverned",
     "commitFilesGoverned",
     "getGitOpsJob",
@@ -31,6 +33,8 @@ BRIDGE_OPERATIONS = (
 OPERATION_REQUIRED_FIELDS = {
     "createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
     "acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
+    "bootstrapWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
+    "renewWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
     "commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
     "commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
     "getGitOpsJob": ("job_id",),
@@ -46,6 +50,49 @@ OPERATION_REQUIRED_FIELDS = {
     "getControlPlaneSnapshot": (),
 }
 
+BRIDGE_OPERATION_CONTRACTS = {
+    "acquireWriterLease": {
+        "method": "POST",
+        "path": "/v2/writer-lease",
+        "required_fields": OPERATION_REQUIRED_FIELDS["acquireWriterLease"],
+        "response_fields": (
+            "task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
+            "create_target_branch", "worktree_mode", "cas_ref_update_required",
+        ),
+    },
+    "bootstrapWriterLease": {
+        "method": "POST",
+        "path": "/v2/writer-lease/bootstrap",
+        "required_fields": OPERATION_REQUIRED_FIELDS["bootstrapWriterLease"],
+        "response_fields": (
+            "task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
+            "create_target_branch", "worktree_mode", "cas_ref_update_required",
+            "recovered", "bootstrapped", "runtime_rehydrated",
+            "durable_branch_lease_present", "idempotent",
+        ),
+    },
+    "renewWriterLease": {
+        "method": "POST",
+        "path": "/v2/writer-lease/renew",
+        "required_fields": OPERATION_REQUIRED_FIELDS["renewWriterLease"],
+        "response_fields": (
+            "task_id", "job_id", "generation", "lease_seconds", "lease_expires_at",
+            "recovered", "bootstrapped", "runtime_rehydrated",
+            "durable_branch_lease_present", "idempotent",
+        ),
+    },
+    "commitFileGoverned": {
+        "method": "POST", "path": "/v2/commit-file",
+        "required_fields": OPERATION_REQUIRED_FIELDS["commitFileGoverned"],
+        "response_fields": ("job_id", "commit_oid", "commit_visible", "branch_mode", "target_branch", "generation"),
+    },
+    "commitFilesGoverned": {
+        "method": "POST", "path": "/v2/commit-files",
+        "required_fields": OPERATION_REQUIRED_FIELDS["commitFilesGoverned"],
+        "response_fields": ("job_id", "commit_oid", "commit_visible", "branch_mode", "target_branch", "generation"),
+    },
+}
+
 
 class RuntimeDriftError(RuntimeError):
     pass
@@ -122,6 +169,14 @@ class BridgeV2Adapter:
             "legacy_writes_enabled": False,
             "governed_writes_enabled": True,
             "operations": list(BRIDGE_OPERATIONS),
+            "operation_contracts": {
+                operation: {
+                    **contract,
+                    "required_fields": list(contract["required_fields"]),
+                    "response_fields": list(contract["response_fields"]),
+                }
+                for operation, contract in BRIDGE_OPERATION_CONTRACTS.items()
+            },
             "force_push": False,
             "main_write": False,
             "tag_write": False,
@@ -151,6 +206,8 @@ class BridgeV2Adapter:
 
     def createGovernedTask(self, request: Mapping[str, Any]) -> Any: return self._call("createGovernedTask", request)
     def acquireWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("acquireWriterLease", request)
+    def bootstrapWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("bootstrapWriterLease", request)
+    def renewWriterLease(self, request: Mapping[str, Any]) -> Any: return self._call("renewWriterLease", request)
     def commitFileGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFileGoverned", request)
     def commitFilesGoverned(self, request: Mapping[str, Any]) -> Any: return self._call("commitFilesGoverned", request)
     def getGitOpsJob(self, request: Mapping[str, Any]) -> Any: return self._call("getGitOpsJob", request)

active-runtime/app/aios_continuation_executor.py

@@ -329,6 +329,7 @@ class ContinuationPermissions:
 @dataclass(frozen=True, slots=True)
 class ContinuationRuntimeBinding:
     request: WriterLeaseRequestV2
+    job_id: str
     workspace_path: Path
     lease_not_before: datetime
     lease_expires_at: datetime
@@ -389,13 +390,20 @@ class ContinuationExecutor:
     def observe_generation(self, branch: str, generation: int) -> None:
         self._observed_generations[branch] = max(generation, self._observed_generations.get(branch, 0))
 
-    def _max_generation(self, branch: str) -> int:
+    def _max_generation(self, branch: str, *, exclude_job_id: str | None = None) -> int:
         observed = self._observed_generations.get(branch, 0)
         with self.controller.store._connect() as connection:
-            row = connection.execute(
-                "SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
-                (branch,),
-            ).fetchone()
+            if exclude_job_id is None:
+                row = connection.execute(
+                    "SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
+                    (branch,),
+                ).fetchone()
+            else:
+                row = connection.execute(
+                    "SELECT MAX(generation) AS generation FROM branch_leases "
+                    "WHERE branch=? AND job_id!=?",
+                    (branch, exclude_job_id),
+                ).fetchone()
         durable = int(row["generation"] or 0)
         return max(observed, durable)
 
@@ -404,6 +412,7 @@ class ContinuationExecutor:
         return {
             "schema": WriterLeaseRequestV2.schema,
             "task_id": request.task_id,
+            "job_id": request.job_id,
             "repository": request.repository,
             "branch_mode": request.branch_mode,
             "base_oid": request.base_oid,
@@ -413,7 +422,7 @@ class ContinuationExecutor:
             "lease_seconds": request.lease_seconds,
         }
 
-    def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> None:
+    def _validate_preflight(self, binding: ContinuationRuntimeBinding, *, job_id: str | None = None) -> None:
         request = binding.request
         normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
         if normalized != {**self._request_mapping(request), "create_target_branch": False, "worktree_mode": "detached_at_expected_head", "cas_ref_update_required": True}:
@@ -426,7 +435,7 @@ class ContinuationExecutor:
             raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
         if now >= binding.lease_expires_at:
             raise ContinuationExecutionError("LEASE_EXPIRED")
-        if request.generation <= self._max_generation(request.target_branch):
+        if request.generation <= self._max_generation(request.target_branch, exclude_job_id=job_id):
             raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
         if not self.workspace.is_detached():
             raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
@@ -452,39 +461,60 @@ class ContinuationExecutor:
             "generation": request.generation,
         }
 
-    def execute(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit) -> dict[str, object]:
-        payload = commit.payload()
-        submitted = self.controller.submit(self._job_request(binding, commit, payload), payload=payload)
-        current = self.controller.store.read(submitted.job_id)
-        if JobState(current["state"]) == JobState.COMPLETED:
+    def execute(
+        self,
+        binding: ContinuationRuntimeBinding,
+        commit: ContinuationCommit,
+        *,
+        before_complete: Callable[[str], None] | None = None,
+    ) -> dict[str, object]:
+        commit.payload()
+        if not binding.job_id or binding.request.job_id != binding.job_id:
+            raise ContinuationExecutionError("RUNTIME_LINEAGE_MISSING_JOB_ID")
+        current = self.controller.store.read(binding.job_id)
+        current_state = JobState(current["state"])
+        if current_state == JobState.COMPLETED:
             return self.controller.store._receipt(current, duplicate=True).as_dict()
-        self._validate_preflight(binding)
+        if current_state == JobState.REMOTE_VERIFIED:
+            receipt = json.loads(str(current["receipt_json"] or "{}"))
+            commit_oid = str(receipt.get("authority_oid") or "")
+            if not OID_RE.fullmatch(commit_oid):
+                raise ContinuationExecutionError("GOVERNED_COMMIT_OID_INVALID")
+            if before_complete is not None:
+                before_complete(commit_oid)
+            return self.controller.advance(
+                binding.job_id,
+                JobState.COMPLETED,
+                side_effect_state="REMOTE_VERIFIED",
+                receipt=receipt,
+            ).as_dict()
+        self._validate_preflight(binding, job_id=binding.job_id)
         request = binding.request
         projected = False
         try:
             self.credential.project()
             projected = True
-            row = self.controller.store.read(submitted.job_id)
+            row = self.controller.store.read(binding.job_id)
             if JobState(row["state"]) == JobState.REQUESTED:
                 self.controller.advance(
-                    submitted.job_id,
+                    binding.job_id,
                     JobState.AUTHORIZED,
                     side_effect_state="AUTHORIZED",
                     receipt={"task_id": request.task_id, "original_grant_consumed": True},
                 )
-                row = self.controller.store.read(submitted.job_id)
+                row = self.controller.store.read(binding.job_id)
             if JobState(row["state"]) == JobState.AUTHORIZED:
                 self.controller.lease_job(
-                    submitted.job_id,
-                    lease_id=f"lease-{submitted.job_id}",
+                    binding.job_id,
+                    lease_id=f"lease-{binding.job_id}",
                     lineage_anchor_oid=request.base_oid,
                     current_head_oid=request.expected_branch_oid,
                     generation=request.generation,
                     lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(),
                     expected_revision=int(row["revision"]),
                 )
-            self.controller.advance(submitted.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
-            self.controller.advance(submitted.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
+            self.controller.advance(binding.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
+            self.controller.advance(binding.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
             new_oid = self.workspace.commit_files(commit.files, commit.message)
             if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
                 raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
@@ -492,17 +522,22 @@ class ContinuationExecutor:
             if authority_oid != new_oid:
                 raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
             self.controller.advance(
-                submitted.job_id,
+                binding.job_id,
                 JobState.LOCAL_VERIFIED,
                 side_effect_state="LOCAL_VERIFIED",
                 receipt={"authority_oid": authority_oid, "force_push": False},
             )
             try:
-                self.controller.advance(submitted.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
+                self.controller.advance(binding.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
                 self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
             except Exception:
-                return self._remote_pending(submitted.job_id, authority_oid, new_oid)
-            return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
+                return self._remote_pending(binding.job_id, authority_oid, new_oid)
+            return self._finish_remote_verification(
+                binding.job_id,
+                authority_oid,
+                new_oid,
+                before_complete=before_complete,
+            )
         finally:
             if projected:
                 self.credential.revoke()
@@ -525,7 +560,14 @@ class ContinuationExecutor:
         )
         return result.as_dict()
 
-    def _finish_remote_verification(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
+    def _finish_remote_verification(
+        self,
+        job_id: str,
+        authority_oid: str,
+        new_oid: str,
+        *,
+        before_complete: Callable[[str], None] | None = None,
+    ) -> dict[str, object]:
         row = self.controller.store.read(job_id)
         branch = row["branch"]
         request = json.loads(row["request_json"])["binding"]
@@ -542,8 +584,20 @@ class ContinuationExecutor:
             "reconciliation_decision": decision.action.value,
             "force_push": False,
         }
-        self.controller.advance(job_id, JobState.REMOTE_VERIFIED, side_effect_state="REMOTE_VERIFIED", receipt=receipt)
-        return self.controller.advance(job_id, JobState.COMPLETED, side_effect_state="REMOTE_VERIFIED", receipt=receipt).as_dict()
+        self.controller.advance(
+            job_id,
+            JobState.REMOTE_VERIFIED,
+            side_effect_state="REMOTE_VERIFIED",
+            receipt=receipt,
+        )
+        if before_complete is not None:
+            before_complete(new_oid)
+        return self.controller.advance(
+            job_id,
+            JobState.COMPLETED,
+            side_effect_state="REMOTE_VERIFIED",
+            receipt=receipt,
+        ).as_dict()
 
     def reconcile(self, job_id: str) -> dict[str, object]:
         row = self.controller.store.read(job_id)

active-runtime/app/aios_continuation_lease.py

@@ -49,6 +49,7 @@ class WriterLeaseRequestV2:
     target_branch: str
     generation: int
     lease_seconds: int = 900
+    job_id: str = ""
 
     schema = "aios.task-writer-lease.v2"
     accepted_schemas = frozenset({"aios.task-writer-lease.v2", "aios.task-writer-lease-request.v2"})
@@ -56,6 +57,8 @@ class WriterLeaseRequestV2:
     def validate(self, probe: BranchProbe) -> None:
         if not self.task_id or self.repository != "aios/aios":
             raise ContinuationError("request_identity_invalid")
+        if not self.job_id or not self.job_id.startswith("job-"):
+            raise ContinuationError("job_id_invalid")
         if self.branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}:
             raise ContinuationError("branch_mode_invalid")
         _oid(self.base_oid, "base_oid")

active-runtime/app/aios_gitops_job_controller.py

@@ -27,6 +27,19 @@ class ServiceUnavailable(JobError):
     pass
 
 
+@dataclass(frozen=True, slots=True)
+class LeaseRenewalResult:
+    """Durable outcome of one writer-lease recovery attempt."""
+
+    renewed: bool
+    idempotent: bool
+    durable: bool
+    lease_expires_at: str | None
+
+    def __bool__(self) -> bool:
+        return self.renewed or self.idempotent
+
+
 class BindingConflict(JobError):
     pass
 
@@ -66,6 +79,14 @@ def utc_now() -> str:
     return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
 
 
+def parse_utc(value: str) -> datetime:
+    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
+    parsed = datetime.fromisoformat(normalized)
+    if parsed.tzinfo is None:
+        raise ValueError("LEASE_EXPIRY_INVALID")
+    return parsed.astimezone(timezone.utc)
+
+
 def canonical_json(value: Any) -> bytes:
     return (
         json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n"
@@ -404,13 +425,16 @@ class GitOpsJobController:
         self.store._mirror(result)
         return self.store._receipt(result, duplicate=True)
 
-    def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE") -> None:
+    def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE", lease_expires_at: str | None = None) -> None:
         if len(current_head_oid) != 40 or generation < 1:
             raise JobError("LEASE_BINDING_INVALID")
         try:
             with self.store._connect() as connection:
                 connection.execute("BEGIN IMMEDIATE")
-                updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, state, lease_id, expected_revision))
+                if lease_expires_at is None:
+                    updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, state, lease_id, expected_revision))
+                else:
+                    updated = connection.execute("UPDATE branch_leases SET current_head_oid=?,generation=?,lease_expires_at=?,state=?,revision=revision+1 WHERE lease_id=? AND revision=?", (current_head_oid, generation, lease_expires_at, state, lease_id, expected_revision))
                 if updated.rowcount != 1:
                     raise JobError("LEASE_REVISION_CAS_CONFLICT")
                 connection.commit()
@@ -419,6 +443,94 @@ class GitOpsJobController:
         except sqlite3.Error as exc:
             raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
 
+    def renew_lease(
+        self,
+        *,
+        branch: str,
+        lineage_anchor_oid: str,
+        current_head_oid: str,
+        generation: int,
+        lease_expires_at: str,
+    ) -> LeaseRenewalResult:
+        if (
+            not branch
+            or len(lineage_anchor_oid) != 40
+            or len(current_head_oid) != 40
+            or generation < 1
+            or not lease_expires_at.strip()
+        ):
+            raise JobError("LEASE_BINDING_INVALID")
+        try:
+            with self.store._connect() as connection:
+                connection.execute("BEGIN IMMEDIATE")
+                rows = connection.execute(
+                    "SELECT lease_id, revision, generation, state, lease_expires_at "
+                    "FROM branch_leases "
+                    "WHERE branch=? AND lineage_anchor_oid=? AND current_head_oid=? "
+                    "AND generation=? AND state='ACTIVE' ORDER BY revision DESC",
+                    (branch, lineage_anchor_oid, current_head_oid, generation),
+                ).fetchall()
+                if not rows:
+                    connection.commit()
+                    return LeaseRenewalResult(False, False, False, None)
+                if len(rows) != 1:
+                    raise JobError("LEASE_BINDING_CONFLICT")
+                row = rows[0]
+                try:
+                    current_expires_at = parse_utc(str(row["lease_expires_at"]))
+                    requested_expires_at = parse_utc(lease_expires_at)
+                except (TypeError, ValueError) as exc:
+                    raise JobError("LEASE_EXPIRY_INVALID") from exc
+                now = datetime.now(timezone.utc)
+                if requested_expires_at <= now:
+                    raise JobError("LEASE_EXPIRY_INVALID")
+                if current_expires_at > now:
+                    connection.commit()
+                    return LeaseRenewalResult(
+                        False,
+                        True,
+                        True,
+                        str(row["lease_expires_at"]),
+                    )
+                updated = connection.execute(
+                    "UPDATE branch_leases SET lease_expires_at=?,revision=revision+1 "
+                    "WHERE lease_id=? AND revision=? AND generation=? "
+                    "AND state='ACTIVE' AND lease_expires_at=?",
+                    (
+                        lease_expires_at,
+                        row["lease_id"],
+                        int(row["revision"]),
+                        generation,
+                        row["lease_expires_at"],
+                    ),
+                )
+                if updated.rowcount != 1:
+                    current = connection.execute(
+                        "SELECT lease_expires_at, generation, state FROM branch_leases "
+                        "WHERE lease_id=?",
+                        (row["lease_id"],),
+                    ).fetchone()
+                    if (
+                        current is not None
+                        and int(current["generation"]) == generation
+                        and current["state"] == "ACTIVE"
+                        and parse_utc(str(current["lease_expires_at"])) > now
+                    ):
+                        connection.commit()
+                        return LeaseRenewalResult(
+                            False,
+                            True,
+                            True,
+                            str(current["lease_expires_at"]),
+                        )
+                    raise JobError("LEASE_REVISION_CAS_CONFLICT")
+                connection.commit()
+                return LeaseRenewalResult(True, False, True, lease_expires_at)
+        except JobError:
+            raise
+        except sqlite3.Error as exc:
+            raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
+
     def advance(self, job_id: str, state: JobState, *, side_effect_state: str, receipt: Mapping[str, Any] | None = None, expected_revision: int | None = None) -> JobReceipt:
         return self._advance(
             job_id,

active-runtime/app/aios_model_api_staging.py

@@ -112,8 +112,6 @@ READ_ONLY_GET_PATHS = frozenset({
     '/admin/ai-evolution/api/data',
     '/api/release-consistency', '/api/tag-publication-status',
     '/api/release-acceptance',
-    '/api/governed-write-capabilities', '/api/control-plane-snapshot',
-    '/api/online-agent-task', '/api/online-agent-task-events',
     '/getReleaseCandidate', '/getReleasePlan', '/getReleaseAcceptance',
     '/getApprovalEnvelope', '/getReleaseSaga', '/getCapabilityStatus',
     '/getRecoveryStatus',
@@ -448,181 +446,24 @@ class H(BaseHTTPRequestHandler):
             return self.sendj(body, exc.code)
         except (urllib.error.URLError, TimeoutError) as exc:
             return self.sendj({'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__}, 503)
-    def bridge_v2_json(self, route, data=None, method='POST'):
-        base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
-        payload = None if data is None else json.dumps(
-            data, sort_keys=True, separators=(',', ':')
-        ).encode('utf-8')
-        request = urllib.request.Request(
-            base + route,
-            data=payload,
-            headers={'Content-Type': 'application/json'} if payload is not None else {},
-            method=method,
-        )
-        try:
-            with urllib.request.urlopen(request, timeout=30) as response:
-                result = json.loads(response.read().decode('utf-8'))
-                return self.sendj(result, response.status)
-        except urllib.error.HTTPError as exc:
-            try:
-                body = json.loads(exc.read().decode('utf-8'))
-            except Exception:
-                body = {'error': 'bridge_http_error'}
-            return self.sendj(body, exc.code)
-        except (urllib.error.URLError, TimeoutError) as exc:
-            return self.sendj(
-                {'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__},
-                503,
-            )
-
-    def control_plane_snapshot(self):
-        """Build a non-sensitive snapshot locally when Bridge v2 has no snapshot HTTP route."""
-        base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
-
-        def bridge_get(route):
-            request = urllib.request.Request(
-                base + route,
-                headers={'Accept': 'application/json'},
-                method='GET',
-            )
-            with urllib.request.urlopen(request, timeout=15) as response:
-                if response.status != 200:
-                    raise RuntimeError('bridge_read_http_' + str(response.status))
-                value = json.loads(response.read().decode('utf-8'))
-                if not isinstance(value, dict):
-                    raise RuntimeError('bridge_read_not_object')
-                return value
-
-        try:
-            runtime = bridge_get('/v2/runtime')
-            capabilities = bridge_get('/v2/capabilities')
-            repository = repository_snapshot(
-                git_dir=AUTHORITATIVE_GIT_DIR,
-                repository_identity=REPOSITORY_IDENTITY,
-            )
-            if not repository.get('repository_available'):
-                return self.sendj({
-                    'error': 'AUTHORITATIVE_REPOSITORY_UNAVAILABLE',
-                    'query_error': repository.get('query_error'),
-                }, 503)
-            baseline = dict(BASELINE_AUTHORITY.read())
-        except BaselineAuthorityError as exc:
-            return self.sendj({
-                'error': 'RECONCILIATION_REQUIRED',
-                'detail': str(exc),
-            }, 503)
-        except urllib.error.HTTPError as exc:
-            return self.sendj({
-                'error': 'GOVERNED_BRIDGE_UNAVAILABLE',
-                'detail': 'HTTP_' + str(exc.code),
-            }, 503)
-        except (urllib.error.URLError, TimeoutError, OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
-            return self.sendj({
-                'error': 'GOVERNED_BRIDGE_UNAVAILABLE',
-                'detail': type(exc).__name__,
-            }, 503)
-
-        legacy = capabilities.get('legacy_writes_enabled')
-        if legacy is None:
-            legacy = capabilities.get('legacy_global_writes_enabled')
-        payload = {
-            'schema': 'aios.control-plane-snapshot.aggregate.v1',
-            'snapshot_source': 'model-api-local-aggregate',
-            'version': {
-                'target_release': runtime.get('control_plane_release_version'),
-            },
-            'runtime': {
-                'bridge_schema_version': runtime.get('bridge_schema_version'),
-                'control_plane_release_version': runtime.get('control_plane_release_version'),
-                'runtime_release_commit': runtime.get('runtime_release_commit'),
-                'runtime_path': runtime.get('runtime_path'),
-                'artifact_sha256': runtime.get('artifact_sha256'),
-                'drift_status': runtime.get('drift_status'),
-                'deployment_state': runtime.get('deployment_state'),
-            },
-            'repository': {
-                'identity': runtime.get('repository_identity') or runtime.get('repository') or REPOSITORY_IDENTITY,
-                'authority_git_dir': runtime.get('authoritative_git_dir') or repository.get('authoritative_git_dir'),
-                'authority_main_oid': repository.get('authoritative_head_oid'),
-            },
-            'production': {
-                'baseline_version': baseline.get('current_baseline'),
-                'target_oid': baseline.get('target_oid'),
-                'deployment_state': baseline.get('status'),
-                'baseline_generation': baseline.get('generation'),
-                'baseline_sha256': baseline.get('authority_sha256'),
-            },
-            'task': {},
-            'capabilities': {
-                'legacy_writes_enabled': legacy,
-                'governed_writes_enabled': capabilities.get('governed_writes_enabled'),
-                'force_push': capabilities.get('force_push'),
-                'main_write': capabilities.get('main_write'),
-                'tag_write': capabilities.get('tag_write'),
-                'baseline_write': capabilities.get('baseline_write'),
-                'production_write': capabilities.get('production_write'),
-            },
-            'credential_broker': {
-                'status': 'SERVER_SIDE_ONLY',
-                'provider': 'BridgeV2RuntimeCredentialProjection',
-            },
-            'secret_material_exposed': False,
-        }
-        return self.sendj(payload)
-
-    def agent_task_control(self, method, route, data=None):
-        token_path = Path(os.getenv(
-            'AIOS_AGENT_TASK_CONTROL_TOKEN_FILE',
-            '/run/credentials/aios-gitops-model-api-v2r2c1-staging.service/agent_task_control_token',
-        ))
-        if token_path.is_symlink() or not token_path.is_file():
-            return self.sendj({'error': 'AGENT_TASK_CONTROL_CREDENTIAL_UNAVAILABLE'}, 503)
-        token = token_path.read_text(encoding='utf-8').strip()
-        if len(token) < 32:
-            return self.sendj({'error': 'AGENT_TASK_CONTROL_CREDENTIAL_INVALID'}, 503)
-        base = os.getenv('AIOS_AGENT_TASK_API_URL', 'http://127.0.0.1:18940').rstrip('/')
-        payload = None if data is None else json.dumps(
-            data, sort_keys=True, separators=(',', ':')
-        ).encode('utf-8')
-        headers = {'Authorization': 'Bearer ' + token, 'Accept': 'application/json'}
-        if payload is not None:
-            headers['Content-Type'] = 'application/json'
-        request = urllib.request.Request(
-            base + route, data=payload, headers=headers, method=method
-        )
-        try:
-            with urllib.request.urlopen(request, timeout=30) as response:
-                result = json.loads(response.read().decode('utf-8'))
-                return self.sendj(result, response.status)
-        except urllib.error.HTTPError as exc:
-            try:
-                body = json.loads(exc.read().decode('utf-8'))
-            except Exception:
-                body = {'error': 'agent_task_http_error'}
-            return self.sendj(body, exc.code)
-        except (urllib.error.URLError, TimeoutError) as exc:
-            return self.sendj(
-                {'error': 'AGENT_TASK_API_UNAVAILABLE', 'detail': type(exc).__name__},
-                503,
-            )
-
+    def do_POST(self):
+        path = urllib.parse.urlparse(self.path).path
+        if path in {
+            '/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
+            '/buildProductionArtifact', '/executeApprovedProductionRelease',
+            '/getReleaseConsistency', '/runReleaseAcceptance',
+        }:
+            actor = self.need('tag:create')
+            if not actor:
+                return
+            return self.bridge_release(path, self.body())
+        if reject_non_read_http(self, 'POST', path):
+            return
+        return self.sendj({'error': 'not_found', 'path': path}, 404)
     def do_GET(self):
         u=urllib.parse.urlparse(self.path); path=u.path.rstrip('/') if u.path != '/' else u.path
         if path not in READ_ONLY_GET_PATHS: return self.sendj({'error':'not_found','path':path},404)
         q=urllib.parse.parse_qs(u.query)
-        if path == '/api/governed-write-capabilities':
-            return self.bridge_v2_json('/v2/capabilities', method='GET')
-        if path == '/api/control-plane-snapshot':
-            return self.control_plane_snapshot()
-        if path in ('/api/online-agent-task', '/api/online-agent-task-events'):
-            actor = self.need('agent-task:control')
-            if not actor:
-                return
-            task_id = (q.get('task_id') or [''])[0]
-            if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{2,179}', task_id):
-                return self.sendj({'error': 'task_id_invalid'}, 400)
-            suffix = '/events' if path.endswith('-events') else ''
-            return self.agent_task_control('GET', '/v1/tasks/' + task_id + suffix)
         try:
             if path in ('/health','/api/health'):
                 return self.sendj({
@@ -859,45 +700,6 @@ class H(BaseHTTPRequestHandler):
         return self.sendj({'error':'not_found','path':path},404)
     def do_POST(self):
         path=urllib.parse.urlparse(self.path).path.rstrip('/')
-        governed_routes = {
-            '/createGovernedTask': ('/v2/governed-task', 'branch:create'),
-            '/acquireWriterLease': ('/v2/writer-lease', 'file:write'),
-            '/commitFileGoverned': ('/v2/commit-file', 'file:write'),
-            '/commitFilesGoverned': ('/v2/commit-files', 'file:write'),
-            '/getGitOpsJob': ('/v2/job', 'file:write'),
-            '/resumeGitOpsJob': ('/v2/job/resume', 'file:write'),
-            '/reconcileGitOpsJob': ('/v2/job/reconcile', 'file:write'),
-            '/createTagApprovalRequest': ('/v2/tag/approval-request', 'tag:create'),
-            '/requestOrCreateTag': ('/v2/tag/publish', 'tag:create'),
-            '/getTagPublicationStatus': ('/v2/tag/status', 'tag:create'),
-            '/buildProductionArtifact': ('/v2/release/artifact', 'tag:create'),
-            '/executeApprovedProductionRelease': ('/v2/release/execute', 'tag:create'),
-            '/getReleaseConsistency': ('/v2/release/consistency', 'tag:create'),
-            '/runReleaseAcceptance': ('/v2/release/acceptance', 'tag:create'),
-        }
-        if path in governed_routes:
-            route, scope = governed_routes[path]
-            actor = self.need(scope)
-            if not actor:
-                return
-            return self.bridge_v2_json(route, self.body(), method='POST')
-        if path in {
-            '/createOnlineAgentTask', '/registerOnlineAgent',
-            '/verifyOnlineAgentTask', '/cancelOnlineAgentTask',
-        }:
-            actor = self.need('agent-task:control')
-            if not actor:
-                return
-            data = self.body()
-            if path == '/createOnlineAgentTask':
-                return self.agent_task_control('POST', '/v1/tasks', data)
-            if path == '/registerOnlineAgent':
-                return self.agent_task_control('POST', '/v1/agents/register', data)
-            task_id = str(data.pop('task_id', '') or '')
-            if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{2,179}', task_id):
-                return self.sendj({'error': 'task_id_invalid'}, 400)
-            suffix = '/verify' if path == '/verifyOnlineAgentTask' else '/cancel'
-            return self.agent_task_control('POST', '/v1/tasks/' + task_id + suffix, data)
         if reject_non_read_http(self, 'POST', path): return
         data=self.body()
         try:

active-runtime/app/authorized_task_writer_provisioning.py

@@ -964,6 +964,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
     required = {
         "schema",
         "task_id",
+        "job_id",
         "repository",
         "branch_mode",
         "base_oid",
@@ -976,6 +977,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
         raise ProvisioningError("LEASE_REQUEST_V2_FIELDS_REJECTED", 409)
     request = WriterLeaseRequestV2(
         task_id=str(raw["task_id"]),
+        job_id=str(raw["job_id"]),
         repository=str(raw["repository"]),
         branch_mode=str(raw["branch_mode"]),
         base_oid=str(raw["base_oid"]),
@@ -988,6 +990,7 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
     return {
         "schema": WriterLeaseRequestV2.schema,
         "task_id": request.task_id,
+        "job_id": request.job_id,
         "repository": request.repository,
         "branch_mode": request.branch_mode,
         "base_oid": request.base_oid,

deploy/nginx/aios-agent-task-api.conf.example

@@ -1,46 +0,0 @@
-# External agent ingress. TLS termination is assumed on the enclosing server.
-# Control endpoints (/v1/agents/register, POST /v1/tasks, verify/cancel/events)
-# are intentionally NOT exposed here; the AIOS control plane reaches them on
-# loopback with its server-side credential.
-
-location = /agent-tasks/health {
-    rewrite ^/agent-tasks/(.*)$ /$1 break;
-    proxy_pass http://127.0.0.1:18940;
-    proxy_set_header Host $host;
-    proxy_set_header X-Forwarded-Proto $scheme;
-}
-
-location = /agent-tasks/v1/tasks/poll {
-    limit_except POST { deny all; }
-    rewrite ^/agent-tasks/(.*)$ /$1 break;
-    proxy_pass http://127.0.0.1:18940;
-    proxy_http_version 1.1;
-    proxy_set_header Authorization $http_authorization;
-    proxy_set_header Host $host;
-    proxy_set_header X-Forwarded-Proto $scheme;
-    client_max_body_size 1m;
-    proxy_read_timeout 35s;
-}
-
-location ~ ^/agent-tasks/v1/tasks/[A-Za-z0-9][A-Za-z0-9._:-]{2,179}/(claim|heartbeat|result)$ {
-    limit_except POST { deny all; }
-    rewrite ^/agent-tasks/(.*)$ /$1 break;
-    proxy_pass http://127.0.0.1:18940;
-    proxy_http_version 1.1;
-    proxy_set_header Authorization $http_authorization;
-    proxy_set_header Host $host;
-    proxy_set_header X-Forwarded-Proto $scheme;
-    client_max_body_size 1m;
-    proxy_read_timeout 35s;
-}
-
-location ~ ^/agent-tasks/v1/tasks/[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$ {
-    limit_except GET { deny all; }
-    rewrite ^/agent-tasks/(.*)$ /$1 break;
-    proxy_pass http://127.0.0.1:18940;
-    proxy_http_version 1.1;
-    proxy_set_header Authorization $http_authorization;
-    proxy_set_header Host $host;
-    proxy_set_header X-Forwarded-Proto $scheme;
-    proxy_read_timeout 35s;
-}

deploy/systemd/aios-agent-task-api.service

@@ -1,36 +0,0 @@
-[Unit]
-Description=AIOS Online Agent Task API
-After=network-online.target
-Wants=network-online.target
-
-[Service]
-Type=simple
-User=aios-agent-task
-Group=aios-agent-task
-WorkingDirectory=@RUNTIME_ROOT@
-Environment=PYTHONUNBUFFERED=1
-Environment=PYTHONDONTWRITEBYTECODE=1
-Environment=PYTHONPATH=@RUNTIME_ROOT@/app
-Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
-Environment=AIOS_AGENT_TASK_HOST=127.0.0.1
-Environment=AIOS_AGENT_TASK_PORT=18940
-Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=/run/credentials/aios-agent-task-api.service/control_token
-LoadCredential=control_token:/etc/aios-gitops/agent-task-control.token
-ExecStart=/usr/bin/python3 @RUNTIME_ROOT@/app/aios_agent_task_api.py
-Restart=on-failure
-RestartSec=2
-NoNewPrivileges=true
-PrivateTmp=true
-ProtectHome=true
-ProtectSystem=strict
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectControlGroups=true
-LockPersonality=true
-RestrictSUIDSGID=true
-RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
-ReadOnlyPaths=@RUNTIME_ROOT@
-ReadWritePaths=/var/lib/aios-agent-task
-
-[Install]
-WantedBy=multi-user.target

deploy/systemd/aios-gitops-model-api-v2r2c1-staging.service

@@ -28,9 +28,6 @@ Environment=AIOS_GITOPS_MODEL_API_SERVICE=aios-gitops-model-api-v2r2c1-staging.s
 Environment=AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false
 Environment=AIOS_SILENT_RELEASE_STATE_ROOT=/var/lib/aios-gitops/model-api-staging/silent-release
 Environment=AIOS_BRIDGE_V2_URL=http://127.0.0.1:18892
-Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
-Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=/run/credentials/aios-gitops-model-api-v2r2c1-staging.service/agent_task_control_token
-LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
 Environment=AIOS_GITOPS_HOST=127.0.0.1
 Environment=AIOS_GITOPS_PORT=18893
 Environment=PYTHONPATH=@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os

docs/gitops/openapi-agent-task-v1.yaml

@@ -1,295 +0,0 @@
-openapi: 3.1.0
-info:
-  title: AIOS Online Agent Task API
-  version: 1.0.0
-  description: >
-    Durable online task distribution for OpenClaw, Hermes, and other external
-    agents. Agent credentials and lease tokens are bearer credentials and must
-    never be committed to Git.
-servers:
-  - url: /agent-tasks
-paths:
-  /health:
-    get:
-      operationId: agentTaskHealth
-      responses:
-        "200":
-          description: Service health
-  /v1/agents/register:
-    post:
-      operationId: registerOnlineAgent
-      security:
-        - bearerAuth: []
-      description: Control-plane only. Returns the agent token once.
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/RegisterAgentRequest"
-      responses:
-        "201":
-          description: Registered
-  /v1/tasks:
-    post:
-      operationId: createOnlineAgentTask
-      security:
-        - bearerAuth: []
-      description: Control-plane only.
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/CreateTaskRequest"
-      responses:
-        "201":
-          description: Task created
-  /v1/tasks/poll:
-    post:
-      operationId: pollOnlineAgentTasks
-      security:
-        - bearerAuth: []
-      description: Agent operation. Returns capability-compatible queued tasks.
-      requestBody:
-        content:
-          application/json:
-            schema:
-              type: object
-              properties:
-                limit:
-                  type: integer
-                  minimum: 1
-                  maximum: 100
-                  default: 20
-      responses:
-        "200":
-          description: Available tasks
-  /v1/tasks/{task_id}:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    get:
-      operationId: getOnlineAgentTaskDirect
-      security:
-        - bearerAuth: []
-      responses:
-        "200":
-          description: Task status
-  /v1/tasks/{task_id}/claim:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    post:
-      operationId: claimOnlineAgentTask
-      security:
-        - bearerAuth: []
-      requestBody:
-        content:
-          application/json:
-            schema:
-              type: object
-              properties:
-                lease_seconds:
-                  type: integer
-                  minimum: 30
-                  maximum: 3600
-                  default: 300
-      responses:
-        "200":
-          description: >
-            Claim succeeded. The lease_token is returned once and must be kept
-            in runtime memory/secret storage, never Git.
-        "409":
-          description: Already claimed or otherwise not claimable
-  /v1/tasks/{task_id}/heartbeat:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    post:
-      operationId: heartbeatOnlineAgentTask
-      security:
-        - bearerAuth: []
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/HeartbeatRequest"
-      responses:
-        "200":
-          description: Lease renewed
-  /v1/tasks/{task_id}/result:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    post:
-      operationId: submitOnlineAgentResult
-      security:
-        - bearerAuth: []
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/ResultRequest"
-      responses:
-        "200":
-          description: Result accepted into VERIFYING
-  /v1/tasks/{task_id}/verify:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    post:
-      operationId: verifyOnlineAgentTaskDirect
-      security:
-        - bearerAuth: []
-      description: Control-plane only.
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/VerifyRequest"
-      responses:
-        "200":
-          description: Verification decision applied
-  /v1/tasks/{task_id}/cancel:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    post:
-      operationId: cancelOnlineAgentTaskDirect
-      security:
-        - bearerAuth: []
-      description: Control-plane only.
-      requestBody:
-        content:
-          application/json:
-            schema:
-              type: object
-              properties:
-                reason:
-                  type: string
-      responses:
-        "200":
-          description: Cancelled or already terminal
-  /v1/tasks/{task_id}/events:
-    parameters:
-      - $ref: "#/components/parameters/TaskId"
-    get:
-      operationId: getOnlineAgentTaskEventsDirect
-      security:
-        - bearerAuth: []
-      description: Control-plane only.
-      responses:
-        "200":
-          description: Audit event list
-components:
-  securitySchemes:
-    bearerAuth:
-      type: http
-      scheme: bearer
-  parameters:
-    TaskId:
-      name: task_id
-      in: path
-      required: true
-      schema:
-        type: string
-        pattern: "^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$"
-  schemas:
-    RegisterAgentRequest:
-      type: object
-      additionalProperties: false
-      required: [agent_id, agent_type, agent_version, capabilities]
-      properties:
-        agent_id:
-          type: string
-        agent_type:
-          type: string
-          examples: [openclaw]
-        agent_version:
-          type: string
-        capabilities:
-          type: array
-          items:
-            type: string
-        metadata:
-          type: object
-          additionalProperties: true
-    CreateTaskRequest:
-      type: object
-      additionalProperties: false
-      required:
-        - goal
-        - instructions
-        - success_criteria
-        - idempotency_key
-      properties:
-        task_id:
-          type: string
-        goal:
-          type: string
-        instructions:
-          type: string
-        success_criteria:
-          type: array
-          items: {}
-        required_capabilities:
-          type: array
-          items:
-            type: string
-        constraints:
-          type: object
-          additionalProperties: true
-        metadata:
-          type: object
-          additionalProperties: true
-        deadline:
-          type: string
-          format: date-time
-        priority:
-          type: integer
-          minimum: -100
-          maximum: 100
-          default: 0
-        max_attempts:
-          type: integer
-          minimum: 1
-          maximum: 20
-          default: 3
-        idempotency_key:
-          type: string
-    HeartbeatRequest:
-      type: object
-      additionalProperties: false
-      required: [lease_token]
-      properties:
-        lease_token:
-          type: string
-        lease_seconds:
-          type: integer
-          minimum: 30
-          maximum: 3600
-          default: 300
-        progress:
-          type: object
-          additionalProperties: true
-    ResultRequest:
-      type: object
-      additionalProperties: false
-      required: [lease_token, idempotency_key, result]
-      properties:
-        lease_token:
-          type: string
-        idempotency_key:
-          type: string
-        result:
-          type: object
-          additionalProperties: true
-    VerifyRequest:
-      type: object
-      additionalProperties: false
-      required: [decision]
-      properties:
-        decision:
-          type: string
-          enum: [SUCCEEDED, NEEDS_REWORK, FAILED]
-        verification:
-          type: object
-          additionalProperties: true

docs/gitops/openapi-chatgpt-governed-agent-task-extension.yaml

@@ -1,326 +0,0 @@
-openapi: 3.1.0
-info:
-  title: AIOS ChatGPT Governed Write and Agent Task Extension
-  version: 1.0.0
-  description: >
-    Merge these paths and schemas into docs/gitops/openapi-chatgpt-actions.yaml.
-    This extension deliberately exposes governed Bridge v2 operations rather
-    than reopening legacy AIOS_GITOPS_WRITES_ENABLED.
-paths:
-  /api/governed-write-capabilities:
-    get:
-      operationId: getGovernedWriteCapabilities
-      responses:
-        "200":
-          description: Governed and legacy write capability status
-  /api/control-plane-snapshot:
-    get:
-      operationId: getControlPlaneSnapshot
-      responses:
-        "200":
-          description: Non-sensitive control-plane snapshot
-  /createGovernedTask:
-    post:
-      operationId: createGovernedTask
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/CreateGovernedTaskRequest"
-      responses:
-        "200":
-          description: Governed task binding
-  /acquireWriterLease:
-    post:
-      operationId: acquireWriterLease
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/WriterLeaseRequest"
-      responses:
-        "200":
-          description: Writer lease binding
-  /commitFileGoverned:
-    post:
-      operationId: commitFileGoverned
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/CommitFileGovernedRequest"
-      responses:
-        "200":
-          description: Durable GitOps job receipt
-  /commitFilesGoverned:
-    post:
-      operationId: commitFilesGoverned
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/CommitFilesGovernedRequest"
-      responses:
-        "200":
-          description: Durable GitOps job receipt
-  /getGitOpsJob:
-    post:
-      operationId: getGitOpsJob
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required: [job_id]
-              properties:
-                job_id: {type: string}
-      responses:
-        "200":
-          description: Job receipt
-  /resumeGitOpsJob:
-    post:
-      operationId: resumeGitOpsJob
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required: [job_id, expected_revision]
-              properties:
-                job_id: {type: string}
-                expected_revision: {type: integer}
-      responses:
-        "200":
-          description: Resumed job receipt
-  /reconcileGitOpsJob:
-    post:
-      operationId: reconcileGitOpsJob
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required: [job_id, expected_revision]
-              properties:
-                job_id: {type: string}
-                expected_revision: {type: integer}
-      responses:
-        "200":
-          description: Reconciliation receipt
-  /createOnlineAgentTask:
-    post:
-      operationId: createOnlineAgentTask
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/CreateOnlineAgentTaskRequest"
-      responses:
-        "200":
-          description: Online task created
-        "201":
-          description: Online task created
-  /registerOnlineAgent:
-    post:
-      operationId: registerOnlineAgent
-      x-openai-isConsequential: true
-      description: >
-        Control operation. The returned agent token is secret material and must
-        be delivered through runtime secret storage, never Git.
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required: [agent_id, agent_type, agent_version, capabilities]
-              properties:
-                agent_id: {type: string}
-                agent_type: {type: string}
-                agent_version: {type: string}
-                capabilities:
-                  type: array
-                  items: {type: string}
-                metadata:
-                  type: object
-                  additionalProperties: true
-      responses:
-        "201":
-          description: Agent registered
-  /api/online-agent-task:
-    get:
-      operationId: getOnlineAgentTask
-      parameters:
-        - name: task_id
-          in: query
-          required: true
-          schema: {type: string}
-      responses:
-        "200":
-          description: Task status
-  /api/online-agent-task-events:
-    get:
-      operationId: getOnlineAgentTaskEvents
-      parameters:
-        - name: task_id
-          in: query
-          required: true
-          schema: {type: string}
-      responses:
-        "200":
-          description: Task events
-  /verifyOnlineAgentTask:
-    post:
-      operationId: verifyOnlineAgentTask
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required: [task_id, decision]
-              properties:
-                task_id: {type: string}
-                decision:
-                  type: string
-                  enum: [SUCCEEDED, NEEDS_REWORK, FAILED]
-                verification:
-                  type: object
-                  additionalProperties: true
-      responses:
-        "200":
-          description: Verification applied
-  /cancelOnlineAgentTask:
-    post:
-      operationId: cancelOnlineAgentTask
-      x-openai-isConsequential: true
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required: [task_id]
-              properties:
-                task_id: {type: string}
-                reason: {type: string}
-      responses:
-        "200":
-          description: Task cancelled
-components:
-  schemas:
-    Oid40:
-      type: string
-      pattern: "^[0-9a-f]{40}$"
-    CreateGovernedTaskRequest:
-      type: object
-      additionalProperties: false
-      required:
-        [task_id, repository, base_oid, target_branch, risk, description, idempotency_key]
-      properties:
-        task_id: {type: string}
-        repository: {type: string, const: aios/aios}
-        base_oid: {$ref: "#/components/schemas/Oid40"}
-        target_branch:
-          type: string
-          pattern: "^agent/"
-        risk:
-          type: string
-          enum: [low, medium, high, engineering]
-        description: {type: string}
-        idempotency_key: {type: string}
-        expected_branch_oid: {$ref: "#/components/schemas/Oid40"}
-        generation: {type: integer, minimum: 1}
-    WriterLeaseRequest:
-      type: object
-      additionalProperties: false
-      required:
-        [schema, task_id, repository, branch_mode, base_oid, expected_branch_oid,
-         target_branch, generation, lease_seconds]
-      properties:
-        schema: {type: string}
-        task_id: {type: string}
-        repository: {type: string, const: aios/aios}
-        branch_mode:
-          type: string
-          enum: [CREATE_NEW, CONTINUE_EXISTING]
-        base_oid: {$ref: "#/components/schemas/Oid40"}
-        expected_branch_oid: {$ref: "#/components/schemas/Oid40"}
-        target_branch: {type: string, pattern: "^agent/"}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-    CommitFileGovernedRequest:
-      type: object
-      additionalProperties: false
-      required:
-        [task_id, repository, expected_branch_oid, target_branch, generation,
-         idempotency_key, path, content_base64, message]
-      properties:
-        task_id: {type: string}
-        repository: {type: string, const: aios/aios}
-        expected_branch_oid: {$ref: "#/components/schemas/Oid40"}
-        target_branch: {type: string, pattern: "^agent/"}
-        generation: {type: integer, minimum: 1}
-        idempotency_key: {type: string}
-        path: {type: string}
-        content_base64: {type: string}
-        message: {type: string}
-    CommitFilesGovernedRequest:
-      type: object
-      additionalProperties: false
-      required:
-        [task_id, repository, expected_branch_oid, target_branch, generation,
-         idempotency_key, files, message]
-      properties:
-        task_id: {type: string}
-        repository: {type: string, const: aios/aios}
-        expected_branch_oid: {$ref: "#/components/schemas/Oid40"}
-        target_branch: {type: string, pattern: "^agent/"}
-        generation: {type: integer, minimum: 1}
-        idempotency_key: {type: string}
-        files:
-          type: object
-          additionalProperties:
-            type: string
-            description: Base64 encoded file bytes
-        message: {type: string}
-    CreateOnlineAgentTaskRequest:
-      type: object
-      additionalProperties: false
-      required: [goal, instructions, success_criteria, idempotency_key]
-      properties:
-        task_id: {type: string}
-        goal: {type: string}
-        instructions: {type: string}
-        success_criteria:
-          type: array
-          items: {}
-        required_capabilities:
-          type: array
-          items: {type: string}
-        constraints:
-          type: object
-          additionalProperties: true
-        metadata:
-          type: object
-          additionalProperties: true
-        deadline: {type: string, format: date-time}
-        priority: {type: integer, minimum: -100, maximum: 100}
-        max_attempts: {type: integer, minimum: 1, maximum: 20}
-        idempotency_key: {type: string}

runtime/agent-observability/records/exec_d75446452723.json

@@ -1,20 +0,0 @@
-{
-  "execution_id": "exec_d75446452723",
-  "user_id": "user1",
-  "visible_scope": "group1",
-  "entry_agent": "openclaw",
-  "path": [
-    "openclaw",
-    "hermes",
-    "codex"
-  ],
-  "used_hermes": true,
-  "closed_by": "codex",
-  "duration_ms": 0,
-  "input_tokens": 100,
-  "output_tokens": 200,
-  "total_tokens": 300,
-  "status": "completed",
-  "started_at": "2026-08-09T17:13:34.996470+00:00",
-  "ended_at": "2026-08-09T17:13:34.996489+00:00"
-}

runtime/task-center/tasks.jsonl

@@ -1,2 +0,0 @@
-{"task_id": "task-command_watcher-CMD-test", "title": "CMD-test", "assigned_by": "command-watcher", "executor": "openclaw", "source": "command_watcher", "status": "pending", "priority": "normal", "created_at": "2026-08-09T17:13:47.285193+00:00", "updated_at": "2026-08-09T17:13:47.285198+00:00", "progress": {"percent": 5, "current_step": "pending", "completed_steps": [], "remaining_steps": [], "blocked_reason": null}, "notification": {"required": true, "sent": false, "channel": null, "sent_at": null}, "evidence": ["ops/openclaw/CMD-test.md"], "metadata": {"bucket": "active"}}
-{"task_id": "task-agent_execution-e1", "title": "Agent execution e1", "assigned_by": "hermes", "executor": "hermes", "source": "agent_execution", "status": "completed", "priority": "normal", "created_at": "2026-08-09T17:13:47.285822+00:00", "updated_at": "2026-08-09T17:13:47.285826+00:00", "progress": {"percent": 100, "current_step": "completed", "completed_steps": [], "remaining_steps": [], "blocked_reason": null}, "notification": {"required": true, "sent": false, "channel": null, "sent_at": null}, "evidence": ["e1"], "metadata": {"execution_id": "e1", "entry_agent": "hermes", "status": "completed"}}

scripts/agent_task_client.py

@@ -1,112 +0,0 @@
-#!/usr/bin/env python3
-"""Small CLI used by OpenClaw/Hermes operators to validate AIOS online tasks."""
-from __future__ import annotations
-
-import argparse
-import json
-from pathlib import Path
-import sys
-from urllib.error import HTTPError, URLError
-from urllib.request import Request, urlopen
-
-
-def read_token(value: str | None, path: str | None) -> str:
-    if value:
-        return value.strip()
-    if path:
-        return Path(path).read_text(encoding="utf-8").strip()
-    raise SystemExit("token or token-file is required")
-
-
-def request(base: str, method: str, path: str, token: str, body=None):
-    payload = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
-    req = Request(
-        base.rstrip("/") + path,
-        data=payload,
-        method=method,
-        headers={
-            "Authorization": "Bearer " + token,
-            "Accept": "application/json",
-            **({"Content-Type": "application/json"} if payload is not None else {}),
-        },
-    )
-    try:
-        with urlopen(req, timeout=30) as response:
-            return response.status, json.loads(response.read().decode("utf-8"))
-    except HTTPError as exc:
-        raw = exc.read()
-        try:
-            value = json.loads(raw.decode("utf-8"))
-        except Exception:
-            value = {"error": "HTTP_" + str(exc.code)}
-        return exc.code, value
-    except (URLError, TimeoutError) as exc:
-        raise SystemExit("network_error:" + type(exc).__name__) from exc
-
-
-def main() -> int:
-    ap = argparse.ArgumentParser()
-    ap.add_argument("--base", required=True)
-    ap.add_argument("--token")
-    ap.add_argument("--token-file")
-    sub = ap.add_subparsers(dest="command", required=True)
-
-    p = sub.add_parser("poll")
-    p.add_argument("--limit", type=int, default=20)
-
-    p = sub.add_parser("claim")
-    p.add_argument("task_id")
-    p.add_argument("--lease-seconds", type=int, default=300)
-
-    p = sub.add_parser("heartbeat")
-    p.add_argument("task_id")
-    p.add_argument("--lease-token", required=True)
-    p.add_argument("--lease-seconds", type=int, default=300)
-    p.add_argument("--progress-json", default="{}")
-
-    p = sub.add_parser("result")
-    p.add_argument("task_id")
-    p.add_argument("--lease-token", required=True)
-    p.add_argument("--idempotency-key", required=True)
-    p.add_argument("--result-json", required=True)
-
-    p = sub.add_parser("status")
-    p.add_argument("task_id")
-
-    args = ap.parse_args()
-    token = read_token(args.token, args.token_file)
-
-    if args.command == "poll":
-        status, value = request(args.base, "POST", "/v1/tasks/poll", token, {"limit": args.limit})
-    elif args.command == "claim":
-        status, value = request(
-            args.base, "POST", f"/v1/tasks/{args.task_id}/claim", token,
-            {"lease_seconds": args.lease_seconds},
-        )
-    elif args.command == "heartbeat":
-        status, value = request(
-            args.base, "POST", f"/v1/tasks/{args.task_id}/heartbeat", token,
-            {
-                "lease_token": args.lease_token,
-                "lease_seconds": args.lease_seconds,
-                "progress": json.loads(args.progress_json),
-            },
-        )
-    elif args.command == "result":
-        status, value = request(
-            args.base, "POST", f"/v1/tasks/{args.task_id}/result", token,
-            {
-                "lease_token": args.lease_token,
-                "idempotency_key": args.idempotency_key,
-                "result": json.loads(args.result_json),
-            },
-        )
-    else:
-        status, value = request(args.base, "GET", f"/v1/tasks/{args.task_id}", token)
-
-    print(json.dumps({"http_status": status, "response": value}, ensure_ascii=False, indent=2))
-    return 0 if 200 <= status < 300 else 1
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/agent_task_online_smoke.py

@@ -1,193 +0,0 @@
-#!/usr/bin/env python3
-"""End-to-end online task feasibility smoke test.
-
-Runs through the public agent path plus control operations:
-create -> poll -> claim -> heartbeat -> result -> verify -> readback.
-It never performs Git, tag, Baseline, or production mutations.
-"""
-from __future__ import annotations
-
-import argparse
-from datetime import datetime, timezone
-import json
-from pathlib import Path
-import secrets
-from urllib.error import HTTPError, URLError
-from urllib.request import Request, urlopen
-
-
-def token(path: Path) -> str:
-    value = path.read_text(encoding="utf-8").strip()
-    if len(value) < 32:
-        raise SystemExit("credential_invalid:" + str(path))
-    return value
-
-
-def call(base: str, method: str, route: str, credential: str, body=None):
-    payload = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
-    headers = {
-        "Authorization": "Bearer " + credential,
-        "Accept": "application/json",
-    }
-    if payload is not None:
-        headers["Content-Type"] = "application/json"
-    request = Request(base.rstrip("/") + route, data=payload, headers=headers, method=method)
-    try:
-        with urlopen(request, timeout=30) as response:
-            return response.status, json.loads(response.read().decode("utf-8"))
-    except HTTPError as exc:
-        try:
-            value = json.loads(exc.read().decode("utf-8"))
-        except Exception:
-            value = {"error": "HTTP_" + str(exc.code)}
-        return exc.code, value
-    except (URLError, TimeoutError) as exc:
-        raise SystemExit("network_error:" + type(exc).__name__) from exc
-
-
-def require(step: str, status: int, value, expected_status: str | None = None):
-    if not 200 <= status < 300:
-        raise SystemExit(f"{step}_failed:http={status}:body={json.dumps(value, ensure_ascii=False)}")
-    if expected_status and value.get("status") != expected_status:
-        raise SystemExit(
-            f"{step}_state_failed:expected={expected_status}:"
-            f"observed={value.get('status')}"
-        )
-    print(json.dumps({"step": step, "http": status, "response": value}, ensure_ascii=False))
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--control-base", required=True,
-                        help="Loopback/model-control base that exposes /v1 control routes")
-    parser.add_argument("--agent-base", required=True,
-                        help="Public agent base, e.g. https://host/agent-tasks")
-    parser.add_argument("--control-token-file", required=True, type=Path)
-    parser.add_argument("--agent-token-file", required=True, type=Path)
-    args = parser.parse_args()
-
-    control = token(args.control_token_file)
-    agent = token(args.agent_token_file)
-    task_id = "TASK-ONLINE-SMOKE-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + secrets.token_hex(3)
-    create_key = "smoke-create:" + task_id
-    result_key = "smoke-result:" + task_id
-
-    s, v = call(
-        args.control_base, "POST", "/v1/tasks", control,
-        {
-            "task_id": task_id,
-            "goal": "Validate AIOS online task distribution end to end",
-            "instructions": (
-                "Do not mutate Git or production. Return a structured result "
-                "containing marker ONLINE_TASK_FLOW_OK."
-            ),
-            "success_criteria": [
-                {"field": "marker", "equals": "ONLINE_TASK_FLOW_OK"},
-                {"field": "agent_observed_task_id", "equals": task_id},
-            ],
-            "required_capabilities": ["online-task-smoke"],
-            "constraints": {
-                "git_mutation": False,
-                "production_mutation": False,
-                "network_scope": "AIOS task endpoint only",
-            },
-            "priority": 100,
-            "max_attempts": 3,
-            "idempotency_key": create_key,
-        },
-    )
-    require("create", s, v, "QUEUED")
-
-    s, v = call(args.agent_base, "POST", "/v1/tasks/poll", agent, {"limit": 100})
-    require("poll", s, v)
-    if task_id not in [item.get("task_id") for item in v.get("tasks", [])]:
-        raise SystemExit("poll_failed:created_task_not_visible")
-
-    s, v = call(
-        args.agent_base, "POST", f"/v1/tasks/{task_id}/claim", agent,
-        {"lease_seconds": 120},
-    )
-    require("claim", s, v, "CLAIMED")
-    lease = v.get("lease_token")
-    if not lease:
-        raise SystemExit("claim_failed:lease_token_missing")
-
-    s, v = call(
-        args.agent_base, "POST", f"/v1/tasks/{task_id}/heartbeat", agent,
-        {
-            "lease_token": lease,
-            "lease_seconds": 120,
-            "progress": {"stage": "executing", "percent": 50},
-        },
-    )
-    require("heartbeat", s, v, "RUNNING")
-
-    s, v = call(
-        args.agent_base, "POST", f"/v1/tasks/{task_id}/result", agent,
-        {
-            "lease_token": lease,
-            "idempotency_key": result_key,
-            "result": {
-                "status": "completed",
-                "marker": "ONLINE_TASK_FLOW_OK",
-                "agent_observed_task_id": task_id,
-                "summary": "Agent received, claimed, heartbeated and returned the online task.",
-                "artifacts": [],
-            },
-        },
-    )
-    require("result", s, v, "VERIFYING")
-
-    # Re-submit the identical result to prove idempotency.
-    s2, v2 = call(
-        args.agent_base, "POST", f"/v1/tasks/{task_id}/result", agent,
-        {
-            "lease_token": lease,
-            "idempotency_key": result_key,
-            "result": {
-                "status": "completed",
-                "marker": "ONLINE_TASK_FLOW_OK",
-                "agent_observed_task_id": task_id,
-                "summary": "Agent received, claimed, heartbeated and returned the online task.",
-                "artifacts": [],
-            },
-        },
-    )
-    require("result_idempotent_replay", s2, v2, "VERIFYING")
-    if v2.get("idempotent") is not True:
-        raise SystemExit("result_idempotency_failed")
-
-    s, v = call(
-        args.control_base, "POST", f"/v1/tasks/{task_id}/verify", control,
-        {
-            "decision": "SUCCEEDED",
-            "verification": {
-                "marker_match": "PASS",
-                "task_id_match": "PASS",
-                "online_round_trip": "PASS",
-            },
-        },
-    )
-    require("verify", s, v, "SUCCEEDED")
-
-    s, v = call(args.control_base, "GET", f"/v1/tasks/{task_id}", control)
-    require("final_control_readback", s, v, "SUCCEEDED")
-
-    # The executing Agent must also be able to observe the final terminal state.
-    s, v = call(args.agent_base, "GET", f"/v1/tasks/{task_id}", agent)
-    require("final_agent_readback", s, v, "SUCCEEDED")
-
-    print(json.dumps({
-        "status": "ONLINE_TASK_FLOW_VERIFIED",
-        "task_id": task_id,
-        "final_state": v.get("status"),
-        "force_push_used": False,
-        "tag_rewritten": False,
-        "direct_main_push": False,
-        "production_changed": False,
-    }, ensure_ascii=False, indent=2))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/build_agent_task_api.py

@@ -1,100 +0,0 @@
-#!/usr/bin/env python3
-"""Build deterministic immutable staging artifact for AIOS Agent Task API."""
-from __future__ import annotations
-
-import argparse
-import hashlib
-import json
-from pathlib import Path
-import tarfile
-import tempfile
-
-
-REQUIRED = (
-    "active-runtime/app/aios_agent_task_protocol.py",
-    "active-runtime/app/aios_agent_task_api.py",
-    "docs/gitops/openapi-agent-task-v1.yaml",
-)
-
-
-def sha256(path: Path) -> str:
-    digest = hashlib.sha256()
-    with path.open("rb") as source:
-        for chunk in iter(lambda: source.read(1024 * 1024), b""):
-            digest.update(chunk)
-    return digest.hexdigest()
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--source-root", required=True, type=Path)
-    parser.add_argument("--output-root", required=True, type=Path)
-    parser.add_argument("--source-oid", required=True)
-    args = parser.parse_args()
-
-    oid = args.source_oid.strip().lower()
-    if len(oid) != 40 or any(ch not in "0123456789abcdef" for ch in oid):
-        raise SystemExit("source oid must be exact 40 hex")
-    source = args.source_root.resolve(strict=True)
-    output = args.output_root.resolve()
-    for relative in REQUIRED:
-        if not (source / relative).is_file():
-            raise SystemExit("required source missing: " + relative)
-
-    output.mkdir(parents=True, exist_ok=True)
-    archive = output / f"aios-agent-task-api-{oid}.tar.gz"
-    manifest = {
-        "schema": "aios.agent-task-artifact.v1",
-        "source_oid": oid,
-        "repository_identity": "aios/aios",
-        "deployment_state": "STAGING",
-        "runtime_mode": "IMMUTABLE_STAGING",
-        "entrypoint": "app/aios_agent_task_api.py",
-        "listen": "127.0.0.1:18940",
-        "plaintext_credentials_in_artifact": False,
-        "included_paths": list(REQUIRED),
-    }
-
-    with tempfile.TemporaryDirectory(prefix="agent-task-build-", dir=str(output)) as temp:
-        stage = Path(temp) / "runtime"
-        for relative in REQUIRED:
-            src = source / relative
-            if relative.startswith("active-runtime/"):
-                dest = stage / Path(relative).relative_to("active-runtime")
-            else:
-                dest = stage / relative
-            dest.parent.mkdir(parents=True, exist_ok=True)
-            dest.write_bytes(src.read_bytes())
-        (stage / "manifest.json").write_text(
-            json.dumps(manifest, sort_keys=True, indent=2) + "\n",
-            encoding="utf-8",
-        )
-        with tarfile.open(archive, "w:gz", format=tarfile.PAX_FORMAT) as tar:
-            for path in sorted(stage.rglob("*")):
-                arcname = str(path.relative_to(stage.parent)).replace("\\", "/")
-                info = tar.gettarinfo(str(path), arcname=arcname)
-                info.mtime = 0
-                info.uid = info.gid = 0
-                info.uname = info.gname = "root"
-                if path.is_file():
-                    with path.open("rb") as fh:
-                        tar.addfile(info, fh)
-                else:
-                    tar.addfile(info)
-
-    receipt = {
-        **manifest,
-        "artifact_path": str(archive),
-        "artifact_sha256": sha256(archive),
-    }
-    receipt_path = archive.with_suffix(archive.suffix + ".receipt.json")
-    receipt_path.write_text(
-        json.dumps(receipt, sort_keys=True, indent=2) + "\n",
-        encoding="utf-8",
-    )
-    print(json.dumps(receipt, sort_keys=True, indent=2))
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

tests/test_agent_task_model_api_integration.py

@@ -1,48 +0,0 @@
-from pathlib import Path
-
-ROOT = Path(__file__).resolve().parents[1]
-
-
-def test_model_api_exposes_governed_writer_without_legacy_global_enable():
-    source = (ROOT / "active-runtime/app/aios_model_api_staging.py").read_text(encoding="utf-8")
-    assert "'/createGovernedTask': ('/v2/governed-task'" in source
-    assert "'/commitFilesGoverned': ('/v2/commit-files'" in source
-    assert "path == '/api/governed-write-capabilities'" in source
-    assert "AIOS_GITOPS_WRITES_ENABLED" not in source or "legacy_global_writes_enabled" in source
-    assert "'writes_enabled': False" in source
-    assert "bridge_v2_json" in source
-
-
-def test_model_api_routes_online_task_control_through_loopback_credential():
-    source = (ROOT / "active-runtime/app/aios_model_api_staging.py").read_text(encoding="utf-8")
-    assert "AIOS_AGENT_TASK_API_URL" in source
-    assert "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE" in source
-    assert "/createOnlineAgentTask" in source
-    assert "/verifyOnlineAgentTask" in source
-    assert "/cancelOnlineAgentTask" in source
-    assert "agent-task:control" in source
-
-
-def test_online_task_service_keeps_control_endpoints_off_public_nginx():
-    nginx = (ROOT / "deploy/nginx/aios-agent-task-api.conf.example").read_text(encoding="utf-8")
-    assert "/agent-tasks/v1/tasks/poll" in nginx
-    assert "(claim|heartbeat|result)" in nginx
-    assert "/v1/agents/register" not in nginx.replace(
-        "# Control endpoints (/v1/agents/register, POST /v1/tasks, verify/cancel/events)", ""
-    )
-
-
-def test_online_task_systemd_uses_runtime_credential_file():
-    unit = (ROOT / "deploy/systemd/aios-agent-task-api.service").read_text(encoding="utf-8")
-    assert "LoadCredential=control_token:" in unit
-    assert "/run/credentials/aios-agent-task-api.service/control_token" in unit
-    assert "ReadWritePaths=/var/lib/aios-agent-task" in unit
-    assert "127.0.0.1" in unit
-
-
-def test_control_plane_snapshot_is_local_aggregate_not_missing_bridge_route():
-    source = (ROOT / "active-runtime/app/aios_model_api_staging.py").read_text(encoding="utf-8")
-    assert "def control_plane_snapshot(self):" in source
-    assert "'snapshot_source': 'model-api-local-aggregate'" in source
-    assert "return self.control_plane_snapshot()" in source
-    assert "bridge_v2_json('/v2/control-plane-snapshot'" not in source

tests/test_agent_task_protocol.py

@@ -1,317 +0,0 @@
-from __future__ import annotations
-
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-import threading
-
-import pytest
-
-from aios_agent_task_protocol import TaskProtocolError, TaskStore
-
-
-class Clock:
-    def __init__(self) -> None:
-        self.value = datetime(2026, 8, 9, 12, 0, tzinfo=timezone.utc)
-
-    def __call__(self) -> datetime:
-        return self.value
-
-    def advance(self, seconds: int) -> None:
-        self.value += timedelta(seconds=seconds)
-
-
-@pytest.fixture
-def env(tmp_path: Path):
-    clock = Clock()
-    store = TaskStore(tmp_path / "tasks.sqlite3", now=clock, token_pepper="test-pepper")
-    a1 = store.register_agent(
-        agent_id="openclaw-1",
-        agent_type="openclaw",
-        agent_version="test",
-        capabilities=["shell", "git", "python"],
-    )
-    a2 = store.register_agent(
-        agent_id="hermes-1",
-        agent_type="hermes",
-        agent_version="test",
-        capabilities=["shell", "git", "python"],
-    )
-    return store, clock, a1, a2
-
-
-def create_task(store: TaskStore, task_id: str = "TASK-ONLINE-001"):
-    return store.create_task(
-        task_id=task_id,
-        goal="Create a deterministic validation artifact",
-        instructions="Write validation.txt containing ONLINE_TASK_OK.",
-        success_criteria=[
-            {"type": "file_exists", "path": "validation.txt"},
-            {"type": "contains", "path": "validation.txt", "value": "ONLINE_TASK_OK"},
-        ],
-        required_capabilities=["shell"],
-        constraints={"network": "optional"},
-        metadata={"suite": "online-task"},
-        priority=10,
-        max_attempts=3,
-        idempotency_key="create:" + task_id,
-    )
-
-
-def test_full_flow_requires_server_verification(env):
-    store, _clock, a1, _a2 = env
-    create_task(store)
-
-    polled = store.poll_tasks(agent_id=a1["agent_id"])
-    assert [item["task_id"] for item in polled] == ["TASK-ONLINE-001"]
-
-    claim = store.claim_task(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_seconds=300,
-    )
-    assert claim["status"] == "CLAIMED"
-    assert claim["lease_token"].startswith("lease_")
-
-    running = store.heartbeat(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        progress={"percent": 50},
-        lease_seconds=300,
-    )
-    assert running["status"] == "RUNNING"
-
-    submitted = store.submit_result(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        idempotency_key="result:1",
-        result={
-            "status": "completed",
-            "summary": "Created validation.txt",
-            "artifacts": [{"path": "validation.txt", "sha256": "0" * 64}],
-        },
-    )
-    assert submitted["status"] == "VERIFYING"
-    assert submitted["status"] != "SUCCEEDED"
-
-    verified = store.verify_task(
-        task_id="TASK-ONLINE-001",
-        decision="SUCCEEDED",
-        verification={"checks": {"content": "PASS"}},
-    )
-    assert verified["status"] == "SUCCEEDED"
-
-
-def test_atomic_claim_allows_only_one_agent(env):
-    store, _clock, a1, a2 = env
-    create_task(store)
-    outcomes: list[tuple[str, str]] = []
-    barrier = threading.Barrier(2)
-
-    def runner(agent_id: str) -> None:
-        barrier.wait()
-        try:
-            store.claim_task(
-                task_id="TASK-ONLINE-001",
-                agent_id=agent_id,
-                lease_seconds=300,
-            )
-            outcomes.append((agent_id, "CLAIMED"))
-        except TaskProtocolError as exc:
-            outcomes.append((agent_id, exc.code))
-
-    t1 = threading.Thread(target=runner, args=(a1["agent_id"],))
-    t2 = threading.Thread(target=runner, args=(a2["agent_id"],))
-    t1.start()
-    t2.start()
-    t1.join()
-    t2.join()
-
-    assert sum(result == "CLAIMED" for _, result in outcomes) == 1
-    assert sum(result in {"TASK_NOT_CLAIMABLE", "TASK_CLAIM_CONFLICT"} for _, result in outcomes) == 1
-
-
-def test_expired_lease_is_requeued_and_reclaimable(env):
-    store, clock, a1, a2 = env
-    create_task(store)
-    first = store.claim_task(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_seconds=30,
-    )
-    assert first["attempt"] == 1
-
-    clock.advance(31)
-    available = store.poll_tasks(agent_id=a2["agent_id"])
-    assert available[0]["status"] == "QUEUED"
-
-    second = store.claim_task(
-        task_id="TASK-ONLINE-001",
-        agent_id=a2["agent_id"],
-        lease_seconds=30,
-    )
-    assert second["attempt"] == 2
-    assert second["claimed_by"] == a2["agent_id"]
-
-
-def test_wrong_lease_token_is_rejected(env):
-    store, _clock, a1, _a2 = env
-    create_task(store)
-    store.claim_task(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_seconds=300,
-    )
-    with pytest.raises(TaskProtocolError, match="TASK_LEASE_TOKEN_INVALID"):
-        store.heartbeat(
-            task_id="TASK-ONLINE-001",
-            agent_id=a1["agent_id"],
-            lease_token="lease_wrong",
-            progress={},
-        )
-
-
-def test_result_submission_is_idempotent(env):
-    store, _clock, a1, _a2 = env
-    create_task(store)
-    claim = store.claim_task(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_seconds=300,
-    )
-    payload = {"status": "completed", "summary": "ok"}
-    first = store.submit_result(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        idempotency_key="result:same",
-        result=payload,
-    )
-    second = store.submit_result(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        idempotency_key="result:same",
-        result=payload,
-    )
-    assert first["status"] == "VERIFYING"
-    assert second["status"] == "VERIFYING"
-    assert second["idempotent"] is True
-
-    with pytest.raises(TaskProtocolError, match="IDEMPOTENCY_CONFLICT"):
-        store.submit_result(
-            task_id="TASK-ONLINE-001",
-            agent_id=a1["agent_id"],
-            lease_token=claim["lease_token"],
-            idempotency_key="result:same",
-            result={"status": "completed", "summary": "different"},
-        )
-
-
-def test_needs_rework_returns_to_queue(env):
-    store, _clock, a1, _a2 = env
-    create_task(store)
-    claim = store.claim_task(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_seconds=300,
-    )
-    store.submit_result(
-        task_id="TASK-ONLINE-001",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        idempotency_key="result:1",
-        result={"status": "completed"},
-    )
-    result = store.verify_task(
-        task_id="TASK-ONLINE-001",
-        decision="NEEDS_REWORK",
-        verification={"reason": "success criteria not met"},
-    )
-    assert result["status"] == "QUEUED"
-    assert result["claimed_by"] is None
-
-
-def test_capability_filter(env):
-    store, _clock, a1, _a2 = env
-    store.create_task(
-        task_id="TASK-GPU-001",
-        goal="GPU task",
-        instructions="Run GPU workload",
-        success_criteria=[],
-        required_capabilities=["gpu"],
-        idempotency_key="create:gpu",
-    )
-    assert store.poll_tasks(agent_id=a1["agent_id"]) == []
-
-
-def test_create_task_idempotency_conflict(env):
-    store, _clock, _a1, _a2 = env
-    first = create_task(store)
-    replay = create_task(store)
-    assert first["task_id"] == replay["task_id"]
-    assert replay["idempotent"] is True
-
-    with pytest.raises(TaskProtocolError, match="IDEMPOTENCY_CONFLICT"):
-        store.create_task(
-            task_id="TASK-ONLINE-002",
-            goal="different",
-            instructions="different",
-            success_criteria=[],
-            idempotency_key="create:TASK-ONLINE-001",
-        )
-
-
-def test_terminal_success_retains_executing_agent_for_final_readback(env):
-    store, _clock, a1, _a2 = env
-    create_task(store, "TASK-TERMINAL-READBACK")
-    claim = store.claim_task(
-        task_id="TASK-TERMINAL-READBACK",
-        agent_id=a1["agent_id"],
-        lease_seconds=120,
-    )
-    store.submit_result(
-        task_id="TASK-TERMINAL-READBACK",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        idempotency_key="terminal-result",
-        result={"status": "completed"},
-    )
-    final = store.verify_task(
-        task_id="TASK-TERMINAL-READBACK",
-        decision="SUCCEEDED",
-        verification={"criteria": "PASS"},
-    )
-    assert final["status"] == "SUCCEEDED"
-    assert final["claimed_by"] == a1["agent_id"]
-
-
-def test_needs_rework_releases_agent_ownership_for_reclaim(env):
-    store, _clock, a1, a2 = env
-    create_task(store, "TASK-REWORK-OWNERSHIP")
-    claim = store.claim_task(
-        task_id="TASK-REWORK-OWNERSHIP",
-        agent_id=a1["agent_id"],
-        lease_seconds=120,
-    )
-    store.submit_result(
-        task_id="TASK-REWORK-OWNERSHIP",
-        agent_id=a1["agent_id"],
-        lease_token=claim["lease_token"],
-        idempotency_key="rework-owner-result",
-        result={"status": "completed"},
-    )
-    requeued = store.verify_task(
-        task_id="TASK-REWORK-OWNERSHIP",
-        decision="NEEDS_REWORK",
-        verification={"criteria": "FAIL"},
-    )
-    assert requeued["status"] == "QUEUED"
-    assert requeued["claimed_by"] is None
-    reclaimed = store.claim_task(
-        task_id="TASK-REWORK-OWNERSHIP",
-        agent_id=a2["agent_id"],
-        lease_seconds=120,
-    )
-    assert reclaimed["claimed_by"] == a2["agent_id"]

tests/test_bridge_v2_governed_http_bootstrap.py

@@ -2,6 +2,7 @@ import base64
 from pathlib import Path
 
 from aios_authority_git_executor import AuthorityGitExecutor
+from aios_chatgpt_bridge_v2 import BridgeV2Adapter, OPERATION_REQUIRED_FIELDS
 from aios_chatgpt_bridge_v2_server import BridgeHandler, RuntimeConfig
 from aios_continuation_executor import GiteaContinuationRemote, ServerSideCredentialProjection
 from aios_continuation_lease import WriterLeaseRequestV2
@@ -26,6 +27,7 @@ class Probe:
 
 def test_create_new_lease_is_exact_base_and_absent():
     request = WriterLeaseRequestV2(
+        job_id="job-bridge-runtime-lineage-test",
         task_id="TASK-P0-BRIDGE-V21-DURABLE-RECONCILIATION-20260809",
         repository="aios/aios",
         branch_mode="CREATE_NEW",
@@ -39,6 +41,7 @@ def test_create_new_lease_is_exact_base_and_absent():
 
 def test_create_new_lease_rejects_existing_target():
     request = WriterLeaseRequestV2(
+        job_id="job-bridge-runtime-lineage-test",
         task_id="task",
         repository="aios/aios",
         branch_mode="CREATE_NEW",
@@ -75,7 +78,7 @@ def test_slash_branch_uses_paginated_collection():
 def test_http_route_contract_is_complete():
     expected = {
         "/v2/runtime", "/v2/capabilities", "/v2/governed-task",
-        "/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
+        "/v2/writer-lease", "/v2/writer-lease/bootstrap", "/v2/writer-lease/renew", "/v2/commit-file", "/v2/commit-files",
         "/v2/job", "/v2/job/resume", "/v2/job/reconcile",
         "/v2/tag/approval-request",
         "/v2/tag/publish",
@@ -88,3 +91,121 @@ def test_http_route_contract_is_complete():
     from aios_chatgpt_bridge_v2_server import ROUTES
 
     assert set(ROUTES) == expected
+
+
+def test_renew_writer_lease_is_exposed_in_action_and_capability_contracts():
+    adapter = BridgeV2Adapter(runtime_status={})
+    capabilities = adapter.getGovernedWriteCapabilities()
+    contract = capabilities["operation_contracts"]["renewWriterLease"]
+
+    assert contract["method"] == "POST"
+    assert contract["path"] == "/v2/writer-lease/renew"
+    assert tuple(contract["required_fields"]) == OPERATION_REQUIRED_FIELDS["renewWriterLease"]
+    assert "runtime_rehydrated" in contract["response_fields"]
+
+    action_schema = Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
+    assert "/v2/writer-lease/renew:" in action_schema
+    assert "operationId: renewWriterLease" in action_schema
+    assert "BridgeWriterLeaseRecoveryRequest" in action_schema
+    assert "BridgeWriterLeaseRecoveryResponse" in action_schema
+    assert {
+        capabilities["operation_contracts"][operation]["path"]
+        for operation in (
+            "acquireWriterLease",
+            "renewWriterLease",
+            "commitFileGoverned",
+            "commitFilesGoverned",
+        )
+    } == {
+        "/v2/writer-lease",
+        "/v2/writer-lease/renew",
+        "/v2/commit-file",
+        "/v2/commit-files",
+    }
+
+
+def test_bootstrap_writer_lease_is_exposed_in_action_and_capability_contracts():
+    adapter = BridgeV2Adapter(runtime_status={})
+    capabilities = adapter.getGovernedWriteCapabilities()
+    contract = capabilities["operation_contracts"]["bootstrapWriterLease"]
+
+    assert contract["method"] == "POST"
+    assert contract["path"] == "/v2/writer-lease/bootstrap"
+    assert tuple(contract["required_fields"]) == OPERATION_REQUIRED_FIELDS["bootstrapWriterLease"]
+    assert "bootstrapped" in contract["response_fields"]
+    assert "durable_branch_lease_present" in contract["response_fields"]
+
+    action_schema = Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
+    assert "/v2/writer-lease/bootstrap:" in action_schema
+    assert "operationId: bootstrapWriterLease" in action_schema
+    assert "bootstrapped: {type: boolean}" in action_schema
+    assert "durable_branch_lease_present: {type: boolean}" in action_schema
+
+
+
+def test_http_allows_explicit_job_id_on_lease_and_commit_requests():
+    handler = BridgeHandler.__new__(BridgeHandler)
+    lease_body = {
+        "schema": WriterLeaseRequestV2.schema,
+        "task_id": "task",
+        "job_id": "job-explicit-runtime-lineage",
+        "repository": "aios/aios",
+        "branch_mode": "CONTINUE_EXISTING",
+        "base_oid": "a" * 40,
+        "expected_branch_oid": "a" * 40,
+        "target_branch": "agent/explicit-runtime-lineage",
+        "generation": 2,
+        "lease_seconds": 120,
+    }
+    handler._allowed("acquireWriterLease", lease_body)
+    handler._allowed(
+        "commitFilesGoverned",
+        {
+            "task_id": "task",
+            "job_id": "job-explicit-runtime-lineage",
+            "repository": "aios/aios",
+            "expected_branch_oid": "a" * 40,
+            "target_branch": "agent/explicit-runtime-lineage",
+            "generation": 2,
+            "idempotency_key": "commit-explicit-runtime-lineage",
+            "files": {"docs/x.txt": "eA=="},
+            "message": "test",
+        },
+    )
+
+
+def test_http_allows_omitted_job_id_for_durable_server_resolution():
+    handler = BridgeHandler.__new__(BridgeHandler)
+    lease_body = {
+        "schema": WriterLeaseRequestV2.schema,
+        "task_id": "task",
+        "repository": "aios/aios",
+        "branch_mode": "CONTINUE_EXISTING",
+        "base_oid": "a" * 40,
+        "expected_branch_oid": "a" * 40,
+        "target_branch": "agent/implicit-runtime-lineage",
+        "generation": 2,
+        "lease_seconds": 120,
+    }
+    handler._allowed("acquireWriterLease", lease_body)
+    handler._allowed(
+        "commitFilesGoverned",
+        {
+            "task_id": "task",
+            "repository": "aios/aios",
+            "expected_branch_oid": "a" * 40,
+            "target_branch": "agent/implicit-runtime-lineage",
+            "generation": 2,
+            "idempotency_key": "commit-implicit-runtime-lineage",
+            "files": {"docs/x.txt": "eA=="},
+            "message": "test",
+        },
+    )
+
+
+def test_capability_contracts_return_exact_job_id_lineage():
+    contracts = BridgeV2Adapter(runtime_status={}).getGovernedWriteCapabilities()["operation_contracts"]
+    for operation in ("acquireWriterLease", "bootstrapWriterLease", "renewWriterLease"):
+        assert "job_id" in contracts[operation]["response_fields"]
+    for operation in ("commitFileGoverned", "commitFilesGoverned"):
+        assert "job_id" in contracts[operation]["response_fields"]

tests/test_continuation_runtime_normalization.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+from pathlib import Path
+import re
+import json
+from types import SimpleNamespace
+
+import pytest
+import yaml
+
+from aios_chatgpt_bridge_v2_server import GovernedRuntime
+from aios_continuation_executor import ContinuationExecutor
+from aios_continuation_lease import WriterLeaseRequestV2
+from aios_gitops_job_controller import JobState
+
+
+TASK = "continuation-normalization-test"
+JOB = "job-continuation-normalization-test"
+REPO = "aios/aios"
+BASE = "a" * 40
+BRANCH = "agent/continuation-normalization-test"
+
+
+class Authority:
+    def __init__(self, head: str | None) -> None:
+        self.head = head
+
+    def read_ref(self, ref: str) -> str | None:
+        assert ref == "refs/heads/" + BRANCH
+        return self.head
+
+
+def lease(**overrides: object) -> WriterLeaseRequestV2:
+    values = {
+        "task_id": TASK,
+        "job_id": JOB,
+        "repository": REPO,
+        "branch_mode": "CREATE_NEW",
+        "base_oid": BASE,
+        "expected_branch_oid": BASE,
+        "target_branch": BRANCH,
+        "generation": 1,
+        "lease_seconds": 120,
+    }
+    values.update(overrides)
+    return WriterLeaseRequestV2(**values)
+
+
+def runtime(*, head: str | None = BASE, **binding_overrides: object) -> GovernedRuntime:
+    value = GovernedRuntime.__new__(GovernedRuntime)
+    value.authority = Authority(head)
+    binding = {
+        "task_id": TASK,
+        "job_id": JOB,
+        "repository": REPO,
+        "base_oid": BASE,
+        "expected_branch_oid": BASE,
+        "target_branch": BRANCH,
+        "generation": 1,
+        "branch_mode": "CREATE_NEW",
+    }
+    binding.update(binding_overrides)
+    value.bindings = {TASK: binding}
+    return value
+
+
+def test_same_lineage_create_new_normalizes_to_continue_existing() -> None:
+    normalized = runtime()._normalize_runtime_lease(lease())
+    assert normalized.branch_mode == "CONTINUE_EXISTING"
+    assert normalized.task_id == TASK
+    assert normalized.job_id == JOB
+    assert normalized.repository == REPO
+    assert normalized.target_branch == BRANCH
+    assert normalized.expected_branch_oid == BASE
+    assert normalized.generation == 1
+
+
+def test_unprovisioned_create_new_stays_create_new() -> None:
+    assert runtime(head=None)._normalize_runtime_lease(lease()).branch_mode == "CREATE_NEW"
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    (
+        ("repository", "other/repo"),
+        ("base_oid", "b" * 40),
+        ("expected_branch_oid", "b" * 40),
+        ("target_branch", "agent/foreign"),
+        ("generation", 2),
+    ),
+)
+def test_binding_mismatch_fails_closed(field: str, value: object) -> None:
+    with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
+        runtime(**{field: value})._normalize_runtime_lease(lease())
+
+
+def test_drifted_existing_head_fails_closed() -> None:
+    with pytest.raises(RuntimeError, match="EXPECTED_BRANCH_OID_CONFLICT"):
+        runtime(head="b" * 40)._normalize_runtime_lease(lease())
+
+
+def test_openapi_contract_parses_and_requires_exact_commit_lineage() -> None:
+    document = yaml.safe_load(
+        Path("docs/gitops/openapi-chatgpt-actions.yaml").read_text(encoding="utf-8")
+    )
+    schema = document["components"]["schemas"]["BridgeCommitResponse"]
+    assert {
+        "job_id",
+        "commit_oid",
+        "commit_visible",
+        "branch_mode",
+        "target_branch",
+        "generation",
+    } <= set(schema["required"])
+
+
+def test_changed_production_files_pass_secret_scan() -> None:
+    paths = [
+        Path("active-runtime/app/aios_chatgpt_bridge_v2_server.py"),
+        Path("active-runtime/app/aios_continuation_executor.py"),
+        Path("active-runtime/app/aios_continuation_lease.py"),
+        Path("active-runtime/app/authorized_task_writer_provisioning.py"),
+        Path("active-runtime/app/aios_chatgpt_bridge_v2.py"),
+        Path("docs/gitops/openapi-chatgpt-actions.yaml"),
+    ]
+    expression = re.compile(
+        r"(?i)(BEGIN (RSA|OPENSSH|EC) PRIVATE KEY|"
+        r"authorization:\\s*bearer\\s+[A-Za-z0-9._-]{12,}|"
+        r"api[_-]?key\\s*[:=]\\s*['\\\"][^'\\\"]{12,})"
+    )
+    hits = []
+    for path in paths:
+        for number, line in enumerate(
+            path.read_text(encoding="utf-8", errors="replace").splitlines(), 1
+        ):
+            if expression.search(line):
+                hits.append(f"{path}:{number}")
+    assert hits == []
+
+
+def test_remote_verified_job_is_not_completed_when_head_cas_callback_fails() -> None:
+    class Store:
+        def read(self, _job_id):
+            return {
+                "branch": BRANCH,
+                "request_json": json.dumps(
+                    {"binding": {"expected_branch_oid": BASE}}
+                ),
+            }
+
+    class Controller:
+        def __init__(self):
+            self.store = Store()
+            self.states = []
+
+        def advance(self, _job_id, state, **_kwargs):
+            self.states.append(state)
+            return SimpleNamespace(as_dict=lambda: {"state": state.value})
+
+    executor = ContinuationExecutor.__new__(ContinuationExecutor)
+    executor.controller = Controller()
+    executor._observe_remote = lambda _branch, oid: (oid, oid, True)
+
+    def fail_cas(_oid):
+        raise RuntimeError("LEASE_CAS_CONFLICT")
+
+    with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
+        executor._finish_remote_verification(
+            JOB,
+            "b" * 40,
+            "b" * 40,
+            before_complete=fail_cas,
+        )
+
+    assert executor.controller.states == [JobState.REMOTE_VERIFIED]

tests/test_writer_lease_recovery.py

@@ -0,0 +1,708 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+import hashlib
+from dataclasses import asdict
+from pathlib import Path
+import sqlite3
+import json
+from types import SimpleNamespace
+import threading
+
+import pytest
+
+from aios_chatgpt_bridge_v2_server import GovernedRuntime
+from aios_continuation_executor import ContinuationPermissions, ContinuationRuntimeBinding
+from aios_continuation_lease import WriterLeaseRequestV2
+from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore
+from aios_governed_task_bootstrap import GovernedTaskBootstrap
+
+
+TASK_ID = "gov-writer-lease-recovery-test"
+BRANCH = "agent/writer-lease-recovery-test"
+BASE = "a" * 40
+JOB_ID_PREFIX = "writer-lease-runtime-lineage"
+
+
+class LocalAuthority:
+    def __init__(self, head: str | None = None) -> None:
+        self.head = head
+
+    def read_ref(self, _ref: str) -> str | None:
+        return self.head
+
+    def commit_exists(self, _oid: str) -> bool:
+        return True
+
+    def ancestry_check(self, _older: str, _newer: str) -> bool:
+        return True
+
+
+def writer_request(task_id: str = TASK_ID, **overrides: object) -> dict[str, object]:
+    request: dict[str, object] = {
+        "schema": WriterLeaseRequestV2.schema,
+        "task_id": task_id,
+        "repository": "aios/aios",
+        "branch_mode": "CREATE_NEW",
+        "base_oid": BASE,
+        "expected_branch_oid": BASE,
+        "target_branch": BRANCH,
+        "generation": 1,
+        "lease_seconds": 120,
+    }
+    request.update(overrides)
+    return request
+
+
+def make_runtime(tmp_path: Path, task_id: str = TASK_ID, *, authority_head: str | None = None) -> GovernedRuntime:
+    runtime = GovernedRuntime.__new__(GovernedRuntime)
+    runtime.config = SimpleNamespace(repository="aios/aios", state_root=tmp_path)
+    runtime.authority = LocalAuthority(authority_head)
+    runtime.bindings = {
+        task_id: {
+            "task_id": task_id,
+            "repository": "aios/aios",
+            "base_oid": BASE,
+            "expected_branch_oid": BASE,
+            "target_branch": BRANCH,
+            "branch_mode": "CREATE_NEW",
+            "generation": 1,
+        }
+    }
+    runtime.runtimes = {}
+    runtime.controllers = {}
+    runtime.bootstrap = GovernedTaskBootstrap(
+        runtime.authority,
+        tmp_path,
+        remote_name="test",
+        token_path=tmp_path / "token",
+        ssh_key_path=tmp_path / "key",
+    )
+    runtime.bindings = {}
+    runtime.create_task(
+        {
+            "task_id": task_id,
+            "repository": "aios/aios",
+            "base_oid": BASE,
+            "expected_branch_oid": BASE,
+            "target_branch": BRANCH,
+            "risk": "engineering",
+            "description": "governed runtime lineage test",
+            "idempotency_key": JOB_ID_PREFIX + "-" + task_id,
+        }
+    )
+    job_root = runtime._job_root(task_id)
+    runtime.controllers[task_id] = GitOpsJobController(
+        JobStore(job_root, payload_root=job_root / "payloads")
+    )
+
+    def build(
+        lease: WriterLeaseRequestV2,
+        *,
+        job_id: str,
+        lease_not_before: datetime | None = None,
+        lease_expires_at: datetime | None = None,
+    ) -> tuple[ContinuationRuntimeBinding, GitOpsJobController, object]:
+        controller = runtime.controllers.get(lease.task_id)
+        if controller is None:
+            root = runtime._job_root(lease.task_id)
+            controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
+        now = datetime.now(timezone.utc)
+        return (
+            ContinuationRuntimeBinding(
+                request=lease,
+                job_id=job_id,
+                workspace_path=tmp_path / "worktrees" / lease.task_id,
+                lease_not_before=lease_not_before or now - timedelta(seconds=1),
+                lease_expires_at=lease_expires_at or now + timedelta(seconds=lease.lease_seconds),
+                original_receipt_valid=True,
+                original_grant_consumed=True,
+                permissions=ContinuationPermissions(),
+            ),
+            controller,
+            object(),
+        )
+
+    runtime._build_runtime = build
+    return runtime
+
+
+def create_durable_lease(
+    runtime: GovernedRuntime,
+    *,
+    expires_at: str = "2020-01-01T00:00:00+00:00",
+) -> GitOpsJobController:
+    root = runtime._job_root(TASK_ID)
+    controller = GitOpsJobController(JobStore(root, payload_root=root / "payloads"))
+    receipt = SimpleNamespace(job_id=runtime.bindings[TASK_ID]["job_id"])
+    controller.lease_job(
+        receipt.job_id,
+        lease_id="writer-lease-recovery",
+        lineage_anchor_oid=BASE,
+        current_head_oid=BASE,
+        generation=1,
+        lease_expires_at=expires_at,
+        expected_revision=0,
+    )
+    runtime.controllers[TASK_ID] = controller
+    return controller
+
+
+def durable_row(runtime: GovernedRuntime) -> tuple[str, int]:
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        row = connection.execute(
+            "SELECT lease_expires_at, revision FROM branch_leases WHERE lease_id=?",
+            ("writer-lease-recovery",),
+        ).fetchone()
+    assert row is not None
+    return row
+
+
+def test_first_bootstrap_refreshes_only_runtime_without_durable_lease(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+
+    result = runtime.bootstrap_writer_lease(writer_request())
+
+    assert result["bootstrapped"] is True
+    assert result["runtime_rehydrated"] is False
+    assert result["durable_branch_lease_present"] is False
+    assert result["idempotent"] is False
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
+
+
+def test_duplicate_bootstrap_is_idempotent_without_consuming_generation(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+
+    first = runtime.bootstrap_writer_lease(writer_request())
+    second = runtime.bootstrap_writer_lease(writer_request())
+
+    assert first["bootstrapped"] is True
+    assert first["generation"] == 1
+    assert second["bootstrapped"] is False
+    assert second["idempotent"] is True
+    assert second["generation"] == 1
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
+
+
+def test_commit_after_bootstrap_reuses_the_exact_runtime_binding(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+
+    class RecordingWorkspace:
+
+        def __init__(self) -> None:
+            self.files: dict[str, bytes] = {}
+
+        def materialize_files(self, files: dict[str, bytes]) -> None:
+            self.files = dict(files)
+
+    class RecordingExecutor:
+
+        def __init__(self) -> None:
+            self.workspace = RecordingWorkspace()
+            self.binding = None
+            self.commit = None
+
+        def execute(self, binding: object, commit: object) -> dict[str, object]:
+            self.binding = binding
+            self.commit = commit
+            return {'task_id': binding.request.task_id, 'generation': binding.request.generation, 'files': sorted(self.workspace.files)}
+    executor = RecordingExecutor()
+    original_build = runtime._build_runtime
+
+    def build(lease: WriterLeaseRequestV2, *, job_id: str, lease_not_before: datetime | None=None, lease_expires_at: datetime | None=None) -> tuple[ContinuationRuntimeBinding, GitOpsJobController, object]:
+        binding, controller, _ = original_build(lease, job_id=job_id, lease_not_before=lease_not_before, lease_expires_at=lease_expires_at)
+        return (binding, controller, executor)
+    runtime._build_runtime = build
+    runtime.bootstrap_writer_lease(writer_request())
+    lease_response = runtime.acquire_lease(writer_request())
+    binding, controller, executor = runtime.runtimes[TASK_ID, 1]
+    assert lease_response['job_id'] == binding.job_id
+    assert lease_response['durable_branch_lease_present'] is True
+    _original_execute = executor.execute
+
+    def _execute_with_governed_commit_contract(*args, **kwargs):
+        _before_complete = kwargs.pop('before_complete', None)
+        _legacy = dict(_original_execute(*args, **kwargs))
+        _result = {**_legacy, 'job_id': binding.job_id, 'state': 'COMPLETED', 'side_effect_state': 'REMOTE_VERIFIED', 'receipt': {'authority_oid': 'b' * 40, 'git_remote_oid': 'b' * 40, 'api_branch_oid': 'b' * 40, 'commit_visible': True, 'remote_verified': True, 'force_push': False}}
+        if _before_complete is not None:
+            _before_complete('b' * 40)
+        return _result
+    executor.execute = _execute_with_governed_commit_contract
+    _bound_job_id = binding.job_id
+    result = runtime.commit_files({'task_id': TASK_ID, 'repository': 'aios/aios', 'generation': 1, 'target_branch': BRANCH, 'expected_branch_oid': BASE, 'idempotency_key': 'commit-after-bootstrap-v1', 'path': 'docs/recovery.txt', 'content_base64': 'YQ==', 'message': 'test commit after bootstrap'})
+    assert result['task_id'] == TASK_ID
+    assert result['generation'] == 1
+    assert result['files'] == ['docs/recovery.txt']
+    assert executor.binding.request.task_id == TASK_ID
+    assert executor.binding.request.generation == 1
+    database = runtime._job_root(TASK_ID) / 'state.sqlite3'
+    with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
+        assert connection.execute('SELECT COUNT(*) FROM branch_leases').fetchone()[0] == 1
+        assert connection.execute('PRAGMA foreign_key_check(branch_leases)').fetchall() == []
+    assert result['commit_oid'] == 'b' * 40
+    assert result['commit_visible'] is True
+    assert result['branch_mode'] == 'CONTINUE_EXISTING'
+    assert result['target_branch'] == BRANCH
+    assert int(result['generation']) == 1
+    with controller.store._connect() as connection:
+        _persisted = connection.execute('SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?', ('lease-' + str(_bound_job_id),)).fetchone()
+    assert _persisted is not None
+    assert _persisted[0] == 'b' * 40
+    assert int(_persisted[1]) == 1
+    fresh_runtime = make_runtime(tmp_path, authority_head='b' * 40)
+    fresh_response = fresh_runtime.bootstrap_writer_lease(writer_request(branch_mode='CONTINUE_EXISTING', expected_branch_oid='b' * 40))
+    fresh_binding, _, _ = fresh_runtime.runtimes[TASK_ID, 1]
+    assert fresh_response['job_id'] == str(_bound_job_id)
+    assert fresh_response['runtime_rehydrated'] is True
+    assert fresh_binding.job_id == str(_bound_job_id)
+    assert fresh_binding.request.expected_branch_oid == 'b' * 40
+    with fresh_runtime.controllers[TASK_ID].store._connect() as connection:
+        _fresh_lease = connection.execute('SELECT job_id,current_head_oid,revision FROM branch_leases WHERE lease_id=?', ('lease-' + str(_bound_job_id),)).fetchone()
+    assert tuple(_fresh_lease) == (str(_bound_job_id), 'b' * 40, 1)
+
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "error"),
+    (
+        ("repository", "other/repository", "LEASE_BINDING_CONFLICT"),
+        ("base_oid", "b" * 40, "LEASE_BINDING_CONFLICT"),
+        ("expected_branch_oid", "b" * 40, "LEASE_BINDING_CONFLICT"),
+        ("target_branch", "agent/wrong-target", "LEASE_BINDING_CONFLICT"),
+        ("generation", 2, "GENERATION_BINDING_CONFLICT"),
+    ),
+)
+def test_bootstrap_rejects_any_nonmatching_binding_field(
+    tmp_path: Path, field: str, value: object, error: str
+) -> None:
+    runtime = make_runtime(tmp_path)
+
+    with pytest.raises(RuntimeError, match=error):
+        runtime.bootstrap_writer_lease(writer_request(**{field: value}))
+
+
+def test_bootstrap_rejects_unknown_task_ownership(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+
+    with pytest.raises(RuntimeError, match="TASK_OWNERSHIP_INVALID"):
+        runtime.bootstrap_writer_lease(writer_request(task_id="unknown-task"))
+
+
+def test_renew_missing_durable_lease_is_not_bootstrap(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+
+    with pytest.raises(RuntimeError, match="LEASE_NOT_FOUND"):
+        runtime.renew_writer_lease(writer_request())
+
+
+def test_renew_is_generation_scoped_before_durable_lookup(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    create_durable_lease(runtime)
+
+    with pytest.raises(RuntimeError, match="GENERATION_BINDING_CONFLICT"):
+        runtime.renew_writer_lease(writer_request(generation=2))
+    assert durable_row(runtime) == ("2020-01-01T00:00:00+00:00", 0)
+
+
+def test_restart_rehydrates_only_from_matching_durable_lease(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    create_durable_lease(runtime)
+    runtime.runtimes.clear()
+
+    result = runtime.renew_writer_lease(writer_request())
+
+    expires_at, revision = durable_row(runtime)
+    assert result["recovered"] is True
+    assert result["runtime_rehydrated"] is True
+    assert result["durable_branch_lease_present"] is True
+    assert revision == 1
+    assert datetime.fromisoformat(expires_at) > datetime.now(timezone.utc)
+
+
+def test_duplicate_renewal_is_idempotent_after_first_durable_recovery(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    create_durable_lease(runtime)
+
+    first = runtime.renew_writer_lease(writer_request())
+    second = runtime.renew_writer_lease(writer_request())
+
+    assert first["recovered"] is True
+    assert first["idempotent"] is False
+    assert second["recovered"] is False
+    assert second["idempotent"] is True
+    assert durable_row(runtime)[1] == 1
+
+
+def test_update_lease_keeps_expiry_without_optional_argument_and_keeps_cas(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    controller = create_durable_lease(runtime)
+
+    controller.update_lease(
+        "writer-lease-recovery",
+        expected_revision=0,
+        current_head_oid=BASE,
+        generation=1,
+    )
+    assert durable_row(runtime) == ("2020-01-01T00:00:00+00:00", 1)
+    with pytest.raises(JobError, match="LEASE_REVISION_CAS_CONFLICT"):
+        controller.update_lease(
+            "writer-lease-recovery",
+            expected_revision=0,
+            current_head_oid=BASE,
+            generation=1,
+        )
+
+
+def test_two_runtime_connections_resolve_renewal_cas_race_idempotently(tmp_path: Path) -> None:
+    first = make_runtime(tmp_path)
+    first_controller = create_durable_lease(first)
+    second = make_runtime(tmp_path)
+    second_controller = GitOpsJobController(
+        JobStore(second._job_root(TASK_ID), payload_root=second._job_root(TASK_ID) / "payloads")
+    )
+    second.controllers[TASK_ID] = second_controller
+    gate = threading.Barrier(2)
+
+    def gate_update(controller: GitOpsJobController) -> None:
+        original = controller.update_lease
+
+        def update(*args: object, **kwargs: object) -> None:
+            gate.wait(timeout=5)
+            original(*args, **kwargs)
+
+        controller.update_lease = update
+
+    gate_update(first_controller)
+    gate_update(second_controller)
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(pool.map(lambda runtime: runtime.renew_writer_lease(writer_request()), (first, second)))
+
+    assert sorted((item["recovered"], item["idempotent"]) for item in results) == [
+        (False, True),
+        (True, False),
+    ]
+    assert durable_row(first)[1] == 1
+
+
+def test_bootstrap_preserves_exact_governed_job_lineage(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
+
+    response = runtime.bootstrap_writer_lease(writer_request())
+    binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
+
+    assert response["job_id"] == expected_job_id
+    assert binding.job_id == expected_job_id
+    assert binding.request.job_id == expected_job_id
+
+
+def test_restart_rehydrate_preserves_exact_governed_job_lineage(tmp_path: Path) -> None:
+    first = make_runtime(tmp_path)
+    expected_job_id = first.bindings[TASK_ID]["job_id"]
+    first.acquire_lease(writer_request())
+
+    runtime = make_runtime(tmp_path)
+    assert runtime is not first
+    assert runtime.runtimes == {}
+    response = runtime.bootstrap_writer_lease(writer_request())
+    binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
+
+    assert response["job_id"] == expected_job_id
+    assert binding.job_id == expected_job_id
+    assert binding.request.job_id == expected_job_id
+
+
+def test_wrong_governed_job_id_is_denied_before_runtime_or_lease_side_effect(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+
+    with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
+        runtime.bootstrap_writer_lease(writer_request(job_id="job-wrong-lineage"))
+
+    assert runtime.runtimes == {}
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0] == 0
+        assert connection.execute("PRAGMA foreign_key_check(branch_leases)").fetchall() == []
+
+
+def test_acquire_creates_exact_governed_writer_lease_with_fk_and_cas(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
+
+    response = runtime.acquire_lease(writer_request())
+
+    assert response["job_id"] == expected_job_id
+    assert response["durable_branch_lease_present"] is True
+    assert response["idempotent"] is False
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        foreign_keys = connection.execute(
+            "PRAGMA foreign_key_list(branch_leases)"
+        ).fetchall()
+        row = connection.execute(
+            "SELECT job_id,branch,lineage_anchor_oid,current_head_oid,generation,state,revision "
+            "FROM branch_leases"
+        ).fetchone()
+        job = connection.execute(
+            "SELECT state,side_effect_state,revision FROM jobs WHERE job_id=?",
+            (expected_job_id,),
+        ).fetchone()
+        assert connection.execute("PRAGMA foreign_key_check(branch_leases)").fetchall() == []
+    assert any(
+        fk[2] == "jobs" and fk[3] == "job_id" and fk[4] == "job_id"
+        for fk in foreign_keys
+    )
+    assert row == (expected_job_id, BRANCH, BASE, BASE, 1, "ACTIVE", 0)
+    assert job == ("LEASED", "LEASED", 2)
+
+
+def test_duplicate_acquire_observes_same_durable_writer_lease(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
+
+    first = runtime.acquire_lease(writer_request())
+    second = runtime.acquire_lease(writer_request())
+
+    assert first["job_id"] == expected_job_id
+    assert second["job_id"] == expected_job_id
+    assert second["idempotent"] is True
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        leases = connection.execute(
+            "SELECT lease_id,job_id,revision FROM branch_leases"
+        ).fetchall()
+        job = connection.execute(
+            "SELECT state,revision FROM jobs WHERE job_id=?",
+            (expected_job_id,),
+        ).fetchone()
+    assert leases == [("lease-" + expected_job_id, expected_job_id, 0)]
+    assert job == ("LEASED", 2)
+
+
+def test_wrong_job_acquire_is_denied_before_authorize_or_lease(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
+
+    with pytest.raises(RuntimeError, match="LEASE_BINDING_CONFLICT"):
+        runtime.acquire_lease(writer_request(job_id="job-wrong-lineage"))
+
+    database = runtime._job_root(TASK_ID) / "state.sqlite3"
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True) as connection:
+        job = connection.execute(
+            "SELECT state,side_effect_state,revision FROM jobs WHERE job_id=?",
+            (expected_job_id,),
+        ).fetchone()
+        lease_count = connection.execute("SELECT COUNT(*) FROM branch_leases").fetchone()[0]
+    assert job == ("REQUESTED", "NONE", 0)
+    assert lease_count == 0
+
+
+def test_omitted_request_job_id_resolves_the_exact_durable_parent(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
+    raw = dict(writer_request())
+
+    response = runtime.acquire_lease(raw)
+    binding, _, _ = runtime.runtimes[(TASK_ID, 1)]
+
+    assert response["job_id"] == expected_job_id
+    assert binding.job_id == expected_job_id
+    assert binding.request.job_id == expected_job_id
+
+
+def test_durable_job_expected_head_mismatch_is_denied(tmp_path: Path) -> None:
+    runtime = make_runtime(tmp_path)
+    expected_job_id = runtime.bindings[TASK_ID]["job_id"]
+    controller = runtime.controllers[TASK_ID]
+    with controller.store._connect() as connection:
+        row = connection.execute(
+            "SELECT request_json FROM jobs WHERE job_id=?", (expected_job_id,)
+        ).fetchone()
+        request = json.loads(row[0])
+        request["expected_branch_oid"] = "c" * 40
+        request["binding"]["expected_branch_oid"] = "c" * 40
+        connection.execute(
+            "UPDATE jobs SET request_json=? WHERE job_id=?",
+            (json.dumps(request, sort_keys=True, separators=(",", ":")), expected_job_id),
+        )
+        connection.commit()
+
+    with pytest.raises(RuntimeError, match="RUNTIME_GOVERNED_JOB_UNRESOLVED"):
+        runtime.bootstrap_writer_lease(writer_request())
+
+    assert runtime.runtimes == {}
+
+
+def _commit_ready_runtime(tmp_path: Path):
+    runtime = make_runtime(tmp_path)
+    runtime.bootstrap_writer_lease(writer_request())
+    binding, controller, _ = runtime.runtimes[(TASK_ID, 1)]
+
+    class Workspace:
+        def materialize_files(self, _files):
+            return None
+
+    class Executor:
+        def __init__(self):
+            self.workspace = Workspace()
+            self.calls = 0
+
+        def execute(self, _binding, _commit, *, before_complete=None):
+            self.calls += 1
+            result = {
+                "job_id": binding.job_id,
+                "state": "COMPLETED",
+                "side_effect_state": "REMOTE_VERIFIED",
+                "receipt": {
+                    "authority_oid": "b" * 40,
+                    "git_remote_oid": "b" * 40,
+                    "api_branch_oid": "b" * 40,
+                    "commit_visible": True,
+                    "remote_verified": True,
+                    "force_push": False,
+                },
+            }
+            if before_complete is not None:
+                before_complete("b" * 40)
+            return result
+
+    executor = Executor()
+    runtime.runtimes[(TASK_ID, 1)] = (binding, controller, executor)
+    runtime.acquire_lease(writer_request())
+    return runtime, binding, controller, executor
+
+
+def _commit_request():
+    return {
+        "task_id": TASK_ID,
+        "repository": "aios/aios",
+        "generation": 1,
+        "target_branch": BRANCH,
+        "expected_branch_oid": BASE,
+        "idempotency_key": "commit-cas-test-v1",
+        "path": "docs/cas.txt",
+        "content_base64": "YQ==",
+        "message": "test commit cas",
+    }
+
+
+def test_commit_lease_revision_cas_conflict_is_preserved(tmp_path: Path) -> None:
+    runtime, _binding, controller, _executor = _commit_ready_runtime(tmp_path)
+
+    def conflict(*_args, **_kwargs):
+        raise JobError("LEASE_REVISION_CAS_CONFLICT")
+
+    controller.update_lease = conflict
+    with pytest.raises(RuntimeError, match="LEASE_CAS_CONFLICT"):
+        runtime.commit_files(_commit_request())
+
+
+def test_commit_divergent_durable_head_fails_before_executor(tmp_path: Path) -> None:
+    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
+    controller.update_lease(
+        "lease-" + binding.job_id,
+        expected_revision=0,
+        current_head_oid="c" * 40,
+        generation=1,
+    )
+
+    with pytest.raises(RuntimeError, match="LEASE_HEAD_RECONCILIATION_REQUIRED"):
+        runtime.commit_files(_commit_request())
+    assert executor.calls == 0
+
+
+def test_remote_pending_never_advances_durable_head(tmp_path: Path) -> None:
+    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
+
+    def pending(_binding, _commit, **_kwargs):
+        executor.calls += 1
+        return {
+            "job_id": binding.job_id,
+            "state": "RECONCILIATION_REQUIRED",
+            "side_effect_state": "REMOTE_PUSH_PENDING",
+            "receipt": {
+                "authority_oid": "b" * 40,
+                "desired_remote_oid": "b" * 40,
+                "remote_verified": False,
+            },
+        }
+
+    executor.execute = pending
+    with pytest.raises(RuntimeError, match="RECONCILIATION_REQUIRED"):
+        runtime.commit_files(_commit_request())
+    with controller.store._connect() as connection:
+        row = connection.execute(
+            "SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
+            ("lease-" + binding.job_id,),
+        ).fetchone()
+    assert tuple(row) == (BASE, 0)
+
+
+def test_retry_accepts_exact_head_already_advanced_at_remote_verified_boundary(tmp_path: Path) -> None:
+    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
+    commit_oid = "b" * 40
+    controller.update_lease(
+        "lease-" + binding.job_id,
+        expected_revision=0,
+        current_head_oid=commit_oid,
+        generation=1,
+    )
+    receipt = {
+        "authority_oid": commit_oid,
+        "git_remote_oid": commit_oid,
+        "api_branch_oid": commit_oid,
+        "commit_visible": True,
+        "remote_verified": True,
+        "force_push": False,
+    }
+    with controller.store._connect() as connection:
+        connection.execute(
+            "UPDATE jobs SET state='REMOTE_VERIFIED',side_effect_state='REMOTE_VERIFIED',"
+            "receipt_json=? WHERE job_id=?",
+            (json.dumps(receipt, sort_keys=True, separators=(",", ":")), binding.job_id),
+        )
+        connection.commit()
+
+    result = runtime.commit_files(_commit_request())
+
+    assert result["commit_oid"] == commit_oid
+    assert executor.calls == 1
+    with controller.store._connect() as connection:
+        row = connection.execute(
+            "SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
+            ("lease-" + binding.job_id,),
+        ).fetchone()
+    assert tuple(row) == (commit_oid, 1)
+
+
+def test_completed_response_with_invalid_oid_uses_canonical_oid_error(tmp_path: Path) -> None:
+    runtime, binding, controller, executor = _commit_ready_runtime(tmp_path)
+
+    def invalid_oid(_binding, _commit, **_kwargs):
+        executor.calls += 1
+        return {
+            "job_id": binding.job_id,
+            "state": "COMPLETED",
+            "side_effect_state": "REMOTE_VERIFIED",
+            "receipt": {
+                "authority_oid": "not-an-oid",
+                "commit_visible": True,
+                "remote_verified": True,
+            },
+        }
+
+    executor.execute = invalid_oid
+    with pytest.raises(RuntimeError, match="GOVERNED_COMMIT_OID_INVALID"):
+        runtime.commit_files(_commit_request())
+    with controller.store._connect() as connection:
+        row = connection.execute(
+            "SELECT current_head_oid,revision FROM branch_leases WHERE lease_id=?",
+            ("lease-" + binding.job_id,),
+        ).fetchone()
+    assert tuple(row) == (BASE, 0)