fix(control-plane): repair caller admission runtime regressions

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

active-runtime/app/aios_caller_admission_projection.py

@@ -0,0 +1,133 @@
+"""Private Unix-socket caller projection and rehydration client.
+
+The broker response body is deliberately public metadata only. The short-lived
+credential projection is delivered in a private response header, consumed
+inside the model API process, and never returned by an HTTP API or persisted.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import http.client
+import json
+from pathlib import Path
+import socket
+from typing import Any, Mapping
+
+CALLER = "chatgpt-governed-implementer-singlehost"
+SCOPE = "release:candidate"
+CREDENTIAL_HEADER = "X-AIOS-Caller-Projection-Credential"
+
+
+class CallerProjectionError(RuntimeError):
+    def __init__(self, code: str, status: int = 503) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+@dataclass(frozen=True)
+class ProjectionResult:
+    public: dict[str, Any]
+    credential_loaded: bool
+
+
+class _UnixHTTPConnection(http.client.HTTPConnection):
+    def __init__(self, socket_path: Path, timeout: float) -> None:
+        super().__init__("localhost", timeout=timeout)
+        self.socket_path = socket_path
+
+    def connect(self) -> None:
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.settimeout(self.timeout)
+        try:
+            sock.connect(str(self.socket_path))
+        except Exception:
+            sock.close()
+            raise
+        self.sock = sock
+
+
+class CallerProjectionClient:
+    def __init__(self, socket_path: Path, *, timeout: float = 2.0) -> None:
+        self.socket_path = Path(socket_path)
+        self.timeout = timeout
+
+    @staticmethod
+    def _idempotency_key(caller: str, scope: str) -> str:
+        digest = hashlib.sha256((caller + "\0" + scope).encode()).hexdigest()
+        return "caller-projection-" + digest[:32] + "-v1"
+
+    def rehydrate(
+        self,
+        caller: str,
+        scope: str,
+        request: Mapping[str, Any] | None = None,
+    ) -> ProjectionResult:
+        if caller != CALLER:
+            raise CallerProjectionError("CALLER_MISMATCH", 403)
+        if scope != SCOPE:
+            raise CallerProjectionError("SCOPE_MISMATCH", 403)
+        body = {
+            "caller": caller,
+            "scope": scope,
+            "idempotency_key": self._idempotency_key(caller, scope),
+            "request": dict(request or {}),
+        }
+        connection = _UnixHTTPConnection(self.socket_path, self.timeout)
+        try:
+            connection.request(
+                "POST",
+                "/v1/project-caller",
+                body=json.dumps(body, sort_keys=True, separators=(",", ":")).encode(),
+                headers={
+                    "Content-Type": "application/json",
+                    "Accept": "application/json",
+                    "X-AIOS-Account": caller,
+                },
+            )
+            response = connection.getresponse()
+            status = response.status
+            raw = response.read()
+            credential = response.getheader(CREDENTIAL_HEADER, "")
+        except (OSError, TimeoutError, http.client.HTTPException) as exc:
+            raise CallerProjectionError("PROJECTION_BROKER_UNAVAILABLE", 503) from exc
+        finally:
+            connection.close()
+
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except Exception as exc:
+            raise CallerProjectionError("PROJECTION_RESPONSE_INVALID", 503) from exc
+        if not isinstance(value, Mapping):
+            raise CallerProjectionError("PROJECTION_RESPONSE_INVALID", 503)
+        if status != 200:
+            code = str(value.get("error") or "PROJECTION_DENIED")
+            raise CallerProjectionError(code, 403 if status < 500 else 503)
+        if any(
+            str(key).lower() in {
+                "credential",
+                "credential_material",
+                "token",
+                "bearer",
+                "secret",
+                "password",
+            }
+            for key in value
+        ):
+            raise CallerProjectionError("PROJECTION_SECRET_FIELD_REJECTED", 503)
+        if value.get("caller") != CALLER or value.get("scope") != SCOPE:
+            raise CallerProjectionError("PROJECTION_BINDING_INVALID", 503)
+        if value.get("projection_state") != "ACKED":
+            raise CallerProjectionError("PROJECTION_STATE_INVALID", 503)
+        if value.get("credential_state") != "REHYDRATED":
+            raise CallerProjectionError("CREDENTIAL_REHYDRATION_INVALID", 503)
+        if value.get("rehydration_state") not in {"GENERATED", "RECOVERED"}:
+            raise CallerProjectionError("REHYDRATION_STATE_INVALID", 503)
+        if value.get("effective_scopes") != [SCOPE]:
+            raise CallerProjectionError("PROJECTION_SCOPE_INVALID", 503)
+        if value.get("secret_exposed") is not False:
+            raise CallerProjectionError("PROJECTION_SECRET_EXPOSURE", 503)
+        if not isinstance(credential, str) or len(credential) < 32:
+            raise CallerProjectionError("CREDENTIAL_REHYDRATION_MISSING", 503)
+        return ProjectionResult(public=dict(value), credential_loaded=True)

active-runtime/app/aios_model_api_staging.py

@@ -3,6 +3,13 @@ from __future__ import annotations
 import hashlib, hmac, json, os, re, subprocess, sys, time, urllib.parse, urllib.request, uuid
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
+
+from aios_caller_admission_projection import (
+    CALLER,
+    SCOPE,
+    CallerProjectionClient,
+    CallerProjectionError,
+)
 from baseline_authority import BaselineAuthority, BaselineAuthorityError
 from typing import Any
 
@@ -89,6 +96,11 @@ SILENT_RELEASE_PUBLIC_KEYS = Path(os.getenv(
     'AIOS_SILENT_RELEASE_PUBLIC_KEYS',
     '/etc/aios-gitops/silent-release-owner-public-keys.json',
 ))
+CALLER_PROJECTION_SOCKET = Path(os.getenv(
+    'AIOS_OWNER_AUTH_BROKER_SOCKET',
+    '/run/aios-owner-authorization-broker/broker.sock',
+))
+CALLER_PROJECTION_CLIENT = CallerProjectionClient(CALLER_PROJECTION_SOCKET)
 monitor = ActionMonitor()
 
 BASE_MODELS = [
@@ -112,6 +124,9 @@ READ_ONLY_GET_PATHS = frozenset({
     '/admin/ai-evolution/api/data',
     '/api/release-consistency', '/api/tag-publication-status',
     '/api/release-acceptance',
+    '/api/governed-write-capabilities', '/api/control-plane-snapshot',
+    '/api/caller-capabilities',
+    '/api/online-agent-task', '/api/online-agent-task-events',
     '/getReleaseCandidate', '/getReleasePlan', '/getReleaseAcceptance',
     '/getApprovalEnvelope', '/getReleaseSaga', '/getCapabilityStatus',
     '/getRecoveryStatus',
@@ -407,14 +422,113 @@ class H(BaseHTTPRequestHandler):
         except Exception: return {'_raw':raw}
     def actor(self):
         auth=self.headers.get('Authorization',''); tok=auth[7:].strip() if auth.startswith('Bearer ') else self.headers.get('X-AIOS-Token','')
-        rec=authenticate(tok)
-        if rec: return {'ok':True,'account':safe(rec.get('account') or rec.get('name') or 'token-user'),'role':rec.get('role','user'),'scopes':set(rec.get('scopes') or [])}
-        return {'ok':False,'account':'anonymous','role':'anonymous','scopes':set()}
-    def need(self, scope):
-        a=self.actor()
-        if not a['ok']: self.sendj({'error':'unauthorized','hint':'Bearer token required'},401); return None
-        if '*' not in a['scopes'] and scope not in a['scopes']: self.sendj({'error':'forbidden','required_scope':scope,'account':a['account']},403); return None
+        try:
+            rec=authenticate(tok)
+        except Exception:
+            rec=None
+        if not isinstance(rec, dict) or not rec:
+            return {
+                'ok': False,
+                'account': 'anonymous',
+                'role': 'anonymous',
+                'scopes': set(),
+                'static_scopes': set(),
+                'projection': None,
+                'credential_loaded': False,
+            }
+        scopes = set(rec.get('scopes') or [])
+        return {
+            'ok': True,
+            'account': safe(rec.get('account') or rec.get('name') or 'token-user'),
+            'role': rec.get('role', 'user'),
+            'scopes': scopes,
+            'static_scopes': set(scopes),
+            'projection': None,
+            'credential_loaded': False,
+        }
+    def need(self, scope, *, projection_request=None):
+        a = self.actor()
+        if not a['ok']:
+            self.sendj({'error': 'unauthorized', 'hint': 'Bearer token required'}, 401)
+            return None
+        if a['account'] == CALLER and scope == SCOPE:
+            try:
+                result = CALLER_PROJECTION_CLIENT.rehydrate(
+                    a['account'],
+                    scope,
+                    projection_request,
+                )
+            except CallerProjectionError as exc:
+                self.sendj(
+                    {
+                        'error': 'caller_projection_denied',
+                        'projection_state': 'DENIED',
+                        'detail': exc.code,
+                        'account': a['account'],
+                    },
+                    exc.status,
+                )
+                return None
+            projection = result.public
+            a['scopes'].update(projection.get('effective_scopes') or [])
+            a['projection'] = projection
+            a['credential_loaded'] = result.credential_loaded
+        if '*' not in a['scopes'] and scope not in a['scopes']:
+            self.sendj(
+                {
+                    'error': 'forbidden',
+                    'required_scope': scope,
+                    'account': a['account'],
+                },
+                403,
+            )
+            return None
         return a
+    def caller_capabilities(self):
+        actor = self.actor()
+        if not actor['ok']:
+            return self.sendj(
+                {'error': 'unauthorized', 'hint': 'Bearer token required'},
+                401,
+            )
+        effective = set(actor['static_scopes'])
+        projection_state = 'NOT_REQUIRED'
+        credential_state = 'NOT_REQUIRED'
+        rehydration_state = 'NOT_APPLICABLE'
+        admitted = True
+        projection_error = None
+        if actor['account'] == CALLER:
+            try:
+                result = CALLER_PROJECTION_CLIENT.rehydrate(
+                    actor['account'],
+                    SCOPE,
+                    {},
+                )
+            except CallerProjectionError as exc:
+                admitted = False
+                projection_state = 'DENIED'
+                credential_state = 'NOT_REHYDRATED'
+                rehydration_state = 'FAILED'
+                projection_error = exc.code
+            else:
+                effective.update(result.public.get('effective_scopes') or [])
+                projection_state = result.public['projection_state']
+                credential_state = result.public['credential_state']
+                rehydration_state = result.public['rehydration_state']
+        return self.sendj(
+            {
+                'schema': 'aios.caller-capability-introspection.v1',
+                'caller_identity': actor['account'],
+                'admitted': admitted,
+                'effective_scopes': sorted(effective),
+                'projection_state': projection_state,
+                'credential_state': credential_state,
+                'rehydration_state': rehydration_state,
+                'projection_error': projection_error,
+                'secret_exposed': False,
+            }
+        )
+
     def bridge_release(self, path, data):
         route = {
             '/createTagApprovalRequest': '/v2/tag/approval-request',
@@ -446,24 +560,183 @@ class H(BaseHTTPRequestHandler):
             return self.sendj(body, exc.code)
         except (urllib.error.URLError, TimeoutError) as exc:
             return self.sendj({'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__}, 503)
-    def do_POST(self):
-        path = urllib.parse.urlparse(self.path).path
-        if path in {
-            '/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
-            '/buildProductionArtifact', '/executeApprovedProductionRelease',
-            '/getReleaseConsistency', '/runReleaseAcceptance',
-        }:
-            actor = self.need('tag:create')
-            if not actor:
-                return
-            return self.bridge_release(path, self.body())
-        if reject_non_read_http(self, 'POST', path):
-            return
-        return self.sendj({'error': 'not_found', 'path': path}, 404)
+    def bridge_v2_json(self, route, data=None, method='POST'):
+        base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
+        payload = None if data is None else json.dumps(
+            data, sort_keys=True, separators=(',', ':')
+        ).encode('utf-8')
+        request = urllib.request.Request(
+            base + route,
+            data=payload,
+            headers={'Content-Type': 'application/json'} if payload is not None else {},
+            method=method,
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                result = json.loads(response.read().decode('utf-8'))
+                return self.sendj(result, response.status)
+        except urllib.error.HTTPError as exc:
+            try:
+                body = json.loads(exc.read().decode('utf-8'))
+            except Exception:
+                body = {'error': 'bridge_http_error'}
+            return self.sendj(body, exc.code)
+        except (urllib.error.URLError, TimeoutError) as exc:
+            return self.sendj(
+                {'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__},
+                503,
+            )
+
+    def control_plane_snapshot(self):
+        """Build a non-sensitive snapshot locally when Bridge v2 has no snapshot HTTP route."""
+        base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
+
+        def bridge_get(route):
+            request = urllib.request.Request(
+                base + route,
+                headers={'Accept': 'application/json'},
+                method='GET',
+            )
+            with urllib.request.urlopen(request, timeout=15) as response:
+                if response.status != 200:
+                    raise RuntimeError('bridge_read_http_' + str(response.status))
+                value = json.loads(response.read().decode('utf-8'))
+                if not isinstance(value, dict):
+                    raise RuntimeError('bridge_read_not_object')
+                return value
+
+        try:
+            runtime = bridge_get('/v2/runtime')
+            capabilities = bridge_get('/v2/capabilities')
+            repository = repository_snapshot(
+                git_dir=AUTHORITATIVE_GIT_DIR,
+                repository_identity=REPOSITORY_IDENTITY,
+            )
+            if not repository.get('repository_available'):
+                return self.sendj({
+                    'error': 'AUTHORITATIVE_REPOSITORY_UNAVAILABLE',
+                    'query_error': repository.get('query_error'),
+                }, 503)
+            baseline = dict(BASELINE_AUTHORITY.read())
+        except BaselineAuthorityError as exc:
+            return self.sendj({
+                'error': 'RECONCILIATION_REQUIRED',
+                'detail': str(exc),
+            }, 503)
+        except urllib.error.HTTPError as exc:
+            return self.sendj({
+                'error': 'GOVERNED_BRIDGE_UNAVAILABLE',
+                'detail': 'HTTP_' + str(exc.code),
+            }, 503)
+        except (urllib.error.URLError, TimeoutError, OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
+            return self.sendj({
+                'error': 'GOVERNED_BRIDGE_UNAVAILABLE',
+                'detail': type(exc).__name__,
+            }, 503)
+
+        legacy = capabilities.get('legacy_writes_enabled')
+        if legacy is None:
+            legacy = capabilities.get('legacy_global_writes_enabled')
+        payload = {
+            'schema': 'aios.control-plane-snapshot.aggregate.v1',
+            'snapshot_source': 'model-api-local-aggregate',
+            'version': {
+                'target_release': runtime.get('control_plane_release_version'),
+            },
+            'runtime': {
+                'bridge_schema_version': runtime.get('bridge_schema_version'),
+                'control_plane_release_version': runtime.get('control_plane_release_version'),
+                'runtime_release_commit': runtime.get('runtime_release_commit'),
+                'runtime_path': runtime.get('runtime_path'),
+                'artifact_sha256': runtime.get('artifact_sha256'),
+                'drift_status': runtime.get('drift_status'),
+                'deployment_state': runtime.get('deployment_state'),
+            },
+            'repository': {
+                'identity': runtime.get('repository_identity') or runtime.get('repository') or REPOSITORY_IDENTITY,
+                'authority_git_dir': runtime.get('authoritative_git_dir') or repository.get('authoritative_git_dir'),
+                'authority_main_oid': repository.get('authoritative_head_oid'),
+            },
+            'production': {
+                'baseline_version': baseline.get('current_baseline'),
+                'target_oid': baseline.get('target_oid'),
+                'deployment_state': baseline.get('status'),
+                'baseline_generation': baseline.get('generation'),
+                'baseline_sha256': baseline.get('authority_sha256'),
+            },
+            'task': {},
+            'capabilities': {
+                'legacy_writes_enabled': legacy,
+                'governed_writes_enabled': capabilities.get('governed_writes_enabled'),
+                'force_push': capabilities.get('force_push'),
+                'main_write': capabilities.get('main_write'),
+                'tag_write': capabilities.get('tag_write'),
+                'baseline_write': capabilities.get('baseline_write'),
+                'production_write': capabilities.get('production_write'),
+            },
+            'credential_broker': {
+                'status': 'SERVER_SIDE_ONLY',
+                'provider': 'BridgeV2RuntimeCredentialProjection',
+            },
+            'secret_material_exposed': False,
+        }
+        return self.sendj(payload)
+
+    def agent_task_control(self, method, route, data=None):
+        token_path = Path(os.getenv(
+            'AIOS_AGENT_TASK_CONTROL_TOKEN_FILE',
+            '/run/credentials/aios-gitops-model-api-v2r2c1-staging.service/agent_task_control_token',
+        ))
+        if token_path.is_symlink() or not token_path.is_file():
+            return self.sendj({'error': 'AGENT_TASK_CONTROL_CREDENTIAL_UNAVAILABLE'}, 503)
+        token = token_path.read_text(encoding='utf-8').strip()
+        if len(token) < 32:
+            return self.sendj({'error': 'AGENT_TASK_CONTROL_CREDENTIAL_INVALID'}, 503)
+        base = os.getenv('AIOS_AGENT_TASK_API_URL', 'http://127.0.0.1:18940').rstrip('/')
+        payload = None if data is None else json.dumps(
+            data, sort_keys=True, separators=(',', ':')
+        ).encode('utf-8')
+        headers = {'Authorization': 'Bearer ' + token, 'Accept': 'application/json'}
+        if payload is not None:
+            headers['Content-Type'] = 'application/json'
+        request = urllib.request.Request(
+            base + route, data=payload, headers=headers, method=method
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                result = json.loads(response.read().decode('utf-8'))
+                return self.sendj(result, response.status)
+        except urllib.error.HTTPError as exc:
+            try:
+                body = json.loads(exc.read().decode('utf-8'))
+            except Exception:
+                body = {'error': 'agent_task_http_error'}
+            return self.sendj(body, exc.code)
+        except (urllib.error.URLError, TimeoutError) as exc:
+            return self.sendj(
+                {'error': 'AGENT_TASK_API_UNAVAILABLE', 'detail': type(exc).__name__},
+                503,
+            )
+
     def do_GET(self):
         u=urllib.parse.urlparse(self.path); path=u.path.rstrip('/') if u.path != '/' else u.path
         if path not in READ_ONLY_GET_PATHS: return self.sendj({'error':'not_found','path':path},404)
         q=urllib.parse.parse_qs(u.query)
+        if path == '/api/governed-write-capabilities':
+            return self.bridge_v2_json('/v2/capabilities', method='GET')
+        if path == '/api/control-plane-snapshot':
+            return self.control_plane_snapshot()
+        if path == '/api/caller-capabilities':
+            return self.caller_capabilities()
+        if path in ('/api/online-agent-task', '/api/online-agent-task-events'):
+            actor = self.need('agent-task:control')
+            if not actor:
+                return
+            task_id = (q.get('task_id') or [''])[0]
+            if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{2,179}', task_id):
+                return self.sendj({'error': 'task_id_invalid'}, 400)
+            suffix = '/events' if path.endswith('-events') else ''
+            return self.agent_task_control('GET', '/v1/tasks/' + task_id + suffix)
         try:
             if path in ('/health','/api/health'):
                 return self.sendj({
@@ -700,12 +973,58 @@ class H(BaseHTTPRequestHandler):
         return self.sendj({'error':'not_found','path':path},404)
     def do_POST(self):
         path=urllib.parse.urlparse(self.path).path.rstrip('/')
+        governed_routes = {
+            '/createGovernedTask': ('/v2/governed-task', 'branch:create'),
+            '/acquireWriterLease': ('/v2/writer-lease', 'file:write'),
+            '/commitFileGoverned': ('/v2/commit-file', 'file:write'),
+            '/commitFilesGoverned': ('/v2/commit-files', 'file:write'),
+            '/getGitOpsJob': ('/v2/job', 'file:write'),
+            '/resumeGitOpsJob': ('/v2/job/resume', 'file:write'),
+            '/reconcileGitOpsJob': ('/v2/job/reconcile', 'file:write'),
+            '/createTagApprovalRequest': ('/v2/tag/approval-request', 'tag:create'),
+            '/requestOrCreateTag': ('/v2/tag/publish', 'tag:create'),
+            '/getTagPublicationStatus': ('/v2/tag/status', 'tag:create'),
+            '/buildProductionArtifact': ('/v2/release/artifact', 'tag:create'),
+            '/executeApprovedProductionRelease': ('/v2/release/execute', 'tag:create'),
+            '/getReleaseConsistency': ('/v2/release/consistency', 'tag:create'),
+            '/runReleaseAcceptance': ('/v2/release/acceptance', 'tag:create'),
+        }
+        if path in governed_routes:
+            route, scope = governed_routes[path]
+            actor = self.need(scope)
+            if not actor:
+                return
+            return self.bridge_v2_json(route, self.body(), method='POST')
+        if path in {
+            '/createOnlineAgentTask', '/registerOnlineAgent',
+            '/verifyOnlineAgentTask', '/cancelOnlineAgentTask',
+        }:
+            actor = self.need('agent-task:control')
+            if not actor:
+                return
+            data = self.body()
+            if path == '/createOnlineAgentTask':
+                return self.agent_task_control('POST', '/v1/tasks', data)
+            if path == '/registerOnlineAgent':
+                return self.agent_task_control('POST', '/v1/agents/register', data)
+            task_id = str(data.pop('task_id', '') or '')
+            if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{2,179}', task_id):
+                return self.sendj({'error': 'task_id_invalid'}, 400)
+            suffix = '/verify' if path == '/verifyOnlineAgentTask' else '/cancel'
+            return self.agent_task_control('POST', '/v1/tasks/' + task_id + suffix, data)
         if reject_non_read_http(self, 'POST', path): return
         data=self.body()
         try:
             if path == '/createReleaseCandidate':
-                a=self.need('release:candidate')
-                if not a: return
+                a = self.need(
+                    'release:candidate',
+                    projection_request={
+                        'source_commit': str(data.get('source_commit') or ''),
+                        'version': str(data.get('version') or ''),
+                    },
+                )
+                if not a:
+                    return
                 return self.sendj(
                     silent_release_service().create_release_candidate(data)
                 )
@@ -996,7 +1315,7 @@ def api_spec():
     return {
         'version': 'V2R2C1-IMMUTABLE-STAGING',
         'openapi': '/api/openapi.yaml',
-        'read': ['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],
+        'read': ['GET /api/caller-capabilities','GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],
         'write': ['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl','POST /createReleaseCandidate','POST /submitOwnerApproval','POST /startDeployment'],
         'policy': {
             'bearer_required_for_write': True,
@@ -1011,6 +1330,9 @@ def api_spec():
             'silent_release_shadow_default': True,
             'runtime_mode': 'IMMUTABLE_STAGING',
             'governed_release_routing': 'bridge-v2',
+            'caller_admission_projection': 'durable-owner-policy-server-side-ack',
+            'caller_projection_scope': SCOPE,
+            'caller_projection_caller': CALLER,
         },
     }
 

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

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

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

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

deploy/systemd/aios-gitops-model-api-owner-caller-projection.conf.example

@@ -0,0 +1,7 @@
+# Existing model API drop-in. Install only with the governed runtime package.
+[Service]
+SupplementaryGroups=gitops aios-executor-exchange aios-autonomous-agent-worker
+Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=/run/aios-owner-authorization-broker/broker.sock
+Environment=AIOS_OWNER_AUTH_BROKER_ACCOUNT=chatgpt-governed-implementer-singlehost
+Environment=AIOS_OWNER_AUTH_BROKER_CALLER=chatgpt-governed-implementer-singlehost
+Environment=AIOS_OWNER_AUTH_BROKER_SCOPE=release:candidate

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

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

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

@@ -1,46 +0,0 @@
-[Unit]
-Description=AIOS generic governed Online Agent Worker (2026-08-15)
-After=aios-agent-task-api.service aios-online-agent-worker-register-20260815.service
-Wants=aios-online-agent-worker-register-20260815.service
-
-[Service]
-Type=simple
-User=aios-engineering-worker
-Group=aios-engineering-worker
-SupplementaryGroups=gitops
-LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
-Environment=PYTHONUNBUFFERED=1
-Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
-Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
-Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-generic-20260815
-Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
-Environment=AIOS_AGENT_VERSION=2026-08-15.3-generic
-Environment=AIOS_AUTHORITY_GIT_DIR=/data/aios-gitops/repo.git
-Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-worker-20260815
-Environment=AIOS_WORKER_WORKSPACE_ROOT=/var/lib/aios-online-agent-worker-20260815/workspaces
-Environment=AIOS_WORKER_POLL_SECONDS=5
-Environment=AIOS_WORKER_HEARTBEAT_SECONDS=10
-Environment=AIOS_WORKER_LEASE_SECONDS=300
-ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-online-agent-worker-20260815/online_agent_worker.py
-Restart=always
-RestartSec=5
-TimeoutStopSec=15
-NoNewPrivileges=true
-PrivateTmp=true
-ProtectSystem=strict
-ProtectHome=true
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectControlGroups=true
-LockPersonality=true
-RestrictSUIDSGID=true
-RestrictRealtime=true
-RestrictNamespaces=true
-MemoryDenyWriteExecute=true
-RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
-ReadOnlyPaths=/data/aios-gitops/repo.git
-ReadWritePaths=/var/lib/aios-online-agent-worker-20260815
-UMask=0027
-
-[Install]
-WantedBy=multi-user.target

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

@@ -1,36 +0,0 @@
-[Unit]
-Description=AIOS generic Online Agent register-or-recover (2026-08-15)
-After=network-online.target aios-gitops-model-api.service aios-agent-task-api.service
-Wants=network-online.target
-Before=aios-online-agent-worker-20260815.service
-
-[Service]
-Type=oneshot
-RemainAfterExit=yes
-User=root
-Group=root
-LoadCredential=bridge_control_token:/run/credentials/aios-gitops-model-api.service/agent_task_control_token
-Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
-Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-generic-20260815
-Environment=AIOS_AGENT_TYPE=trusted-online-engineering-worker
-Environment=AIOS_AGENT_VERSION=2026-08-15.3-generic
-Environment=AIOS_WORKER_ENCRYPTED_CREDENTIAL=/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token
-Environment=AIOS_WORKER_REGISTRATION_STATE=/var/lib/aios-online-agent-worker-20260815/registration
-ExecStart=/usr/bin/python3 -B /usr/local/lib/aios-online-agent-worker-20260815/register.py
-NoNewPrivileges=true
-PrivateTmp=true
-ProtectSystem=strict
-ProtectHome=true
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectControlGroups=true
-LockPersonality=true
-RestrictSUIDSGID=true
-RestrictRealtime=true
-RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
-ReadWritePaths=/etc/credstore.encrypted /var/lib/aios-online-agent-worker-20260815
-UMask=0077
-TimeoutStartSec=120
-
-[Install]
-WantedBy=multi-user.target

deploy/systemd/aios-owner-durable-authorization-broker.service

@@ -0,0 +1,49 @@
+[Unit]
+Description=AIOS durable Owner authorization broker
+After=network.target aios-chatgpt-bridge-v2-staging.service
+Wants=aios-chatgpt-bridge-v2-staging.service
+
+[Service]
+Type=simple
+User=root
+Group=aios-executor-exchange
+RuntimeDirectory=aios-owner-authorization-broker
+RuntimeDirectoryMode=0750
+StateDirectory=aios-owner-authorization-broker
+StateDirectoryMode=0750
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY=/etc/aios-gitops/owner-durable-authorization.json
+Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_REVOCATIONS=/etc/aios-gitops/owner-durable-authorization-revocations.json
+Environment=AIOS_OWNER_AUTH_BROKER_ACCOUNT=chatgpt-governed-implementer-singlehost
+Environment=AIOS_OWNER_AUTH_BROKER_ALLOWED_USERS=aios-autonomous-agent-worker,aiman
+Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
+ExecStart=/usr/bin/python3 -B @RUNTIME_ROOT@/active-runtime/app/aios_durable_owner_authorization.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+LockPersonality=yes
+RestrictSUIDSGID=yes
+RestrictRealtime=yes
+RestrictNamespaces=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=127.0.0.0/8
+ReadOnlyPaths=/etc/aios-gitops/owner-durable-authorization.json
+ReadOnlyPaths=/etc/aios-gitops/owner-durable-authorization-revocations.json
+ReadOnlyPaths=@RUNTIME_ROOT@
+InaccessiblePaths=/data/aios-gitops /home
+UMask=0007
+
+[Install]
+WantedBy=multi-user.target

docs/gitops/caller-admission-projection.md

@@ -0,0 +1,88 @@
+# Caller admission projection and rehydration
+
+This runbook covers the exact durable Owner policy and the exact API caller:
+
+- caller: \`chatgpt-governed-implementer-singlehost\`
+- scope: \`release:candidate\`
+- policy source: \`/etc/aios-gitops/owner-durable-authorization.json\`
+- revocation source: \`/etc/aios-gitops/owner-durable-authorization-revocations.json\`
+
+## Admission source
+
+The model API first authenticates the existing caller token from the static
+registry. For this exact caller and exact release scope, static authentication
+is not sufficient. The model API calls the existing Unix-socket Owner broker.
+The broker reloads and validates the durable policy on every projection request,
+including account, capability, optional expiry, and revocation.
+
+Other callers retain their existing static-scope behavior. The target caller
+does not receive a permanent \`release:candidate\` entry in the static registry,
+and it is never treated as a release-worker account.
+
+## Projection lifecycle and ACK semantics
+
+The broker records only non-secret projection metadata in its existing durable
+state database. The state sequence is:
+
+\`PENDING\` -> \`ACKED\`
+
+The ACK is produced by the broker's server-side projection consumer after the
+exact policy and request binding have passed validation. The public response
+contains projection state, effective scope, policy digest, expiry and
+rehydration state. The short-lived credential material is returned only in a
+private Unix-socket response header to the model API process. It is not written
+to the response body, logs, diagnostics, Git, task metadata or terminal output.
+
+## Credential rehydration, expiry and revocation
+
+The credential projection is generated in memory for a short TTL and is
+discarded after the model API admission check. The durable row stores no raw
+credential and no bearer token. After a broker or model API restart, the next
+admission request creates a new short-lived credential from the durable
+projection metadata; this is reported as \`RECOVERED\`.
+
+Every request reloads the policy. Missing, malformed, expired or revoked policy
+fails closed. A caller mismatch, scope mismatch, invalid projection state,
+broker outage, missing credential header or unexpected secret field also fails
+closed. No static scope fallback is allowed for the target caller.
+
+## Diagnosis
+
+Use the read-only API with the caller's existing credential:
+
+\`GET /api/caller-capabilities\`
+
+The response is safe to display and contains:
+
+- \`caller_identity\`
+- \`admitted\`
+- \`effective_scopes\`
+- \`projection_state\`
+- \`credential_state\`
+- \`rehydration_state\`
+- \`secret_exposed=false\`
+
+It never contains a bearer, token, credential value or credential hash.
+
+For a release candidate request, the model API also binds the exact
+\`source_commit\` and \`version\` to the durable policy before invoking the
+trusted release service. The caller cannot supply release evidence fields.
+
+## Failure states and troubleshooting
+
+- \`PROJECTION_BROKER_UNAVAILABLE\`: verify the broker unit, socket path and
+  the model API supplementary groups; do not add a static scope.
+- \`OWNER_GRANT_REVOKED\` or \`POLICY_EXPIRED\`: restore a valid Owner policy
+  through governed policy management; do not edit broker state.
+- \`CALLER_MISMATCH\` or \`SCOPE_MISMATCH\`: verify the authenticated account and
+  requested exact scope.
+- \`PROJECTION_STATE_INVALID\`: inspect the durable projection through the
+  broker's read-only diagnostics and fail closed until it is reconciled.
+- \`CREDENTIAL_REHYDRATION_MISSING\`: restart only the related control-plane
+  service after capturing pre-state and then repeat the read-only diagnosis.
+- \`trusted candidate builder is not configured\`: Gate A admission is working,
+  but the separate existing TrustedCandidateBuilder runtime must be restored
+  before Gate B can create a candidate.
+
+No Owner Approval, Deployment, Host Seed, Host Activation, Projector,
+continuation task, Tag or Baseline action is part of this path.

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

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

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

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

runtime/agent-observability/records/exec_d75446452723.json

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

runtime/task-center/tasks.jsonl

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

scripts/agent_task_client.py

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

scripts/agent_task_online_smoke.py

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

scripts/build_agent_task_api.py

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

services/aios-online-agent-worker/README.md

@@ -1,54 +0,0 @@
-# Generic Governed Online Agent Worker
-
-This worker is a pull-based, restartable consumer for `aios.agent-task.v1`.
-The service configuration contains only stable runtime identity, capability
-names, API endpoints, heartbeat timing, and the credential projection. It does
-not contain a task id, repository base OID, target branch, or plan-file path.
-
-## Dynamic task contract
-
-Engineering tasks carry `metadata.execution` with schema
-`aios.online-agent.execution.v2`. The payload supplies the repository, exact
-base OID, target branch, detached-worktree policy, adapter name, explicit file
-change set, verification commands, and governed commit idempotency keys.
-Task prose is descriptive only and is never passed to a shell.
-
-The worker accepts only structured operations:
-
-- `deterministic`: allow-listed file changes and test commands;
-- `codex`: fail-closed until a task supplies an explicit Codex file/command
-  policy;
-- `hermes`: fail-closed because the current Hermes runtime is a user-chat
-  service, not an engineering executor.
-
-All command execution uses `shell=False`, fixed program allowlists, relative
-paths beneath an ephemeral detached worktree, and output redaction. Direct
-`git branch`, `update-ref`, `push`, `commit`, `reset`, `clean`, main,
-production, tag, and baseline mutations are rejected by the worker.
-
-## Governed write path
-
-The worker calls the Bridge in this order:
-
-1. `POST /v2/governed-task`;
-2. `POST /v2/writer-lease`;
-3. `POST /v2/commit-files`;
-4. read-only authority-ref visibility verification.
-
-CREATE_NEW is represented by the task payload and is normalized by the Bridge
-lease authority. The worker never creates an authoritative ref itself.
-
-## Lifecycle and recovery
-
-The runtime registers with `register_or_recover`, receives its agent token
-once, and stores only the encrypted systemd credential. The consumer loops over
-poll, atomically claims one task, sends the first heartbeat, emits
-`TASK_STARTED`, and continues heartbeats while executing.
-
-State is atomically stored per task without agent or lease token material. If
-the process stops after a governed commit but before result submission, the
-next lease attempt reuses the stable governed idempotency key and submits the
-stored commit evidence without issuing a second commit.
-
-The control plane must independently verify submitted results; worker-local
-success is not production or mainline authorization.

services/aios-online-agent-worker/aios_online_agent_worker/__init__.py

@@ -1,3 +0,0 @@
-"""Generic governed Online Agent Worker runtime."""
-
-__all__ = ["worker"]

services/aios-online-agent-worker/aios_online_agent_worker/worker.py

@@ -1,937 +0,0 @@
-#!/usr/bin/env python3
-"""Generic, restartable Online Agent Worker.
-
-The worker is deliberately task-agnostic.  A task's signed/durable payload is
-the only source of repository, branch, change, verification, and governance
-parameters.  There is no task id, base OID, branch name, or plan file in the
-runtime configuration.
-
-The worker performs local work only in an ephemeral detached worktree.  All
-authoritative ref updates go through the governed Bridge API; direct branch,
-update-ref, push, main, production, tag, and baseline mutations are rejected.
-"""
-
-from __future__ import annotations
-
-import base64
-from dataclasses import dataclass
-from datetime import datetime, timezone
-from abc import ABC, abstractmethod
-import hashlib
-import json
-import os
-from pathlib import Path
-import re
-import shutil
-import subprocess
-import tempfile
-import threading
-import time
-import urllib.error
-import urllib.request
-from typing import Any, Mapping
-from urllib.parse import unquote
-
-
-API_BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
-BRIDGE_BASE = os.environ.get("AIOS_GOVERNED_WRITE_API_URL", "http://127.0.0.1:18892").rstrip("/")
-AGENT_ID = os.environ.get(
-    "AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-generic-20260815"
-)
-AGENT_TYPE = os.environ.get("AIOS_AGENT_TYPE", "trusted-online-engineering-worker")
-AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.3-generic")
-AUTHORITY_GIT_DIR = Path(
-    os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git")
-)
-STATE_ROOT = Path(
-    os.environ.get("AIOS_WORKER_STATE_ROOT", "/var/lib/aios-online-agent-worker-20260815")
-)
-WORKSPACE_ROOT = Path(
-    os.environ.get(
-        "AIOS_WORKER_WORKSPACE_ROOT", "/var/lib/aios-online-agent-worker-20260815/workspaces"
-    )
-)
-GIT_CONFIG_GLOBAL = STATE_ROOT / "gitconfig"
-POLL_SECONDS = int(os.environ.get("AIOS_WORKER_POLL_SECONDS", "5"))
-HEARTBEAT_SECONDS = int(os.environ.get("AIOS_WORKER_HEARTBEAT_SECONDS", "20"))
-LEASE_SECONDS = int(os.environ.get("AIOS_WORKER_LEASE_SECONDS", "300"))
-MAX_TASK_BYTES = 512 * 1024
-MAX_FILE_BYTES = 2 * 1024 * 1024
-OID_RE = re.compile(r"^[0-9a-f]{40}$")
-SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
-ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
-BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
-PATH_RE = re.compile(r"^[^\\\x00-\x1f\x7f]+$")
-SENSITIVE_RE = re.compile(
-    r"(?i)(authorization\s*:\s*bearer\s+\S+|(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*\S+)"
-)
-
-
-class WorkerError(RuntimeError):
-    def __init__(self, code: str, *, phase: str, details: Mapping[str, Any] | None = None) -> None:
-        super().__init__(code)
-        self.code = code
-        self.phase = phase
-        self.details = dict(details or {})
-
-
-def now() -> str:
-    return datetime.now(timezone.utc).isoformat()
-
-
-def digest_bytes(value: bytes) -> str:
-    return hashlib.sha256(value).hexdigest()
-
-
-def digest_json(value: Any) -> str:
-    return digest_bytes(
-        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
-    )
-
-
-def sanitize(value: Any) -> Any:
-    if isinstance(value, Mapping):
-        cleaned: dict[str, Any] = {}
-        for key, inner in value.items():
-            lowered = str(key).lower()
-            if any(marker in lowered for marker in ("token", "secret", "credential", "password", "authorization")):
-                cleaned[str(key)] = "[REDACTED]"
-            else:
-                cleaned[str(key)] = sanitize(inner)
-        return cleaned
-    if isinstance(value, list):
-        return [sanitize(item) for item in value]
-    if isinstance(value, tuple):
-        return [sanitize(item) for item in value]
-    if isinstance(value, str):
-        return SENSITIVE_RE.sub("[REDACTED]", value)
-    return value
-
-
-def excerpt(value: str, limit: int = 800) -> str:
-    value = str(sanitize(value))
-    return value if len(value) <= limit else value[: limit // 2] + "...[TRUNCATED]..." + value[-limit // 2 :]
-
-
-def audit(event_type: str, **fields: Any) -> None:
-    STATE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
-    path = STATE_ROOT / "events.jsonl"
-    record = {
-        "event_type": event_type,
-        "created_at": now(),
-        "agent_id": AGENT_ID,
-        "credential_material_exposed": False,
-        "lease_material_exposed": False,
-        **sanitize(fields),
-    }
-    with path.open("a", encoding="utf-8") as handle:
-        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
-    os.chmod(path, 0o640)
-    print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
-
-
-def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o640) -> None:
-    path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
-    if path.parent.is_symlink() or path.is_symlink():
-        raise WorkerError("STATE_PATH_SYMLINK_REJECTED", phase="state")
-    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
-    try:
-        os.fchmod(fd, mode)
-        with os.fdopen(fd, "w", encoding="utf-8") as handle:
-            json.dump(sanitize(dict(value)), handle, ensure_ascii=False, sort_keys=True)
-            handle.write("\n")
-            handle.flush()
-            os.fsync(handle.fileno())
-        os.replace(temporary, path)
-        os.chmod(path, mode)
-    finally:
-        if os.path.exists(temporary):
-            os.unlink(temporary)
-
-
-def safe_relative_path(value: Any, *, field: str = "path") -> str:
-    if not isinstance(value, str) or not value or len(value) > 240:
-        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
-    if unquote(value) != value or not PATH_RE.fullmatch(value):
-        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
-    if value.startswith(("/", "~")) or re.match(r"^[A-Za-z]:", value):
-        raise WorkerError(f"{field.upper()}_ABSOLUTE", phase="validation")
-    parts = value.replace("\\", "/").split("/")
-    if any(part in {"", ".", ".."} for part in parts):
-        raise WorkerError(f"{field.upper()}_TRAVERSAL", phase="validation")
-    return "/".join(parts)
-
-
-def resolve_under(root: Path, relative: str, *, write: bool = False) -> Path:
-    relative = safe_relative_path(relative)
-    root = root.resolve()
-    candidate = (root / relative).resolve(strict=False)
-    if candidate != root and root not in candidate.parents:
-        raise WorkerError("WORKSPACE_ESCAPE", phase="validation")
-    current = root
-    for part in Path(relative).parts:
-        current /= part
-        if current.exists() and current.is_symlink():
-            raise WorkerError("WORKSPACE_SYMLINK_REJECTED", phase="validation")
-    if write and candidate.exists() and candidate.is_symlink():
-        raise WorkerError("WRITE_SYMLINK_REJECTED", phase="validation")
-    return candidate
-
-
-def validate_oid(value: Any, field: str = "oid") -> str:
-    if not isinstance(value, str) or not OID_RE.fullmatch(value):
-        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
-    return value
-
-
-def validate_sha(value: Any, field: str = "sha256") -> str:
-    if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
-        raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
-    return value
-
-
-def validate_branch(value: Any) -> str:
-    if not isinstance(value, str) or not 3 <= len(value) <= 180:
-        raise WorkerError("TARGET_BRANCH_INVALID", phase="validation")
-    if value in {"main", "master", "HEAD"} or value.startswith(
-        ("main/", "master/", "refs/", "tags/", "tag/", "release/", "releases/", "baseline/", "production/")
-    ):
-        raise WorkerError("PROTECTED_BRANCH_REJECTED", phase="validation")
-    if (
-        value.startswith("/")
-        or value.endswith(("/", ".", ".lock"))
-        or ".." in value
-        or "//" in value
-        or "@{" in value
-        or any(ch in value for ch in " ~^:?*[\\")
-        or any(part in {"", ".", ".."} for part in value.split("/"))
-        or not BRANCH_RE.fullmatch(value)
-    ):
-        raise WorkerError("TARGET_BRANCH_INVALID", phase="validation")
-    return value
-
-
-def validate_program(argv: Any, *, internal_git: bool = False) -> list[str]:
-    if not isinstance(argv, list) or not argv or not all(isinstance(item, str) and item for item in argv):
-        raise WorkerError("STRUCTURED_ARGV_REQUIRED", phase="validation")
-    command = argv[0]
-    if command not in {"/usr/bin/git", "/usr/bin/python3"}:
-        raise WorkerError("PROGRAM_NOT_ALLOWLISTED", phase="validation", details={"program": Path(command).name})
-    if command.endswith("python3"):
-        if any(item in {"-c", "-", "--command"} for item in argv):
-            raise WorkerError("PYTHON_INLINE_CODE_REJECTED", phase="validation")
-        if "-m" in argv:
-            index = argv.index("-m")
-            if index + 1 >= len(argv) or argv[index + 1] not in {"pytest", "py_compile"}:
-                raise WorkerError("PYTHON_MODULE_NOT_ALLOWLISTED", phase="validation")
-    if command.endswith("git") and not internal_git:
-        rejected = {"push", "update-ref", "receive-pack", "send-pack", "reset", "clean", "branch", "commit", "merge", "rebase"}
-        if any(item in rejected for item in argv[1:]):
-            raise WorkerError("GIT_MUTATION_NOT_ALLOWLISTED", phase="validation")
-    return list(argv)
-
-
-def run_command(
-    label: str,
-    argv: list[str],
-    *,
-    cwd: Path,
-    timeout: int,
-    internal_git: bool = False,
-    allowed_returncodes: tuple[int, ...] = (0,),
-) -> dict[str, Any]:
-    validate_program(argv, internal_git=internal_git)
-    if not 1 <= timeout <= 3600:
-        raise WorkerError("COMMAND_TIMEOUT_INVALID", phase=label)
-    started = time.monotonic()
-    try:
-        completed = subprocess.run(
-            argv,
-            cwd=str(cwd),
-            text=True,
-            capture_output=True,
-            check=False,
-            shell=False,
-            timeout=timeout,
-            env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)},
-        )
-    except subprocess.TimeoutExpired as exc:
-        raise WorkerError("COMMAND_TIMEOUT", phase=label, details={"timeout_seconds": timeout}) from exc
-    result = {
-        "label": label,
-        "argv": [Path(item).name if index == 0 else item for index, item in enumerate(argv)],
-        "exit_code": completed.returncode,
-        "stdout_sha256": digest_bytes(completed.stdout.encode("utf-8", "replace")),
-        "stderr_sha256": digest_bytes(completed.stderr.encode("utf-8", "replace")),
-        "stdout_excerpt": excerpt(completed.stdout),
-        "stderr_excerpt": excerpt(completed.stderr),
-        "duration_ms": int((time.monotonic() - started) * 1000),
-    }
-    audit("COMMAND_RESULT", **result)
-    if completed.returncode not in allowed_returncodes:
-        raise WorkerError("COMMAND_FAILED", phase=label, details={"exit_code": completed.returncode, "stderr": excerpt(completed.stderr, 400)})
-    return result
-
-
-@dataclass(frozen=True)
-class ExecutionEnvelope:
-    task_id: str
-    repository: str
-    base_oid: str
-    target_branch: str
-    adapter: str
-    change_set: tuple[dict[str, Any], ...]
-    verification: tuple[dict[str, Any], ...]
-    governance: dict[str, Any]
-    workspace: dict[str, Any]
-    digest: str
-
-    @classmethod
-    def from_task(cls, task: Mapping[str, Any]) -> "ExecutionEnvelope":
-        task_id = str(task.get("task_id") or "")
-        if not ID_RE.fullmatch(task_id):
-            raise WorkerError("TASK_ID_INVALID", phase="validation")
-        metadata = task.get("metadata")
-        if not isinstance(metadata, Mapping):
-            raise WorkerError("TASK_METADATA_REQUIRED", phase="validation")
-        execution = metadata.get("execution")
-        if not isinstance(execution, Mapping):
-            raise WorkerError("DYNAMIC_EXECUTION_PAYLOAD_REQUIRED", phase="validation")
-        if execution.get("schema") != "aios.online-agent.execution.v2":
-            raise WorkerError("EXECUTION_SCHEMA_INVALID", phase="validation")
-        repository = execution.get("repository")
-        if not isinstance(repository, str) or not repository:
-            raise WorkerError("REPOSITORY_INVALID", phase="validation")
-        base_oid = validate_oid(execution.get("base_oid"), "base_oid")
-        target_branch = validate_branch(execution.get("target_branch"))
-        workspace = execution.get("workspace")
-        if not isinstance(workspace, Mapping):
-            raise WorkerError("WORKSPACE_POLICY_REQUIRED", phase="validation")
-        workspace_value = dict(workspace)
-        if workspace_value.get("authority_git_dir") != str(AUTHORITY_GIT_DIR):
-            raise WorkerError("AUTHORITY_ROOT_INVALID", phase="validation")
-        if workspace_value.get("read_only_repository") is not True:
-            raise WorkerError("REPOSITORY_WRITE_POLICY_INVALID", phase="validation")
-        if workspace_value.get("cleanup_policy") not in {"always", "cleanup_on_success"}:
-            raise WorkerError("WORKSPACE_CLEANUP_POLICY_INVALID", phase="validation")
-        adapter_value = execution.get("adapter")
-        if isinstance(adapter_value, Mapping):
-            adapter = str(adapter_value.get("name") or "")
-        else:
-            adapter = str(adapter_value or "")
-        if adapter not in {"deterministic", "codex", "hermes"}:
-            raise WorkerError("RUNTIME_ADAPTER_UNSUPPORTED", phase="validation")
-        change_set = execution.get("change_set")
-        if not isinstance(change_set, list) or not change_set or len(change_set) > 128:
-            raise WorkerError("CHANGE_SET_INVALID", phase="validation")
-        normalized_changes: list[dict[str, Any]] = []
-        for item in change_set:
-            if not isinstance(item, Mapping):
-                raise WorkerError("CHANGE_OPERATION_INVALID", phase="validation")
-            operation = str(item.get("operation") or "")
-            relative = safe_relative_path(item.get("path"))
-            normalized = dict(item)
-            normalized["operation"] = operation
-            normalized["path"] = relative
-            if operation == "write_file":
-                content = item.get("content")
-                if not isinstance(content, str) or len(content.encode("utf-8")) > MAX_FILE_BYTES:
-                    raise WorkerError("WRITE_CONTENT_INVALID", phase="validation")
-            elif operation == "replace_text":
-                if not isinstance(item.get("old"), str) or not item.get("old") or not isinstance(item.get("new"), str):
-                    raise WorkerError("REPLACE_TEXT_INVALID", phase="validation")
-                if len(str(item["old"]).encode("utf-8")) > MAX_FILE_BYTES:
-                    raise WorkerError("REPLACE_TEXT_INVALID", phase="validation")
-            elif operation == "delete_file":
-                pass
-            else:
-                raise WorkerError("CHANGE_OPERATION_UNSUPPORTED", phase="validation", details={"operation": operation})
-            expected = item.get("expected_sha256")
-            if expected is not None:
-                validate_sha(expected, "expected_sha256")
-            normalized_changes.append(normalized)
-        verification = execution.get("verification") or []
-        if not isinstance(verification, list) or len(verification) > 128:
-            raise WorkerError("VERIFICATION_INVALID", phase="validation")
-        normalized_verification: list[dict[str, Any]] = []
-        for item in verification:
-            if not isinstance(item, Mapping):
-                raise WorkerError("VERIFICATION_COMMAND_INVALID", phase="validation")
-            label = str(item.get("label") or "verification")
-            argv = validate_program(item.get("argv"))
-            cwd_value = item.get("cwd")
-            cwd = "." if cwd_value in {None, "", "."} else safe_relative_path(cwd_value, field="cwd")
-            timeout = item.get("timeout_seconds", 120)
-            if not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 <= timeout <= 3600:
-                raise WorkerError("VERIFICATION_TIMEOUT_INVALID", phase="validation")
-            normalized_verification.append({"label": label, "argv": argv, "cwd": cwd, "timeout_seconds": timeout})
-        governance = execution.get("governance")
-        if not isinstance(governance, Mapping):
-            raise WorkerError("GOVERNANCE_POLICY_REQUIRED", phase="validation")
-        governance_value = dict(governance)
-        governed_task_id = str(governance_value.get("task_id") or f"{task_id}-writer")
-        if not ID_RE.fullmatch(governed_task_id):
-            raise WorkerError("GOVERNED_TASK_ID_INVALID", phase="validation")
-        idempotency_key = str(governance_value.get("idempotency_key") or f"{task_id}-governed-v1")
-        commit_key = str(governance_value.get("commit_idempotency_key") or f"{task_id}-commit-v1")
-        for field, value in (("idempotency_key", idempotency_key), ("commit_idempotency_key", commit_key)):
-            if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,159}", value):
-                raise WorkerError(f"{field.upper()}_INVALID", phase="validation")
-        governance_value.update(
-            {
-                "task_id": governed_task_id,
-                "idempotency_key": idempotency_key,
-                "commit_idempotency_key": commit_key,
-                "message": str(governance_value.get("message") or f"chore(gitops): execute task {task_id}"),
-            }
-        )
-        for field in ("main_write", "force_push", "production_write", "tag_write", "baseline_write", "protected_ref_bypass"):
-            if governance_value.get(field, False) is not False:
-                raise WorkerError("GOVERNANCE_POLICY_REJECTED", phase="validation", details={"field": field})
-        allowed_files = governance_value.get("allowed_files")
-        if allowed_files is not None:
-            if not isinstance(allowed_files, list):
-                raise WorkerError("ALLOWED_FILES_INVALID", phase="validation")
-            allowed = {safe_relative_path(item, field="allowed_file") for item in allowed_files}
-            if not {item["path"] for item in normalized_changes}.issubset(allowed):
-                raise WorkerError("CHANGE_OUTSIDE_ALLOWED_FILES", phase="validation")
-            governance_value["allowed_files"] = sorted(allowed)
-        return cls(
-            task_id=task_id,
-            repository=repository,
-            base_oid=base_oid,
-            target_branch=target_branch,
-            adapter=adapter,
-            change_set=tuple(normalized_changes),
-            verification=tuple(normalized_verification),
-            governance=governance_value,
-            workspace=workspace_value,
-            digest=digest_json(execution),
-        )
-
-
-class RuntimeAdapter(ABC):
-    name: str
-
-    @abstractmethod
-    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
-        raise NotImplementedError
-
-
-class DeterministicExecutor(RuntimeAdapter):
-    name = "deterministic"
-
-    @staticmethod
-    def _write_atomic(path: Path, content: str) -> None:
-        path.parent.mkdir(mode=0o750, parents=True, exist_ok=True)
-        fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
-        try:
-            with os.fdopen(fd, "w", encoding="utf-8") as handle:
-                handle.write(content)
-                handle.flush()
-                os.fsync(handle.fileno())
-            os.replace(temporary, path)
-        finally:
-            if os.path.exists(temporary):
-                os.unlink(temporary)
-
-    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
-        changed: set[str] = set()
-        for change in envelope.change_set:
-            relative = str(change["path"])
-            path = resolve_under(worktree, relative, write=True)
-            expected = change.get("expected_sha256")
-            current = path.read_bytes() if path.is_file() else None
-            if expected is not None and (current is None or digest_bytes(current) != expected):
-                raise WorkerError("WORKTREE_SOURCE_DRIFT", phase="implementing", details={"path": relative})
-            operation = change["operation"]
-            if operation == "write_file":
-                self._write_atomic(path, str(change["content"]))
-            elif operation == "replace_text":
-                if current is None:
-                    raise WorkerError("REPLACE_SOURCE_MISSING", phase="implementing", details={"path": relative})
-                original = current.decode("utf-8")
-                old = str(change["old"])
-                count = original.count(old)
-                expected_count = int(change.get("count", 1))
-                if count != expected_count:
-                    raise WorkerError("REPLACE_TEXT_MATCH_COUNT_INVALID", phase="implementing", details={"path": relative, "count": count})
-                self._write_atomic(path, original.replace(old, str(change["new"])))
-            elif operation == "delete_file":
-                if path.exists():
-                    path.unlink()
-            changed.add(relative)
-            audit("REPOSITORY_CHANGE_APPLIED", operation=operation, path=relative, sha256=digest_bytes(path.read_bytes()) if path.exists() else None)
-        commands: list[dict[str, Any]] = []
-        if changed:
-            commands.append(run_command("git_intent_to_add", ["/usr/bin/git", "-C", str(worktree), "add", "-N", "--", *sorted(changed)], cwd=worktree, timeout=120, internal_git=True))
-        commands.append(run_command("git_diff_check", ["/usr/bin/git", "-C", str(worktree), "diff", "--check"], cwd=worktree, timeout=120))
-        commands.append(run_command("git_diff_stat", ["/usr/bin/git", "-C", str(worktree), "diff", "--no-ext-diff", "--stat"], cwd=worktree, timeout=120))
-        tests: list[dict[str, Any]] = []
-        for item in envelope.verification:
-            cwd = worktree if item["cwd"] == "." else resolve_under(worktree, item["cwd"])
-            tests.append(run_command(item["label"], item["argv"], cwd=cwd, timeout=item["timeout_seconds"]))
-        return {
-            "changed_files": sorted(changed),
-            "commands": commands,
-            "tests": tests,
-            "adapter": self.name,
-        }
-
-
-class CodexExecutor(RuntimeAdapter):
-    name = "codex"
-
-    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
-        raise WorkerError(
-            "CODEX_ADAPTER_REQUIRES_EXPLICIT_POLICY",
-            phase="adapter",
-            details={"reason": "Codex is not allowed to mutate a governed worktree without an explicit file/command policy"},
-        )
-
-
-class HermesExecutor(RuntimeAdapter):
-    name = "hermes"
-
-    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
-        raise WorkerError(
-            "HERMES_ENGINEERING_ADAPTER_UNAVAILABLE",
-            phase="adapter",
-            details={"reason": "The live Hermes service is a user-chat runtime, not an engineering executor"},
-        )
-
-
-def adapter_for(name: str) -> RuntimeAdapter:
-    return {"deterministic": DeterministicExecutor, "codex": CodexExecutor, "hermes": HermesExecutor}[name]()
-
-
-class HttpClient:
-    @staticmethod
-    def call(base: str, method: str, route: str, token: str | None, body: Mapping[str, Any] | None = None, timeout: int = 30) -> tuple[int, dict[str, Any]]:
-        payload = None if body is None else json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
-        headers = {"Accept": "application/json"}
-        if token:
-            headers["Authorization"] = "Bearer " + token
-        if payload is not None:
-            headers["Content-Type"] = "application/json"
-        request = urllib.request.Request(base + route, data=payload, headers=headers, method=method)
-        try:
-            with urllib.request.urlopen(request, timeout=timeout) as response:
-                value = json.loads(response.read().decode("utf-8"))
-                return response.status, value if isinstance(value, dict) else {}
-        except urllib.error.HTTPError as exc:
-            raw = exc.read(8192).decode("utf-8", "replace")
-            try:
-                value = json.loads(raw)
-            except json.JSONDecodeError:
-                value = {"error": "HTTP_ERROR"}
-            return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
-        except (urllib.error.URLError, TimeoutError):
-            return 599, {"error": "SERVICE_UNAVAILABLE"}
-
-
-class AgentTaskClient:
-    def __init__(self, token: str) -> None:
-        self.token = token
-
-    def poll(self) -> list[dict[str, Any]]:
-        status, response = HttpClient.call(API_BASE, "POST", "/v1/tasks/poll", self.token, {"limit": 100})
-        if status != 200:
-            raise WorkerError("TASK_POLL_FAILED", phase="poll", details={"api_status": status, "error": response.get("error")})
-        tasks = response.get("tasks")
-        return [dict(item) for item in tasks if isinstance(item, Mapping)] if isinstance(tasks, list) else []
-
-    def claim(self, task_id: str) -> tuple[int, dict[str, Any]]:
-        return HttpClient.call(API_BASE, "POST", f"/v1/tasks/{task_id}/claim", self.token, {"lease_seconds": LEASE_SECONDS})
-
-    def heartbeat(self, task_id: str, lease_token: str, stage: str) -> tuple[int, dict[str, Any]]:
-        return HttpClient.call(
-            API_BASE,
-            "POST",
-            f"/v1/tasks/{task_id}/heartbeat",
-            self.token,
-            {"lease_token": lease_token, "lease_seconds": LEASE_SECONDS, "progress": {"stage": stage, "worker_version": AGENT_VERSION}},
-        )
-
-    def result(self, task_id: str, lease_token: str, idempotency_key: str, result: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
-        return HttpClient.call(
-            API_BASE,
-            "POST",
-            f"/v1/tasks/{task_id}/result",
-            self.token,
-            {"lease_token": lease_token, "idempotency_key": idempotency_key, "result": dict(result)},
-        )
-
-
-class BridgeClient:
-    def call(self, route: str, body: Mapping[str, Any]) -> tuple[int, dict[str, Any]]:
-        return HttpClient.call(BRIDGE_BASE, "POST", route, None, body, timeout=60)
-
-
-class LeaseHeartbeat:
-    def __init__(self, client: AgentTaskClient, task_id: str, lease_token: str) -> None:
-        self.client = client
-        self.task_id = task_id
-        self.lease_token = lease_token
-        self.stage = "claimed"
-        self.failed: str | None = None
-        self.stop_event = threading.Event()
-        self.lock = threading.Lock()
-        self.thread = threading.Thread(target=self._run, name="online-agent-heartbeat", daemon=True)
-
-    def first(self) -> None:
-        status, response = self.client.heartbeat(self.task_id, self.lease_token, self.stage)
-        if not 200 <= status < 300:
-            self.failed = str(response.get("error") or status)
-            raise WorkerError("TASK_HEARTBEAT_FAILED", phase="heartbeat", details={"api_status": status, "error": self.failed})
-        audit("TASK_HEARTBEAT", task_id=self.task_id, stage=self.stage, api_status=status)
-
-    def start(self) -> None:
-        self.thread.start()
-
-    def set_stage(self, stage: str) -> None:
-        with self.lock:
-            self.stage = stage
-
-    def _run(self) -> None:
-        while not self.stop_event.wait(HEARTBEAT_SECONDS):
-            with self.lock:
-                stage = self.stage
-            try:
-                status, response = self.client.heartbeat(self.task_id, self.lease_token, stage)
-                if not 200 <= status < 300:
-                    self.failed = str(response.get("error") or status)
-                    audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, api_status=status, error=self.failed)
-                else:
-                    audit("TASK_HEARTBEAT", task_id=self.task_id, stage=stage, api_status=status)
-            except Exception as exc:  # pragma: no cover - defensive runtime guard
-                self.failed = type(exc).__name__
-                audit("TASK_HEARTBEAT_FAILED", task_id=self.task_id, stage=stage, error=self.failed)
-
-    def ensure(self) -> None:
-        if self.failed:
-            raise WorkerError("TASK_LEASE_LOST", phase="heartbeat", details={"error": self.failed})
-
-    def close(self) -> None:
-        self.stop_event.set()
-        if self.thread.ident is not None:
-            self.thread.join(timeout=2)
-
-
-def load_agent_token() -> str:
-    directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
-    path = directory / "agent_token"
-    if not path.is_file() or path.is_symlink():
-        raise WorkerError("AGENT_RUNTIME_CREDENTIAL_UNAVAILABLE", phase="startup")
-    value = path.read_text(encoding="utf-8").strip()
-    if len(value) < 24:
-        raise WorkerError("AGENT_RUNTIME_CREDENTIAL_INVALID", phase="startup")
-    return value
-
-
-def ensure_roots() -> None:
-    if AUTHORITY_GIT_DIR.is_symlink() or STATE_ROOT.is_symlink() or WORKSPACE_ROOT.is_symlink():
-        raise WorkerError("ROOT_SYMLINK_REJECTED", phase="startup")
-    if not AUTHORITY_GIT_DIR.is_dir():
-        raise WorkerError("AUTHORITY_GIT_DIR_UNAVAILABLE", phase="startup")
-    STATE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
-    WORKSPACE_ROOT.mkdir(mode=0o750, parents=True, exist_ok=True)
-    if GIT_CONFIG_GLOBAL.is_symlink():
-        raise WorkerError("GIT_CONFIG_SYMLINK_REJECTED", phase="startup")
-    expected = f"[safe]\n\tdirectory = {AUTHORITY_GIT_DIR}\n"
-    if GIT_CONFIG_GLOBAL.exists() and GIT_CONFIG_GLOBAL.read_text(encoding="utf-8") != expected:
-        raise WorkerError("GIT_CONFIG_BINDING_CONFLICT", phase="startup")
-    if not GIT_CONFIG_GLOBAL.exists():
-        GIT_CONFIG_GLOBAL.write_text(expected, encoding="utf-8")
-        os.chmod(GIT_CONFIG_GLOBAL, 0o640)
-
-
-def task_state_path(task_id: str) -> Path:
-    if not ID_RE.fullmatch(task_id):
-        raise WorkerError("TASK_ID_INVALID", phase="state")
-    return STATE_ROOT / "tasks" / f"{task_id}.json"
-
-
-def load_task_state(task_id: str, execution_digest: str) -> dict[str, Any] | None:
-    path = task_state_path(task_id)
-    if not path.is_file() or path.is_symlink():
-        return None
-    try:
-        value = json.loads(path.read_text(encoding="utf-8"))
-    except Exception:
-        raise WorkerError("TASK_STATE_INVALID", phase="state")
-    if not isinstance(value, dict):
-        raise WorkerError("TASK_STATE_INVALID", phase="state")
-    if value.get("execution_digest") not in {None, execution_digest}:
-        raise WorkerError("TASK_PAYLOAD_DRIFT", phase="recovery")
-    return value
-
-
-def save_task_state(task_id: str, value: Mapping[str, Any]) -> None:
-    atomic_json(task_state_path(task_id), value)
-
-
-def new_run_root() -> Path:
-    ensure_roots()
-    root = Path(tempfile.mkdtemp(prefix="task-", dir=str(WORKSPACE_ROOT)))
-    os.chmod(root, 0o750)
-    return root
-
-
-def cleanup_run_root(path: Path) -> None:
-    root = WORKSPACE_ROOT.resolve()
-    candidate = path.resolve(strict=False)
-    if candidate == root or root not in candidate.parents:
-        raise WorkerError("CLEANUP_SCOPE_INVALID", phase="cleanup")
-    if candidate.exists() or candidate.is_symlink():
-        shutil.rmtree(candidate)
-
-
-def prepare_worktree(run_root: Path, base_oid: str) -> Path:
-    clone = run_root / "repository"
-    worktree = run_root / "worktree"
-    run_command("workspace_clone", ["/usr/bin/git", "clone", "--no-local", "--no-hardlinks", str(AUTHORITY_GIT_DIR), str(clone)], cwd=run_root, timeout=300, internal_git=True)
-    run_command("workspace_base_probe", ["/usr/bin/git", "-C", str(clone), "rev-parse", "--verify", base_oid + "^{commit}"], cwd=run_root, timeout=60)
-    run_command("workspace_worktree_create", ["/usr/bin/git", "-C", str(clone), "worktree", "add", "--detach", str(worktree), base_oid], cwd=run_root, timeout=300, internal_git=True)
-    return worktree
-
-
-def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeClient, heartbeat: LeaseHeartbeat) -> dict[str, Any]:
-    governance = envelope.governance
-    create_body = {
-        "task_id": governance["task_id"],
-        "repository": envelope.repository,
-        "base_oid": envelope.base_oid,
-        "target_branch": envelope.target_branch,
-        "risk": str(governance.get("risk") or "engineering"),
-        "description": str(governance.get("description") or f"Governed execution for {envelope.task_id}"),
-        "idempotency_key": governance["idempotency_key"],
-    }
-    status, response = bridge.call("/v2/governed-task", create_body)
-    if not 200 <= status < 300:
-        raise WorkerError("GOVERNED_TASK_CREATE_FAILED", phase="committing", details={"api_status": status, "error": response.get("error")})
-    generation = int(response.get("generation") or 0)
-    expected_branch_oid = str(response.get("expected_branch_oid") or envelope.base_oid)
-    branch_mode = str(response.get("branch_mode") or "CREATE_NEW")
-    lease_body = {
-        "schema": "aios.task-writer-lease.v2",
-        "task_id": governance["task_id"],
-        "repository": envelope.repository,
-        "branch_mode": branch_mode,
-        "base_oid": envelope.base_oid,
-        "expected_branch_oid": expected_branch_oid,
-        "target_branch": envelope.target_branch,
-        "generation": generation,
-        "lease_seconds": int(governance.get("lease_seconds") or 900),
-    }
-    status, lease_response = bridge.call("/v2/writer-lease", lease_body)
-    if not 200 <= status < 300:
-        raise WorkerError("WRITER_LEASE_ACQUIRE_FAILED", phase="committing", details={"api_status": status, "error": lease_response.get("error")})
-    heartbeat.ensure()
-    files: dict[str, str] = {}
-    paths = governance.get("allowed_files") or [item["path"] for item in envelope.change_set]
-    for relative in sorted({safe_relative_path(item, field="allowed_file") for item in paths}):
-        path = resolve_under(worktree, relative)
-        if not path.is_file():
-            raise WorkerError("COMMIT_FILE_MISSING", phase="committing", details={"path": relative})
-        files[relative] = base64.b64encode(path.read_bytes()).decode("ascii")
-    commit_body = {
-        "task_id": governance["task_id"],
-        "repository": envelope.repository,
-        "expected_branch_oid": expected_branch_oid,
-        "target_branch": envelope.target_branch,
-        "generation": generation,
-        "idempotency_key": governance["commit_idempotency_key"],
-        "files": files,
-        "message": governance["message"],
-    }
-    status, commit_response = bridge.call("/v2/commit-files", commit_body)
-    if not 200 <= status < 300:
-        raise WorkerError("GOVERNED_COMMIT_FAILED", phase="committing", details={"api_status": status, "error": commit_response.get("error")})
-    commit_oid = commit_response.get("commit_oid")
-    if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
-        raise WorkerError("GOVERNED_COMMIT_OID_INVALID", phase="committing")
-    visible_oid: str | None = None
-    for _ in range(5):
-        probe = subprocess.run(
-            ["/usr/bin/git", "--git-dir", str(AUTHORITY_GIT_DIR), "rev-parse", "--verify", f"refs/heads/{envelope.target_branch}"],
-            text=True,
-            capture_output=True,
-            check=False,
-            env={"PATH": "/usr/bin:/bin", "LC_ALL": "C.UTF-8", "GIT_CONFIG_GLOBAL": str(GIT_CONFIG_GLOBAL)},
-            timeout=30,
-        )
-        if probe.returncode == 0:
-            visible_oid = probe.stdout.strip()
-            if visible_oid == commit_oid:
-                break
-        time.sleep(1)
-    if visible_oid != commit_oid:
-        raise WorkerError("GIT_VISIBILITY_MISMATCH", phase="verification", details={"commit_oid": commit_oid, "visible_oid": visible_oid})
-    return {
-        "governed_task_id": governance["task_id"],
-        "branch_mode": branch_mode,
-        "generation": generation,
-        "target_branch": envelope.target_branch,
-        "commit_oid": commit_oid,
-        "visible_oid": visible_oid,
-        "commit_idempotency_key": governance["commit_idempotency_key"],
-        "api_commit_visible": commit_response.get("commit_visible"),
-        "direct_main_write": False,
-        "force_push": False,
-        "production_write": False,
-        "protected_ref_bypass": False,
-    }
-
-
-def build_result(task_id: str, envelope: ExecutionEnvelope, execution: Mapping[str, Any], governed: Mapping[str, Any]) -> dict[str, Any]:
-    return {
-        "status": "SUCCEEDED",
-        "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
-        "task_id": task_id,
-        "execution_backend": "generic-online-agent-worker-v2",
-        "adapter": envelope.adapter,
-        "execution_digest": envelope.digest,
-        "repository": envelope.repository,
-        "base_oid": envelope.base_oid,
-        "target_branch": envelope.target_branch,
-        "changed_files": execution.get("changed_files", []),
-        "commands": [{k: item.get(k) for k in ("label", "exit_code", "stdout_sha256", "stderr_sha256", "duration_ms")} for item in execution.get("commands", [])],
-        "tests": [{k: item.get(k) for k in ("label", "exit_code", "stdout_sha256", "stderr_sha256", "duration_ms")} for item in execution.get("tests", [])],
-        "governed": dict(governed),
-        "workspace_cleanup": "PASS",
-        "secret_material_exposed": False,
-        "direct_main_write": False,
-        "force_push": False,
-        "production_write": False,
-        "protected_ref_bypass": False,
-    }
-
-
-class WorkerSupervisor:
-    def __init__(self, token: str) -> None:
-        self.agent = AgentTaskClient(token)
-        self.bridge = BridgeClient()
-
-    def process_claim(self, task: Mapping[str, Any], claim: Mapping[str, Any]) -> None:
-        task_id = str(claim.get("task_id") or task.get("task_id") or "")
-        lease_token = claim.get("lease_token")
-        if not isinstance(lease_token, str) or len(lease_token) < 24:
-            raise WorkerError("LEASE_TOKEN_NOT_RETURNED", phase="claim")
-        audit("TASK_CLAIMED", task_id=task_id, attempt=claim.get("attempt"), revision=claim.get("revision"))
-        heartbeat = LeaseHeartbeat(self.agent, task_id, lease_token)
-        try:
-            heartbeat.first()
-            heartbeat.set_stage("started")
-            heartbeat.start()
-            audit("TASK_STARTED", task_id=task_id, attempt=claim.get("attempt"))
-            envelope = ExecutionEnvelope.from_task(task)
-            state = load_task_state(task_id, envelope.digest) or {"task_id": task_id, "execution_digest": envelope.digest, "phase": "claimed"}
-            governed = state.get("governed") if isinstance(state.get("governed"), Mapping) else None
-            if isinstance(governed, Mapping) and isinstance(governed.get("commit_oid"), str) and OID_RE.fullmatch(governed["commit_oid"]):
-                result = state.get("result")
-                if not isinstance(result, Mapping):
-                    result = {"status": "SUCCEEDED", "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW", "task_id": task_id, "execution_digest": envelope.digest, "governed": dict(governed), "recovered_without_duplicate_commit": True, "direct_main_write": False, "production_write": False, "force_push": False, "protected_ref_bypass": False}
-                audit("TASK_RECOVERED_COMMIT", task_id=task_id, commit_oid=governed["commit_oid"], duplicate_commit=False)
-            else:
-                run_root = new_run_root()
-                try:
-                    worktree = prepare_worktree(run_root, envelope.base_oid)
-                    state.update({"phase": "workspace_prepared", "execution_digest": envelope.digest, "attempt": claim.get("attempt")})
-                    save_task_state(task_id, state)
-                    adapter = adapter_for(envelope.adapter)
-                    execution = adapter.execute(envelope, worktree)
-                    state.update({"phase": "verification_passed", "execution": execution})
-                    save_task_state(task_id, state)
-                    heartbeat.ensure()
-                    heartbeat.set_stage("committing")
-                    governed = governed_commit(envelope, worktree, self.bridge, heartbeat)
-                    result = build_result(task_id, envelope, execution, governed)
-                    state.update({"phase": "commit_done", "governed": governed, "result": result})
-                    save_task_state(task_id, state)
-                finally:
-                    cleanup_run_root(run_root)
-                    audit("WORKSPACE_CLEANED", task_id=task_id, cleanup="PASS")
-            heartbeat.ensure()
-            heartbeat.set_stage("result")
-            result_key = str(result.get("governed", {}).get("commit_idempotency_key") or f"{task_id}-result-{envelope.digest[:16]}")
-            status, response = self.agent.result(task_id, lease_token, result_key, result)
-            if not 200 <= status < 300:
-                raise WorkerError("TASK_RESULT_SUBMIT_FAILED", phase="result", details={"api_status": status, "error": response.get("error")})
-            audit("TASK_RESULT_SUBMITTED", task_id=task_id, api_status=status, result_idempotency_key=result_key, commit_oid=result.get("governed", {}).get("commit_oid"))
-        except WorkerError as exc:
-            audit("TASK_FAILED", task_id=task_id, error=exc.code, phase=exc.phase, details=exc.details)
-            failure_key = f"{task_id}-attempt-{claim.get('attempt', 0)}-failure"
-            failure = {
-                "status": "FAILED",
-                "verification_state": "UNVERIFIED",
-                "task_id": task_id,
-                "execution_backend": "generic-online-agent-worker-v2",
-                "error": exc.code,
-                "phase": exc.phase,
-                "details": exc.details,
-                "direct_main_write": False,
-                "force_push": False,
-                "production_write": False,
-                "protected_ref_bypass": False,
-                "secret_material_exposed": False,
-            }
-            try:
-                status, response = self.agent.result(task_id, lease_token, failure_key, failure)
-                audit("TASK_FAILURE_RESULT_SUBMITTED", task_id=task_id, api_status=status, error=response.get("error"))
-            except Exception as submit_exc:  # pragma: no cover - defensive runtime guard
-                audit("TASK_FAILURE_RESULT_SUBMIT_FAILED", task_id=task_id, error=type(submit_exc).__name__)
-        finally:
-            heartbeat.close()
-
-    def once(self) -> int:
-        try:
-            tasks = self.agent.poll()
-            audit("TASK_POLL", visible_count=len(tasks))
-        except WorkerError as exc:
-            audit("TASK_POLL_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
-            return 10
-        dynamic_tasks: list[dict[str, Any]] = []
-        for task in tasks:
-            metadata = task.get("metadata")
-            execution = metadata.get("execution") if isinstance(metadata, Mapping) else None
-            if not isinstance(execution, Mapping) or execution.get("schema") != "aios.online-agent.execution.v2":
-                audit("TASK_SKIPPED_NON_DYNAMIC", task_id=task.get("task_id"), reason="DYNAMIC_EXECUTION_PAYLOAD_REQUIRED")
-                continue
-            dynamic_tasks.append(task)
-        for task in dynamic_tasks:
-            task_id = str(task.get("task_id") or "")
-            status, claim = self.agent.claim(task_id)
-            if status == 409:
-                audit("TASK_CLAIM_SKIPPED", task_id=task_id, api_status=status, error=claim.get("error"))
-                continue
-            if status != 200:
-                audit("TASK_CLAIM_FAILED", task_id=task_id, api_status=status, error=claim.get("error"))
-                continue
-            self.process_claim(task, claim)
-            return 0
-        return 0
-
-    def run_forever(self) -> int:
-        ensure_roots()
-        audit("WORKER_STARTED", task_agnostic=True, api_base=API_BASE, bridge_base=BRIDGE_BASE, adapter_names=["deterministic", "codex", "hermes"], runtime_only_credentials=True)
-        while True:
-            self.once()
-            time.sleep(POLL_SECONDS)
-
-
-def main() -> int:
-    try:
-        token = load_agent_token()
-        supervisor = WorkerSupervisor(token)
-        if os.environ.get("AIOS_WORKER_ONCE") == "1":
-            ensure_roots()
-            return supervisor.once()
-        return supervisor.run_forever()
-    except WorkerError as exc:
-        audit("WORKER_START_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
-        return 20
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

services/aios-online-agent-worker/register.py

@@ -1,157 +0,0 @@
-#!/usr/bin/env python3
-"""Register or recover the generic Online Agent using runtime-only secrets."""
-
-from __future__ import annotations
-
-import json
-import os
-from pathlib import Path
-import subprocess
-import tempfile
-import urllib.error
-import urllib.request
-from datetime import datetime, timezone
-from typing import Any
-
-
-BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
-AGENT_ID = os.environ.get("AIOS_AGENT_ID", "online-engineering-worker-cn-network-mkt-generic-20260815")
-AGENT_TYPE = os.environ.get("AIOS_AGENT_TYPE", "trusted-online-engineering-worker")
-AGENT_VERSION = os.environ.get("AIOS_AGENT_VERSION", "2026-08-15.3-generic")
-PROJECTION_PROVIDER = "BridgeV2RuntimeCredentialProjection"
-STATE_DIR = Path(os.environ.get("AIOS_WORKER_REGISTRATION_STATE", "/var/lib/aios-online-agent-worker-20260815/registration"))
-CREDENTIAL_PATH = Path(os.environ.get("AIOS_WORKER_ENCRYPTED_CREDENTIAL", "/etc/credstore.encrypted/aios-online-agent-worker-20260815/agent_token"))
-IDENTITY_PATH = STATE_DIR / "identity.json"
-AUDIT_PATH = STATE_DIR / "events.jsonl"
-CAPABILITIES = [
-    "online_agent_task_api", "task.poll", "task.claim", "task.heartbeat", "task.result",
-    "engineering.consumer", "engineering.execute", "workspace.worktree", "git.read", "git.diff",
-    "git.test", "governed.write.branch", "governed.write.file",
-]
-
-
-def now() -> str:
-    return datetime.now(timezone.utc).isoformat()
-
-
-def audit(event_type: str, **fields: Any) -> None:
-    STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
-    record = {"event_type": event_type, "created_at": now(), "agent_id": AGENT_ID, "credential_material_exposed": False, **fields}
-    with AUDIT_PATH.open("a", encoding="utf-8") as handle:
-        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
-    os.chmod(AUDIT_PATH, 0o640)
-    print(json.dumps(record, ensure_ascii=False, sort_keys=True), flush=True)
-
-
-def api_call(method: str, route: str, token: str, body: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
-    payload = None if body is None else json.dumps(body, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
-    headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
-    if payload is not None:
-        headers["Content-Type"] = "application/json"
-    request = urllib.request.Request(BASE + route, data=payload, headers=headers, method=method)
-    try:
-        with urllib.request.urlopen(request, timeout=15) as response:
-            value = json.loads(response.read().decode("utf-8"))
-            return response.status, value if isinstance(value, dict) else {}
-    except urllib.error.HTTPError as exc:
-        try:
-            value = json.loads(exc.read(2048).decode("utf-8"))
-        except Exception:
-            value = {"error": "HTTP_ERROR"}
-        return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
-    except (urllib.error.URLError, TimeoutError):
-        return 599, {"error": "AGENT_TASK_API_UNAVAILABLE"}
-
-
-def read_identity() -> dict[str, Any] | None:
-    if not IDENTITY_PATH.is_file() or IDENTITY_PATH.is_symlink():
-        return None
-    try:
-        value = json.loads(IDENTITY_PATH.read_text(encoding="utf-8"))
-    except Exception:
-        return None
-    return value if isinstance(value, dict) else None
-
-
-def identity_matches(value: dict[str, Any] | None) -> bool:
-    return bool(value and value.get("agent_id") == AGENT_ID and value.get("agent_type") == AGENT_TYPE and value.get("agent_version") == AGENT_VERSION and value.get("credential_projection_provider") == PROJECTION_PROVIDER and value.get("task_scope") == "dynamic")
-
-
-def encrypted_credential_is_valid() -> bool:
-    if not CREDENTIAL_PATH.is_file() or CREDENTIAL_PATH.is_symlink():
-        return False
-    result = subprocess.run(["systemd-creds", "decrypt", "--with-key=host", str(CREDENTIAL_PATH), "/dev/null"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=15)
-    return result.returncode == 0
-
-
-def write_identity() -> None:
-    value = {"agent_id": AGENT_ID, "agent_type": AGENT_TYPE, "agent_version": AGENT_VERSION, "task_scope": "dynamic", "credential_projection_provider": PROJECTION_PROVIDER, "credential_delivery": "systemd-LoadCredentialEncrypted", "task_api_url": BASE, "task_api_schema": "aios.agent-task.v1", "execution_backend": "generic-online-agent-worker-v2", "runtime_only_secret_delivery": True, "production_write": False, "main_write": False, "force_push": False, "tag_write": False, "baseline_write": False, "writer_lease": True, "created_at": now()}
-    STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
-    fd, temporary = tempfile.mkstemp(prefix="identity.", dir=str(STATE_DIR))
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as handle:
-            json.dump(value, handle, ensure_ascii=False, sort_keys=True)
-            handle.write("\n")
-        os.chmod(temporary, 0o640)
-        os.replace(temporary, IDENTITY_PATH)
-    finally:
-        if os.path.exists(temporary):
-            os.unlink(temporary)
-
-
-def encrypt_agent_token(agent_token: str) -> None:
-    CREDENTIAL_PATH.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-    fd, temporary = tempfile.mkstemp(prefix=".online-agent-token.", dir=str(CREDENTIAL_PATH.parent))
-    os.close(fd)
-    os.unlink(temporary)
-    try:
-        result = subprocess.run(["systemd-creds", "encrypt", "--with-key=host", "--name=agent_token", "/dev/stdin", temporary], input=(agent_token + "\n").encode("utf-8"), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=20)
-        if result.returncode != 0:
-            raise RuntimeError("AGENT_CREDENTIAL_ENCRYPT_FAILED")
-        os.chmod(temporary, 0o600)
-        os.replace(temporary, CREDENTIAL_PATH)
-        os.chmod(CREDENTIAL_PATH, 0o600)
-    finally:
-        if os.path.exists(temporary):
-            os.unlink(temporary)
-
-
-def register_or_recover() -> str:
-    STATE_DIR.mkdir(mode=0o750, parents=True, exist_ok=True)
-    existing = read_identity()
-    if CREDENTIAL_PATH.exists():
-        if not identity_matches(existing):
-            raise RuntimeError("AGENT_IDENTITY_BINDING_CONFLICT")
-        if not encrypted_credential_is_valid():
-            raise RuntimeError("AGENT_ENCRYPTED_CREDENTIAL_INVALID")
-        audit("AGENT_RECOVERED", credential_delivery="systemd-LoadCredentialEncrypted", credential_projection_provider=PROJECTION_PROVIDER, register_or_recover=True)
-        return "RECOVERED"
-    if existing is not None:
-        raise RuntimeError("AGENT_CREDENTIAL_MISSING_AFTER_PRIOR_REGISTRATION")
-    credential_directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
-    control_path = credential_directory / "bridge_control_token"
-    if not control_path.is_file():
-        raise RuntimeError("BRIDGE_RUNTIME_CONTROL_CREDENTIAL_UNAVAILABLE")
-    control_token = control_path.read_text(encoding="utf-8").strip()
-    if len(control_token) < 32:
-        raise RuntimeError("BRIDGE_RUNTIME_CONTROL_CREDENTIAL_INVALID")
-    status, response = api_call("POST", "/v1/agents/register", control_token, {"agent_id": AGENT_ID, "agent_type": AGENT_TYPE, "agent_version": AGENT_VERSION, "capabilities": CAPABILITIES, "metadata": {"task_scope": "dynamic", "credential_projection_provider": PROJECTION_PROVIDER, "runtime_only_secret_delivery": True, "task_api_schema": "aios.agent-task.v1", "execution_backend": "generic-online-agent-worker-v2", "production_write": False, "main_write": False, "force_push": False, "tag_write": False, "baseline_write": False, "writer_lease": True}})
-    if status == 409 and response.get("error") == "AGENT_ALREADY_EXISTS":
-        raise RuntimeError("AGENT_ALREADY_EXISTS_RECOVERY_CREDENTIAL_UNAVAILABLE")
-    if status != 201:
-        raise RuntimeError("AGENT_REGISTER_FAILED_" + str(response.get("error") or status))
-    agent_token = response.get("agent_token")
-    if not isinstance(agent_token, str) or len(agent_token) < 24:
-        raise RuntimeError("AGENT_TOKEN_NOT_RETURNED")
-    encrypt_agent_token(agent_token)
-    write_identity()
-    audit("AGENT_REGISTERED", credential_delivery="systemd-LoadCredentialEncrypted", credential_projection_provider=PROJECTION_PROVIDER, register_or_recover=True)
-    return "REGISTERED"
-
-
-if __name__ == "__main__":
-    try:
-        raise SystemExit(0 if register_or_recover() in {"REGISTERED", "RECOVERED"} else 20)
-    except Exception as exc:
-        audit("AGENT_REGISTRATION_FAILED", error=type(exc).__name__)
-        raise SystemExit(20)

services/aios-online-agent-worker/test_worker_contract.py

@@ -1,79 +0,0 @@
-from __future__ import annotations
-
-import os
-from pathlib import Path
-
-from aios_online_agent_worker.worker import ExecutionEnvelope, WorkerError, sanitize  # noqa: E402
-
-
-BASE = "a" * 40
-
-
-def payload(task_id: str, branch: str, *, base_oid: str = BASE) -> dict[str, object]:
-    return {
-        "task_id": task_id,
-        "metadata": {
-            "execution": {
-                "schema": "aios.online-agent.execution.v2",
-                "repository": "aios/aios",
-                "base_oid": base_oid,
-                "target_branch": branch,
-                "adapter": "deterministic",
-                "workspace": {
-                    "authority_git_dir": str(Path(os.environ.get("AIOS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git"))),
-                    "read_only_repository": True,
-                    "cleanup_policy": "always",
-                },
-                "change_set": [
-                    {
-                        "operation": "write_file",
-                        "path": "tests/generated_marker.py",
-                        "content": "def test_marker():\n    assert True\n",
-                        "expected_sha256": None,
-                    }
-                ],
-                "verification": [
-                    {"label": "compile", "argv": ["/usr/bin/python3", "-m", "py_compile", "tests/generated_marker.py"], "cwd": ".", "timeout_seconds": 60}
-                ],
-                "governance": {
-                    "task_id": f"{task_id}-writer",
-                    "idempotency_key": f"{task_id}-governed-v1",
-                    "commit_idempotency_key": f"{task_id}-commit-v1",
-                    "allowed_files": ["tests/generated_marker.py"],
-                    "main_write": False,
-                    "force_push": False,
-                    "production_write": False,
-                    "tag_write": False,
-                    "baseline_write": False,
-                    "protected_ref_bypass": False,
-                },
-            }
-        },
-    }
-
-
-def test_dynamic_payload_is_not_bound_to_task_or_base() -> None:
-    first = ExecutionEnvelope.from_task(payload("task-alpha-20260815", "agent/alpha-20260815"))
-    second = ExecutionEnvelope.from_task(payload("task-beta-20260815", "agent/beta-20260815", base_oid="b" * 40))
-    assert first.task_id != second.task_id
-    assert first.base_oid != second.base_oid
-    assert first.target_branch != second.target_branch
-
-
-def test_protected_branch_is_rejected() -> None:
-    try:
-        ExecutionEnvelope.from_task(payload("task-main-20260815", "main"))
-    except WorkerError as exc:
-        assert exc.code == "PROTECTED_BRANCH_REJECTED"
-    else:
-        raise AssertionError("protected branch accepted")
-
-
-def test_task_payload_does_not_persist_secret_values() -> None:
-    value = sanitize({"agent_token": "agt-secret", "nested": {"authorization": "Bearer secret"}})
-    assert value == {"agent_token": "[REDACTED]", "nested": {"authorization": "[REDACTED]"}}
-
-
-def test_execution_digest_is_stable() -> None:
-    value = payload("task-digest-20260815", "agent/digest-20260815")
-    assert ExecutionEnvelope.from_task(value).digest == ExecutionEnvelope.from_task(value).digest

tests/test_agent_task_model_api_integration.py

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

tests/test_agent_task_protocol.py

@@ -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_caller_admission_projection.py

@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+SPEC = importlib.util.spec_from_file_location(
+    "aios_durable_owner_authorization",
+    APP / "aios_durable_owner_authorization.py",
+)
+BROKER_MODULE = importlib.util.module_from_spec(SPEC)
+import sys
+sys.modules[SPEC.name] = BROKER_MODULE
+assert SPEC.loader is not None
+SPEC.loader.exec_module(BROKER_MODULE)
+
+AuthorizationError = BROKER_MODULE.AuthorizationError
+Broker = BROKER_MODULE.Broker
+Constraints = BROKER_MODULE.Constraints
+DurablePolicy = BROKER_MODULE.DurablePolicy
+ReceiptStore = BROKER_MODULE.ReceiptStore
+Upstream = BROKER_MODULE.Upstream
+
+CALLER = "chatgpt-governed-implementer-singlehost"
+SCOPE = "release:candidate"
+SOURCE = "257c7a8df433f75b364bcaf69cbbc32e8c2e5d38"
+
+
+def policy(*, expires_at=None, capabilities=None):
+    return DurablePolicy(
+        grant_id="grant-caller-projection-v1",
+        owner_identity="owner-durable-policy",
+        account=CALLER,
+        capabilities=frozenset(capabilities or {SCOPE}),
+        issued_at=datetime.now(timezone.utc) - timedelta(minutes=1),
+        not_before=datetime.now(timezone.utc) - timedelta(minutes=1),
+        persistent_until_revoked=True,
+        constraints=Constraints(
+            repository="aios/aios",
+            account=CALLER,
+            allowed_target_prefix="agent/",
+            release_version_prefix="V2R2C",
+            max_task_risk="engineering",
+        ),
+        policy_sha256="a" * 64,
+        expires_at=expires_at,
+    )
+
+
+def make_broker(root):
+    return Broker(
+        policy_path=root / "policy.json",
+        revocation_path=root / "revocations.json",
+        store=ReceiptStore(root / "receipts.sqlite3"),
+        upstream=Upstream("http://127.0.0.1:1", "not-used"),
+        expected_account=CALLER,
+        expected_peer_user="test-peer",
+        allowed_peer_users=frozenset({"test-peer"}),
+    )
+
+
+class CallerProjectionTests(unittest.TestCase):
+    def project(self, broker, request=None):
+        return broker.project_caller(
+            CALLER,
+            SCOPE,
+            request or {"source_commit": SOURCE, "version": "V2R2C2"},
+            idempotency_key="caller-projection-test-v1",
+        )
+
+    def test_positive_projection_ack_and_effective_scope(self):
+        broker = make_broker(Path(tempfile.mkdtemp()))
+        with patch.object(broker, "policy", return_value=policy()):
+            value = self.project(broker)
+        credential = value.pop("_credential")
+        self.assertGreaterEqual(len(credential), 32)
+        self.assertEqual("ACKED", value["projection_state"])
+        self.assertEqual("REHYDRATED", value["credential_state"])
+        self.assertEqual([SCOPE], value["effective_scopes"])
+        self.assertFalse(value["secret_exposed"])
+
+    def test_secret_material_is_not_durable(self):
+        root = Path(tempfile.mkdtemp())
+        broker = make_broker(root)
+        with patch.object(broker, "policy", return_value=policy()):
+            value = self.project(broker)
+        credential = value.pop("_credential")
+        durable = (root / "receipts.sqlite3").read_bytes()
+        self.assertNotIn(credential.encode(), durable)
+        self.assertNotIn("credential_material", durable.decode("utf-8", "ignore"))
+        self.assertFalse(value["secret_exposed"])
+
+    def test_caller_mismatch_is_denied(self):
+        broker = make_broker(Path(tempfile.mkdtemp()))
+        with self.assertRaises(AuthorizationError) as caught:
+            broker.project_caller(
+                "other-caller",
+                SCOPE,
+                {},
+                idempotency_key="caller-projection-test-v1",
+            )
+        self.assertEqual("CALLER_MISMATCH", caught.exception.code)
+
+    def test_scope_mismatch_is_denied(self):
+        broker = make_broker(Path(tempfile.mkdtemp()))
+        with self.assertRaises(AuthorizationError) as caught:
+            broker.project_caller(
+                CALLER,
+                "release:deploy",
+                {},
+                idempotency_key="caller-projection-test-v1",
+            )
+        self.assertEqual("SCOPE_MISMATCH", caught.exception.code)
+
+    def test_policy_missing_revoked_and_expired_fail_closed(self):
+        broker = make_broker(Path(tempfile.mkdtemp()))
+        for code in ("POLICY_FILE_INVALID", "OWNER_GRANT_REVOKED", "POLICY_EXPIRED"):
+            with self.subTest(code=code):
+                with patch.object(
+                    broker,
+                    "policy",
+                    side_effect=AuthorizationError(code, 403),
+                ):
+                    with self.assertRaises(AuthorizationError) as caught:
+                        self.project(broker)
+                self.assertEqual(code, caught.exception.code)
+
+    def test_invalid_projection_state_fails_closed(self):
+        with self.assertRaises(AuthorizationError) as caught:
+            Broker.validate_caller_projection(
+                {
+                    "caller": CALLER,
+                    "scope": SCOPE,
+                    "projection_state": "CORRUPT",
+                    "credential_state": "REHYDRATED",
+                    "rehydration_state": "RECOVERED",
+                    "effective_scopes": [SCOPE],
+                    "secret_exposed": False,
+                }
+            )
+        self.assertEqual("PROJECTION_STATE_INVALID", caught.exception.code)
+
+    def test_restart_rehydrates_from_durable_metadata(self):
+        root = Path(tempfile.mkdtemp())
+        first = make_broker(root)
+        with patch.object(first, "policy", return_value=policy()):
+            initial = self.project(first)
+        first_credential = initial.pop("_credential")
+        second = make_broker(root)
+        with patch.object(second, "policy", return_value=policy()):
+            recovered = self.project(second)
+        second_credential = recovered.pop("_credential")
+        self.assertNotEqual(first_credential, second_credential)
+        self.assertEqual("RECOVERED", recovered["rehydration_state"])
+        self.assertEqual("ACKED", recovered["projection_state"])
+
+    def test_expired_policy_loader_rejects(self):
+        root = Path(tempfile.mkdtemp())
+        policy_path = root / "policy.json"
+        payload = {
+            "schema": "aios.owner-durable-authorization.v1",
+            "grant_id": "grant-caller-projection-v1",
+            "owner_identity": "owner-durable-policy",
+            "account": CALLER,
+            "capabilities": [SCOPE],
+            "issued_at": "2026-08-22T00:00:00Z",
+            "not_before": "2026-08-22T00:00:00Z",
+            "expires_at": "2026-08-22T01:00:00Z",
+            "persistent_until_revoked": True,
+            "constraints": {
+                "repository": "aios/aios",
+                "allowed_target_prefix": "agent/",
+                "release_version_prefix": "V2R2C",
+                "max_task_risk": "engineering",
+            },
+        }
+        policy_path.write_text(json.dumps(payload), encoding="utf-8")
+        with patch.object(BROKER_MODULE, "validate_root_file", lambda path, **kwargs: path):
+            with self.assertRaises(AuthorizationError) as caught:
+                DurablePolicy.load(policy_path)
+        self.assertEqual("POLICY_EXPIRED", caught.exception.code)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_caller_admission_runtime_regression.py

@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+import json
+import sys
+import types
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+if str(APP) not in sys.path:
+    sys.path.insert(0, str(APP))
+
+try:
+    import orchestrator.action_monitor
+except ModuleNotFoundError:
+    _orchestrator = types.ModuleType("orchestrator")
+    _action_monitor = types.ModuleType("orchestrator.action_monitor")
+    _action_monitor.ActionMonitor = type("ActionMonitor", (), {})
+    _orchestrator.__path__ = []
+    sys.modules["orchestrator"] = _orchestrator
+    sys.modules["orchestrator.action_monitor"] = _action_monitor
+
+import aios_durable_owner_authorization as broker_api
+import aios_model_api_staging as model_api
+
+
+CALLER = "chatgpt-governed-implementer-singlehost"
+SCOPE = "release:candidate"
+
+
+def _invoke_capabilities(
+    headers,
+    *,
+    auth_result=None,
+    auth_side_effect=None,
+    projection_client=None,
+):
+    handler = object.__new__(model_api.H)
+    handler.headers = headers
+    responses = []
+    handler.sendj = lambda value, code=200: responses.append((value, code))
+    if auth_side_effect is not None:
+        auth_patch = patch.object(
+            model_api, "authenticate", side_effect=auth_side_effect
+        )
+    else:
+        auth_patch = patch.object(
+            model_api, "authenticate", return_value=auth_result
+        )
+    client = (
+        projection_client
+        if projection_client is not None
+        else model_api.CALLER_PROJECTION_CLIENT
+    )
+    with auth_patch, patch.object(model_api, "CALLER_PROJECTION_CLIENT", client):
+        model_api.H.caller_capabilities(handler)
+    assert len(responses) == 1
+    return responses[0]
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {},
+        {"Authorization": "Bearer invalid"},
+        {"Authorization": "Bearer unknown"},
+        {"X-AIOS-Token": "rec-none"},
+    ],
+)
+def test_anonymous_invalid_unknown_and_rec_none_are_structured_401(headers):
+    payload, status = _invoke_capabilities(headers, auth_result=None)
+    assert status == 401
+    assert payload == {"error": "unauthorized", "hint": "Bearer token required"}
+    assert "Traceback" not in json.dumps(payload)
+
+
+def test_authenticate_exception_is_fail_closed_without_socket_abort():
+    payload, status = _invoke_capabilities(
+        {"Authorization": "Bearer malformed"},
+        auth_side_effect=RuntimeError("malformed"),
+    )
+    assert status == 401
+    assert payload["error"] == "unauthorized"
+    assert "Traceback" not in json.dumps(payload)
+
+
+def test_authorized_capability_introspection_remains_secret_free():
+    row = {
+        "account": CALLER,
+        "role": "user",
+        "scopes": ["agent-task:control", "branch:create", "file:write"],
+    }
+    result = SimpleNamespace(
+        public={
+            "projection_state": "ACKED",
+            "credential_state": "ACTIVE",
+            "rehydration_state": "REHYDRATED",
+            "effective_scopes": [SCOPE],
+        },
+        credential_loaded=True,
+    )
+
+    class _ProjectionClient:
+        def rehydrate(self, caller, scope, request):
+            assert caller == CALLER
+            assert scope == SCOPE
+            assert request == {}
+            return result
+
+    payload, status = _invoke_capabilities(
+        {"Authorization": "Bearer valid"},
+        auth_result=row,
+        projection_client=_ProjectionClient(),
+    )
+    assert status == 200
+    assert payload["caller_identity"] == CALLER
+    assert payload["admitted"] is True
+    assert SCOPE in payload["effective_scopes"]
+    assert payload["projection_state"] == "ACKED"
+    assert payload["credential_state"] == "ACTIVE"
+    assert payload["rehydration_state"] == "REHYDRATED"
+    assert payload["secret_exposed"] is False
+    assert "_credential" not in json.dumps(payload)
+
+
+class _FakeBroker:
+    def __init__(self):
+        self.calls = []
+
+    def project_caller(self, caller, scope, request, *, idempotency_key):
+        self.calls.append((caller, scope, request, idempotency_key))
+        return {
+            "caller": caller,
+            "scope": scope,
+            "projection_state": "ACKED",
+            "credential_state": "ACTIVE",
+            "rehydration_state": "REHYDRATED",
+            "effective_scopes": [SCOPE],
+            "_credential": "x" * 40,
+        }
+
+
+def _dispatch_broker(path="/v1/project-caller", *, peer=True, account=True):
+    broker = _FakeBroker()
+    handler = object.__new__(broker_api.Handler)
+    handler.path = path
+    handler.broker = broker
+    handler.peer_authorized = lambda: peer
+    handler.account_authorized = lambda: account
+    handler.body = lambda: {
+        "caller": CALLER,
+        "scope": SCOPE,
+        "idempotency_key": "runtime-regression-test-20260823",
+        "request": {
+            "repository": "aios/aios",
+            "target_branch": "agent/runtime-regression-test",
+            "release_version": "V2R2C2",
+        },
+    }
+    responses = []
+    handler.send_json = lambda value, status, internal_credential=None: responses.append(
+        (value, status, internal_credential)
+    )
+    broker_api.Handler.do_POST(handler)
+    return responses, broker
+
+
+def test_existing_project_caller_route_reaches_handler_without_public_secret():
+    responses, broker = _dispatch_broker()
+    assert len(responses) == 1
+    value, status, internal_credential = responses[0]
+    assert status == 200
+    assert broker.calls
+    assert broker.calls[0][0:2] == (CALLER, SCOPE)
+    assert internal_credential == "x" * 40
+    assert "_credential" not in value
+    assert "x" * 40 not in json.dumps(value)
+
+
+def test_unauthorized_peer_is_denied_and_unregistered_route_is_not_added():
+    responses, broker = _dispatch_broker(peer=False)
+    assert responses[0][1] == 403
+    assert broker.calls == []
+
+    responses, broker = _dispatch_broker(path="/v1/not-registered")
+    assert responses[0][1] == 404
+    assert broker.calls == []