chore(gitops): execute task bridge-execution-admission-repair-20260823

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,142 @@
+"""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"
+GOVERNED_TASK_SCOPE = "governed-task:project"
+PROJECTED_SCOPES = frozenset({GOVERNED_TASK_SCOPE, SCOPE})
+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 not in PROJECTED_SCOPES:
+            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)
+        effective_scopes = value.get("effective_scopes")
+        if (
+            not isinstance(effective_scopes, list)
+            or len(effective_scopes) != len(set(effective_scopes))
+            or not all(isinstance(item, str) for item in effective_scopes)
+            or scope not in effective_scopes
+            or not set(effective_scopes).issubset(PROJECTED_SCOPES)
+        ):
+            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_chatgpt_actions_api.py

@@ -22,7 +22,6 @@ from gitea_tag_publication import (
     TagPublisher,
     redact,
     validate_version,
-    validate_gitea_api_url,
 )
 from orchestrator.action_monitor import ActionMonitor
 from silent_release_hardening import (
@@ -309,7 +308,8 @@ def resolve_commit_oid(value: str | None, field: str = 'target_oid') -> str:
         raise ValueError(f'{field} does not resolve to the exact approved commit') from exc
 
 def existing_tag_target(tag: str) -> str | None:
-    validate_version(tag)
+    if not re.fullmatch(r'V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$', str(tag or '')):
+        raise ValueError('invalid release version')
     try:
         result = run_authoritative_git(
             ['rev-parse', '--verify', f'refs/tags/{tag}^{{commit}}'],
@@ -510,15 +510,7 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({'error':redact(exc)},413)
                 except (AuthoritativeGitError, RuntimeError) as exc:
                     return self.sendj({'error':redact(exc)},503)
-                payload = {'from': frm, 'to': to, 'diff': diff}
-                payload.update({
-                    key: diff[key]
-                    for key in (
-                        'from_oid', 'to_oid', 'changed_paths', 'stats',
-                        'total_size_bytes', 'truncated', 'patch_limit_bytes',
-                    )
-                })
-                return self.sendj(payload)
+                return self.sendj({'from':frm,'to':to,'diff':diff})
             if path in ('/api/patch','/api/patches','/api/branches','/patch'):
                 snapshot = repository_snapshot(
                     git_dir=AUTHORITATIVE_GIT_DIR,
@@ -781,10 +773,7 @@ class H(BaseHTTPRequestHandler):
         a=self.need('tag:create');
         if not a: return
         ver=data.get('version') or data.get('tag')
-        try:
-            validate_version(ver)
-        except ValueError:
-            return self.sendj({'error':'version must match VxRxCx'},400)
+        if not re.match(r'^V\d+R\d+C\d+(-[A-Za-z0-9_.-]+)?$', ver or ''): return self.sendj({'error':'version must match VxRxCx'},400)
         if not data.get('target_oid') or not data.get('evidence_closure_oid') or not data.get('approval_plan_id'):
             return self.sendj({
                 'error':'production_tag_evidence_required',
@@ -958,7 +947,6 @@ def api_spec():
     return {'version':'V1R1C0-chatgpt-actions','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'],'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,'direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True,'silent_release_execution_enabled':SILENT_RELEASE_EXECUTION_ENABLED,'silent_release_shadow_default':True}}
 
 def main():
-    validate_gitea_api_url(os.getenv('AIOS_GITOPS_GITEA_API_URL', ''))
     print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
     ThreadingHTTPServer((HOST,PORT), H).serve_forever()
 if __name__ == '__main__': main()

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -46,75 +46,6 @@ OPERATION_REQUIRED_FIELDS = {
     "getControlPlaneSnapshot": (),
 }
 
-_OPERATION_PATHS = {
-    "createGovernedTask": "/v2/governed-task",
-    "acquireWriterLease": "/v2/writer-lease",
-    "commitFileGoverned": "/v2/commit-file",
-    "commitFilesGoverned": "/v2/commit-files",
-    "getGitOpsJob": "/v2/job",
-    "resumeGitOpsJob": "/v2/job/resume",
-    "reconcileGitOpsJob": "/v2/job/reconcile",
-    "getTagPublicationStatus": "/v2/tag/status",
-    "getReleaseConsistency": "/v2/release/consistency",
-    "runReleaseAcceptance": "/v2/release/acceptance",
-}
-
-
-def operation_contracts() -> dict[str, dict[str, object]]:
-    """Return the runtime contract consumed by adapters and schema tooling."""
-    return {
-        operation: {
-            "method": "POST",
-            "path": _OPERATION_PATHS.get(operation),
-            "required_fields": list(fields),
-        }
-        for operation, fields in OPERATION_REQUIRED_FIELDS.items()
-    }
-
-
-def render_openapi_contract_fragment() -> str:
-    required = "\n".join(
-        f"                - {field}"
-        for field in OPERATION_REQUIRED_FIELDS["commitFilesGoverned"]
-    )
-    return (
-        "  /v2/commit-files:\n"
-        "    post:\n"
-        "      operationId: commitFilesGoverned\n"
-        "      summary: Submit one atomic governed commit containing multiple files\n"
-        "      requestBody:\n"
-        "        required: true\n"
-        "        content:\n"
-        "          application/json:\n"
-        "            schema:\n"
-        "              type: object\n"
-        "              additionalProperties: false\n"
-        "              required:\n"
-        f"{required}\n"
-        "              properties:\n"
-        "                task_id: {type: string}\n"
-        "                repository: {type: string, example: aios/aios}\n"
-        "                expected_branch_oid: {type: string, pattern: '^[0-9a-f]{40}$'}\n"
-        "                target_branch: {type: string, pattern: '^agent/[A-Za-z0-9._/-]+$'}\n"
-        "                generation: {type: integer, minimum: 1}\n"
-        "                idempotency_key: {type: string}\n"
-        "                files:\n"
-        "                  type: array\n"
-        "                  minItems: 1\n"
-        "                  items:\n"
-        "                    type: object\n"
-        "                    additionalProperties: false\n"
-        "                    required: [path, content_base64]\n"
-        "                    properties:\n"
-        "                      path: {type: string, minLength: 1, maxLength: 512}\n"
-        "                      content_base64: {type: string}\n"
-        "                message: {type: string, minLength: 1, maxLength: 200}\n"
-        "      responses:\n"
-        "        '200': {description: Governed commit job and verified branch result}\n"
-        "        '400': {description: Contract or policy rejection}\n"
-        "        '409': {description: Branch CAS or idempotency conflict}\n"
-    )
-
 
 class RuntimeDriftError(RuntimeError):
     pass
@@ -191,7 +122,6 @@ class BridgeV2Adapter:
             "legacy_writes_enabled": False,
             "governed_writes_enabled": True,
             "operations": list(BRIDGE_OPERATIONS),
-            "operation_contracts": operation_contracts(),
             "force_push": False,
             "main_write": False,
             "tag_write": False,

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -14,11 +14,10 @@ from typing import Any, Mapping
 
 from aios_authority_git_executor import AuthorityGitExecutor
 from aios_chatgpt_bridge_v2 import BridgeV2Adapter, BridgeV2Error, OPERATION_REQUIRED_FIELDS
-from gitea_tag_publication import validate_gitea_api_url
 from aios_continuation_executor import AuthorityGitAdapter, ContinuationCommit, ContinuationExecutor, ContinuationPermissions, ContinuationRuntimeBinding, DetachedGitWorkspace, GiteaContinuationRemote, ServerSideCredentialProjection
 from aios_continuation_lease import WriterLeaseRequestV2
 from aios_control_plane_snapshot import build_control_plane_snapshot
-from aios_gitops_job_controller import GitOpsJobController, JobState, JobStore, _RECOVERABLE
+from aios_gitops_job_controller import GitOpsJobController, JobStore
 from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
 from governed_production_release import build_executor_config
 
@@ -53,7 +52,7 @@ class RuntimeConfig:
     ssh_key_path: Path
     remote_name: str = "gitea-http"
     repository: str = "aios/aios"
-    api_base: str = ""
+    api_base: str = "https://g.cnnetmkt.top/git/api/v1"
     v2_task_id: str = "TASK-P0-CHATGPT-BRIDGE-V2-CONTROL-PLANE-CONVERGENCE-20260809"
     v2_branch: str = "agent/p0-chatgpt-bridge-v2-control-plane-convergence-20260809"
     target_version: str = "V2R2C1"
@@ -281,8 +280,6 @@ class GovernedRuntime:
             return controller.store._receipt(row, duplicate=True).as_dict()
         if action == "resume":
             return controller.resume(job_id).as_dict()
-        if JobState(row["state"]) not in _RECOVERABLE:
-            return controller.store._receipt(row, duplicate=True).as_dict()
         return self._executor_from_row(row, controller).reconcile(job_id)
 
     def _snapshot_task(self) -> dict[str, Any]:
@@ -421,7 +418,7 @@ def main() -> None:
         token_path,
         ssh_key_path,
         remote_name=os.environ.get("AIOS_GITEA_REMOTE", "gitea-http"),
-        api_base=validate_gitea_api_url(os.environ.get("AIOS_GITEA_API_BASE", "")),
+        api_base=os.environ.get("AIOS_GITEA_API_BASE", "https://g.cnnetmkt.top/git/api/v1"),
         target_version=os.environ.get("AIOS_CONTROL_PLANE_TARGET_VERSION", "V2R2C1"),
     )
     runtime = GovernedRuntime(config)

active-runtime/app/aios_gitops_job_controller.py

@@ -41,8 +41,6 @@ class JobState(str, Enum):
     REMOTE_SYNCING = "REMOTE_SYNCING"
     REMOTE_VERIFIED = "REMOTE_VERIFIED"
     COMPLETED = "COMPLETED"
-    FAILED = "FAILED"
-    CANCELLED = "CANCELLED"
     RECOVERABLE_FAILURE = "RECOVERABLE_FAILURE"
     RECONCILING = "RECONCILING"
     RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED"
@@ -61,12 +59,7 @@ _ORDER = {
     JobState.COMPLETED: 8,
 }
 _RECOVERABLE = {JobState.RECOVERABLE_FAILURE, JobState.RECONCILIATION_REQUIRED}
-_TERMINAL = {
-    JobState.COMPLETED,
-    JobState.FAILED,
-    JobState.CANCELLED,
-    JobState.MANUAL_OWNER_REVIEW_REQUIRED,
-}
+_TERMINAL = {JobState.COMPLETED, JobState.MANUAL_OWNER_REVIEW_REQUIRED}
 
 
 def utc_now() -> str:

active-runtime/app/aios_gitops_model_api.py

@@ -5,8 +5,6 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Any
 
-from release_version import validate_release_version
-
 ROOT = Path(os.getenv('AIOS_GITOPS_ROOT', '/data/aios-gitops'))
 WORKSPACE = Path(os.getenv('AIOS_GITOPS_WORKSPACE', str(ROOT / 'workspace')))
 DB_PATH = Path(os.getenv('AIOS_GITOPS_DB', str(ROOT / 'runtime' / 'gitops_model.sqlite3')))
@@ -220,11 +218,7 @@ class Handler(BaseHTTPRequestHandler):
                 a = self.require_write(['owner']);
                 if not a: return
                 ver=data.get('version') or data.get('tag')
-                try:
-            validate_release_version(ver)
-        except ValueError:
-            self.send_json({'error':'version must match VxRxCx'},400)
-            return
+                if not re.match(r'^V\d+R\d+C\d+(-[a-zA-Z0-9_.-]+)?$', ver or ''): self.send_json({'error':'version must match VxRxCx'},400); return
                 msg=data.get('message') or f'release {ver}'; out=git_cmd(['tag','-a',ver,'-m',msg]);
                 if data.get('push',False): git_cmd(['push','origin',ver])
                 level='V' if re.match(r'^V[1-9]\d*R0C0', ver) else ('R' if re.match(r'^V\d+R[1-9]\d*C0', ver) else 'C'); exec_sql('INSERT OR REPLACE INTO gitops_versions(version,tag,level,summary,status,created_by,created_at) VALUES(?,?,?,?,?,?,?)',(ver,ver,level,msg,'ready',a['username'],now())); exec_sql('UPDATE gitops_baseline SET active_version=?,active_tag=?,updated_at=? WHERE id=?',(ver,ver,now(),'current')); audit(a['username'],'git_tag',ver,{'level':level}); self.send_json({'status':'tagged','version':ver,'git':out}); return

active-runtime/app/aios_model_api_staging.py

@@ -3,6 +3,15 @@ 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,
+    GOVERNED_TASK_SCOPE,
+    PROJECTED_SCOPES,
+    CallerProjectionClient,
+    CallerProjectionError,
+)
 from baseline_authority import BaselineAuthority, BaselineAuthorityError
 from typing import Any
 
@@ -23,7 +32,6 @@ from gitea_tag_publication import (
     TagPublisher,
     redact,
     validate_version,
-    validate_gitea_api_url,
 )
 from orchestrator.action_monitor import ActionMonitor
 from silent_release_hardening import (
@@ -90,6 +98,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 = [
@@ -113,6 +126,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',
@@ -320,7 +336,8 @@ def resolve_commit_oid(value: str | None, field: str = 'target_oid') -> str:
         raise ValueError(f'{field} does not resolve to the exact approved commit') from exc
 
 def existing_tag_target(tag: str) -> str | None:
-    validate_version(tag)
+    if not re.fullmatch(r'V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$', str(tag or '')):
+        raise ValueError('invalid release version')
     try:
         result = run_authoritative_git(
             ['rev-parse', '--verify', f'refs/tags/{tag}^{{commit}}'],
@@ -407,14 +424,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 in PROJECTED_SCOPES:
+            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'],
+                    GOVERNED_TASK_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 +562,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({
@@ -553,15 +828,7 @@ class H(BaseHTTPRequestHandler):
                     return self.sendj({'error':redact(exc)},413)
                 except (AuthoritativeGitError, RuntimeError) as exc:
                     return self.sendj({'error':redact(exc)},503)
-                payload = {'from': frm, 'to': to, 'diff': diff}
-                payload.update({
-                    key: diff[key]
-                    for key in (
-                        'from_oid', 'to_oid', 'changed_paths', 'stats',
-                        'total_size_bytes', 'truncated', 'patch_limit_bytes',
-                    )
-                })
-                return self.sendj(payload)
+                return self.sendj({'from':frm,'to':to,'diff':diff})
             if path in ('/api/patch','/api/patches','/api/branches','/patch'):
                 snapshot = repository_snapshot(
                     git_dir=AUTHORITATIVE_GIT_DIR,
@@ -708,12 +975,60 @@ 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]
+            if path == '/createGovernedTask' and self.actor()['account'] == CALLER:
+                scope = GOVERNED_TASK_SCOPE
+            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)
                 )
@@ -830,10 +1145,7 @@ class H(BaseHTTPRequestHandler):
         a=self.need('tag:create');
         if not a: return
         ver=data.get('version') or data.get('tag')
-        try:
-            validate_version(ver)
-        except ValueError:
-            return self.sendj({'error':'version must match VxRxCx'},400)
+        if not re.match(r'^V\d+R\d+C\d+(-[A-Za-z0-9_.-]+)?$', ver or ''): return self.sendj({'error':'version must match VxRxCx'},400)
         if not data.get('target_oid') or not data.get('evidence_closure_oid') or not data.get('approval_plan_id'):
             return self.sendj({
                 'error':'production_tag_evidence_required',
@@ -1007,7 +1319,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,
@@ -1022,12 +1334,14 @@ 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,
         },
     }
 
 
 def main():
-    validate_gitea_api_url(os.getenv('AIOS_GITOPS_GITEA_API_URL', ''))
     print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
     ThreadingHTTPServer((HOST,PORT), H).serve_forever()
 if __name__ == '__main__': main()

active-runtime/app/authoritative_git.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 import os
 import re
 import subprocess
-import time
 import urllib.parse
 from pathlib import Path
 from typing import Iterable
@@ -307,76 +306,14 @@ def authoritative_diff(
     from_oid = validate_git_revision(from_revision, "from", git_dir=path)
     to_oid = validate_git_revision(to_revision, "to", git_dir=path)
     args = ["diff", "--no-ext-diff", "--no-textconv", from_oid, to_oid]
-    patch_limit = max(1, min(_MAX_OUTPUT, 256 * 1024))
-    process = subprocess.Popen(
-        ["git", "--git-dir", str(path), *args],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
-    )
-    chunks: list[bytes] = []
-    stored_bytes = 0
-    total_bytes = 0
-    started = time.monotonic()
-    try:
-        while process.stdout is not None:
-            chunk = process.stdout.read(64 * 1024)
-            if not chunk:
-                break
-            total_bytes += len(chunk)
-            if stored_bytes < patch_limit:
-                remaining = patch_limit - stored_bytes
-                selected = chunk[:remaining]
-                chunks.append(selected)
-                stored_bytes += len(selected)
-            if time.monotonic() - started > _TIMEOUT:
-                process.kill()
-                raise AuthoritativeGitError("GIT_REF_QUERY_TIMEOUT")
-        stderr = (
-            process.stderr.read(8192).decode("utf-8", errors="replace")
-            if process.stderr is not None
-            else ""
-        )
-        returncode = process.wait(timeout=max(0.1, _TIMEOUT - (time.monotonic() - started)))
-    except (OSError, subprocess.TimeoutExpired) as exc:
-        process.kill()
-        process.wait()
-        raise AuthoritativeGitError("GIT_REF_QUERY_FAILED") from exc
-    if returncode != 0:
-        raise AuthoritativeGitError(_redact(stderr.strip() or "GIT_REF_QUERY_FAILED")[:200])
-    try:
-        names_result = run_authoritative_git(
-            ["diff", "--no-ext-diff", "--no-textconv", "--name-status", from_oid, to_oid],
-            git_dir=path,
-        )
-        changed_paths = [
-            line.split("\t", 1)[-1]
-            for line in names_result.stdout.splitlines()
-            if line.strip()
-        ]
-    except AuthoritativeGitOutputTooLarge:
-        changed_paths = []
-    try:
-        stats_result = run_authoritative_git(
-            ["diff", "--no-ext-diff", "--no-textconv", "--stat", from_oid, to_oid],
-            git_dir=path,
-        )
-        stats = stats_result.stdout.strip()
-    except AuthoritativeGitOutputTooLarge:
-        stats = "GIT_STAT_OUTPUT_TRUNCATED"
-    patch = b"".join(chunks).decode("utf-8", errors="replace")
+    result = run_authoritative_git(args, git_dir=path)
     return {
         "code": 0,
-        "stdout": patch,
+        "stdout": result.stdout,
         "stderr": "",
         "cmd": ["git", f"--git-dir={path}", *args],
         "from_oid": from_oid,
         "to_oid": to_oid,
-        "changed_paths": changed_paths,
-        "stats": stats,
-        "total_size_bytes": total_bytes,
-        "truncated": total_bytes > patch_limit,
-        "patch_limit_bytes": patch_limit,
     }
 
 

active-runtime/app/gitea_tag_publication.py

@@ -8,7 +8,6 @@ import json
 import os
 import re
 import subprocess
-import sys
 import tempfile
 import time
 import urllib.error
@@ -24,15 +23,10 @@ try:
     from authoritative_git import AuthoritativeGitError, run_authoritative_git
 except ModuleNotFoundError:
     from .authoritative_git import AuthoritativeGitError, run_authoritative_git
-try:
-    from release_version import RELEASE_VERSION_RE, validate_release_version
-except ModuleNotFoundError:
-    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from release_version import RELEASE_VERSION_RE, validate_release_version
 
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
-VERSION_RE = RELEASE_VERSION_RE
+VERSION_RE = re.compile(r"^V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$")
 DEPLOYMENT_EVIDENCE_RE = re.compile(r"^(?:sha256:[0-9a-f]{64}|git:[0-9a-f]{40})$")
 FINGERPRINT_FIELDS = (
     "version",
@@ -89,14 +83,9 @@ class PublicationManualOwnerReviewRequired(
 
 
 def validate_version(version: str) -> str:
-    return validate_release_version(version)
-
-
-def validate_gitea_api_url(api_url: str) -> str:
-    value = str(api_url or "").strip().rstrip("/")
-    parsed = urllib.parse.urlsplit(value)
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
-        raise ValueError("Gitea API URL must be HTTP(S)")
+    value = str(version or "")
+    if not VERSION_RE.fullmatch(value):
+        raise ValueError("invalid release version")
     return value
 
 
@@ -422,11 +411,13 @@ class GiteaTagClient:
         timeout: float = 15,
         token_line: int = 1,
     ):
-        self.api_url = validate_gitea_api_url(api_url)
+        self.api_url = api_url.rstrip("/")
         self.repository = repository.strip("/")
         self.token_file = token_file
         self.timeout = timeout
         self.token_line = token_line
+        if not self.api_url.startswith(("https://", "http://")):
+            raise ValueError("Gitea API URL must be HTTP(S)")
         if self.repository.count("/") != 1:
             raise ValueError("Gitea repository must be owner/name")
 

active-runtime/app/governed_tag_adapter.py

@@ -34,8 +34,6 @@ from gitea_tag_publication import (
     TagPublisher,
 )
 
-from release_version import validate_release_version
-
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
 
@@ -46,10 +44,8 @@ def load_request() -> dict[str, Any]:
         raise ValueError("request_object_required")
     if value.get("schema") != "aios.production-executor-request.v1":
         raise ValueError("request_schema_invalid")
-    try:
-        validate_release_version(value.get("version", ""))
-    except ValueError as exc:
-        raise ValueError("version_invalid") from exc
+    if not re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", str(value.get("version", ""))):
+        raise ValueError("version_invalid")
     for name in ("target_oid", "evidence_closure_oid"):
         if not OID_RE.fullmatch(str(value.get(name, ""))):
             raise ValueError(f"{name}_invalid")

active-runtime/app/release_metadata_adapter.py

@@ -27,8 +27,6 @@ import urllib.request
 from pathlib import Path
 from typing import Any, Mapping
 
-from release_version import validate_release_version
-
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA_RE = re.compile(r"^[0-9a-f]{64}$")
 PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
@@ -88,10 +86,8 @@ def load_request() -> dict[str, Any]:
         raise ValueError("request_schema_invalid")
     if value.get("repository_identity") != "aios/aios":
         raise ValueError("repository_identity_invalid")
-    try:
-        validate_release_version(value.get("version", ""))
-    except ValueError as exc:
-        raise ValueError("version_invalid") from exc
+    if not re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", str(value.get("version", ""))):
+        raise ValueError("version_invalid")
     for name in ("target_oid", "evidence_closure_oid"):
         if not OID_RE.fullmatch(str(value.get(name, ""))):
             raise ValueError(f"{name}_invalid")

active-runtime/app/silent_release_controller.py

@@ -22,12 +22,11 @@ from enum import Enum
 from pathlib import Path
 from typing import Any, Protocol
 
-from release_version import validate_release_version
-
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 CANDIDATE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{2,127}$")
+VERSION_RE = re.compile(r"^V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?$")
 SECRET_PATTERNS = (
     re.compile(r"(?i)(authorization\s*[:=]\s*)(?:bearer|token)\s+\S+"),
     re.compile(r"(?i)((?:access_)?token\s*[:=]\s*)\S+"),
@@ -770,10 +769,8 @@ class TagApprovalEnvelope:
     def validate(self) -> None:
         if self.action != "production_tag":
             raise ApprovalError("approval does not authorize production tag")
-        try:
-            validate_release_version(self.version)
-        except ValueError as exc:
-            raise ApprovalError("invalid release version") from exc
+        if not VERSION_RE.fullmatch(self.version):
+            raise ApprovalError("invalid release version")
         _validate_oid("tag_oid", self.tag_oid)
         _validate_oid("target_oid", self.target_oid)
         if _parse_utc(self.expires_at) <= datetime.now(timezone.utc):

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-chatgpt-bridge-v2-staging.service

@@ -4,25 +4,14 @@ After=network-online.target
 
 [Service]
 Type=simple
-User=aiman
-Group=gitops
-PermissionsStartOnly=true
-WorkingDirectory=@RUNTIME_ROOT@
+User=aios-bridge-staging
+Group=aios-bridge-staging
+WorkingDirectory=/opt/aios/gitops-control-plane/current
 Environment=AIOS_BRIDGE_SCHEMA_VERSION=aios.chatgpt-gitops-bridge.v2
 Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
-Environment=AIOS_GITOPS_WRITES_ENABLED=false
-Environment=AIOS_GITEA_API_BASE=https://g.cnnetmkt.top/git/api/v1
-Environment=AIOS_GITEA_TOKEN_FILE=/run/credentials/aios-chatgpt-bridge-v2-staging.service/gitea_token
-Environment=AIOS_GITEA_REMOTE=gitea-http
-Environment=AIOS_CONTROL_PLANE_TARGET_VERSION=V2R2C1
-Environment=AIOS_CONTROL_PLANE_TARGET_OID=@SOURCE_OID@
-Environment=AIOS_CONTROL_PLANE_ARTIFACT_SHA256=@ARTIFACT_SHA256@
-Environment=PYTHONPATH=@RUNTIME_ROOT@:@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os
 Environment=AIOS_DEPLOYMENT_STATE=STAGING
-LoadCredential=gitea_token:/run/credentials/aios-governance-executor.service/gitea-governance-executor-token
-ExecStartPre=/usr/bin/install -d -o aiman -g gitops -m 0750 /var/lib/aios-gitops/bridge-v2-staging
-ExecStart=/usr/bin/python3 -B @RUNTIME_ROOT@/app/aios_chatgpt_bridge_v2_server.py
+ExecStart=/usr/bin/python3 /opt/aios/gitops-control-plane/current/bridge-v2-staging-server.py
 NoNewPrivileges=true
 PrivateTmp=true
 ProtectSystem=strict

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

@@ -7,9 +7,6 @@ Wants=network-online.target
 Type=simple
 User=aiman
 Group=gitops
-PermissionsStartOnly=true
-ExecStartPre=/usr/bin/install -d -o aiman -g gitops -m 0750 /var/lib/aios-gitops/model-api-staging
-ExecStartPre=/usr/bin/install -d -o aiman -g gitops -m 0750 /var/lib/aios-gitops/model-api-staging/silent-release
 WorkingDirectory=@RUNTIME_ROOT@
 Environment=PYTHONUNBUFFERED=1
 Environment=PYTHONDONTWRITEBYTECODE=1
@@ -30,16 +27,13 @@ Environment=AIOS_GITOPS_DEPLOYMENT_STATE=STAGING
 Environment=AIOS_GITOPS_MODEL_API_SERVICE=aios-gitops-model-api-v2r2c1-staging.service
 Environment=AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false
 Environment=AIOS_SILENT_RELEASE_STATE_ROOT=/var/lib/aios-gitops/model-api-staging/silent-release
-Environment=AIOS_GITOPS_GITEA_API_URL=https://g.cnnetmkt.top/git/api/v1
-Environment=AIOS_GITOPS_GITEA_REPOSITORY=aios/aios
-Environment=AIOS_GITOPS_GITEA_REMOTE=gitea-http
-Environment=AIOS_GITOPS_GITEA_TOKEN_FILE=/run/credentials/aios-gitops-model-api-v2r2c1-staging.service/gitea_token
-Environment=AIOS_GITOPS_GITEA_TOKEN_LINE=1
-LoadCredential=gitea_token:/run/credentials/aios-governance-executor.service/gitea-governance-executor-token
 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@:@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os
+Environment=PYTHONPATH=@RUNTIME_ROOT@/app:@RUNTIME_ROOT@/runtime-agent-os
 ExecStart=/usr/bin/python3 @RUNTIME_ROOT@/app/aios_model_api_staging.py
 Restart=on-failure
 RestartSec=2

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,154 @@
+# Caller admission projection and Bridge execution rehydration
+
+This runbook covers the exact durable Owner policy and exact API caller:
+
+- caller: chatgpt-governed-implementer-singlehost
+- durable policy: /etc/aios-gitops/owner-durable-authorization.json
+- revocations: /etc/aios-gitops/owner-durable-authorization-revocations.json
+- grant: owner-permanent-preparation-20260819
+- durable capabilities: governed-task:project and release:candidate
+
+## Admission source
+
+The model API authenticates the caller with the existing static token registry
+only to identify the caller and retain the existing static scopes. For the
+exact caller, release:candidate and governed-task:project are not permanent
+static-token grants. The model API asks the Unix-socket Owner broker to
+rehydrate the requested durable capability.
+
+The broker reloads the durable policy and revocation file on every request,
+checks the exact caller, requested capability, policy constraints, optional
+expiry and revocation state, and rejects all capabilities outside the durable
+policy. It never uses a release-worker identity or credential.
+
+## Projection lifecycle and ACK semantics
+
+The server-side flow is:
+
+durable Owner policy
+  -> exact caller and capability validation
+  -> caller projection rows for the policy capabilities
+  -> in-memory short-lived credential generation
+  -> projection ACKED / credential REHYDRATED
+  -> model API admission
+  -> Bridge governed-task admission and execution envelope
+
+The existing caller_projections table stores only projection metadata. A
+rehydration request atomically projects the currently authorized caller
+capabilities, so a valid current policy containing both capabilities returns
+both
+governed-task:project and release:candidate in effective_scopes. No raw
+credential, bearer, token, or credential hash is stored.
+
+ACKED means the broker completed the server-side projection transaction and
+validated the returned public state. It is not an Owner Approval, release
+approval, deployment authorization, or release-worker identity.
+
+## Credential rehydration, expiry and revocation
+
+Credential material is generated server-side, sent only in the private Unix
+socket response header to the model API process, consumed for admission, and
+discarded. The public response and introspection never contain it.
+
+Each request reloads policy state. Missing, malformed, expired or revoked
+policy, caller mismatch, capability mismatch, invalid projection state,
+credential generation failure, or broker outage fails closed. Projection rows
+are short-lived metadata and are rehydrated after process restart; the
+rehydration state is GENERATED or RECOVERED.
+
+The exact caller uses governed-task:project for createGovernedTask admission.
+Other callers keep their existing branch:create path. The existing
+agent-task:control, branch:create and file:write static behavior remains
+available and is not expanded by this repair.
+
+## Bridge execution projection
+
+A governed task created through the existing Owner broker -> Bridge path
+receives metadata.execution with schema aios.online-agent.execution.v2 and an
+execution_projection record. A source_capture placeholder with write
+capabilities is intentionally not claimable. The existing deterministic plan
+registry and execution-rebind CAS path must produce the deterministic plan
+before a worker can claim it.
+
+The legitimate execution chain is:
+
+QUEUED -> PROJECTED -> deterministic-plan/rebind CAS -> CLAIMED
+-> heartbeat -> Writer Lease -> CAS branch/file writes -> tests
+-> official verifier -> immutable runtime evidence.
+
+Never write metadata.execution, task events, plan records, ACKs, outbox rows or
+broker database rows by hand. Never bypass the Bridge with SSH/root or a
+release-worker credential.
+
+## Restart recovery and failure states
+
+After broker or model API restart, repeat the read-only capability check and
+make one normal admission request. The broker must regenerate a short-lived
+credential from the durable policy and report RECOVERED without manual token
+operations.
+
+Useful fail-closed states include:
+
+- PROJECTION_BROKER_UNAVAILABLE
+- POLICY_FILE_INVALID
+- POLICY_EXPIRED
+- OWNER_GRANT_REVOKED
+- CALLER_MISMATCH
+- SCOPE_MISMATCH
+- CAPABILITY_NOT_AUTHORIZED
+- PROJECTION_STATE_INVALID
+- CREDENTIAL_REHYDRATION_MISSING
+- TASK_EXECUTION_PROJECTION_MISSING
+- TASK_PERSISTED_BINDING_CONFLICT
+- EXECUTION_REBIND_CAS_MISMATCH
+
+A rejected state is not repaired by editing SQLite, projection_outbox,
+static token registry, task metadata or task events.
+
+## Whoami and effective-scope diagnosis
+
+Use the read-only model 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
+- projection_error
+- secret_exposed=false
+
+For the authorized current policy, effective_scopes must contain:
+
+- agent-task:control
+- branch:create
+- file:write
+- governed-task:project
+- release:candidate
+
+No credential, token, bearer, password or secret value may appear in the
+response, logs, diagnostics, Git diff or Task metadata.
+
+For broker-level diagnosis use the read-only /v1/capabilities endpoint through
+the authorized Unix peer. It must identify the exact account and grant and
+show both durable capabilities; it must not be used to create manual ACKs.
+
+## Release:candidate troubleshooting
+
+For createReleaseCandidate, the caller must pass only source_commit and
+version to the model API. The durable policy constraint requires an exact
+40-hex source commit and version beginning with V2R2C. The model API first
+rehydrates release:candidate and then invokes the existing release service.
+
+If release admission fails, check caller-capabilities, policy validity,
+revocation state, projection_state, credential_state and rehydration_state.
+Do not add release:candidate to static tokens and do not use
+online-release-builder-cn-network-mkt-20260819.
+
+Creating and reading a Release Candidate is the end of this repair. Owner
+Approval, Deployment, Host Seed, Host Activation, Projector, continuation
+Task, Tag and Baseline actions are outside this scope.

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}

release_version.py

@@ -1,21 +0,0 @@
-"""Canonical AIOS release-version parser shared by runtime and release tools."""
-from __future__ import annotations
-
-import re
-
-
-# The suffix preserves already-supported annotated release names while the
-# canonical VnRnCn namespace remains accepted exactly.
-RELEASE_VERSION_PATTERN = r"^V[0-9]+R[0-9]+C[0-9]+(?:-[A-Za-z0-9_.-]+)?$"
-RELEASE_VERSION_RE = re.compile(RELEASE_VERSION_PATTERN)
-
-
-def parse_release_version(version: object) -> str:
-    value = str(version or "").strip()
-    if not RELEASE_VERSION_RE.fullmatch(value):
-        raise ValueError("invalid release version")
-    return value
-
-
-def validate_release_version(version: object) -> str:
-    return parse_release_version(version)

reports/action-contract-regression-remediation-20260825.md

@@ -1,75 +0,0 @@
-# Action Contract Regression Remediation — 2026-08-25
-
-## Scope
-
-Governed task: `fix-action-contracts-20260825-1407`
-Base: `3244f0f17f8845592ac94d585f5ebd65693c9f29`
-Target branch: `agent/fix-action-contracts-20260825-1407`
-
-## Verified regressions
-
-1. `getReleaseConsistency` and `runReleaseAcceptance` fail before evaluation with:
-   `Gitea API URL must be HTTP(S)`.
-   V2R2C1 release evidence at `a8b906d1bc24088c19647e98de6a679268f6d0a2`
-   proves Gitea API readback passed for target `998d30304787e31e5ae607c0bd86b5edb2b0e5ad`.
-   Therefore current staging/contract overlay configuration has regressed.
-
-2. `getReleaseSaga` fails with:
-   `Permission denied: '/var/lib/aios-gitops/silent-release'`.
-   Runtime service identity must have read/traverse access to the durable release state root.
-   Fix ownership/group/ACL at deployment time; do not broaden to world-writable permissions.
-
-3. `getTagPublicationStatus` rejects the authoritative approved version `V2R2C1`
-   as `invalid release version`, while `getGovernedWriteCapabilities` returns the same
-   version as the approved release scope. The two Action contracts are internally inconsistent.
-   Release-version validation must accept the canonical AIOS `^V[0-9]+R[0-9]+C[0-9]+$`
-   namespace (or, preferably, use one shared validator for all release Actions).
-
-4. `commitFilesGoverned` is unusable from the published Action schema.
-   Runtime capability metadata declares required fields:
-   `task_id, repository, expected_branch_oid, target_branch, generation, idempotency_key, files, message`,
-   but the exposed tool contract has no `files` input. Regenerate/publish the Action schema from
-   the same operation contract source and add a contract-parity test.
-
-5. `reconcileGitOpsJob` returned `DURABLE_BINDING_INVALID` for a completed governed
-   single-file job that was otherwise read back as `COMPLETED`.
-   Reconciliation of a valid terminal job should be idempotent/read-only and must not require
-   a stale or absent active lease. Add a regression case for terminal completed jobs.
-
-6. `getDiff` returns `ResponseTooLargeError` / `GIT_OUTPUT_LIMIT_EXCEEDED` for large diffs.
-   Add server-side bounded output: path/stat summary plus truncation metadata, or pagination.
-   The Action must never fail merely because a valid repository diff is large.
-
-## Required implementation changes
-
-- Centralize release-version parsing and reuse it in tag status, candidate, consistency, and acceptance.
-- Validate/normalize Gitea API URL at service startup and fail health checks with a configuration-specific diagnostic.
-- Ensure the deployment unit projects a valid HTTP(S) Gitea API base URL into the staging overlay.
-- Correct durable release-state directory ownership/ACL for the runtime service account.
-- Generate the external Action schema from the runtime operation contract; parity-test required fields.
-- Make terminal reconciliation idempotent and independent of expired active writer leases.
-- Bound diff output and return `truncated=true` with deterministic metadata rather than transport failure.
-
-## Acceptance gates
-
-All must pass on the same deployed overlay:
-
-- `getReleaseConsistency(V2R2C1, 998d303..., a8b906d..., approved_plan)` returns a structured result, not URL error.
-- `runReleaseAcceptance(...)` reaches policy evaluation, not URL error.
-- `getTagPublicationStatus("V2R2C1")` returns structured publication state.
-- `getReleaseSaga(nonexistent_candidate)` returns NOT_FOUND/empty-state semantics, never filesystem permission failure.
-- Published `commitFilesGoverned` schema exposes `files` and a two-file isolated branch commit succeeds.
-- Reconcile of a completed governed job is idempotent.
-- Large diff returns bounded/truncated data rather than transport exception.
-- Existing protected-ref controls remain unchanged: no direct main/tag/baseline/production write.
-
-## Proven stable reference
-
-V2R2C1 evidence closure (`a8b906d1bc24088c19647e98de6a679268f6d0a2`) records:
-- Gitea Git/API readback PASS
-- staging runtime PASS
-- restart recovery PASS
-- 59 mandatory tests passed / 0 failed
-
-This evidence should be used as the compatibility baseline when repairing the current
-`contract-overlay-v2` / `scheme-a-323b615b...` staging layer.

runbooks/agent-fix-action-contracts-20260825.md

@@ -1,207 +0,0 @@
-# Agent Execution Contract — Action Contract Regression Repair
-
-## Objective
-
-Fully repair the current AIOS Action runtime/contract overlay regressions on the single server, without modifying protected refs directly and without introducing new infrastructure. The task is not complete until the deployed staging/action overlay passes all acceptance gates.
-
-## Authority and scope
-
-- Repository: `aios/aios`
-- Base at task creation: `3244f0f17f8845592ac94d585f5ebd65693c9f29`
-- Governed task: `fix-action-contracts-20260825-1407`
-- Target branch: `agent/fix-action-contracts-20260825-1407`
-- Protected refs must not be directly written: `main`, release tags, baseline registry, production runtime.
-- No new SaaS, database, agent framework, or monitoring platform may be introduced.
-
-## Non-interactive rules
-
-The Agent must run the whole flow unnattended:
-
-1. Collect facts.
-2. Determine root causes.
-3. Make the smallest correct changes.
-4. Run unit/integration/contract tests.
-5. Deploy to staging only.
-6. Restart only the affected non-production services.
-7. Run the full Action smoke suite.
-8. On failure, automatically collect logs, diagnose, fix, and retry.
-9. If a change causes a new regression, automatically roll back the staging change to the previous known-good configuration and record the failure.
-10. Stop only if an owner decision is required for a high-risk, irreversible, or production-boundary action.
-
-## Fact collection — required before editing
-
-Record all of the following in the final evidence bundle:
-
-- `id`, `hostname`, `uname -a`, `date --iso-seconds`
-- `git rev-parse --show HEAD` for authority worktree and running overlay source
-- `git status --porcelaine=v2`
-- `git remote -v`
-- current systemd service unit(s), drop-ins, Environment/EnvironmentFile sources for GitOps/Bredge/release services
-- `systemctl show <affected-service>` including `User=`, `Group=`, `Environment=`, `EnvironmentFile=`, `ExecStart=`
-- resolved Gitea API base URL after env/config expansion; redact secrets but not scheme/host
-- `stat -c, `namei -l`, `getfacl`, als -ld on `/var/lib/aios-gitops/silent-release` and its parents
-- find Action OpenAPI/schema generator and runtime operation definitions for `commitFilesGoverned`
-- find the release version validator(s) and all callers of `getTagPublicationStatus`
-- find `reconcileGitOpsJob` implementation and durable binding validation logic
-- find `getDiff  implementation and transport output limits
-
-Do not edit any file until the actual implementation and configuration sources have been located.
-
-## Repair 1 — Gitea API URL
-
-Symptoms:
-
-- `getReleaseConsistency` → `Gitea API URL must be HTTP(S)`
-- `runReleaseAcceptance` → `Gitea API URL must be HTTP(ÊX
-
-Required behavior:
-
-- Determine which env/config key is projected into the current contract overlay.
-- Compare the deployed overlay with V2R2C1 known-good configuration/deployment manifests.
-- Restore a valid `HTTP://` or `HTTPS://` Gitea API base URL.
-- Add startup config validation so invalid schemes fail at health/startup with a config-specific message, not during a release check.
-- Never hard-code a different Gitea host if the existing deployment already has a single authoritative source.
-
-## Repair 2 — durable release state permissions
-
-Symptom:
-
-- `getReleaseSaga` → `Permission denied: '/var/lib/aios-gitops/silent-release'`
-
-Required behavior:
-
-- Identify the actual runtime service User/Group.
-- Grant only the required traverse/read/write permissions for the durable state directories the service must use.
-- Prefer ownership + group mode or a specific ACL; keep the directory non-world-writable.
-- Persist the permission in deployment/install logic so restart/reboot does not regress.
-- Verify create/read/update of a staging-only saga state file under the service identity.
-
-## Repair 3 — release version contract
-
-Symptom:
-
-- `getTagPublicationStatus("V2R2C1")` → `invalid release version`
-- The same runtime exposes `V2R2C1` as `approved_release_version`.
-
-Required behavior:
-
-- Locate all version validators; consolidate onto one canonical parser.
-- Accept the AIOS canonical namespace `^V[0-9]+R[0-9]+C[0-9]+$`.
-- Preserve any additional already-supported legacy formats unless they are security-sensitive.
- - Add contract tests proving that the version accepted by capabilities is accepted by status, candidate, consistency, acceptance, and tag publication.
-
-## Repair 4 — `commitFilesGoverned` schema parity
-
-Symptom:
-
-- Runtime capability metadata requires `files`.
-- Published Action tool schema does not expose `files`.
-- Call reaches Bridge and fails `BRIDGE_REQUEST_FIELDS_MISSING`.
-
-Required behavior:
-
-- Find the canonical runtime operation contract.
-- Generate the external tool schema from the same source of truth; do not manually copy a parallel schema.
-- `files` must be required and must represent a list of file objects with at least `path` and `content_base64`.
-- Add a parity test that compares runtime required fields with published Action required fields.
-- Add a real governed commit test that writes two test files to an isolated `agent/` ibranch and verifies remote OID.
-
-## Repair 5 — terminal reconciliation
-
-Symptom:
-
-- A governed job read back as `COMPLETED' but `reconcileGitOpsJob` failed `DURABLE_BINDING_INVALID`.
-
-Required behavior:
-
-- Reconciling a terminal successful job must be idempotent.
-- An expired writer lease must not invalidate a durable binding for a job that is already terminal.
-- Active-job reconciliation may require a live lease; terminal read-back must not.
-- Add cases for COMPLETED, FAILED, CANCELLED / equivalent terminal states.
-
-## Repair 6 — bounded `getDiff` output
-
-Symptom:
-
-- Large valid diffs fail with `ResponseTooLargeError` or `GIT_OUTPUT_LIMIT_EXCEEDED`.
-
-Required behavior:
-
-- Do not load unbounded diff text into the Action response.
-- Return bounded summary data: commit identities, changed paths/stats, total bytes, and `truncated: true` when the full patch exceeds the safe limit.
-- If pagination already exists in the contract use it; otherwise prefer deterministic truncation over a breaking schema change.
-- Add tests for small, empty, and oversize diffs.
-
-## Staging deployment procedure
-
-1. Run the existing mandatory test suite before modification; record baseline.
-2. Apply code and config changes only on the governed branch/worktree.
-3. Run focused regression tests.
-4. Run the full mandatory suite.
-5. Build the staging contract overlay using the repository's existing build/deploy path.
-6. Deploy to staging/non-production only.
-7. If systemd units are changed: `systemctl daemon-reload`, then restart only the affected staging services.
-8. Wait for the existing health endpoint/runtime status to become healthy.
-9. Run the acceptance suite below.
-
-If any step fails:
-- collect `journalctl` logs and relevant app logs;
-- diagnose root cause;
-- automatically repair and retry when non-destructive;
-- if a new change breaks staging, restore the previous staging artifact/config and restart the affected service;
-- never silently continue after a failed gate.
-
-## Acceptance suite
-
-All of the following must pass on the same deployed staging overlay:
-
-- Runtime status is `active` and healthy after restart.
-- Gitea Git/API readback works from the runtime service identity.
-- `getReleaseConsistency(V2R2C1, 998d30304787e31e5ae607c0bd86b5edb2b0e5ad, a8b906d1bc24088c19647e98de6a679268f6d0a2, approved_plan) returns a structured result and reaches evaluation.
-- `runReleaseAcceptance` for the same identities reaches policy evaluation, not URL validation error.
-- `getTagPublicationStatus("V2R2C1")` returns structured state, not `invalid release version`.
-- `getReleaseSaga` for a nonexistent candidate returns NOT_FOUND/empty-state semantics, not filesystem permission failure.
-- `commitFilesGoverned` exposes its required `files` parameter in the published Action schema.
- - A real two-file governed commit to an isolated `agent/` branch succeeds and remote branch OID matches.
-- Resuming a COMPLETED job returns the existing `JOB_TERMINAL` semantics or idempotent success as designed.
-- Reconciling a COMPLETED governed job is successful/idempotent and does not require an active lease.
-- Small/empty `getDiff  still works.
-- Large `getDiff` returns bounded output with truncation metadata instead of a transport exception.
-- Existing protected-ref controls remain unchange: no direct main/tag/baseline/production write.
-
-## Restart / reboot persistence
-
-After the first full pass:
-
-- Restart the staging Action/Bridge/release services one more time without changing files.
-- Rerun the acceptance suite.
-- Permissions, Gitea URL, and schema must still be correct after restart.
-- If the repository already has a safe staging reboot test, run it as well.
-
-## Evidence and completion
-
-Create a single durable report in the repository, e.g. `reports/action-contract-repair-evidence-20260825.md`, containing:
-
-- pre-change facts
-- root cause for each regression
-- files changed and why
-- config/permission changes
-- test commands and exact results
-- staging deployment identity (commit/artifact/config digest)
-- restart-persistence results
-- final Action smoke results
-- any remaining risks
-
-Commit all source, test, deployment config, and runbook changes to the governed task branch. Do not mark the task SUCCEEDED until all acceptance gates pass on the actually deployed staging overlay.
-
-## Automatic rollback conditions
-
-Automatically roll back the staging deployment if any of the following occur:
-
-- service fails to become healthy after restart
-- protected-ref write boundary is broken
-- Gitea readback that worked before the change stops working
-- durable saga state cannot be read or recovered
-- mandatory test suite regresses below the pre-change baseline
-
-Do not automatically roll back an existing production release. Rollback here refers only to the staging overlay/service configuration changes made by this task.

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

scripts/build_v2r2c1_model_api_staging.py

@@ -5,7 +5,6 @@ import hashlib
 import json
 import os
 from pathlib import Path
-import sys
 import tarfile
 import tempfile
 
@@ -19,18 +18,6 @@ REQUIRED_APP = (
     'silent_release_controller.py',
     'silent_release_hardening.py',
     'aios_gitops_job_controller.py',
-    'aios_chatgpt_bridge_v2.py',
-    'aios_chatgpt_bridge_v2_server.py',
-    'governed_production_release.py',
-    'v2r2c1_production_adapter.py',
-    'v2r1c7_production_adapter.py',
-    'aios_governed_task_bootstrap.py',
-    'aios_continuation_executor.py',
-    'aios_continuation_lease.py',
-    'aios_authority_git_executor.py',
-    'aios_gitops_reconciliation_core.py',
-    'authorized_task_writer_provisioning.py',
-    'signer_contract_compatibility.py',
 )
 OPTIONAL_APP = (
     'aios_chatgpt_actions_api.py',
@@ -69,18 +56,11 @@ def main() -> int:
     source = args.source_root.resolve(strict=True)
     output = args.output_root.resolve()
     app = source / 'active-runtime' / 'app'
-    sys.path.insert(0, str(app))
-    from aios_chatgpt_bridge_v2 import render_openapi_contract_fragment
-
     runtime = source / 'runtime' / 'agent-os'
     docs = source / 'docs' / 'gitops'
     if not app.is_dir():
         raise SystemExit('active-runtime/app missing')
     paths = []
-    root_version = source / 'release_version.py'
-    if not root_version.is_file():
-        raise SystemExit('required source dependency missing: release_version.py')
-    paths.append((Path('release_version.py'), root_version))
     for name in REQUIRED_APP:
         path = app / name
         if not path.is_file():
@@ -97,8 +77,6 @@ def main() -> int:
         paths.append((Path('runtime-agent-os') / rel, path))
     openapi = docs / 'openapi-chatgpt-actions.yaml'
     if openapi.is_file():
-        if '  /v2/commit-files:' not in openapi.read_text(encoding='utf-8'):
-            raise SystemExit('published OpenAPI is missing generated commitFilesGoverned contract')
         paths.append((Path('docs') / 'gitops' / 'openapi-chatgpt-actions.yaml', openapi))
     for folder in ('config/model_registry', 'config/model_priority', 'patches', 'versions'):
         root = source / folder

scripts/release_version.sh

@@ -9,16 +9,10 @@ WORKSPACE="${AIOS_GITOPS_WORKSPACE:-/data/aios-gitops/workspace}"
 TAG_REMOTE="${AIOS_GITOPS_GITEA_REMOTE:-gitea}"
 BRANCH_REMOTE="${AIOS_GITOPS_BRANCH_REMOTE:-gitea}"
 
-SCRIPT_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)"
-if ! PYTHONPATH="$SCRIPT_ROOT:${PYTHONPATH:-}" python3 - "$VERSION" <<'PY'
-import sys
-from release_version import validate_release_version
-validate_release_version(sys.argv[1])
-PY
-then
-  echo "ERROR: version must match VxRxCx" >&2
-  exit 64
-fi
+case "$VERSION" in
+  V[0-9]*R[0-9]*C[0-9]*) ;;
+  *) echo "ERROR: version must match VxRxCx" >&2; exit 64 ;;
+esac
 
 for value in "$TARGET_OID" "$EVIDENCE_CLOSURE_OID"; do
   [[ "$value" =~ ^[0-9a-f]{40}$ ]] || {

services/aios-release-controller/aios_release_controller/manifest.py

@@ -1,9 +1,8 @@
 from __future__ import annotations
 import hashlib,json,re,tarfile
-from release_version import RELEASE_VERSION_RE
 from pathlib import Path
 from typing import Any
-OID_RE=re.compile(r"^[0-9a-f]{40}$"); SHA_RE=re.compile(r"^[0-9a-f]{64}$"); VERSION_RE=RELEASE_VERSION_RE
+OID_RE=re.compile(r"^[0-9a-f]{40}$"); SHA_RE=re.compile(r"^[0-9a-f]{64}$"); VERSION_RE=re.compile(r"^V[0-9]+R[0-9]+C[0-9]+(?:-[A-Za-z0-9_.-]+)?$")
 REQUIRED_SOURCE={"schema","version","repository","base_ref","base_oid","binding_mode","release_readiness","production_deployed","tag_created","baseline_changed","production_service","production_port","rollback_metadata","controller_schema","required_checks"}
 REQUEST_KEYS={"schema","version","repository","base_oid","target_oid","artifact_uri","artifact_sha256","artifact_size","evidence_closure_oid","approval_plan_id","execute","production_deployed","production_service","production_port","rollback_metadata","current_tag_oid","source_manifest_sha256"}
 class ManifestValidationError(ValueError):

tests/contract-integration-a.txt

@@ -1 +0,0 @@
-action-contract-integration-a

tests/contract-integration-b.txt

@@ -1 +0,0 @@
-action-contract-integration-b

tests/test_action_contract_regressions.py

@@ -1,116 +0,0 @@
-from __future__ import annotations
-
-import json
-import sys
-from pathlib import Path
-
-import pytest
-
-ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(ROOT))
-
-from aios_chatgpt_bridge_v2 import (  # noqa: E402
-    OPERATION_REQUIRED_FIELDS,
-    operation_contracts,
-    render_openapi_contract_fragment,
-)
-from aios_chatgpt_bridge_v2_server import GovernedRuntime  # noqa: E402
-from aios_gitops_job_controller import JobState  # noqa: E402
-from gitea_tag_publication import validate_gitea_api_url  # noqa: E402
-from release_version import validate_release_version  # noqa: E402
-
-
-def test_canonical_release_version_is_shared_and_accepts_v2r2c1():
-    assert validate_release_version("V2R2C1") == "V2R2C1"
-    assert validate_release_version("V2R2C1-repair") == "V2R2C1-repair"
-    with pytest.raises(ValueError):
-        validate_release_version("v2r2c1")
-
-
-def test_gitea_url_validation_is_startup_safe():
-    assert validate_gitea_api_url("https://g.cnnetmkt.top/git/api/v1") == (
-        "https://g.cnnetmkt.top/git/api/v1"
-    )
-    with pytest.raises(ValueError, match="HTTP"):
-        validate_gitea_api_url("g.cnnetmkt.top/git/api/v1")
-
-
-def test_commit_files_runtime_and_published_schema_have_same_required_fields():
-    runtime = operation_contracts()["commitFilesGoverned"]["required_fields"]
-    assert runtime == list(OPERATION_REQUIRED_FIELDS["commitFilesGoverned"])
-    fragment = render_openapi_contract_fragment()
-    published = (ROOT / "docs/gitops/openapi-chatgpt-actions.yaml").read_text()
-    assert fragment in published
-    for field in runtime:
-        assert f"                - {field}" in fragment
-    assert "required: [path, content_base64]" in fragment
-
-
-class _Store:
-    def __init__(self, row):
-        self.row = row
-        self.receipt_calls = 0
-
-    def read(self, _job_id):
-        return self.row
-
-    def _receipt(self, row, duplicate=False):
-        self.receipt_calls += 1
-        return type("Receipt", (), {"as_dict": lambda _self: {
-            "job_id": row["job_id"],
-            "state": row["state"],
-            "duplicate": duplicate,
-        }})()
-
-
-class _Controller:
-    def __init__(self, row):
-        self.store = _Store(row)
-
-
-def _runtime_for(row):
-    runtime = object.__new__(GovernedRuntime)
-    controller = _Controller(row)
-    runtime._controller_for_job = lambda _job_id: controller
-    runtime._executor_from_row = lambda *_args: (_ for _ in ()).throw(
-        AssertionError("terminal reconcile must not reconstruct an executor")
-    )
-    return runtime, controller
-
-
-@pytest.mark.parametrize(
-    "state",
-    [
-        JobState.COMPLETED.value,
-        JobState.FAILED.value,
-        JobState.CANCELLED.value,
-        JobState.MANUAL_OWNER_REVIEW_REQUIRED.value,
-    ],
-)
-def test_terminal_reconcile_is_idempotent_without_durable_binding(state):
-    row = {
-        "job_id": "job-terminal",
-        "state": state,
-        "request_json": json.dumps({}),
-    }
-    runtime, controller = _runtime_for(row)
-    result = GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile")
-    assert result["duplicate"] is True
-    assert controller.store.receipt_calls == 1
-
-
-def test_active_reconcile_still_uses_executor():
-    row = {
-        "job_id": "job-active",
-        "state": JobState.RECONCILIATION_REQUIRED.value,
-        "request_json": json.dumps({}),
-    }
-    runtime, controller = _runtime_for(row)
-    class Executor:
-        def reconcile(self, job_id):
-            return {"job_id": job_id, "state": "COMPLETED"}
-    runtime._executor_from_row = lambda *_args: Executor()
-    assert GovernedRuntime.job(runtime, {"job_id": row["job_id"]}, "reconcile") == {
-        "job_id": "job-active",
-        "state": "COMPLETED",
-    }

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,215 @@
+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"
+GOVERNED_TASK_SCOPE = "governed-task:project"
+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 {GOVERNED_TASK_SCOPE, 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(
+            [GOVERNED_TASK_SCOPE, SCOPE],
+            value["effective_scopes"],
+        )
+        self.assertFalse(value["secret_exposed"])
+
+    def test_governed_task_projection_uses_durable_capability(self):
+        broker = make_broker(Path(tempfile.mkdtemp()))
+        with patch.object(broker, "policy", return_value=policy()):
+            value = broker.project_caller(
+                CALLER,
+                GOVERNED_TASK_SCOPE,
+                {},
+                idempotency_key="caller-projection-governed-task-v1",
+            )
+        value.pop("_credential")
+        self.assertEqual(GOVERNED_TASK_SCOPE, value["scope"])
+        self.assertEqual(
+            [GOVERNED_TASK_SCOPE, 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,193 @@
+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"
+GOVERNED_TASK_SCOPE = "governed-task:project"
+
+
+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": [GOVERNED_TASK_SCOPE, SCOPE],
+        },
+        credential_loaded=True,
+    )
+
+    class _ProjectionClient:
+        def rehydrate(self, caller, scope, request):
+            assert caller == CALLER
+            assert scope == GOVERNED_TASK_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 GOVERNED_TASK_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 == []