fix: restore governed capability and exact branch contracts

active-runtime/app/aios_agent_task_api.py

@@ -0,0 +1,315 @@
+#!/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

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

@@ -0,0 +1,20 @@
+"""Shared exact branch-name validation for governed Git operations."""
+
+from __future__ import annotations
+
+import re
+
+
+# Keep the existing governed branch envelope. Validation rejects invalid input;
+# it never shortens or normalizes a caller's branch string.
+BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+
+
+def is_valid_governed_branch(value: object) -> bool:
+    return isinstance(value, str) and BRANCH_RE.fullmatch(value) is not None
+
+
+def require_exact_governed_branch(value: object, *, field: str = "target_branch") -> str:
+    if not is_valid_governed_branch(value):
+        raise ValueError(f"{field}_invalid")
+    return value

active-runtime/app/aios_capability_contract.py

@@ -0,0 +1,159 @@
+"""Canonical fail-closed capability semantics for task eligibility."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+
+_TASK_ALIASES: dict[str, frozenset[str]] = {
+    "repository_read": frozenset({"repo.read"}),
+    "git.read": frozenset({"repo.read"}),
+    "repo.read": frozenset({"repo.read"}),
+    "test_execution": frozenset({"test.execute"}),
+    "git.test": frozenset({"test.execute"}),
+    "test.execute": frozenset({"test.execute"}),
+    "governed_git_write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
+    "governed.git.write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
+    "governed.write.branch": frozenset({"governed.git.write.branch"}),
+    "governed.write.file": frozenset({"governed.git.write.file"}),
+    "shell": frozenset({"runtime.shell"}),
+    "git": frozenset({"runtime.git"}),
+    "python": frozenset({"runtime.python"}),
+    "code.read": frozenset({"legacy.code.read"}),
+    "code.review": frozenset({"legacy.code.review"}),
+    "code.write": frozenset({"legacy.code.write"}),
+    "code.write.patch": frozenset({"legacy.code.write.patch"}),
+    "test.python": frozenset({"legacy.test.python"}),
+    "test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
+    "git.commit": frozenset({"legacy.git.commit"}),
+    "git.diff.read": frozenset({"legacy.git.diff.read"}),
+    "git.force_push": frozenset({"legacy.git.force_push"}),
+    "deploy.production": frozenset({"legacy.deploy.production"}),
+    "secret.transform": frozenset({"legacy.secret.transform"}),
+    "secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
+    "runtime.shadow": frozenset({"legacy.runtime.shadow"}),
+    "review.security": frozenset({"legacy.review.security"}),
+    "review.api-contract": frozenset({"legacy.review.api-contract"}),
+    "gpu.train": frozenset({"legacy.gpu.train"}),
+}
+
+_WORKER_ALIASES: dict[str, frozenset[str]] = {
+    "git.read": frozenset({"repo.read"}),
+    "repo.read": frozenset({"repo.read"}),
+    "repository_read": frozenset({"repo.read"}),
+    "git.test": frozenset({"test.execute"}),
+    "test.execute": frozenset({"test.execute"}),
+    "test_execution": frozenset({"test.execute"}),
+    "governed.write.branch": frozenset({"governed.git.write.branch"}),
+    "governed.git.write.branch": frozenset({"governed.git.write.branch"}),
+    "governed.write.file": frozenset({"governed.git.write.file"}),
+    "governed.git.write.file": frozenset({"governed.git.write.file"}),
+    "shell": frozenset({"runtime.shell"}),
+    "git": frozenset({"runtime.git"}),
+    "python": frozenset({"runtime.python"}),
+    "code.read": frozenset({"legacy.code.read"}),
+    "code.review": frozenset({"legacy.code.review"}),
+    "code.write": frozenset({"legacy.code.write"}),
+    "code.write.patch": frozenset({"legacy.code.write.patch"}),
+    "test.python": frozenset({"legacy.test.python"}),
+    "test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
+    "git.commit": frozenset({"legacy.git.commit"}),
+    "git.diff.read": frozenset({"legacy.git.diff.read"}),
+    "git.force_push": frozenset({"legacy.git.force_push"}),
+    "deploy.production": frozenset({"legacy.deploy.production"}),
+    "secret.transform": frozenset({"legacy.secret.transform"}),
+    "secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
+    "runtime.shadow": frozenset({"legacy.runtime.shadow"}),
+    "review.security": frozenset({"legacy.review.security"}),
+    "review.api-contract": frozenset({"legacy.review.api-contract"}),
+    "gpu.train": frozenset({"legacy.gpu.train"}),
+}
+
+_UNKNOWN_PREFIX = "exact:"
+
+
+def _clean(value: object) -> str:
+    if not isinstance(value, str):
+        raise ValueError("capability must be a string")
+    text = value.strip().lower()
+    if not text:
+        raise ValueError("capability must not be empty")
+    return text
+
+
+def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
+    """Map task-facing names to one semantic vocabulary.
+
+    Unknown names remain explicit exact requirements. They are never treated
+    as wildcards or granted by a failed normalization.
+    """
+
+    result: set[str] = set()
+    for value in values:
+        text = _clean(value)
+        if "*" in text:
+            result.add(_UNKNOWN_PREFIX + text)
+        else:
+            result.update(_TASK_ALIASES.get(text, frozenset({_UNKNOWN_PREFIX + text})))
+    return frozenset(result)
+
+
+def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
+    """Map registered worker names to effective semantic capabilities."""
+
+    result: set[str] = set()
+    for value in values:
+        text = _clean(value)
+        if "*" in text:
+            continue
+        mapped = _WORKER_ALIASES.get(text)
+        if mapped is not None:
+            result.update(mapped)
+    if {
+        "governed.git.write.branch",
+        "governed.git.write.file",
+    }.issubset(result):
+        result.add("governed.git.write")
+    return frozenset(result)
+
+
+def has_capabilities(
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+) -> bool:
+    """Return the single eligibility decision used by poll and claim."""
+
+    required = normalize_required_capabilities(required_capabilities)
+    effective = normalize_worker_capabilities(worker_capabilities)
+    if any(value.startswith(_UNKNOWN_PREFIX) for value in required):
+        return False
+    return required.issubset(effective)
+
+
+def required_capabilities_subset(
+    granted_requirements: Iterable[object],
+    required_requirements: Iterable[object],
+) -> bool:
+    """Check semantic capability containment for delegation constraints."""
+    granted = normalize_required_capabilities(granted_requirements)
+    required = normalize_required_capabilities(required_requirements)
+    if any(value.startswith(_UNKNOWN_PREFIX) for value in granted | required):
+        return False
+    return required.issubset(granted)
+
+
+def normalized_capability_matrix(
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+) -> dict[str, object]:
+    """Expose non-secret decision evidence for tests and diagnostics."""
+
+    worker_values = tuple(worker_capabilities)
+    required_values = tuple(required_capabilities)
+    worker = normalize_worker_capabilities(worker_values)
+    required = normalize_required_capabilities(required_values)
+    return {
+        "worker_effective": sorted(worker),
+        "task_required": sorted(required),
+        "eligible": has_capabilities(worker_values, required_values),
+    }

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -13,8 +13,6 @@ BRIDGE_OPERATIONS = (
     "getGovernedWriteCapabilities",
     "createGovernedTask",
     "acquireWriterLease",
-    "bootstrapWriterLease",
-    "renewWriterLease",
     "commitFileGoverned",
     "commitFilesGoverned",
     "getGitOpsJob",
@@ -33,8 +31,6 @@ 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",),
@@ -50,49 +46,6 @@ 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
@@ -169,14 +122,6 @@ 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,
@@ -206,8 +151,6 @@ 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,7 +329,6 @@ 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
@@ -390,20 +389,13 @@ 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, *, exclude_job_id: str | None = None) -> int:
+    def _max_generation(self, branch: str) -> int:
         observed = self._observed_generations.get(branch, 0)
         with self.controller.store._connect() as connection:
-            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()
+            row = connection.execute(
+                "SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
+                (branch,),
+            ).fetchone()
         durable = int(row["generation"] or 0)
         return max(observed, durable)
 
@@ -412,7 +404,6 @@ 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,
@@ -422,7 +413,7 @@ class ContinuationExecutor:
             "lease_seconds": request.lease_seconds,
         }
 
-    def _validate_preflight(self, binding: ContinuationRuntimeBinding, *, job_id: str | None = None) -> None:
+    def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> 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}:
@@ -435,7 +426,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, exclude_job_id=job_id):
+        if request.generation <= self._max_generation(request.target_branch):
             raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
         if not self.workspace.is_detached():
             raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
@@ -453,6 +444,7 @@ class ContinuationExecutor:
             "repository": request.repository,
             "operation": "CONTINUATION_COMMIT_FILES",
             "branch": request.target_branch,
+            "target_branch": request.target_branch,
             "expected_branch_oid": request.expected_branch_oid,
             "base_oid": request.base_oid,
             "branch_mode": request.branch_mode,
@@ -461,60 +453,39 @@ class ContinuationExecutor:
             "generation": request.generation,
         }
 
-    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:
+    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:
             return self.controller.store._receipt(current, duplicate=True).as_dict()
-        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)
+        self._validate_preflight(binding)
         request = binding.request
         projected = False
         try:
             self.credential.project()
             projected = True
-            row = self.controller.store.read(binding.job_id)
+            row = self.controller.store.read(submitted.job_id)
             if JobState(row["state"]) == JobState.REQUESTED:
                 self.controller.advance(
-                    binding.job_id,
+                    submitted.job_id,
                     JobState.AUTHORIZED,
                     side_effect_state="AUTHORIZED",
                     receipt={"task_id": request.task_id, "original_grant_consumed": True},
                 )
-                row = self.controller.store.read(binding.job_id)
+                row = self.controller.store.read(submitted.job_id)
             if JobState(row["state"]) == JobState.AUTHORIZED:
                 self.controller.lease_job(
-                    binding.job_id,
-                    lease_id=f"lease-{binding.job_id}",
+                    submitted.job_id,
+                    lease_id=f"lease-{submitted.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(binding.job_id, JobState.WORKTREE_READY, side_effect_state="WORKTREE_READY")
-            self.controller.advance(binding.job_id, JobState.EXECUTING, side_effect_state="IN_PROGRESS")
+            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")
             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")
@@ -522,22 +493,17 @@ class ContinuationExecutor:
             if authority_oid != new_oid:
                 raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
             self.controller.advance(
-                binding.job_id,
+                submitted.job_id,
                 JobState.LOCAL_VERIFIED,
                 side_effect_state="LOCAL_VERIFIED",
                 receipt={"authority_oid": authority_oid, "force_push": False},
             )
             try:
-                self.controller.advance(binding.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
+                self.controller.advance(submitted.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(binding.job_id, authority_oid, new_oid)
-            return self._finish_remote_verification(
-                binding.job_id,
-                authority_oid,
-                new_oid,
-                before_complete=before_complete,
-            )
+                return self._remote_pending(submitted.job_id, authority_oid, new_oid)
+            return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
         finally:
             if projected:
                 self.credential.revoke()
@@ -560,14 +526,7 @@ class ContinuationExecutor:
         )
         return result.as_dict()
 
-    def _finish_remote_verification(
-        self,
-        job_id: str,
-        authority_oid: str,
-        new_oid: str,
-        *,
-        before_complete: Callable[[str], None] | None = None,
-    ) -> dict[str, object]:
+    def _finish_remote_verification(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
         row = self.controller.store.read(job_id)
         branch = row["branch"]
         request = json.loads(row["request_json"])["binding"]
@@ -584,20 +543,8 @@ 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,
-        )
-        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()
+        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()
 
     def reconcile(self, job_id: str) -> dict[str, object]:
         row = self.controller.store.read(job_id)

active-runtime/app/aios_continuation_lease.py

@@ -12,6 +12,8 @@ from datetime import datetime, timedelta, timezone
 import re
 from typing import Protocol
 
+from aios_branch_binding import BRANCH_RE
+
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 TARGET_BRANCH = "agent/autonomous-release-campaign-v1-20260807"
@@ -49,7 +51,6 @@ 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"})
@@ -57,12 +58,12 @@ 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")
         _oid(self.expected_branch_oid, "expected_branch_oid")
+        if not isinstance(self.target_branch, str) or BRANCH_RE.fullmatch(self.target_branch) is None:
+            raise ContinuationError("target_branch_invalid")
         if self.target_branch in {"main", "master"} or self.target_branch.startswith(("refs/", "tags/")):
             raise ContinuationError("protected_branch_rejected")
         if not isinstance(self.generation, int) or isinstance(self.generation, bool) or self.generation < 1:

active-runtime/app/aios_gitops_job_controller.py

@@ -18,6 +18,8 @@ import tempfile
 from datetime import datetime, timezone
 from typing import Any, Callable, Mapping
 
+from aios_branch_binding import is_valid_governed_branch
+
 
 class JobError(RuntimeError):
     pass
@@ -27,19 +29,6 @@ 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
 
@@ -79,14 +68,6 @@ 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"
@@ -263,8 +244,26 @@ class JobStore:
             raise ServiceUnavailable("JOB_DB_UNAVAILABLE") from exc
         if row is None:
             raise JobError("JOB_NOT_FOUND")
+        self._validate_persisted_binding(row)
         return dict(row)
 
+    @staticmethod
+    def _validate_persisted_binding(row: Mapping[str, Any]) -> None:
+        try:
+            raw = json.loads(str(row["request_json"]))
+        except (KeyError, TypeError, ValueError) as exc:
+            raise JobError("DURABLE_BINDING_INVALID") from exc
+        binding = raw.get("binding", raw)
+        if not isinstance(binding, Mapping):
+            raise JobError("DURABLE_BINDING_INVALID")
+        canonical_branch = row["branch"]
+        if not is_valid_governed_branch(canonical_branch):
+            raise JobError("DURABLE_BINDING_INVALID")
+        for field in ("branch", "target_branch"):
+            value = binding.get(field)
+            if value is not None and (not isinstance(value, str) or value != canonical_branch):
+                raise JobError("TARGET_BRANCH_BINDING_CONFLICT")
+
     def _receipt(self, row: Mapping[str, Any], *, duplicate: bool) -> JobReceipt:
         state = JobState(row["state"])
         if state == JobState.COMPLETED:
@@ -297,6 +296,11 @@ class GitOpsJobController:
             raise JobError("PAYLOAD_DIGEST_INVALID")
         if request["branch"] in {"main", "master"} or request["branch"].startswith(("refs/", "tags/")):
             raise JobError("PROTECTED_BRANCH_REJECTED")
+        target_branch = request.get("target_branch")
+        if target_branch is not None and (
+            not isinstance(target_branch, str) or target_branch != request["branch"]
+        ):
+            raise JobError("TARGET_BRANCH_BINDING_CONFLICT")
         binding = {field: str(request[field]) for field in fields}
         # Operation-specific, non-secret binding inputs must survive the
         # durable round trip and participate in idempotency. Arbitrary request
@@ -323,7 +327,7 @@ class GitOpsJobController:
                 continue
             if isinstance(value, bool) or not isinstance(value, (str, int)):
                 raise JobError("JOB_BINDING_OPTION_INVALID")
-            text = str(value).strip()
+            text = str(value) if field == "target_branch" else str(value).strip()
             if not text or len(text) > 512:
                 raise JobError("JOB_BINDING_OPTION_INVALID")
             binding[field] = text
@@ -425,16 +429,13 @@ 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", lease_expires_at: str | None = None) -> None:
+    def update_lease(self, lease_id: str, *, expected_revision: int, current_head_oid: str, generation: int, state: str = "ACTIVE") -> 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")
-                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))
+                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 updated.rowcount != 1:
                     raise JobError("LEASE_REVISION_CAS_CONFLICT")
                 connection.commit()
@@ -443,94 +444,6 @@ 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_governed_task_bootstrap.py

@@ -78,7 +78,7 @@ class GovernedTaskBootstrap:
             if generation < 1: raise BootstrapError("GENERATION_INVALID")
         payload = json.dumps(values, sort_keys=True, separators=(",", ":")).encode()
         _, controller = self._store(values["task_id"])
-        receipt = controller.submit({"idempotency_key": values["idempotency_key"], "campaign_id": values["task_id"], "task_id": values["task_id"], "repository": values["repository"], "operation": "GOVERNED_TASK_CREATE", "branch": values["target_branch"], "expected_branch_oid": expected, "payload_digest": hashlib.sha256(payload).hexdigest(), "lineage_anchor_oid": values["base_oid"]}, payload=payload)
+        receipt = controller.submit({"idempotency_key": values["idempotency_key"], "campaign_id": values["task_id"], "task_id": values["task_id"], "repository": values["repository"], "operation": "GOVERNED_TASK_CREATE", "branch": values["target_branch"], "target_branch": values["target_branch"], "expected_branch_oid": expected, "payload_digest": hashlib.sha256(payload).hexdigest(), "lineage_anchor_oid": values["base_oid"]}, payload=payload)
         binding = GovernedTaskBinding(**values, expected_branch_oid=expected, branch_mode=mode, generation=generation, job_id=receipt.job_id)
         self._write(binding)
         return binding.response()

active-runtime/app/authorized_task_writer_provisioning.py

@@ -964,7 +964,6 @@ def validate_writer_lease_request_v2(raw: Mapping[str, Any], probe: Any) -> dict
     required = {
         "schema",
         "task_id",
-        "job_id",
         "repository",
         "branch_mode",
         "base_oid",
@@ -977,7 +976,6 @@ 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"]),
@@ -990,7 +988,6 @@ 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,

active-runtime/app/task_plane/capabilities.py

@@ -1,36 +1,25 @@
-"""Capability matcher helpers."""
+"""Compatibility facade for the canonical capability contract."""
 
 from __future__ import annotations
 
-from fnmatch import fnmatch
-
-
-def _normalize_capability(capability: str) -> str:
-    return capability.strip().lower()
+from aios_capability_contract import (
+    has_capabilities as _has_capabilities,
+    normalize_worker_capabilities,
+    required_capabilities_subset as _required_capabilities_subset,
+)
 
 
 def capability_matches(granted: str, required: str) -> bool:
-    granted = _normalize_capability(granted)
-    required = _normalize_capability(required)
-    if granted == required:
-        return True
-    # A granted wildcard such as "test.*" should match required "test.unit"
-    if "*" in granted:
-        return fnmatch(required, granted)
-    # A required wildcard is treated as a suffix/prefix requirement for callers.
-    if "*" in required:
-        return fnmatch(granted, required)
-    return False
+    return _has_capabilities([granted], [required])
 
 
 def has_capabilities(granted_caps: list[str], required_caps: list[str]) -> bool:
-    granted = [_normalize_capability(c) for c in granted_caps]
-    for required in required_caps:
-        required = _normalize_capability(required)
-        if not any(capability_matches(g, required) for g in granted):
-            return False
-    return True
+    return _has_capabilities(granted_caps, required_caps)
 
 
 def normalize_capabilities(caps: list[str]) -> list[str]:
-    return sorted({_normalize_capability(cap) for cap in caps})
+    return sorted(normalize_worker_capabilities(caps))
+
+
+def required_capabilities_subset(granted: list[str], required: list[str]) -> bool:
+    return _required_capabilities_subset(granted, required)

active-runtime/app/task_plane/delegation.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from .capabilities import required_capabilities_subset
 from .errors import ConflictError, ForbiddenError, NotFoundError
 from .models import Budget, TaskPlan, TaskRecord
 from .store import TaskPlaneStore
@@ -34,8 +35,9 @@ class DelegationEngine:
     def delegate(self, parent_id: str, child_plan: TaskPlan, *, caller: str) -> TaskRecord:
         self.can_delegate(parent_id, child_plan)
         parent = self.store.get_task(parent_id)
-        if child_plan.required_capabilities and not set(child_plan.required_capabilities).issubset(
-            set(parent.task_plan.required_capabilities or [])
+        if child_plan.required_capabilities and not required_capabilities_subset(
+            parent.task_plan.required_capabilities or [],
+            child_plan.required_capabilities,
         ):
             raise ForbiddenError("child capabilities exceed parent constraints")
         risk_rank = {"low": 1, "medium": 2, "high": 3}

active-runtime/app/task_plane/policy.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from dataclasses import dataclass
 from typing import Callable
 
+from .capabilities import has_capabilities
 from .errors import ForbiddenError, LockedError
 from .models import ProbeEvidence, ResultEnvelope, TaskRecord
 
@@ -50,7 +51,7 @@ class ReadonlyWorkerProfile:
             return False
         if plan.metadata.get("production_side_effects") is not False:
             return False
-        return set(plan.required_capabilities).issubset(self.capabilities)
+        return has_capabilities(list(self.capabilities), plan.required_capabilities)
 
 
 class IdentityVerifier:

docs/gitops/capability-contract.md

@@ -0,0 +1,31 @@
+# AIOS capability contract
+
+The Agent Task API compares task requirements and worker registrations through one
+strict semantic matcher. Task-facing names are normalized to the worker vocabulary
+before both poll and claim; the two paths cannot disagree about eligibility.
+
+| Task-facing requirement | Semantic capability | Worker vocabulary |
+| --- | --- | --- |
+| repository_read | repo.read | git.read |
+| test_execution | test.execute | git.test |
+| governed_git_write | governed.git.write (composite) | governed.write.branch + governed.write.file |
+
+governed_git_write is composite: a worker must have both the branch-write and
+file-write capabilities. A single component never satisfies the requirement.
+
+Unknown names are not granted by normalization. Wildcard, prefix, and glob
+expressions are never expanded. This is fail-closed: an unknown task
+requirement is never eligible, even if a worker happens to report the same
+unknown string.
+
+The same matcher is used by:
+
+- GET /v1/agent/tasks/poll
+- POST /v1/agent/tasks/{task_id}/claim
+- the compatibility facade in task_plane.capabilities
+
+Legacy runtime names such as shell, git, and python remain exact semantic
+capabilities for compatibility; none implies governed write, production, main, or
+unrestricted command execution. Capability aliases are a compatibility boundary
+only; they do not grant additional authority. Capability strings remain separate
+from Owner credentials, Agent tokens, and WriterLease authorization.

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

@@ -0,0 +1,310 @@
+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
+x-aios-capability-contract:
+  description: >
+    Task-facing semantic capability names are normalized against the worker
+    runtime vocabulary before eligibility. Unknown and wildcard names fail
+    closed. The composite governed_git_write requirement requires both
+    governed.write.branch and governed.write.file.
+  task_facing:
+    repository_read: repo.read
+    test_execution: test.execute
+    governed_git_write: governed.git.write
+  worker_vocabulary:
+    git.read: repo.read
+    git.test: test.execute
+    governed.write.branch: governed.git.write.branch
+    governed.write.file: governed.git.write.file

docs/gitops/openapi-authorized-task-writer-fragment.yaml

@@ -35,6 +35,7 @@ components:
         target_branch:
           type: string
           pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'
+          maxLength: 186
         owner_authorization_reference:
           type: string
           minLength: 4

docs/gitops/openapi-chatgpt-actions.yaml

@@ -240,114 +240,6 @@ components:
       additionalProperties: false
       properties:
         candidate_id: {type: string}
-    BridgeWriterLeaseRecoveryRequest:
-      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
-          enum: [aios.task-writer-lease.v2, aios.task-writer-lease-request.v2]
-        task_id: {type: string, minLength: 1, maxLength: 256}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
-        base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-    BridgeWriterLeaseResponse:
-      type: object
-      additionalProperties: false
-      required: [schema, task_id, job_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds, create_target_branch, worktree_mode, cas_ref_update_required]
-      properties:
-        schema: {type: string, enum: [aios.task-writer-lease.v2]}
-        task_id: {type: string}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
-        base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-        create_target_branch: {type: boolean}
-        worktree_mode: {type: string, enum: [detached_at_base_oid, detached_at_expected_head]}
-        cas_ref_update_required: {type: boolean, const: true}
-        lease_expires_at: {type: string, format: date-time}
-    BridgeWriterLeaseRecoveryResponse:
-      type: object
-      additionalProperties: false
-      required: [schema, task_id, job_id, repository, branch_mode, base_oid, expected_branch_oid, target_branch, generation, lease_seconds, lease_expires_at, create_target_branch, worktree_mode, cas_ref_update_required, recovered, bootstrapped, idempotent, runtime_rehydrated, durable_branch_lease_present]
-      properties:
-        schema: {type: string, enum: [aios.task-writer-lease.v2]}
-        task_id: {type: string}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        branch_mode: {type: string, enum: [CREATE_NEW, CONTINUE_EXISTING]}
-        base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        lease_seconds: {type: integer, minimum: 1, maximum: 86400}
-        lease_expires_at: {type: string, format: date-time}
-        create_target_branch: {type: boolean, const: false}
-        worktree_mode: {type: string, enum: [detached_at_base_oid, detached_at_expected_head]}
-        cas_ref_update_required: {type: boolean, const: true}
-        recovered: {type: boolean}
-        bootstrapped: {type: boolean}
-        idempotent: {type: boolean}
-        runtime_rehydrated: {type: boolean}
-        durable_branch_lease_present: {type: boolean}
-    BridgeCommitFileRequest:
-      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, minLength: 1}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        idempotency_key: {type: string, minLength: 1, maxLength: 256}
-        path: {type: string, minLength: 1, maxLength: 512}
-        content_base64: {type: string, format: byte}
-        message: {type: string, minLength: 1, maxLength: 200}
-    BridgeCommitFilesRequest:
-      type: object
-      additionalProperties: false
-      required: [task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message]
-      properties:
-        task_id: {type: string, minLength: 1}
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        repository: {type: string, const: aios/aios}
-        expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-        idempotency_key: {type: string, minLength: 1, maxLength: 256}
-        files: {type: object, additionalProperties: {type: string, format: byte}, minProperties: 1}
-        message: {type: string, minLength: 1, maxLength: 200}
-    BridgeCommitResponse:
-      type: object
-      additionalProperties: true
-      required: [job_id, commit_oid, commit_visible, branch_mode, target_branch, generation]
-      properties:
-        job_id: {type: string, pattern: '^job-[A-Za-z0-9][A-Za-z0-9._-]{7,255}$'}
-        commit_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
-        commit_visible: {type: boolean}
-        branch_mode: {type: string, enum: [CONTINUE_EXISTING]}
-        target_branch: {type: string, pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'}
-        generation: {type: integer, minimum: 1}
-    BridgeGovernedCapabilities:
-      type: object
-      required: [bridge_schema_version, governed_writes_enabled, operations, operation_contracts]
-      properties:
-        bridge_schema_version: {type: string, const: aios.chatgpt-gitops-bridge.v2}
-        governed_writes_enabled: {type: boolean, const: true}
-        operations: {type: array, items: {type: string}}
-        operation_contracts: {type: object, additionalProperties: true}
 paths:
   /createReleaseIntent:
     post:
@@ -675,118 +567,3 @@ paths:
         '200': {description: Safe GitOps plan}
         '401': {description: Unauthorized}
       x-openai-isConsequential: true
-  /v2/capabilities:
-    get:
-      operationId: getGovernedWriteCapabilities
-      summary: Read the governed Bridge v2 capability contract
-      responses:
-        '200':
-          description: Governed operations and their request/response contracts
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeGovernedCapabilities'}
-  /v2/writer-lease:
-    post:
-      operationId: acquireWriterLease
-      summary: Acquire the exact governed writer lease for a task generation
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
-      responses:
-        '200':
-          description: Exact detached worktree lease binding
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeWriterLeaseResponse'}
-        '400': {description: Invalid lease binding}
-        '409': {description: Branch, base, or generation conflict}
-      x-openai-isConsequential: true
-  /v2/writer-lease/bootstrap:
-    post:
-      operationId: bootstrapWriterLease
-      summary: Bootstrap a governed writer runtime without creating a durable lease
-      description: >
-        Staging-only runtime recovery. It validates the exact task, repository,
-        base OID, expected branch OID, target branch, and generation. It must
-        not create a branch, insert a branch lease, or consume a generation.
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
-      responses:
-        '200':
-          description: Runtime binding bootstrapped or returned idempotently
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryResponse'}
-        '400': {description: Invalid lease binding}
-        '409': {description: A matching durable lease already exists or binding conflicts}
-        '503': {description: Runtime or durable state unavailable}
-      x-openai-isConsequential: true
-  /v2/writer-lease/renew:
-    post:
-      operationId: renewWriterLease
-      summary: Recover an expired governed writer lease without changing its binding
-      description: >
-        Renews only the current task generation and exact repository, base OID,
-        expected branch OID, and target branch. Durable renewal is revision-CAS
-        protected; runtime rehydration is reported separately.
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryRequest'}
-      responses:
-        '200':
-          description: Lease recovered or idempotently observed as already active
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeWriterLeaseRecoveryResponse'}
-        '400': {description: Invalid schema, generation, or lease duration}
-        '409': {description: Task, base, repository, branch, generation, or durable lease conflict}
-        '503': {description: Durable job store or runtime unavailable}
-      x-openai-isConsequential: true
-  /v2/commit-file:
-    post:
-      operationId: commitFileGoverned
-      summary: Commit one file through the governed writer lease
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeCommitFileRequest'}
-      responses:
-        '200':
-          description: Authority and remote visibility receipt
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeCommitResponse'}
-        '409': {description: Lease, expected-head, idempotency, or repository conflict}
-        '503': {description: Reconciliation required}
-      x-openai-isConsequential: true
-  /v2/commit-files:
-    post:
-      operationId: commitFilesGoverned
-      summary: Atomically commit a bounded set of files through the governed writer lease
-      security: [{bearerAuth: []}]
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema: {$ref: '#/components/schemas/BridgeCommitFilesRequest'}
-      responses:
-        '200':
-          description: Authority and remote visibility receipt
-          content:
-            application/json:
-              schema: {$ref: '#/components/schemas/BridgeCommitResponse'}
-        '409': {description: Lease, expected-head, idempotency, or repository conflict}
-        '503': {description: Reconciliation required}
-      x-openai-isConsequential: true

docs/gitops/openapi-owner-authorization-broker-fragment.yaml

@@ -38,6 +38,7 @@ components:
         target_branch:
           type: string
           pattern: '^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$'
+          maxLength: 186
         one_time_use:
           type: boolean
           const: true

docs/gitops/writer-lease-binding.md

@@ -0,0 +1,19 @@
+# AIOS WriterLease branch binding
+
+A governed branch name is an opaque contract value and must survive the entire
+lifecycle unchanged:
+
+Task.target_branch -> Job.branch -> Lease.branch -> commit request.target_branch -> actual ref
+
+The binding rules are:
+
+- The value must match the governed agent branch grammar and the configured length limit.
+- No truncation, abbreviation, slugging, whitespace trimming, case folding, or inferred replacement is permitted.
+- If a request carries both branch and target_branch, they must be byte-for-byte equal before a job is persisted or a ref is touched.
+- A WriterLease response is checked against the exact requested branch before editing begins.
+- A persisted binding with conflicting aliases is invalid and fails closed as reconciliation-required state; operators must inspect and repair it through the governed path rather than guessing a branch.
+- Overlong or malformed names are validation errors, never shortened values.
+
+The exact value is also checked at commit preflight alongside repository and
+expected base OID. The proof of completion is the exact branch ref and commit OID,
+not merely a successful HTTP response or an active service.

scripts/build_agent_task_api.py

@@ -0,0 +1,101 @@
+#!/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",
+    "active-runtime/app/aios_capability_contract.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_protocol.py

@@ -0,0 +1,317 @@
+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_branch_binding_contract.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+
+import pytest
+
+from aios_continuation_lease import ContinuationError, WriterLeaseRequestV2
+from aios_gitops_job_controller import GitOpsJobController, JobError, JobStore
+
+
+BASE_OID = "3244f0f17f8845592ac94d585f5ebd65693c9f29"
+FULL_BRANCH = "agent/r4p0-writerlease-correctness-20260816"
+
+
+class CreateNewProbe:
+    def branch_exists(self, branch):
+        return False
+
+    def commit_exists(self, oid):
+        return True
+
+
+def _job_request(target_branch=FULL_BRANCH, branch=FULL_BRANCH):
+    return {
+        "campaign_id": "governance-contract-remediation-20260816",
+        "repository": "aios/aios",
+        "operation": "GOVERNED_COMMIT",
+        "branch": branch,
+        "target_branch": target_branch,
+        "expected_branch_oid": BASE_OID,
+        "payload_digest": "a" * 64,
+        "idempotency_key": "governance-contract-remediation-v1",
+    }
+
+
+def test_writer_lease_branch_round_trip_preserves_full_regression_value():
+    request = WriterLeaseRequestV2(
+        task_id="governance-contract-remediation-20260816",
+        repository="aios/aios",
+        branch_mode="CREATE_NEW",
+        base_oid=BASE_OID,
+        expected_branch_oid=BASE_OID,
+        target_branch=FULL_BRANCH,
+        generation=1,
+        lease_seconds=900,
+    )
+    request.validate(CreateNewProbe())
+    assert request.target_branch == FULL_BRANCH
+
+
+def test_job_binding_round_trip_keeps_branch_and_target_branch_exact(tmp_path):
+    store = JobStore(str(tmp_path / "jobs"))
+    receipt = GitOpsJobController(store).submit(_job_request())
+    binding = store.read(receipt.job_id)
+    assert binding["branch"] == FULL_BRANCH
+    assert json.loads(binding["request_json"])["binding"]["target_branch"] == FULL_BRANCH
+
+
+def test_mismatched_alias_fails_closed_before_persisting_job(tmp_path):
+    root = tmp_path / "jobs"
+    store = JobStore(str(root))
+    request = _job_request(
+        target_branch="agent/r4p0-writerlease-correctness-2026",
+        branch=FULL_BRANCH,
+    )
+    with pytest.raises(JobError, match="TARGET_BRANCH_BINDING_CONFLICT"):
+        GitOpsJobController(store).submit(request)
+
+    connection = sqlite3.connect(root / "state.sqlite3")
+    try:
+        assert connection.execute("SELECT COUNT(*) FROM jobs").fetchone()[0] == 0
+    finally:
+        connection.close()
+
+
+def test_damaged_persisted_alias_fails_closed(tmp_path):
+    root = tmp_path / "jobs"
+    store = JobStore(str(root))
+    receipt = GitOpsJobController(store).submit(_job_request())
+    connection = sqlite3.connect(root / "state.sqlite3")
+    try:
+        damaged = json.dumps({"binding": {**_job_request(), "target_branch": "agent/r4p0-writerlease-correctness-2026"}})
+        connection.execute("UPDATE jobs SET request_json=? WHERE job_id=?", (damaged, receipt.job_id))
+        connection.commit()
+    finally:
+        connection.close()
+    with pytest.raises(JobError, match="TARGET_BRANCH_BINDING_CONFLICT"):
+        store.read(receipt.job_id)
+
+
+def test_overlong_branch_is_rejected_without_shortening():
+    overlong = "agent/" + ("a" * 181)
+    request = WriterLeaseRequestV2(
+        task_id="governance-contract-remediation-20260816",
+        repository="aios/aios",
+        branch_mode="CREATE_NEW",
+        base_oid=BASE_OID,
+        expected_branch_oid=BASE_OID,
+        target_branch=overlong,
+        generation=1,
+        lease_seconds=900,
+    )
+    with pytest.raises(ContinuationError, match="target_branch_invalid"):
+        request.validate(CreateNewProbe())

tests/test_bridge_v2_governed_http_bootstrap.py

@@ -2,7 +2,6 @@ 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
@@ -27,7 +26,6 @@ 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",
@@ -41,7 +39,6 @@ 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",
@@ -78,7 +75,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/writer-lease/bootstrap", "/v2/writer-lease/renew", "/v2/commit-file", "/v2/commit-files",
+        "/v2/writer-lease", "/v2/commit-file", "/v2/commit-files",
         "/v2/job", "/v2/job/resume", "/v2/job/reconcile",
         "/v2/tag/approval-request",
         "/v2/tag/publish",
@@ -91,121 +88,3 @@ 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_capability_contract.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import pytest
+
+from aios_capability_contract import (
+    has_capabilities,
+    normalize_required_capabilities,
+    normalize_worker_capabilities,
+)
+from aios_agent_task_protocol import TaskProtocolError, TaskStore
+from task_plane.capabilities import has_capabilities as legacy_has_capabilities
+
+
+def test_task_worker_aliases_and_composite_write():
+    assert has_capabilities(["git.read"], ["repository_read"])
+    assert has_capabilities(["git.test"], ["test_execution"])
+    assert has_capabilities(
+        ["governed.write.branch", "governed.write.file"],
+        ["governed_git_write"],
+    )
+    assert legacy_has_capabilities(["git.read"], ["repository_read"])
+
+
+@pytest.mark.parametrize(
+    ("required", "worker"),
+    [
+        (["governed_git_write"], ["governed.write.file"]),
+        (["governed_git_write"], ["governed.write.branch"]),
+        (["governed.write.branch"], ["governed.write.file"]),
+        (["production.superuser"], ["git.read"]),
+        (["mystery.capability"], ["mystery.capability"]),
+        (["git.*"], ["git.read"]),
+        (["git.read*"], ["git.read"]),
+    ],
+)
+def test_capability_matching_fails_closed(required, worker):
+    assert not has_capabilities(required, worker)
+
+
+def test_security_boundaries_do_not_expand_capabilities():
+    assert not has_capabilities(
+        ["governed.write.branch", "governed.write.file"], ["production.superuser"]
+    )
+    assert not has_capabilities(
+        ["governed.write.branch", "governed.write.file"], ["main.write"]
+    )
+    assert not has_capabilities(["git.test"], ["runtime.shell"])
+    assert not has_capabilities(["git.read"], ["governed_git_write"])
+
+
+def test_duplicate_aliases_normalize_to_one_semantic_capability():
+    assert normalize_required_capabilities(
+        ["repository_read", "git.read", "repo.read"]
+    ) == {"repo.read"}
+    assert normalize_worker_capabilities(["git.read", "repo.read"]) == {"repo.read"}
+
+
+def test_poll_and_claim_share_the_same_matcher(tmp_path):
+    store = TaskStore(tmp_path / "tasks.sqlite3")
+    store.register_agent(
+        agent_id="agent-1",
+        agent_type="test-worker",
+        agent_version="test",
+        capabilities=["git.read", "git.test"],
+    )
+    task_id = "task-capability-contract"
+    store.create_task(
+        task_id=task_id,
+        goal="capability contract",
+        instructions="run the contract test",
+        success_criteria=[],
+        required_capabilities=["repository_read", "test_execution"],
+        idempotency_key="capability-contract-v1",
+    )
+
+    eligible = store.poll_tasks(agent_id="agent-1", limit=10)
+    assert [row["task_id"] for row in eligible] == [task_id]
+
+    claimed = store.claim_task(task_id=task_id, agent_id="agent-1")
+    assert claimed["task_id"] == task_id
+    assert claimed["claimed_by"] == "agent-1"
+
+
+def test_poll_and_claim_reject_the_same_ineligible_worker(tmp_path):
+    store = TaskStore(tmp_path / "tasks.sqlite3")
+    store.register_agent(
+        agent_id="agent-file-only",
+        agent_type="test-worker",
+        agent_version="test",
+        capabilities=["governed.write.file"],
+    )
+    task_id = "task-write-contract"
+    store.create_task(
+        task_id=task_id,
+        goal="write capability contract",
+        instructions="run the contract test",
+        success_criteria=[],
+        required_capabilities=["governed_git_write"],
+        idempotency_key="write-contract-v1",
+    )
+
+    eligible = store.poll_tasks(agent_id="agent-file-only", limit=10)
+    assert eligible == []
+
+    with pytest.raises(TaskProtocolError, match="AGENT_CAPABILITY_MISMATCH"):
+        store.claim_task(task_id=task_id, agent_id="agent-file-only")

tests/test_continuation_runtime_normalization.py

@@ -1,175 +0,0 @@
-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

@@ -1,708 +0,0 @@
-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)