chore: repair GPT Actions OpenAPI projection to 27 operations

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_initial_execution_projection.py

@@ -192,6 +192,140 @@ class InitialExecutionProjectionRequest:
         }
 
 
+@dataclass(frozen=True)
+class SourceCaptureProjectionRecoveryRequest:
+    """CAS contract for the first durable runtime projection after source capture.
+
+    Some deployed Bridge versions materialize a binding by letting the generic
+    worker perform one read-only ``source_capture`` attempt.  This request does
+    not relax the pristine initial-projection contract above.  It authorizes a
+    separate, evidence-bound transition only after that attempt has been
+    officially verified as ``NEEDS_REWORK`` and returned to ``QUEUED``.
+    """
+
+    expected_status: str
+    expected_revision: int
+    expected_attempt: int
+    expected_metadata_sha256: str
+    expected_required_capabilities_sha256: str
+    previous_verification_sha256: str
+    expected_source_capture_result_sha256: str
+    expected_source_capture_execution_digest: str
+    expected_job_id: str
+    expected_execution_digest: str
+    required_capability: str
+    adapter: str
+    authorization_id: str
+    idempotency_key: str
+    reason: str
+
+    @classmethod
+    def from_mapping(
+        cls, value: Mapping[str, Any]
+    ) -> "SourceCaptureProjectionRecoveryRequest":
+        allowed = {
+            "expected_status",
+            "expected_revision",
+            "expected_attempt",
+            "expected_metadata_sha256",
+            "expected_required_capabilities_sha256",
+            "previous_verification_sha256",
+            "expected_source_capture_result_sha256",
+            "expected_source_capture_execution_digest",
+            "expected_job_id",
+            "expected_execution_digest",
+            "required_capability",
+            "adapter",
+            "authorization_id",
+            "idempotency_key",
+            "reason",
+        }
+        if set(value) != allowed:
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_PROJECTION_RECOVERY_FIELDS_INVALID", 400
+            )
+        status = _require_string(value["expected_status"], "EXPECTED_STATUS_INVALID")
+        revision = value["expected_revision"]
+        attempt = value["expected_attempt"]
+        if (
+            status != "QUEUED"
+            or not isinstance(revision, int)
+            or isinstance(revision, bool)
+            or revision < 2
+            or not isinstance(attempt, int)
+            or isinstance(attempt, bool)
+            or attempt < 1
+        ):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_PROJECTION_RECOVERY_STATE_INVALID", 400
+            )
+        digest_fields = {
+            "expected_metadata_sha256": "METADATA_SHA256_INVALID",
+            "expected_required_capabilities_sha256": "CAPABILITIES_SHA256_INVALID",
+            "previous_verification_sha256": "VERIFICATION_SHA256_INVALID",
+            "expected_source_capture_result_sha256": "SOURCE_CAPTURE_RESULT_SHA256_INVALID",
+            "expected_source_capture_execution_digest": "SOURCE_CAPTURE_EXECUTION_DIGEST_INVALID",
+            "expected_execution_digest": "EXECUTION_DIGEST_INVALID",
+        }
+        digests: dict[str, str] = {}
+        for name, code in digest_fields.items():
+            digests[name] = _require_string(value[name], code, SHA_RE)
+        job_id = _require_string(value["expected_job_id"], "JOB_ID_INVALID", JOB_RE)
+        capability = _require_string(
+            value["required_capability"],
+            "RUNTIME_CAPABILITY_INVALID",
+            RUNTIME_CAPABILITY_RE,
+        )
+        adapter = _require_string(value["adapter"], "RUNTIME_ADAPTER_INVALID")
+        if capability != f"agent.runtime.{adapter}" or adapter not in {"hermes", "openclaw"}:
+            raise InitialExecutionProjectionError("RUNTIME_CAPABILITY_MISMATCH", 400)
+        authorization_id = _require_string(
+            value["authorization_id"], "AUTHORIZATION_ID_INVALID", KEY_RE
+        )
+        idempotency_key = _require_string(
+            value["idempotency_key"], "IDEMPOTENCY_KEY_INVALID", KEY_RE
+        )
+        reason = _require_string(value["reason"], "PROJECTION_REASON_INVALID")
+        if len(reason) > 512:
+            raise InitialExecutionProjectionError("PROJECTION_REASON_INVALID", 400)
+        return cls(
+            status,
+            revision,
+            attempt,
+            digests["expected_metadata_sha256"],
+            digests["expected_required_capabilities_sha256"],
+            digests["previous_verification_sha256"],
+            digests["expected_source_capture_result_sha256"],
+            digests["expected_source_capture_execution_digest"],
+            job_id,
+            digests["expected_execution_digest"],
+            capability,
+            adapter,
+            authorization_id,
+            idempotency_key,
+            reason,
+        )
+
+    def public(self) -> dict[str, Any]:
+        return {
+            "expected_status": self.expected_status,
+            "expected_revision": self.expected_revision,
+            "expected_attempt": self.expected_attempt,
+            "expected_metadata_sha256": self.expected_metadata_sha256,
+            "expected_required_capabilities_sha256": self.expected_required_capabilities_sha256,
+            "previous_verification_sha256": self.previous_verification_sha256,
+            "expected_source_capture_result_sha256": self.expected_source_capture_result_sha256,
+            "expected_source_capture_execution_digest": self.expected_source_capture_execution_digest,
+            "expected_job_id": self.expected_job_id,
+            "expected_execution_digest": self.expected_execution_digest,
+            "required_capability": self.required_capability,
+            "adapter": self.adapter,
+            "authorization_id": self.authorization_id,
+            "idempotency_key": self.idempotency_key,
+            "reason": self.reason,
+        }
+
+
 class CanonicalBridgeInitialExecutionResolver:
     """Resolve one agent-runtime envelope from the canonical governed Bridge."""
 
@@ -377,6 +511,191 @@ class InitialExecutionProjectionStore:
             """
         )
 
+    @staticmethod
+    def _ensure_source_capture_recovery_table(
+        connection: sqlite3.Connection,
+    ) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS task_source_capture_projection_recoveries (
+              task_id TEXT NOT NULL,
+              idempotency_key TEXT NOT NULL,
+              request_sha256 TEXT NOT NULL,
+              authorization_id TEXT NOT NULL,
+              expected_job_id TEXT NOT NULL,
+              source_capture_result_sha256 TEXT NOT NULL,
+              source_capture_execution_digest TEXT NOT NULL,
+              execution_digest TEXT NOT NULL,
+              required_capability TEXT NOT NULL,
+              adapter TEXT NOT NULL,
+              expected_revision INTEGER NOT NULL,
+              applied_revision INTEGER NOT NULL,
+              attempt INTEGER NOT NULL,
+              created_at TEXT NOT NULL,
+              PRIMARY KEY(task_id,idempotency_key)
+            )
+            """
+        )
+
+    @classmethod
+    def _source_capture_event_chain(
+        cls, connection: sqlite3.Connection, task_id: str
+    ) -> None:
+        table, columns = cls._event_table(connection)
+        ordering = "event_id" if "event_id" in columns else "rowid"
+        rows = connection.execute(
+            f'SELECT event_type, payload_json FROM "{table}" '
+            f'WHERE task_id=? ORDER BY {ordering}',
+            (task_id,),
+        ).fetchall()
+        events = [str(row[0]) for row in rows]
+        if (
+            len(events) < 5
+            or events[0] != "TASK_CREATED"
+            or events.count("TASK_CLAIMED") != 1
+            or events.count("RESULT_SUBMITTED") != 1
+            or events.count("TASK_VERIFIED") != 1
+            or events[-1] != "TASK_VERIFIED"
+        ):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_RECOVERY_EVENT_CHAIN_INVALID", 409
+            )
+        claim = events.index("TASK_CLAIMED")
+        result = events.index("RESULT_SUBMITTED")
+        verified = events.index("TASK_VERIFIED")
+        heartbeat_positions = [
+            index for index, event in enumerate(events) if event == "TASK_HEARTBEAT"
+        ]
+        if (
+            not heartbeat_positions
+            or not all(claim < index < result for index in heartbeat_positions)
+            or not claim < result < verified
+            or any(
+                event not in {
+                    "TASK_CREATED",
+                    "TASK_CLAIMED",
+                    "TASK_HEARTBEAT",
+                    "RESULT_SUBMITTED",
+                    "TASK_VERIFIED",
+                }
+                for event in events
+            )
+        ):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_RECOVERY_EVENT_CHAIN_INVALID", 409
+            )
+        try:
+            verification_event = json.loads(rows[-1][1])
+        except (TypeError, json.JSONDecodeError) as exc:
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_RECOVERY_VERIFICATION_EVENT_INVALID", 409
+            ) from exc
+        if (
+            not isinstance(verification_event, Mapping)
+            or verification_event.get("decision") != "NEEDS_REWORK"
+            or verification_event.get("next_state") != "QUEUED"
+        ):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_RECOVERY_VERIFICATION_EVENT_INVALID", 409
+            )
+
+    def _load_source_capture_recovery(
+        self,
+        connection: sqlite3.Connection,
+        task_id: str,
+        request: SourceCaptureProjectionRecoveryRequest,
+    ) -> tuple[sqlite3.Row, dict[str, Any], dict[str, Any]]:
+        columns = self._columns(connection, "tasks")
+        required = {
+            "task_id", "goal", "instructions", "success_criteria_json",
+            "constraints_json", "required_capabilities_json", "metadata_json",
+            "status", "attempt", "revision", "claimed_by", "lease_token_hash",
+            "lease_expires_at", "result_json", "verification_json",
+        }
+        if not required.issubset(columns):
+            raise InitialExecutionProjectionError(
+                "TASK_SCHEMA_SOURCE_CAPTURE_RECOVERY_UNSUPPORTED", 503
+            )
+        task = connection.execute(
+            "SELECT * FROM tasks WHERE task_id=?", (task_id,)
+        ).fetchone()
+        if task is None:
+            raise InitialExecutionProjectionError("TASK_NOT_FOUND", 404)
+        if (
+            str(task["status"]) != request.expected_status
+            or int(task["revision"]) != request.expected_revision
+            or int(task["attempt"]) != request.expected_attempt
+        ):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_PROJECTION_RECOVERY_CAS_MISMATCH", 409
+            )
+        if task["claimed_by"] not in (None, ""):
+            raise InitialExecutionProjectionError("TASK_ALREADY_CLAIMED", 409)
+        if task["lease_token_hash"] not in (None, "") or task["lease_expires_at"] not in (None, ""):
+            raise InitialExecutionProjectionError("ACTIVE_LEASE_DENY", 409)
+        metadata = self._json_value(task["metadata_json"], "TASK_METADATA_INVALID")
+        capabilities = self._json_value(
+            task["required_capabilities_json"], "TASK_CAPABILITIES_INVALID"
+        )
+        result = self._json_value(
+            task["result_json"], "SOURCE_CAPTURE_RESULT_INVALID"
+        )
+        verification = self._json_value(
+            task["verification_json"], "SOURCE_CAPTURE_VERIFICATION_INVALID"
+        )
+        if (
+            not isinstance(metadata, Mapping)
+            or not isinstance(capabilities, list)
+            or not isinstance(result, Mapping)
+            or not isinstance(verification, Mapping)
+        ):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_RECOVERY_STATE_INVALID", 409
+            )
+        metadata = dict(metadata)
+        result = dict(result)
+        verification = dict(verification)
+        if "execution" in metadata or "execution_projection" in metadata:
+            raise InitialExecutionProjectionError("TASK_EXECUTION_ALREADY_PROJECTED", 409)
+        if capabilities:
+            raise InitialExecutionProjectionError(
+                "TASK_INITIAL_CAPABILITIES_MUST_BE_EMPTY", 409
+            )
+        if _sha(metadata) != request.expected_metadata_sha256:
+            raise InitialExecutionProjectionError("TASK_METADATA_DRIFT", 409)
+        if _sha(capabilities) != request.expected_required_capabilities_sha256:
+            raise InitialExecutionProjectionError("TASK_CAPABILITIES_DRIFT", 409)
+        if _sha(result) != request.expected_source_capture_result_sha256:
+            raise InitialExecutionProjectionError("SOURCE_CAPTURE_RESULT_DRIFT", 409)
+        if _sha(verification) != request.previous_verification_sha256:
+            raise InitialExecutionProjectionError("SOURCE_CAPTURE_VERIFICATION_DRIFT", 409)
+        if verification.get("decision") != "NEEDS_REWORK":
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_NEEDS_REWORK_VERIFICATION_REQUIRED", 409
+            )
+        expected_result = {
+            "task_id": task_id,
+            "job_id": request.expected_job_id,
+            "status": "SUCCEEDED",
+            "classification": "NEEDS_REWORK_SOURCE_CAPTURE",
+            "execution_mode": "source_capture",
+            "execution_digest": request.expected_source_capture_execution_digest,
+            "changed_files": [],
+            "writer_lease": "NONE",
+            "verification_state": "NEEDS_REWORK",
+            "direct_main_write": False,
+            "production_write": False,
+            "force_push": False,
+            "protected_ref_bypass": False,
+            "secret_material_exposed": False,
+        }
+        if any(result.get(name) != expected for name, expected in expected_result.items()):
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_RESULT_CONTRACT_INVALID", 409
+            )
+        self._source_capture_event_chain(connection, task_id)
+        return task, metadata, result
+
     def _load_and_validate(
         self,
         connection: sqlite3.Connection,
@@ -632,6 +951,222 @@ class InitialExecutionProjectionStore:
                 "idempotent": False,
             }
 
+    def project_from_source_capture(
+        self,
+        task_id: str,
+        request: SourceCaptureProjectionRecoveryRequest,
+    ) -> dict[str, Any]:
+        task_id = _require_string(task_id, "TASK_ID_INVALID", TASK_RE)
+        if request.authorization_id != _required_authorization_id():
+            raise InitialExecutionProjectionError("AUTHORIZATION_ID_MISMATCH", 403)
+        request_sha = _sha(request.public())
+
+        def replay_result(row: sqlite3.Row) -> dict[str, Any]:
+            return {
+                "task_id": task_id,
+                "status": "QUEUED",
+                "attempt": int(row["attempt"]),
+                "revision": int(row["applied_revision"]),
+                "job_id": row["expected_job_id"],
+                "source_capture_execution_digest": row[
+                    "source_capture_execution_digest"
+                ],
+                "execution_digest": row["execution_digest"],
+                "required_capability": row["required_capability"],
+                "adapter": row["adapter"],
+                "authorization_id": row["authorization_id"],
+                "recovery_from_source_capture": True,
+                "idempotent": True,
+            }
+
+        with self._connect() as connection:
+            self._ensure_source_capture_recovery_table(connection)
+            replay = connection.execute(
+                "SELECT * FROM task_source_capture_projection_recoveries "
+                "WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise InitialExecutionProjectionError("IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return replay_result(replay)
+
+        with self._connect() as connection:
+            task, _, result = self._load_source_capture_recovery(
+                connection, task_id, request
+            )
+            runtime_input = self._runtime_input(task)
+
+        resolved = dict(
+            self.resolver(
+                task_id,
+                request.expected_job_id,
+                request.required_capability,
+                request.adapter,
+                runtime_input,
+            )
+        )
+        binding = GovernedExecutionBinding.from_mapping(resolved)
+        if binding.task_id != task_id or binding.job_id != request.expected_job_id:
+            raise InitialExecutionProjectionError("GOVERNED_BINDING_CONFLICT", 409)
+        result_binding = {
+            "task_id": result.get("task_id"),
+            "job_id": result.get("job_id"),
+            "repository": result.get("repository"),
+            "base_oid": result.get("base_oid"),
+            "expected_branch_oid": result.get("expected_branch_oid"),
+            "target_branch": result.get("target_branch"),
+            "generation": result.get("generation"),
+        }
+        expected_binding = {
+            "task_id": binding.task_id,
+            "job_id": binding.job_id,
+            "repository": binding.repository,
+            "base_oid": binding.base_oid,
+            "expected_branch_oid": binding.expected_branch_oid,
+            "target_branch": binding.target_branch,
+            "generation": binding.generation,
+        }
+        if result_binding != expected_binding:
+            raise InitialExecutionProjectionError(
+                "SOURCE_CAPTURE_GOVERNED_BINDING_CONFLICT", 409
+            )
+        execution = validate_execution_envelope(resolved, binding=binding)
+        if (
+            execution.get("action") != "agent_runtime"
+            or execution.get("adapter") != request.adapter
+            or execution.get("capability_scope") != [request.required_capability]
+            or execution.get("change_set") != []
+            or execution.get("verification") != []
+            or execution.get("governance", {}).get("writer_lease") is not False
+        ):
+            raise InitialExecutionProjectionError("AGENT_RUNTIME_ENVELOPE_INVALID", 409)
+        if execution["execution_digest"] != request.expected_execution_digest:
+            raise InitialExecutionProjectionError("EXECUTION_DIGEST_DRIFT", 409)
+
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._ensure_source_capture_recovery_table(connection)
+            replay = connection.execute(
+                "SELECT * FROM task_source_capture_projection_recoveries "
+                "WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise InitialExecutionProjectionError("IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return replay_result(replay)
+            task, metadata, _ = self._load_source_capture_recovery(
+                connection, task_id, request
+            )
+            new_metadata = dict(metadata)
+            new_metadata["execution"] = execution
+            new_metadata["execution_projection"] = {
+                "schema": "aios.online-agent.task-projection.v1",
+                "task_id": task_id,
+                "job_id": request.expected_job_id,
+                "execution_digest": execution["execution_digest"],
+            }
+            applied_revision = request.expected_revision + 1
+            cursor = connection.execute(
+                """
+                UPDATE tasks
+                   SET metadata_json=?, required_capabilities_json=?,
+                       revision=?, updated_at=?
+                 WHERE task_id=? AND status=? AND revision=? AND attempt=?
+                   AND claimed_by IS NULL AND lease_token_hash IS NULL
+                   AND lease_expires_at IS NULL
+                   AND metadata_json=? AND required_capabilities_json=?
+                   AND result_json=? AND verification_json=?
+                """,
+                (
+                    json.dumps(new_metadata, sort_keys=True, separators=(",", ":")),
+                    json.dumps([request.required_capability], separators=(",", ":")),
+                    applied_revision,
+                    _now(),
+                    task_id,
+                    request.expected_status,
+                    request.expected_revision,
+                    request.expected_attempt,
+                    task["metadata_json"],
+                    task["required_capabilities_json"],
+                    task["result_json"],
+                    task["verification_json"],
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise InitialExecutionProjectionError(
+                    "SOURCE_CAPTURE_PROJECTION_RECOVERY_CAS_MISMATCH", 409
+                )
+            event_payload = {
+                "schema": "aios.task-initial-execution-projection.event.v1",
+                "task_id": task_id,
+                "job_id": request.expected_job_id,
+                "authorization_id": request.authorization_id,
+                "idempotency_key": request.idempotency_key,
+                "reason": request.reason,
+                "action": "agent_runtime",
+                "adapter": request.adapter,
+                "required_capability": request.required_capability,
+                "source_capture_execution_digest": request.expected_source_capture_execution_digest,
+                "source_capture_result_sha256": request.expected_source_capture_result_sha256,
+                "previous_verification_sha256": request.previous_verification_sha256,
+                "execution_digest": execution["execution_digest"],
+                "previous_revision": request.expected_revision,
+                "applied_revision": applied_revision,
+                "attempt": request.expected_attempt,
+                "request_sha256": request_sha,
+                "recovery_from_source_capture": True,
+                "writer_lease": False,
+                "main_write": False,
+                "production_write": False,
+            }
+            self._insert_event(connection, task_id, event_payload)
+            connection.execute(
+                """
+                INSERT INTO task_source_capture_projection_recoveries(
+                  task_id,idempotency_key,request_sha256,authorization_id,
+                  expected_job_id,source_capture_result_sha256,
+                  source_capture_execution_digest,execution_digest,
+                  required_capability,adapter,expected_revision,applied_revision,
+                  attempt,created_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    task_id,
+                    request.idempotency_key,
+                    request_sha,
+                    request.authorization_id,
+                    request.expected_job_id,
+                    request.expected_source_capture_result_sha256,
+                    request.expected_source_capture_execution_digest,
+                    execution["execution_digest"],
+                    request.required_capability,
+                    request.adapter,
+                    request.expected_revision,
+                    applied_revision,
+                    request.expected_attempt,
+                    _now(),
+                ),
+            )
+            connection.commit()
+            return {
+                "task_id": task_id,
+                "status": "QUEUED",
+                "attempt": request.expected_attempt,
+                "revision": applied_revision,
+                "job_id": request.expected_job_id,
+                "source_capture_execution_digest": request.expected_source_capture_execution_digest,
+                "execution_digest": execution["execution_digest"],
+                "required_capability": request.required_capability,
+                "adapter": request.adapter,
+                "authorization_id": request.authorization_id,
+                "recovery_from_source_capture": True,
+                "idempotent": False,
+            }
+
 
 class InitialExecutionProjectionHandler(BaseHTTPRequestHandler):
     store: InitialExecutionProjectionStore | None = None
@@ -666,8 +1201,13 @@ class InitialExecutionProjectionHandler(BaseHTTPRequestHandler):
             self._json({"error": "NOT_FOUND"}, 404)
 
     def do_POST(self) -> None:
-        match = re.fullmatch(r"/v1/tasks/([^/]+)/initial-execution-projection", self.path)
-        if match is None:
+        initial_match = re.fullmatch(
+            r"/v1/tasks/([^/]+)/initial-execution-projection", self.path
+        )
+        recovery_match = re.fullmatch(
+            r"/v1/tasks/([^/]+)/source-capture-projection-recovery", self.path
+        )
+        if initial_match is None and recovery_match is None:
             self._json({"error": "NOT_FOUND"}, 404)
             return
         if not self._authorized():
@@ -685,10 +1225,18 @@ class InitialExecutionProjectionHandler(BaseHTTPRequestHandler):
             value = json.loads(self.rfile.read(length).decode("utf-8"))
             if not isinstance(value, Mapping):
                 raise ValueError
-            request = InitialExecutionProjectionRequest.from_mapping(value)
             if self.store is None:
                 raise InitialExecutionProjectionError("RUNTIME_UNAVAILABLE", 503)
-            result = self.store.project(match.group(1), request)
+            if initial_match is not None:
+                request = InitialExecutionProjectionRequest.from_mapping(value)
+                result = self.store.project(initial_match.group(1), request)
+            else:
+                recovery_request = SourceCaptureProjectionRecoveryRequest.from_mapping(
+                    value
+                )
+                result = self.store.project_from_source_capture(
+                    recovery_match.group(1), recovery_request
+                )
             self._json(result, 200 if result["idempotent"] else 201)
         except InitialExecutionProjectionError as exc:
             self._json({"error": exc.code}, exc.status)

active-runtime/app/aios_agent_task_protocol.py

@@ -0,0 +1,870 @@
+#!/usr/bin/env python3
+"""Durable online task protocol for external AIOS agents.
+
+The store is deliberately independent from Git/Gitea mutation. It provides a
+small durable state machine for publishing work, atomic claim/lease ownership,
+heartbeat, idempotent result submission, explicit server-side verification,
+lease recovery, and audit events.
+
+No plaintext agent credential or task lease token is persisted.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+import hashlib
+import hmac
+import json
+from pathlib import Path
+import secrets
+import sqlite3
+from typing import Any, Callable, Iterable, Mapping
+import uuid
+
+from aios_capability_contract import has_capabilities
+from aios_online_agent_execution import (
+    ExecutionProjectionError,
+    validate_execution_envelope,
+)
+
+
+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"})
+
+
+GOVERNED_WRITE_CAPABILITIES = frozenset({
+    "governed.write.branch",
+    "governed.write.file",
+})
+EXECUTION_PROJECTION_SCHEMA = "aios.online-agent.task-projection.v1"
+
+
+def _validate_write_execution_binding(
+    task_id: str,
+    required_capabilities: list[str],
+    metadata: Mapping[str, Any],
+) -> None:
+    required = set(required_capabilities)
+    if not required.intersection(GOVERNED_WRITE_CAPABILITIES):
+        return
+    execution = metadata.get("execution")
+    projection = metadata.get("execution_projection")
+    if not isinstance(execution, Mapping) or not isinstance(projection, Mapping):
+        raise TaskProtocolError("GOVERNED_EXECUTION_BINDING_REQUIRED", 409)
+    try:
+        canonical = validate_execution_envelope(dict(execution))
+    except (ExecutionProjectionError, TypeError, ValueError, KeyError) as exc:
+        raise TaskProtocolError("EXECUTION_ENVELOPE_INVALID", 409) from exc
+    if canonical.get("task_id") != task_id:
+        raise TaskProtocolError("TASK_BINDING_CONFLICT", 409)
+    if set(canonical.get("capability_scope") or ()) != required:
+        raise TaskProtocolError("EXECUTION_CAPABILITY_SCOPE_CONFLICT", 409)
+    if (
+        projection.get("schema") != EXECUTION_PROJECTION_SCHEMA
+        or projection.get("task_id") != task_id
+        or projection.get("job_id") != canonical.get("job_id")
+        or projection.get("execution_digest") != canonical.get("execution_digest")
+    ):
+        raise TaskProtocolError("EXECUTION_PROJECTION_INVALID", 409)
+
+
+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")
+        _validate_write_execution_binding(task_id, required, metadata_value)
+        payload = {
+            "task_id": task_id,
+            "goal": goal,
+            "instructions": instructions,
+            "success_criteria": criteria,
+            "required_capabilities": required,
+            "constraints": constraints_value,
+            "metadata": metadata_value,
+            "deadline": iso(parsed_deadline) if parsed_deadline else None,
+            "priority": priority,
+            "max_attempts": max_attempts,
+        }
+        payload_digest = digest_json(payload)
+        now = iso(self.now())
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            replay = self._idempotent_lookup(
+                connection,
+                scope="create_task",
+                key=idempotency_key,
+                payload_digest=payload_digest,
+            )
+            if replay is not None:
+                connection.commit()
+                return {**replay, "idempotent": True}
+            try:
+                connection.execute(
+                    "INSERT INTO tasks(task_id,goal,instructions,success_criteria_json,"
+                    "required_capabilities_json,constraints_json,metadata_json,deadline,priority,"
+                    "max_attempts,attempt,status,revision,created_at,updated_at)"
+                    " VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        task_id,
+                        goal,
+                        instructions,
+                        canonical_json(criteria),
+                        canonical_json(required),
+                        canonical_json(constraints_value),
+                        canonical_json(metadata_value),
+                        payload["deadline"],
+                        priority,
+                        max_attempts,
+                        0,
+                        "QUEUED",
+                        1,
+                        now,
+                        now,
+                    ),
+                )
+            except sqlite3.IntegrityError as exc:
+                raise TaskProtocolError("TASK_ID_CONFLICT", 409) from exc
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            result = self._row_task(row)
+            self._idempotent_store(
+                connection,
+                scope="create_task",
+                key=idempotency_key,
+                payload_digest=payload_digest,
+                response=result,
+            )
+            self._event(connection, "TASK_CREATED", task_id=task_id, payload={"priority": priority})
+            connection.commit()
+            return result
+
+    def _expire_locked(self, connection: sqlite3.Connection) -> None:
+        now_dt = self.now()
+        now = iso(now_dt)
+        rows = connection.execute(
+            "SELECT * FROM tasks WHERE status IN ('QUEUED','CLAIMED','RUNNING')"
+        ).fetchall()
+        for row in rows:
+            deadline = parse_time(row["deadline"])
+            if deadline is not None and deadline <= now_dt:
+                connection.execute(
+                    "UPDATE tasks SET status='EXPIRED',lease_token_hash=NULL,"
+                    "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                    (now, row["task_id"]),
+                )
+                self._event(connection, "TASK_EXPIRED", task_id=row["task_id"])
+                continue
+            if row["status"] in ACTIVE_LEASE_STATES:
+                lease_expires = parse_time(row["lease_expires_at"])
+                if lease_expires is not None and lease_expires <= now_dt:
+                    if row["attempt"] >= row["max_attempts"]:
+                        next_state = "FAILED"
+                    else:
+                        next_state = "QUEUED"
+                    if next_state == "FAILED":
+                        connection.execute(
+                            "UPDATE tasks SET status=?,lease_token_hash=NULL,"
+                            "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                            (next_state, now, row["task_id"]),
+                        )
+                    else:
+                        connection.execute(
+                            "UPDATE tasks SET status=?,claimed_by=NULL,lease_token_hash=NULL,"
+                            "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                            (next_state, now, row["task_id"]),
+                        )
+                    self._event(
+                        connection,
+                        "LEASE_EXPIRED",
+                        task_id=row["task_id"],
+                        agent_id=row["claimed_by"],
+                        payload={"next_state": next_state, "attempt": row["attempt"]},
+                    )
+
+    def expire_leases(self) -> int:
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            before = connection.total_changes
+            self._expire_locked(connection)
+            changed = connection.total_changes - before
+            connection.commit()
+            return changed
+
+    def poll_tasks(self, *, agent_id: str, limit: int = 20) -> list[dict[str, Any]]:
+        if not 1 <= int(limit) <= 100:
+            raise TaskProtocolError("LIMIT_INVALID", 400)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._expire_locked(connection)
+            agent = connection.execute(
+                "SELECT capabilities_json FROM agents WHERE agent_id=? AND status='ACTIVE'",
+                (agent_id,),
+            ).fetchone()
+            if agent is None:
+                raise TaskProtocolError("AGENT_NOT_FOUND", 404)
+            capabilities = json.loads(agent["capabilities_json"])
+            rows = connection.execute(
+                "SELECT * FROM tasks WHERE status='QUEUED' ORDER BY priority DESC,created_at ASC"
+            ).fetchall()
+            result = []
+            for row in rows:
+                required = json.loads(row["required_capabilities_json"])
+                if has_capabilities(capabilities, required):
+                    result.append(self._row_task(row, include_result=False))
+                    if len(result) >= int(limit):
+                        break
+            connection.commit()
+            return result
+
+    def _agent_capabilities(self, connection: sqlite3.Connection, agent_id: str) -> set[str]:
+        row = connection.execute(
+            "SELECT capabilities_json FROM agents WHERE agent_id=? AND status='ACTIVE'",
+            (agent_id,),
+        ).fetchone()
+        if row is None:
+            raise TaskProtocolError("AGENT_NOT_FOUND", 404)
+        return set(json.loads(row["capabilities_json"]))
+
+    def claim_task(
+        self,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_seconds: int = 300,
+    ) -> dict[str, Any]:
+        if not isinstance(lease_seconds, int) or isinstance(lease_seconds, bool) or not 30 <= lease_seconds <= 3600:
+            raise TaskProtocolError("LEASE_SECONDS_INVALID", 400)
+        lease_token = "lease_" + secrets.token_urlsafe(32)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._expire_locked(connection)
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["status"] != "QUEUED":
+                raise TaskProtocolError("TASK_NOT_CLAIMABLE", 409)
+            capabilities = self._agent_capabilities(connection, agent_id)
+            required = json.loads(row["required_capabilities_json"])
+            if not has_capabilities(capabilities, required):
+                raise TaskProtocolError("AGENT_CAPABILITY_MISMATCH", 403)
+            attempt = int(row["attempt"]) + 1
+            if attempt > int(row["max_attempts"]):
+                raise TaskProtocolError("TASK_ATTEMPTS_EXHAUSTED", 409)
+            expires = self.now() + timedelta(seconds=lease_seconds)
+            now = iso(self.now())
+            updated = connection.execute(
+                "UPDATE tasks SET status='CLAIMED',claimed_by=?,lease_token_hash=?,"
+                "lease_expires_at=?,attempt=?,revision=revision+1,updated_at=? "
+                "WHERE task_id=? AND status='QUEUED'",
+                (
+                    agent_id,
+                    self._token_hash(lease_token),
+                    iso(expires),
+                    attempt,
+                    now,
+                    task_id,
+                ),
+            )
+            if updated.rowcount != 1:
+                raise TaskProtocolError("TASK_CLAIM_CONFLICT", 409)
+            self._event(
+                connection,
+                "TASK_CLAIMED",
+                task_id=task_id,
+                agent_id=agent_id,
+                payload={"attempt": attempt, "lease_expires_at": iso(expires)},
+            )
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return {
+                **self._row_task(row, include_result=False),
+                "lease_token": lease_token,
+                "lease_token_returned_once": True,
+            }
+
+    def _validate_lease(
+        self,
+        connection: sqlite3.Connection,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_token: str,
+        allowed_states: frozenset[str] = ACTIVE_LEASE_STATES,
+    ) -> sqlite3.Row:
+        self._expire_locked(connection)
+        row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+        if row is None:
+            raise TaskProtocolError("TASK_NOT_FOUND", 404)
+        if row["status"] not in allowed_states:
+            raise TaskProtocolError("TASK_LEASE_NOT_ACTIVE", 409)
+        if row["claimed_by"] != agent_id:
+            raise TaskProtocolError("TASK_LEASE_OWNER_MISMATCH", 403)
+        if not row["lease_token_hash"] or not hmac.compare_digest(
+            row["lease_token_hash"], self._token_hash(lease_token)
+        ):
+            raise TaskProtocolError("TASK_LEASE_TOKEN_INVALID", 403)
+        expires = parse_time(row["lease_expires_at"])
+        if expires is None or expires <= self.now():
+            raise TaskProtocolError("TASK_LEASE_EXPIRED", 409)
+        return row
+
+    def heartbeat(
+        self,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_token: str,
+        progress: Mapping[str, Any] | None = None,
+        lease_seconds: int = 300,
+    ) -> dict[str, Any]:
+        if not 30 <= int(lease_seconds) <= 3600:
+            raise TaskProtocolError("LEASE_SECONDS_INVALID", 400)
+        progress_value = _json_object(progress, "progress", max_bytes=64 * 1024)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._validate_lease(
+                connection,
+                task_id=task_id,
+                agent_id=agent_id,
+                lease_token=lease_token,
+            )
+            expires = self.now() + timedelta(seconds=int(lease_seconds))
+            now = iso(self.now())
+            connection.execute(
+                "UPDATE tasks SET status='RUNNING',lease_expires_at=?,revision=revision+1,"
+                "updated_at=? WHERE task_id=?",
+                (iso(expires), now, task_id),
+            )
+            self._event(
+                connection,
+                "TASK_HEARTBEAT",
+                task_id=task_id,
+                agent_id=agent_id,
+                payload={"progress": progress_value, "lease_expires_at": iso(expires)},
+            )
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(row, include_result=False)
+
+    def submit_result(
+        self,
+        *,
+        task_id: str,
+        agent_id: str,
+        lease_token: str,
+        idempotency_key: str,
+        result: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        idempotency_key = idempotency_key.strip()
+        if not idempotency_key or len(idempotency_key) > 160:
+            raise TaskProtocolError("IDEMPOTENCY_KEY_INVALID", 400)
+        result_value = _json_object(result, "result", max_bytes=1024 * 1024)
+        result_digest = digest_json(result_value)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["result_idempotency_key"] == idempotency_key:
+                if not row["result_digest"] or not hmac.compare_digest(row["result_digest"], result_digest):
+                    raise TaskProtocolError("IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {**self._row_task(row), "idempotent": True}
+            self._validate_lease(
+                connection,
+                task_id=task_id,
+                agent_id=agent_id,
+                lease_token=lease_token,
+            )
+            now = iso(self.now())
+            connection.execute(
+                "UPDATE tasks SET status='VERIFYING',result_json=?,result_digest=?,"
+                "result_idempotency_key=?,lease_token_hash=NULL,lease_expires_at=NULL,"
+                "revision=revision+1,updated_at=? WHERE task_id=?",
+                (
+                    canonical_json(result_value),
+                    result_digest,
+                    idempotency_key,
+                    now,
+                    task_id,
+                ),
+            )
+            self._event(
+                connection,
+                "RESULT_SUBMITTED",
+                task_id=task_id,
+                agent_id=agent_id,
+                payload={"result_digest": result_digest, "idempotency_key": idempotency_key},
+            )
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(row)
+
+    def verify_task(
+        self,
+        *,
+        task_id: str,
+        decision: str,
+        verification: Mapping[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        decision = decision.strip().upper()
+        if decision not in VERIFY_DECISIONS:
+            raise TaskProtocolError("VERIFICATION_DECISION_INVALID", 400)
+        verification_value = _json_object(verification, "verification")
+        verification_value = {"decision": decision, **verification_value}
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["status"] != "VERIFYING":
+                raise TaskProtocolError("TASK_NOT_VERIFYING", 409)
+            if decision == "SUCCEEDED":
+                next_state = "SUCCEEDED"
+            elif decision == "FAILED":
+                next_state = "FAILED"
+            elif int(row["attempt"]) >= int(row["max_attempts"]):
+                next_state = "FAILED"
+                verification_value["reason"] = verification_value.get(
+                    "reason", "MAX_ATTEMPTS_EXHAUSTED"
+                )
+            else:
+                next_state = "QUEUED"
+            now = iso(self.now())
+            if next_state in {"SUCCEEDED", "FAILED"}:
+                connection.execute(
+                    "UPDATE tasks SET status=?,lease_token_hash=NULL,"
+                    "lease_expires_at=NULL,verification_json=?,revision=revision+1,updated_at=? "
+                    "WHERE task_id=?",
+                    (next_state, canonical_json(verification_value), now, task_id),
+                )
+            else:
+                connection.execute(
+                    "UPDATE tasks SET status=?,claimed_by=NULL,lease_token_hash=NULL,"
+                    "lease_expires_at=NULL,verification_json=?,revision=revision+1,updated_at=? "
+                    "WHERE task_id=?",
+                    (next_state, canonical_json(verification_value), now, task_id),
+                )
+            self._event(
+                connection,
+                "TASK_VERIFIED",
+                task_id=task_id,
+                agent_id=row["claimed_by"],
+                payload={"decision": decision, "next_state": next_state},
+            )
+            updated = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(updated)
+
+    def cancel_task(self, *, task_id: str, reason: str = "") -> dict[str, Any]:
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            if row["status"] in TERMINAL_STATES:
+                connection.commit()
+                return self._row_task(row)
+            connection.execute(
+                "UPDATE tasks SET status='CANCELLED',lease_token_hash=NULL,"
+                "lease_expires_at=NULL,revision=revision+1,updated_at=? WHERE task_id=?",
+                (iso(self.now()), task_id),
+            )
+            self._event(connection, "TASK_CANCELLED", task_id=task_id, payload={"reason": reason})
+            updated = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            connection.commit()
+            return self._row_task(updated)
+
+    def get_task(self, task_id: str) -> dict[str, Any]:
+        self.expire_leases()
+        with self._connect() as connection:
+            row = connection.execute("SELECT * FROM tasks WHERE task_id=?", (task_id,)).fetchone()
+            if row is None:
+                raise TaskProtocolError("TASK_NOT_FOUND", 404)
+            return self._row_task(row)
+
+    def get_events(self, task_id: str, *, limit: int = 200) -> list[dict[str, Any]]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                "SELECT * FROM events WHERE task_id=? ORDER BY event_id DESC LIMIT ?",
+                (task_id, max(1, min(int(limit), 1000))),
+            ).fetchall()
+            return [
+                {
+                    "event_id": row["event_id"],
+                    "task_id": row["task_id"],
+                    "agent_id": row["agent_id"],
+                    "event_type": row["event_type"],
+                    "payload": json.loads(row["payload_json"]),
+                    "created_at": row["created_at"],
+                }
+                for row in reversed(rows)
+            ]

active-runtime/app/aios_agent_task_supervisor_coordinator_projection.py

@@ -0,0 +1,655 @@
+"""CAS-bound projection from an unprivileged planner to a host coordinator."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+from typing import Any, Callable, Mapping, Sequence
+import urllib.error
+import urllib.request
+
+from aios_agent_task_initial_execution_projection import (
+    InitialExecutionProjectionError,
+    InitialExecutionProjectionStore,
+    JOB_RE,
+    KEY_RE,
+    MAX_BODY,
+    SHA_RE,
+    TASK_RE,
+    _canonical,
+    _now,
+    _required_authorization_id,
+    _require_string,
+    _sha,
+    load_control_token,
+)
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    validate_execution_envelope,
+)
+
+
+RUNTIME_CAPABILITY = "agent.runtime.openclaw"
+COORDINATOR_CAPABILITY = "engineering.autonomous"
+COORDINATOR_CAPABILITIES = [RUNTIME_CAPABILITY, COORDINATOR_CAPABILITY]
+
+
+@dataclass(frozen=True)
+class SupervisorCoordinatorProjectionRequest:
+    expected_status: str
+    expected_revision: int
+    expected_attempt: int
+    expected_metadata_sha256: str
+    expected_required_capabilities_sha256: str
+    previous_verification_sha256: str
+    expected_planner_result_sha256: str
+    expected_old_execution_digest: str
+    expected_new_execution_digest: str
+    expected_job_id: str
+    authorization_id: str
+    idempotency_key: str
+    reason: str
+
+    @classmethod
+    def from_mapping(
+        cls, value: Mapping[str, Any]
+    ) -> "SupervisorCoordinatorProjectionRequest":
+        allowed = {
+            "expected_status",
+            "expected_revision",
+            "expected_attempt",
+            "expected_metadata_sha256",
+            "expected_required_capabilities_sha256",
+            "previous_verification_sha256",
+            "expected_planner_result_sha256",
+            "expected_old_execution_digest",
+            "expected_new_execution_digest",
+            "expected_job_id",
+            "authorization_id",
+            "idempotency_key",
+            "reason",
+        }
+        if set(value) != allowed:
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_COORDINATOR_PROJECTION_FIELDS_INVALID", 400
+            )
+        status = _require_string(value["expected_status"], "EXPECTED_STATUS_INVALID")
+        revision = value["expected_revision"]
+        attempt = value["expected_attempt"]
+        if (
+            status != "QUEUED"
+            or not isinstance(revision, int)
+            or isinstance(revision, bool)
+            or revision < 2
+            or not isinstance(attempt, int)
+            or isinstance(attempt, bool)
+            or attempt < 1
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_COORDINATOR_PROJECTION_STATE_INVALID", 400
+            )
+        names = (
+            "expected_metadata_sha256",
+            "expected_required_capabilities_sha256",
+            "previous_verification_sha256",
+            "expected_planner_result_sha256",
+            "expected_old_execution_digest",
+            "expected_new_execution_digest",
+        )
+        digests = {
+            name: _require_string(value[name], name.upper() + "_INVALID", SHA_RE)
+            for name in names
+        }
+        job_id = _require_string(value["expected_job_id"], "JOB_ID_INVALID", JOB_RE)
+        authorization_id = _require_string(
+            value["authorization_id"], "AUTHORIZATION_ID_INVALID", KEY_RE
+        )
+        idempotency_key = _require_string(
+            value["idempotency_key"], "IDEMPOTENCY_KEY_INVALID", KEY_RE
+        )
+        reason = _require_string(value["reason"], "PROJECTION_REASON_INVALID")
+        if len(reason) > 512:
+            raise InitialExecutionProjectionError("PROJECTION_REASON_INVALID", 400)
+        return cls(
+            status,
+            revision,
+            attempt,
+            digests["expected_metadata_sha256"],
+            digests["expected_required_capabilities_sha256"],
+            digests["previous_verification_sha256"],
+            digests["expected_planner_result_sha256"],
+            digests["expected_old_execution_digest"],
+            digests["expected_new_execution_digest"],
+            job_id,
+            authorization_id,
+            idempotency_key,
+            reason,
+        )
+
+    def public(self) -> dict[str, Any]:
+        return {
+            "expected_status": self.expected_status,
+            "expected_revision": self.expected_revision,
+            "expected_attempt": self.expected_attempt,
+            "expected_metadata_sha256": self.expected_metadata_sha256,
+            "expected_required_capabilities_sha256": self.expected_required_capabilities_sha256,
+            "previous_verification_sha256": self.previous_verification_sha256,
+            "expected_planner_result_sha256": self.expected_planner_result_sha256,
+            "expected_old_execution_digest": self.expected_old_execution_digest,
+            "expected_new_execution_digest": self.expected_new_execution_digest,
+            "expected_job_id": self.expected_job_id,
+            "authorization_id": self.authorization_id,
+            "idempotency_key": self.idempotency_key,
+            "reason": self.reason,
+        }
+
+
+class CanonicalCoordinatorExecutionResolver:
+    def __init__(self, base_url: str, control_token: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.control_token = control_token
+
+    def __call__(
+        self,
+        task_id: str,
+        expected_job_id: str,
+        runtime_input: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        body = {
+            "task_id": task_id,
+            "job_id": expected_job_id,
+            "required_capabilities": COORDINATOR_CAPABILITIES,
+            "execution": {
+                "action": "agent_runtime",
+                "adapter": "openclaw",
+                "capability_scope": COORDINATOR_CAPABILITIES,
+                "runtime_input": dict(runtime_input),
+            },
+        }
+        request = urllib.request.Request(
+            self.base_url + "/v2/execution-envelope",
+            method="POST",
+            data=_canonical(body),
+            headers={
+                "Authorization": "Bearer " + self.control_token,
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=15) as response:
+                raw, status = response.read(), response.status
+        except urllib.error.HTTPError as exc:
+            raw, status = exc.read(), exc.code
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise InitialExecutionProjectionError(
+                "CANONICAL_COORDINATOR_EXECUTION_RESPONSE_INVALID", 502
+            ) from exc
+        if not isinstance(value, Mapping):
+            raise InitialExecutionProjectionError(
+                "CANONICAL_COORDINATOR_EXECUTION_RESPONSE_INVALID", 502
+            )
+        if status != 200:
+            raise InitialExecutionProjectionError(
+                str(value.get("error") or "CANONICAL_COORDINATOR_EXECUTION_FAILED"),
+                502 if status >= 500 else 409,
+            )
+        return dict(value)
+
+
+class SupervisorCoordinatorProjectionStore(InitialExecutionProjectionStore):
+    def __init__(
+        self,
+        db_path: Path,
+        resolver: Callable[[str, str, Mapping[str, Any]], Mapping[str, Any]],
+    ) -> None:
+        super().__init__(db_path, resolver)  # type: ignore[arg-type]
+        self.coordinator_resolver = resolver
+
+    @staticmethod
+    def _ensure_table(connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS task_supervisor_coordinator_projections (
+              task_id TEXT NOT NULL,
+              idempotency_key TEXT NOT NULL,
+              request_sha256 TEXT NOT NULL,
+              authorization_id TEXT NOT NULL,
+              expected_job_id TEXT NOT NULL,
+              planner_result_sha256 TEXT NOT NULL,
+              old_execution_digest TEXT NOT NULL,
+              new_execution_digest TEXT NOT NULL,
+              expected_revision INTEGER NOT NULL,
+              applied_revision INTEGER NOT NULL,
+              attempt INTEGER NOT NULL,
+              created_at TEXT NOT NULL,
+              PRIMARY KEY(task_id,idempotency_key)
+            )
+            """
+        )
+
+    @classmethod
+    def _planner_event_chain(
+        cls, connection: sqlite3.Connection, task_id: str
+    ) -> None:
+        table, columns = cls._event_table(connection)
+        ordering = "event_id" if "event_id" in columns else "rowid"
+        rows = connection.execute(
+            f'SELECT event_type,payload_json FROM "{table}" '
+            f'WHERE task_id=? ORDER BY {ordering}',
+            (task_id,),
+        ).fetchall()
+        events = [str(row[0]) for row in rows]
+        try:
+            projection = len(events) - 1 - events[::-1].index(
+                "TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED"
+            )
+        except ValueError as exc:
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_PROJECTION_EVENT_MISSING", 409
+            ) from exc
+        tail = events[projection + 1 :]
+        if (
+            len(tail) < 4
+            or tail[0] != "TASK_CLAIMED"
+            or tail.count("TASK_CLAIMED") != 1
+            or tail.count("RESULT_SUBMITTED") != 1
+            or tail.count("TASK_VERIFIED") != 1
+            or tail[-1] != "TASK_VERIFIED"
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_EVENT_CHAIN_INVALID", 409
+            )
+        result = tail.index("RESULT_SUBMITTED")
+        if (
+            not any(event == "TASK_HEARTBEAT" for event in tail[1:result])
+            or any(
+                event not in {
+                    "TASK_CLAIMED",
+                    "TASK_HEARTBEAT",
+                    "RESULT_SUBMITTED",
+                    "TASK_VERIFIED",
+                }
+                for event in tail
+            )
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_EVENT_CHAIN_INVALID", 409
+            )
+        verification_event = json.loads(rows[-1][1])
+        if (
+            verification_event.get("decision") != "NEEDS_REWORK"
+            or verification_event.get("next_state") != "QUEUED"
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_NEEDS_REWORK_EVENT_REQUIRED", 409
+            )
+
+    def _load(
+        self,
+        connection: sqlite3.Connection,
+        task_id: str,
+        request: SupervisorCoordinatorProjectionRequest,
+    ) -> tuple[sqlite3.Row, dict[str, Any], dict[str, Any]]:
+        task = connection.execute(
+            "SELECT * FROM tasks WHERE task_id=?", (task_id,)
+        ).fetchone()
+        if task is None:
+            raise InitialExecutionProjectionError("TASK_NOT_FOUND", 404)
+        if (
+            task["status"] != request.expected_status
+            or int(task["revision"]) != request.expected_revision
+            or int(task["attempt"]) != request.expected_attempt
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_COORDINATOR_PROJECTION_CAS_MISMATCH", 409
+            )
+        if (
+            task["claimed_by"] not in (None, "")
+            or task["lease_token_hash"] not in (None, "")
+            or task["lease_expires_at"] not in (None, "")
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_COORDINATOR_ACTIVE_LEASE_DENY", 409
+            )
+        metadata = self._json_value(task["metadata_json"], "TASK_METADATA_INVALID")
+        capabilities = self._json_value(
+            task["required_capabilities_json"], "TASK_CAPABILITIES_INVALID"
+        )
+        result = self._json_value(task["result_json"], "PLANNER_RESULT_INVALID")
+        verification = self._json_value(
+            task["verification_json"], "PLANNER_VERIFICATION_INVALID"
+        )
+        if (
+            not isinstance(metadata, Mapping)
+            or not isinstance(capabilities, list)
+            or not isinstance(result, Mapping)
+            or not isinstance(verification, Mapping)
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_COORDINATOR_STATE_INVALID", 409
+            )
+        metadata, result, verification = dict(metadata), dict(result), dict(verification)
+        if _sha(metadata) != request.expected_metadata_sha256:
+            raise InitialExecutionProjectionError("TASK_METADATA_DRIFT", 409)
+        if _sha(capabilities) != request.expected_required_capabilities_sha256:
+            raise InitialExecutionProjectionError("TASK_CAPABILITIES_DRIFT", 409)
+        if _sha(result) != request.expected_planner_result_sha256:
+            raise InitialExecutionProjectionError("PLANNER_RESULT_DRIFT", 409)
+        if _sha(verification) != request.previous_verification_sha256:
+            raise InitialExecutionProjectionError("PLANNER_VERIFICATION_DRIFT", 409)
+        old = metadata.get("execution")
+        if not isinstance(old, Mapping):
+            raise InitialExecutionProjectionError("TASK_EXECUTION_METADATA_MISSING", 409)
+        old_execution = validate_execution_envelope(dict(old))
+        if (
+            old_execution.get("action") != "agent_runtime"
+            or old_execution.get("adapter") != "openclaw"
+            or old_execution.get("capability_scope") != [RUNTIME_CAPABILITY]
+            or old_execution.get("execution_digest")
+            != request.expected_old_execution_digest
+            or capabilities != [RUNTIME_CAPABILITY]
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_EXECUTION_CONTRACT_INVALID", 409
+            )
+        expected_result = {
+            "task_id": task_id,
+            "job_id": request.expected_job_id,
+            "status": "SUCCEEDED",
+            "execution_mode": "agent_runtime",
+            "agent_runtime": "openclaw",
+            "execution_digest": request.expected_old_execution_digest,
+            "control_plane_access_granted_to_runtime": False,
+            "credential_material_exposed": False,
+            "secret_material_exposed": False,
+            "changed_files": [],
+            "writer_lease": "NONE",
+            "direct_main_write": False,
+            "production_write": False,
+            "force_push": False,
+            "protected_ref_bypass": False,
+        }
+        if any(result.get(name) != expected for name, expected in expected_result.items()):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_RESULT_CONTRACT_INVALID", 409
+            )
+        if verification.get("decision") != "NEEDS_REWORK":
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_NEEDS_REWORK_REQUIRED", 409
+            )
+        self._planner_event_chain(connection, task_id)
+        return task, metadata, result
+
+    def project(
+        self,
+        task_id: str,
+        request: SupervisorCoordinatorProjectionRequest,
+    ) -> dict[str, Any]:
+        task_id = _require_string(task_id, "TASK_ID_INVALID", TASK_RE)
+        if request.authorization_id != _required_authorization_id():
+            raise InitialExecutionProjectionError("AUTHORIZATION_ID_MISMATCH", 403)
+        request_sha = _sha(request.public())
+        with self._connect() as connection:
+            self._ensure_table(connection)
+            replay = connection.execute(
+                "SELECT * FROM task_supervisor_coordinator_projections "
+                "WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise InitialExecutionProjectionError("IDEMPOTENCY_CONFLICT", 409)
+                return {
+                    "task_id": task_id,
+                    "status": "QUEUED",
+                    "attempt": replay["attempt"],
+                    "revision": replay["applied_revision"],
+                    "execution_digest": replay["new_execution_digest"],
+                    "required_capabilities": COORDINATOR_CAPABILITIES,
+                    "idempotent": True,
+                }
+        with self._connect() as connection:
+            task, _, result = self._load(connection, task_id, request)
+            runtime_input = self._runtime_input(task)
+        resolved = dict(
+            self.coordinator_resolver(task_id, request.expected_job_id, runtime_input)
+        )
+        binding = GovernedExecutionBinding.from_mapping(resolved)
+        if binding.task_id != task_id or binding.job_id != request.expected_job_id:
+            raise InitialExecutionProjectionError("GOVERNED_BINDING_CONFLICT", 409)
+        result_binding = {
+            name: result.get(name)
+            for name in (
+                "task_id",
+                "job_id",
+                "repository",
+                "base_oid",
+                "expected_branch_oid",
+                "target_branch",
+                "generation",
+            )
+        }
+        expected_binding = {
+            "task_id": binding.task_id,
+            "job_id": binding.job_id,
+            "repository": binding.repository,
+            "base_oid": binding.base_oid,
+            "expected_branch_oid": binding.expected_branch_oid,
+            "target_branch": binding.target_branch,
+            "generation": binding.generation,
+        }
+        if result_binding != expected_binding:
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_PLANNER_BINDING_CONFLICT", 409
+            )
+        execution = validate_execution_envelope(resolved, binding=binding)
+        if (
+            execution.get("action") != "agent_runtime"
+            or execution.get("adapter") != "openclaw"
+            or execution.get("capability_scope") != COORDINATOR_CAPABILITIES
+            or execution.get("change_set") != []
+            or execution.get("verification") != []
+            or execution.get("governance", {}).get("writer_lease") is not False
+            or execution.get("execution_digest") != request.expected_new_execution_digest
+        ):
+            raise InitialExecutionProjectionError(
+                "SUPERVISOR_COORDINATOR_ENVELOPE_INVALID", 409
+            )
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._ensure_table(connection)
+            task, metadata, _ = self._load(connection, task_id, request)
+            new_metadata = dict(metadata)
+            new_metadata["execution"] = execution
+            new_metadata["execution_projection"] = {
+                "schema": "aios.online-agent.task-projection.v1",
+                "task_id": task_id,
+                "job_id": request.expected_job_id,
+                "execution_digest": execution["execution_digest"],
+            }
+            applied_revision = request.expected_revision + 1
+            cursor = connection.execute(
+                """
+                UPDATE tasks SET metadata_json=?,required_capabilities_json=?,
+                    revision=?,updated_at=?
+                 WHERE task_id=? AND status=? AND revision=? AND attempt=?
+                   AND claimed_by IS NULL AND lease_token_hash IS NULL
+                   AND lease_expires_at IS NULL AND metadata_json=?
+                   AND required_capabilities_json=? AND result_json=?
+                   AND verification_json=?
+                """,
+                (
+                    json.dumps(new_metadata, sort_keys=True, separators=(",", ":")),
+                    json.dumps(COORDINATOR_CAPABILITIES, separators=(",", ":")),
+                    applied_revision,
+                    _now(),
+                    task_id,
+                    request.expected_status,
+                    request.expected_revision,
+                    request.expected_attempt,
+                    task["metadata_json"],
+                    task["required_capabilities_json"],
+                    task["result_json"],
+                    task["verification_json"],
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise InitialExecutionProjectionError(
+                    "SUPERVISOR_COORDINATOR_PROJECTION_CAS_MISMATCH", 409
+                )
+            event = {
+                "schema": "aios.supervisor-coordinator-projection.event.v1",
+                "task_id": task_id,
+                "job_id": request.expected_job_id,
+                "authorization_id": request.authorization_id,
+                "idempotency_key": request.idempotency_key,
+                "reason": request.reason,
+                "old_execution_digest": request.expected_old_execution_digest,
+                "new_execution_digest": execution["execution_digest"],
+                "required_capabilities": COORDINATOR_CAPABILITIES,
+                "previous_revision": request.expected_revision,
+                "applied_revision": applied_revision,
+                "attempt": request.expected_attempt,
+                "request_sha256": request_sha,
+                "credential_forwarded_to_planner": False,
+                "writer_lease": False,
+                "main_write": False,
+                "production_write": False,
+            }
+            table, columns = self._event_table(connection)
+            values = {
+                "task_id": task_id,
+                "event_type": "TASK_SUPERVISOR_COORDINATOR_PROJECTION_AUTHORIZED",
+                "payload_json": json.dumps(event, sort_keys=True, separators=(",", ":")),
+                "created_at": _now(),
+                "agent_id": "owner-control",
+            }
+            names = [name for name in values if name in columns]
+            connection.execute(
+                f'INSERT INTO "{table}" ({",".join(chr(34)+name+chr(34) for name in names)}) '
+                f'VALUES ({",".join("?" for _ in names)})',
+                tuple(values[name] for name in names),
+            )
+            connection.execute(
+                """
+                INSERT INTO task_supervisor_coordinator_projections(
+                  task_id,idempotency_key,request_sha256,authorization_id,
+                  expected_job_id,planner_result_sha256,old_execution_digest,
+                  new_execution_digest,expected_revision,applied_revision,
+                  attempt,created_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    task_id,
+                    request.idempotency_key,
+                    request_sha,
+                    request.authorization_id,
+                    request.expected_job_id,
+                    request.expected_planner_result_sha256,
+                    request.expected_old_execution_digest,
+                    execution["execution_digest"],
+                    request.expected_revision,
+                    applied_revision,
+                    request.expected_attempt,
+                    _now(),
+                ),
+            )
+            connection.commit()
+            return {
+                "task_id": task_id,
+                "status": "QUEUED",
+                "attempt": request.expected_attempt,
+                "revision": applied_revision,
+                "execution_digest": execution["execution_digest"],
+                "required_capabilities": COORDINATOR_CAPABILITIES,
+                "credential_forwarded_to_planner": False,
+                "idempotent": False,
+            }
+
+
+class Handler(BaseHTTPRequestHandler):
+    store: SupervisorCoordinatorProjectionStore | None = None
+    control_token: str = ""
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def send_json(self, value: Mapping[str, Any], status: int) -> None:
+        raw = _canonical(dict(value))
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def do_GET(self) -> None:
+        if self.path == "/health":
+            self.send_json(
+                {"status": "ok", "service": "aios-supervisor-coordinator-projection"},
+                200,
+            )
+        else:
+            self.send_json({"error": "NOT_FOUND"}, 404)
+
+    def do_POST(self) -> None:
+        match = re.fullmatch(
+            r"/v1/tasks/([^/]+)/supervisor-coordinator-projection", self.path
+        )
+        header = self.headers.get("Authorization", "")
+        if match is None:
+            self.send_json({"error": "NOT_FOUND"}, 404)
+            return
+        if (
+            not header.startswith("Bearer ")
+            or not hmac.compare_digest(header[7:], self.control_token)
+        ):
+            self.send_json({"error": "UNAUTHORIZED"}, 401)
+            return
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+            if length < 1 or length > MAX_BODY:
+                raise InitialExecutionProjectionError("REQUEST_SIZE_INVALID", 413)
+            value = json.loads(self.rfile.read(length).decode("utf-8"))
+            if not isinstance(value, Mapping) or self.store is None:
+                raise InitialExecutionProjectionError("REQUEST_INVALID", 400)
+            request = SupervisorCoordinatorProjectionRequest.from_mapping(value)
+            result = self.store.project(match.group(1), request)
+            self.send_json(result, 200 if result["idempotent"] else 201)
+        except InitialExecutionProjectionError as exc:
+            self.send_json({"error": exc.code}, exc.status)
+        except (ValueError, UnicodeDecodeError, json.JSONDecodeError):
+            self.send_json({"error": "REQUEST_INVALID"}, 400)
+
+
+def serve() -> None:
+    credential = Path(
+        os.environ.get(
+            "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+            str(
+                Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+                / "agent_task_control_token"
+            ),
+        )
+    )
+    token = load_control_token(credential)
+    bridge = os.environ.get("AIOS_GOVERNED_WRITE_API_URL", "http://127.0.0.1:18892")
+    db = Path(os.environ.get("AIOS_AGENT_TASK_DB", "/var/lib/aios-agent-task/state.sqlite3"))
+    Handler.control_token = token
+    Handler.store = SupervisorCoordinatorProjectionStore(
+        db, CanonicalCoordinatorExecutionResolver(bridge, token)
+    )
+    host = os.environ.get("AIOS_SUPERVISOR_COORDINATOR_PROJECTION_HOST", "127.0.0.1")
+    port = int(os.environ.get("AIOS_SUPERVISOR_COORDINATOR_PROJECTION_PORT", "18944"))
+    ThreadingHTTPServer((host, port), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    serve()

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_capability_contract.py

@@ -39,6 +39,7 @@ _TASK_ALIASES: dict[str, frozenset[str]] = {
     "gpu.train": frozenset({"legacy.gpu.train"}),
     "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
     "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+    "engineering.autonomous": frozenset({"engineering.autonomous"}),
     "release:candidate": frozenset({"release:candidate"}),
 }
 
@@ -74,6 +75,7 @@ _WORKER_ALIASES: dict[str, frozenset[str]] = {
     "gpu.train": frozenset({"legacy.gpu.train"}),
     "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
     "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+    "engineering.autonomous": frozenset({"engineering.autonomous"}),
     "release:candidate": frozenset({"release:candidate"}),
 }
 
@@ -154,6 +156,11 @@ _CAPABILITY_CONTRACTS: dict[str, CapabilityContract] = {
         dependencies=("openclaw_runtime",),
         health_requirements=("worker_liveness",),
     ),
+    "engineering.autonomous": CapabilityContract(
+        "engineering.autonomous",
+        dependencies=("agent_task_control_plane",),
+        health_requirements=("worker_liveness",),
+    ),
     "release:candidate": CapabilityContract(
         "release:candidate",
         min_version=1,

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -1,9 +1,10 @@
 """Loopback-only governed Bridge v2 HTTP gateway with durable handlers."""
 from __future__ import annotations
 
-from dataclasses import dataclass, replace
+from dataclasses import asdict, dataclass, replace
 from datetime import datetime, timedelta, timezone
 import base64
+import hmac
 import json
 import os
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -12,6 +13,8 @@ import re
 import sqlite3
 import threading
 from typing import Any, Mapping
+import urllib.error
+import urllib.request
 
 from aios_authority_git_executor import AuthorityGitExecutor
 from aios_chatgpt_bridge_v2 import BridgeV2Adapter, BridgeV2Error, OPERATION_REQUIRED_FIELDS
@@ -19,7 +22,9 @@ from aios_continuation_executor import AuthorityGitAdapter, ContinuationCommit,
 from aios_continuation_lease import WriterLeaseRequestV2
 from aios_control_plane_snapshot import build_control_plane_snapshot
 from aios_gitops_job_controller import GitOpsJobController, JobError, JobState, JobStore
-from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
+from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBinding, GovernedTaskBootstrap, _legacy_safe, _task_key
+from aios_online_agent_execution import ExecutionProjectionError, OnlineTaskProjectionClient
+from aios_online_agent_plan_store import ExecutionPlanError, ExecutionPlanStore
 from governed_production_release import build_executor_config
 
 MAX_BODY = 2 * 1024 * 1024
@@ -27,6 +32,8 @@ ROUTES = {
     "/v2/runtime": "getRuntimeStatus",
     "/v2/capabilities": "getGovernedWriteCapabilities",
     "/v2/governed-task": "createGovernedTask",
+    "/v2/execution-envelope": "projectOnlineAgentExecution",
+    "/v2/execution-plan": "registerOnlineAgentExecutionPlan",
     "/v2/writer-lease": "acquireWriterLease",
     "/v2/writer-lease/bootstrap": "bootstrapWriterLease",
     "/v2/writer-lease/renew": "renewWriterLease",
@@ -62,6 +69,9 @@ class RuntimeConfig:
     registry_root: Path = Path("/var/lib/aios-gitops/runtime/registry-projection")
     baseline_path: Path = Path("/var/lib/aios-gitops-runtime-anchor/baseline/current.json")
     health_url: str = "http://127.0.0.1:18927/health"
+    agent_task_api_url: str = "http://127.0.0.1:18940"
+    agent_task_control_token_path: Path = Path("/run/credentials/aios-chatgpt-bridge-v2-staging.service/agent_task_control_token")
+    online_task_projection_enabled: bool = False
 
 
 def _safe(value: str) -> str:
@@ -74,6 +84,16 @@ class GovernedRuntime:
         config.state_root.mkdir(parents=True, exist_ok=True)
         self.authority = AuthorityGitExecutor(config.authoritative_git_dir, config.state_root / "worktrees")
         self.bootstrap = GovernedTaskBootstrap(self.authority, config.state_root, remote_name=config.remote_name, token_path=config.token_path, ssh_key_path=config.ssh_key_path, api_base=config.api_base)
+        self.online_task_projector = (
+            OnlineTaskProjectionClient(config.agent_task_api_url, config.agent_task_control_token_path)
+            if config.online_task_projection_enabled
+            else None
+        )
+        self.projection_outbox_root = config.state_root / "online-task-projection-outbox"
+        self.projection_outbox_root.mkdir(parents=True, exist_ok=True)
+        self.execution_plan_store = ExecutionPlanStore(config.state_root / "online-agent-execution-plans")
+        self._projection_locks_guard = threading.Lock()
+        self._projection_locks: dict[str, threading.RLock] = {}
         self.runtimes: dict[tuple[str, int], tuple[Any, GitOpsJobController, ContinuationExecutor]] = {}
         self.controllers: dict[str, GitOpsJobController] = {}
         self._recovery_lock = threading.RLock()
@@ -93,6 +113,54 @@ class GovernedRuntime:
             askpass.write_text('#!/bin/sh\ncase "$1" in *Username*) printf "%s\\n" "$GITEA_MACHINE_ACTOR" ;; *) cat "$CREDENTIAL_FILE" ;; esac\n', encoding="utf-8")
             askpass.chmod(0o700)
         os.environ.setdefault("GIT_ASKPASS", str(askpass)); os.environ.setdefault("GIT_ASKPASS_REQUIRE", "force"); os.environ.setdefault("GIT_TERMINAL_PROMPT", "0"); os.environ.setdefault("GITEA_MACHINE_ACTOR", "aios-autonomy-agent")
+        if self.online_task_projector is not None:
+            threading.Thread(target=self._projection_reconciler, name="online-task-projection-reconciler", daemon=True).start()
+
+    def _projection_lock(self, task_id: str) -> threading.RLock:
+        guard = getattr(self, "_projection_locks_guard", None)
+        if guard is None:
+            guard = threading.Lock()
+            self._projection_locks_guard = guard
+            self._projection_locks = {}
+        with guard:
+            return self._projection_locks.setdefault(task_id, threading.RLock())
+
+    def _projection_outbox_path(self, task_id: str) -> Path:
+        return self.projection_outbox_root / (_task_key(task_id) + ".json")
+
+    def _write_projection_outbox(self, task_id: str, projection: Mapping[str, Any], binding: Any) -> Path:
+        path = self._projection_outbox_path(task_id)
+        temporary = path.with_name(path.name + f".{os.getpid()}.{threading.get_ident()}.tmp")
+        temporary.write_text(json.dumps({"task": dict(projection), "binding": asdict(binding)}, sort_keys=True, separators=(",", ":")), encoding="utf-8")
+        temporary.replace(path)
+        return path
+
+    def _deliver_projection_outbox(self, path: Path) -> dict[str, Any]:
+        value = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(value, Mapping) or not isinstance(value.get("task"), Mapping) or not isinstance(value.get("binding"), Mapping):
+            raise RuntimeError("ONLINE_TASK_PROJECTION_OUTBOX_INVALID")
+        binding = self.bootstrap.execution_binding(GovernedTaskBinding(**dict(value["binding"])))
+        if self.online_task_projector is None:
+            raise RuntimeError("ONLINE_TASK_PROJECTION_CONTROL_UNAVAILABLE")
+        created = self.online_task_projector.create(dict(value["task"]), binding=binding)
+        path.unlink(missing_ok=True)
+        return created
+
+    def _projection_reconciler(self) -> None:
+        while True:
+            for path in sorted(self.projection_outbox_root.glob("*.json")):
+                try:
+                    value = json.loads(path.read_text(encoding="utf-8"))
+                    binding = value.get("binding") if isinstance(value, Mapping) else None
+                    task_id = str(binding.get("task_id") or "") if isinstance(binding, Mapping) else ""
+                    if not task_id:
+                        continue
+                    with self._projection_lock(task_id):
+                        if path.exists():
+                            self._deliver_projection_outbox(path)
+                except Exception:
+                    continue
+            threading.Event().wait(5)
 
     def runtime_status(self) -> dict[str, Any]:
         return {
@@ -113,7 +181,10 @@ class GovernedRuntime:
         }
 
     def _job_root(self, task_id: str) -> Path:
-        return self.config.state_root / "jobs" / _safe(task_id)
+        parent = self.config.state_root / "jobs"
+        keyed = parent / _task_key(task_id)
+        legacy = parent / _legacy_safe(task_id)
+        return keyed if keyed.exists() or not legacy.exists() else legacy
 
     def _build_runtime(
         self,
@@ -443,7 +514,85 @@ class GovernedRuntime:
         return result
 
     def create_task(self, request: Mapping[str, Any]) -> dict[str, Any]:
-        result = self.bootstrap.create(request); self.bindings[result["task_id"]] = result; return result
+        online_request = request.get("online_task")
+        if online_request is not None and not isinstance(online_request, Mapping):
+            raise ValueError("ONLINE_TASK_REQUEST_INVALID")
+        execution_request = request.get("execution")
+        if execution_request is not None and not isinstance(execution_request, Mapping):
+            raise ValueError("EXECUTION_REQUEST_INVALID")
+        online_task_projector = getattr(self, "online_task_projector", None)
+        projection_required = (
+            online_request is not None
+            or execution_request is not None
+            or request.get("require_online_task_projection") is True
+        )
+        if online_task_projector is None:
+            if projection_required:
+                raise RuntimeError("ONLINE_TASK_PROJECTION_CONTROL_UNAVAILABLE")
+            return self.bootstrap.create(request)
+        online = dict(online_request or {})
+        online.setdefault("goal", str(request.get("description") or "Governed engineering task"))
+        online.setdefault("instructions", "Execute only the exact governed binding and return server-verifiable evidence.")
+        online.setdefault("success_criteria", [{"id": "governed", "description": "Complete the exact governed task lifecycle."}])
+        online.setdefault("required_capabilities", ["git.read", "git.test", "governed.write.branch", "governed.write.file"])
+        online.setdefault("priority", int(request.get("priority") or 0))
+        online.setdefault("max_attempts", int(request.get("max_attempts") or 10))
+        normalized = dict(request)
+        normalized["online_task"] = online
+        normalized["execution"] = dict(execution_request or {})
+        normalized["require_online_task_projection"] = True
+        result = self.bootstrap.create(normalized)
+        self.bindings[result["task_id"]] = result
+        projection = self.bootstrap.online_task_projection(
+            result["task_id"],
+            online,
+            normalized["execution"],
+            expected_job_id=str(result.get("job_id") or "") or None,
+        )
+        durable_binding = self.bootstrap.get_binding(result["task_id"])
+        with self._projection_lock(result["task_id"]):
+            path = self._write_projection_outbox(result["task_id"], projection, durable_binding)
+            created = self._deliver_projection_outbox(path)
+        return {
+            **result,
+            "online_task_projection": {
+                "status": "PROJECTED",
+                "task_id": created["task_id"],
+                "job_id": result["job_id"],
+                "task_status": created["status"],
+                "execution_digest": projection["metadata"]["execution"]["execution_digest"],
+            },
+        }
+
+    def project_online_agent_execution(self, request: Mapping[str, Any]) -> dict[str, Any]:
+        capabilities = request.get("required_capabilities") or []
+        if not isinstance(capabilities, list) or not all(isinstance(item, str) for item in capabilities):
+            raise ValueError("CAPABILITY_SCOPE_INVALID")
+        execution = request.get("execution")
+        if execution is not None and not isinstance(execution, Mapping):
+            raise ValueError("EXECUTION_REQUEST_INVALID")
+        task_id = str(request["task_id"])
+        durable = self.bootstrap.get_binding(task_id)
+        binding = self.bootstrap.execution_binding(durable)
+        expected_job_id = str(request.get("job_id") or "") or None
+        if expected_job_id and expected_job_id != binding.job_id:
+            raise BootstrapError("JOB_BINDING_CONFLICT")
+        if execution is None:
+            planned = self.execution_plan_store.resolve(binding)
+            if planned is not None:
+                execution = planned["execution"]
+        return self.bootstrap.execution_envelope(
+            task_id,
+            execution,
+            required_capabilities=tuple(capabilities),
+            expected_job_id=expected_job_id,
+        )
+
+    def register_online_agent_execution_plan(self, request: Mapping[str, Any]) -> dict[str, Any]:
+        task_id = str(request["task_id"])
+        durable = self.bootstrap.get_binding(task_id)
+        binding = self.bootstrap.execution_binding(durable)
+        return self.execution_plan_store.register(binding, request)
 
     def acquire_lease(self, request: Mapping[str, Any]) -> dict[str, Any]:
         binding, controller, _ = self._runtime(request)
@@ -1340,7 +1489,19 @@ class GovernedRuntime:
         controller = self._controller_for_job(job_id)
         row = controller.store.read(job_id)
         if action == "get":
-            return controller.store._receipt(row, duplicate=True).as_dict()
+            response = controller.store._receipt(row, duplicate=True).as_dict()
+            try:
+                with controller.store._connect() as connection:
+                    connection.row_factory = sqlite3.Row
+                    leases = connection.execute(
+                        "SELECT lease_id,job_id,branch,lineage_anchor_oid,current_head_oid,generation,"
+                        "lease_expires_at,state,revision FROM branch_leases WHERE job_id=? ORDER BY revision DESC",
+                        (job_id,),
+                    ).fetchall()
+            except sqlite3.Error as exc:
+                raise RuntimeError("JOB_DB_UNAVAILABLE") from exc
+            response["branch_leases"] = [dict(item) for item in leases]
+            return response
         if action == "resume":
             return controller.resume(job_id).as_dict()
         return self._executor_from_row(row, controller).reconcile(job_id)
@@ -1405,6 +1566,8 @@ class GovernedRuntime:
     def handlers(self) -> dict[str, Any]:
         return {
             "createGovernedTask": self.create_task,
+            "projectOnlineAgentExecution": self.project_online_agent_execution,
+            "registerOnlineAgentExecutionPlan": self.register_online_agent_execution_plan,
             "acquireWriterLease": self.acquire_lease,
             "bootstrapWriterLease": self.bootstrap_writer_lease,
             "renewWriterLease": self.renew_writer_lease,
@@ -1434,6 +1597,8 @@ class _HTTPError(RuntimeError): pass
 
 class BridgeHandler(BaseHTTPRequestHandler):
     adapter: BridgeV2Adapter | None = None
+    control_token: str | None = None
+    agent_task_api_url: str = "http://127.0.0.1:18940"
     def log_message(self, *_args: Any) -> None: return
     def _json(self, value: Any, status: int = 200) -> None:
         raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode(); self.send_response(status); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)
@@ -1451,7 +1616,50 @@ class BridgeHandler(BaseHTTPRequestHandler):
         allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | {"schema", "job_id", "lease_seconds", "create_target_branch", "worktree_mode", "cas_ref_update_required", "expected_branch_oid", "generation", "branch_mode"}
         if operation == "commitFilesGoverned": allowed.add("files")
         allowed.update({"owner_approval", "execute"})
+        if operation == "projectOnlineAgentExecution":
+            allowed.update({"execution", "required_capabilities"})
+        if operation == "registerOnlineAgentExecutionPlan":
+            allowed.update({"execution", "source_head_oid", "expected_previous_plan_digest"})
+        if operation == "createGovernedTask":
+            allowed.update({"online_task", "execution", "require_online_task_projection", "priority", "max_attempts", "parent_task_id", "parent_job_id"})
         if set(body) - allowed: raise _HTTPError("UNKNOWN_FIELDS")
+    def _authorized(self, operation: str, body: Mapping[str, Any]) -> bool:
+        header = self.headers.get("Authorization", "")
+        if not header.startswith("Bearer "):
+            return False
+        supplied = header[7:]
+        if self.control_token and hmac.compare_digest(supplied, self.control_token):
+            return True
+        poll = urllib.request.Request(
+            self.agent_task_api_url.rstrip("/") + "/v1/tasks/poll",
+            method="POST",
+            data=b'{"limit":1}',
+            headers={"Authorization": header, "Content-Type": "application/json", "Accept": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(poll, timeout=5) as response:
+                principal = json.loads(response.read().decode("utf-8"))
+        except (OSError, urllib.error.URLError, urllib.error.HTTPError, ValueError):
+            return False
+        agent_id = str(principal.get("agent_id") or "") if isinstance(principal, Mapping) else ""
+        if not agent_id:
+            return False
+        claim_bound = {"acquireWriterLease", "bootstrapWriterLease", "renewWriterLease", "commitFileGoverned", "commitFilesGoverned", "projectOnlineAgentExecution", "getGitOpsJob"}
+        if operation not in claim_bound:
+            return False
+        task_id = str(body.get("task_id") or "")
+        if not task_id:
+            return False
+        task_request = urllib.request.Request(
+            self.agent_task_api_url.rstrip("/") + "/v1/tasks/" + task_id,
+            headers={"Authorization": header, "Accept": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(task_request, timeout=5) as response:
+                task = json.loads(response.read().decode("utf-8"))
+        except (OSError, urllib.error.URLError, urllib.error.HTTPError, ValueError):
+            return False
+        return isinstance(task, Mapping) and task.get("task_id") == task_id and task.get("claimed_by") == agent_id and task.get("status") in {"CLAIMED", "RUNNING"}
     def do_GET(self) -> None:
         if self.adapter is None: self._json({"error": "RUNTIME_UNAVAILABLE"}, 503); return
         if self.path == "/v2/runtime": self._json(self.adapter.getRuntimeStatus()); return
@@ -1462,14 +1670,23 @@ class BridgeHandler(BaseHTTPRequestHandler):
         operation = ROUTES.get(self.path)
         if operation is None or operation in ("getRuntimeStatus", "getGovernedWriteCapabilities") or self.adapter is None: self._json({"error": "NOT_FOUND"}, 404); return
         try:
-            body = self._body(); self._allowed(operation, body); self._json(getattr(self.adapter, operation)(body))
+            body = self._body()
+        except _HTTPError as exc: self._json({"error": str(exc)}, 413 if str(exc) == "REQUEST_TOO_LARGE" else 400); return
+        if not self._authorized(operation, body): self._json({"error": "BRIDGE_CALLER_UNAUTHORIZED"}, 401); return
+        try:
+            self._allowed(operation, body); self._json(getattr(self.adapter, operation)(body))
         except _HTTPError as exc: self._json({"error": str(exc)}, 413 if str(exc) == "REQUEST_TOO_LARGE" else 400)
-        except (BootstrapError, ValueError, BridgeV2Error) as exc: self._json({"error": str(exc)}, 409 if "CONFLICT" in str(exc) else 400)
+        except (BootstrapError, ExecutionProjectionError, ExecutionPlanError, ValueError, BridgeV2Error) as exc: self._json({"error": str(exc)}, 409 if any(token in str(exc) for token in ("CONFLICT", "DRIFT", "CAS")) else 400)
         except Exception as exc: self._json({"error": str(exc) or type(exc).__name__}, 409 if any(token in str(exc) for token in ("CONFLICT", "EXPECTATION", "MONOTONIC")) else 503)
 
 
-def serve(adapter: BridgeV2Adapter, host: str = "127.0.0.1", port: int = 18892) -> None:
-    BridgeHandler.adapter = adapter; ThreadingHTTPServer((host, port), BridgeHandler).serve_forever()
+def serve(adapter: BridgeV2Adapter, host: str = "127.0.0.1", port: int = 18892, *, control_token_path: Path, agent_task_api_url: str) -> None:
+    if not control_token_path.is_file() or control_token_path.is_symlink():
+        raise RuntimeError("BRIDGE_CONTROL_CREDENTIAL_UNAVAILABLE")
+    control_token = control_token_path.read_text(encoding="utf-8").strip()
+    if len(control_token) < 32:
+        raise RuntimeError("BRIDGE_CONTROL_CREDENTIAL_INVALID")
+    BridgeHandler.adapter = adapter; BridgeHandler.control_token = control_token; BridgeHandler.agent_task_api_url = agent_task_api_url; ThreadingHTTPServer((host, port), BridgeHandler).serve_forever()
 
 
 def main() -> None:
@@ -1488,11 +1705,21 @@ def main() -> None:
         remote_name=os.environ.get("AIOS_GITEA_REMOTE", "gitea-http"),
         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"),
+        agent_task_api_url=os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940"),
+        agent_task_control_token_path=Path(
+            os.environ.get(
+                "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+                str(Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials/aios-chatgpt-bridge-v2-staging.service")) / "agent_task_control_token"),
+            )
+        ),
+        online_task_projection_enabled=os.environ.get("AIOS_ONLINE_TASK_PROJECTION_ENABLED", "1") == "1",
     )
     runtime = GovernedRuntime(config)
     serve(
         BridgeV2Adapter(runtime_status=runtime.runtime_status(), handlers=runtime.handlers()),
         port=int(os.environ.get("AIOS_BRIDGE_V2_PORT", "18892")),
+        control_token_path=config.agent_task_control_token_path,
+        agent_task_api_url=config.agent_task_api_url,
     )
 
 

active-runtime/app/aios_continuation_executor.py

@@ -157,7 +157,7 @@ class DetachedGitWorkspace:
     def materialize_files(self, files: Mapping[str, bytes]) -> None:
         dirty = self._dirty_paths()
         expected_paths = set(files)
-        if dirty and dirty != expected_paths:
+        if dirty and not dirty.issubset(expected_paths):
             raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
         for relative, content in files.items():
             target = (self.path / relative).resolve()

active-runtime/app/aios_durable_owner_authorization.py

@@ -11,19 +11,24 @@ or runtime switching.
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
 import argparse
 import hashlib
+import grp
 import hmac
 from http.server import BaseHTTPRequestHandler
 import json
 import os
+import pwd
 from pathlib import Path
 import re
+import secrets
 import sqlite3
+import socket
 import socketserver
 import stat
 import tempfile
+import struct
 import threading
 import urllib.error
 import urllib.request
@@ -34,6 +39,14 @@ RECEIPT_SCHEMA = "aios.owner-durable-authorization-receipt.v1"
 DECISION_SCHEMA = "aios.owner-durable-authorization-decision.v1"
 PROJECTION_SCHEMA = "aios.external-capability-projection.request.v1"
 PROJECTION_LIST_SCHEMA = "aios.external-capability-projection.list.v1"
+CALLER_PROJECTION_SCHEMA = "aios.caller-admission-projection.v1"
+CALLER_PROJECTION_TTL_DEFAULT = 300
+CALLER_PROJECTION_SCOPES = frozenset({
+    "governed-task:project",
+    "release:candidate",
+})
+GATE_AMENDMENT_SCHEMA = "aios.release-preparation.gate-amendment.v1"
+GATE_AMENDMENT_CLASSIFICATION = "SUPERVISOR_GLOBAL_EXTERNAL_ACK_GATE_REMEDIATED"
 
 ALLOWED_PERSISTENT_CAPABILITIES = frozenset({
     "governed-task:project",
@@ -64,6 +77,80 @@ TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
 BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
 JOB_RE = re.compile(r"^job-[0-9a-f]{32}$")
 MAX_BODY = 512 * 1024
+SO_PEERCRED_SIZE = struct.calcsize("3i")
+
+
+def _peer_process_status(pid: int) -> tuple[int, int, set[int]]:
+    status_path = Path(f"/proc/{pid}/status")
+    try:
+        lines = status_path.read_text(encoding="utf-8").splitlines()
+    except (OSError, UnicodeError) as exc:
+        raise AuthorizationError("SOCKPEER_CREDENTIALS_INVALID", 403) from exc
+    fields: dict[str, str] = {}
+    for line in lines:
+        if ":" in line:
+            key, value = line.split(":", 1)
+            if key in {"Uid", "Gid", "Groups"}:
+                fields[key] = value.strip()
+    try:
+        uid = int(fields["Uid"].split()[0])
+        gid = int(fields["Gid"].split()[0])
+        groups = {int(item) for item in fields["Groups"].split()}
+    except (KeyError, IndexError, ValueError) as exc:
+        raise AuthorizationError("SOCKPEER_CREDENTIALS_INVALID", 403) from exc
+    return uid, gid, groups
+
+
+def _peer_process_groups(pid: int) -> set[int]:
+    return _peer_process_status(pid)[2]
+
+
+def peer_credentials_from_connection(connection: object) -> tuple[int, int, int, set[int]]:
+    if not hasattr(socket, "SO_PEERCRED"):
+        raise AuthorizationError("SO_PEERCRED_UNAVAILABLE", 500)
+    try:
+        raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, SO_PEERCRED_SIZE)
+    except OSError as exc:
+        raise AuthorizationError("SO_PEERCRED_UNAVAILABLE", 403) from exc
+    if len(raw) != SO_PEERCRED_SIZE:
+        raise AuthorizationError("SOCKPEER_CREDENTIALS_INVALID", 403)
+    try:
+        pid, uid, gid = struct.unpack("3i", raw)
+    except struct.error as exc:
+        raise AuthorizationError("SOCKPEER_CREDENTIALS_INVALID", 403) from exc
+    if pid <= 1 or uid < 0 or gid < 0:
+        raise AuthorizationError("SOCKPEER_CREDENTIALS_INVALID", 403)
+    process_uid, process_gid, groups = _peer_process_status(pid)
+    if process_uid != uid or process_gid != gid:
+        raise AuthorizationError("SOCKPEER_CREDENTIALS_INVALID", 403)
+    return pid, uid, gid, groups | {gid}
+
+
+def peer_user_from_connection(connection: object) -> str:
+    _pid, uid, _gid, _groups = peer_credentials_from_connection(connection)
+    try:
+        return pwd.getpwuid(uid).pw_name
+    except (KeyError, OSError) as exc:
+        raise AuthorizationError("SO_PEERCRED_UID_UNKNOWN", 403) from exc
+
+
+CANONICAL_BROKER_GROUP = os.environ.get(
+    "AIOS_OWNER_AUTH_BROKER_ALLOWED_GROUP",
+    "aios-autonomous-agent-worker",
+).strip()
+
+
+def authorize_broker_peer(connection: object) -> bool:
+    _pid, _uid, _gid, groups = peer_credentials_from_connection(connection)
+    if not CANONICAL_BROKER_GROUP:
+        raise AuthorizationError("SOCKGROUP_CONFIGURATION_INVALID", 503)
+    try:
+        canonical_gid = grp.getgrnam(CANONICAL_BROKER_GROUP).gr_gid
+    except KeyError as exc:
+        raise AuthorizationError("SOCKGROUP_CONFIGURATION_INVALID", 503) from exc
+    if canonical_gid not in groups:
+        raise AuthorizationError("SOCKGROUP_NOT_AUTHORIZED", 403)
+    return True
 
 
 class AuthorizationError(RuntimeError):
@@ -134,6 +221,7 @@ class DurablePolicy:
     persistent_until_revoked: bool
     constraints: Constraints
     policy_sha256: str
+    expires_at: datetime | None
 
     @classmethod
     def load(
@@ -158,10 +246,14 @@ class DurablePolicy:
             "capabilities",
             "issued_at",
             "not_before",
+            "expires_at",
             "persistent_until_revoked",
             "constraints",
         }
-        if set(value) != allowed_fields or value.get("schema") != POLICY_SCHEMA:
+        if (
+            not set(value).issubset(allowed_fields)
+            or value.get("schema") != POLICY_SCHEMA
+        ):
             raise AuthorizationError("POLICY_FIELDS_INVALID", 503)
 
         grant_id = str(value.get("grant_id") or "")
@@ -190,6 +282,13 @@ class DurablePolicy:
         current = now or datetime.now(timezone.utc)
         if issued_at > not_before or current < not_before:
             raise AuthorizationError("POLICY_NOT_ACTIVE", 403)
+        expires_at = None
+        if value.get("expires_at") is not None:
+            expires_at = parse_time(value.get("expires_at"), field="expires_at")
+            if expires_at <= not_before:
+                raise AuthorizationError("POLICY_EXPIRY_INVALID", 503)
+            if current >= expires_at:
+                raise AuthorizationError("POLICY_EXPIRED", 403)
         if value.get("persistent_until_revoked") is not True:
             raise AuthorizationError("POLICY_PERSISTENCE_INVALID", 503)
 
@@ -243,8 +342,43 @@ class DurablePolicy:
                 max_task_risk=max_risk,
             ),
             policy_sha256=sha256(policy_path.read_bytes()),
+            expires_at=expires_at,
         )
 
+    def validate_caller_scope(
+        self,
+        caller: str,
+        scope: str,
+        request: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        if caller != self.account:
+            raise AuthorizationError("CALLER_MISMATCH", 403)
+        if scope not in CALLER_PROJECTION_SCOPES:
+            raise AuthorizationError("SCOPE_MISMATCH", 403)
+        if scope not in self.capabilities:
+            raise AuthorizationError("CAPABILITY_NOT_AUTHORIZED", 403)
+        if not isinstance(request, Mapping):
+            raise AuthorizationError("CALLER_PROJECTION_REQUEST_INVALID")
+        if scope == "release:candidate":
+            if set(request) != {"source_commit", "version"}:
+                raise AuthorizationError("CALLER_PROJECTION_FIELDS_INVALID")
+            source_commit = str(request.get("source_commit") or "")
+            version = str(request.get("version") or "")
+            if OID_RE.fullmatch(source_commit) is None:
+                raise AuthorizationError("RELEASE_SOURCE_COMMIT_INVALID")
+            if VERSION_RE.fullmatch(version) is None or not version.startswith(
+                self.constraints.release_version_prefix
+            ):
+                raise AuthorizationError("RELEASE_VERSION_NOT_AUTHORIZED", 403)
+        elif request:
+            raise AuthorizationError("CALLER_PROJECTION_FIELDS_INVALID")
+        return {
+            "schema": CALLER_PROJECTION_SCHEMA,
+            "caller": caller,
+            "scope": scope,
+            "request_sha256": sha256(canonical(request)),
+        }
+
     def authorize(self, capability: str, request: Mapping[str, Any]) -> dict[str, Any]:
         if capability not in self.capabilities:
             raise AuthorizationError("CAPABILITY_NOT_AUTHORIZED", 403)
@@ -447,9 +581,160 @@ class ReceiptStore:
                     ON projection_outbox(state, capability, created_at)
                 """
             )
+            con.execute(
+                """
+                CREATE TABLE IF NOT EXISTS caller_projections(
+                  projection_id TEXT PRIMARY KEY,
+                  caller TEXT NOT NULL,
+                  scope TEXT NOT NULL,
+                  grant_id TEXT NOT NULL,
+                  policy_sha256 TEXT NOT NULL,
+                  projection_state TEXT NOT NULL,
+                  credential_state TEXT NOT NULL,
+                  rehydration_state TEXT NOT NULL,
+                  issued_at TEXT NOT NULL,
+                  expires_at TEXT NOT NULL,
+                  updated_at TEXT NOT NULL,
+                  UNIQUE(caller, scope)
+                )
+                """
+            )
+            con.execute(
+                """
+                CREATE TABLE IF NOT EXISTS gate_amendments(
+                  idempotency_key TEXT PRIMARY KEY,
+                  amendment_sha256 TEXT NOT NULL,
+                  amendment_json TEXT NOT NULL,
+                  created_at TEXT NOT NULL
+                )
+                """
+            )
             con.commit()
         self._lock = threading.RLock()
 
+    @staticmethod
+    def caller_projection_id(caller: str, scope: str) -> str:
+        digest = sha256(canonical({"caller": caller, "scope": scope}))
+        return "caller-projection-" + digest[:32]
+
+    def upsert_caller_projection(
+        self,
+        *,
+        caller: str,
+        scope: str,
+        grant_id: str,
+        policy_sha256: str,
+        ttl_seconds: int,
+        effective_scopes: Any,
+    ) -> dict[str, Any]:
+        if not 30 <= ttl_seconds <= 900:
+            raise AuthorizationError("CALLER_PROJECTION_TTL_INVALID", 503)
+        normalized_scopes = sorted({str(item) for item in effective_scopes})
+        if (
+            not normalized_scopes
+            or scope not in normalized_scopes
+            or not set(normalized_scopes).issubset(CALLER_PROJECTION_SCOPES)
+        ):
+            raise AuthorizationError("CALLER_PROJECTION_SCOPES_INVALID", 503)
+        requested_projection_id = self.caller_projection_id(caller, scope)
+        issued = datetime.now(timezone.utc)
+        issued_at = issued.isoformat()
+        expires_at = (issued + timedelta(seconds=ttl_seconds)).isoformat()
+        requested_recovery = False
+        with self._lock, sqlite3.connect(self.path) as con:
+            con.row_factory = sqlite3.Row
+            con.execute("BEGIN IMMEDIATE")
+            for projected_scope in normalized_scopes:
+                projection_id = self.caller_projection_id(caller, projected_scope)
+                row = con.execute(
+                    "SELECT * FROM caller_projections WHERE caller=? AND scope=?",
+                    (caller, projected_scope),
+                ).fetchone()
+                if row is not None and row["projection_state"] not in {"ACKED", "PENDING"}:
+                    con.rollback()
+                    raise AuthorizationError("PROJECTION_STATE_INVALID", 503)
+                recovery = row is not None
+                if projected_scope == scope:
+                    requested_recovery = recovery
+                if row is None:
+                    con.execute(
+                        """
+                        INSERT INTO caller_projections(
+                          projection_id,caller,scope,grant_id,policy_sha256,
+                          projection_state,credential_state,rehydration_state,
+                          issued_at,expires_at,updated_at
+                        ) VALUES(?,?,?,?,?,?,?,?,?,?,?)
+                        """,
+                        (
+                            projection_id,
+                            caller,
+                            projected_scope,
+                            grant_id,
+                            policy_sha256,
+                            "PENDING",
+                            "GENERATING",
+                            "RECOVERING" if recovery else "GENERATING",
+                            issued_at,
+                            expires_at,
+                            issued_at,
+                        ),
+                    )
+                else:
+                    con.execute(
+                        """
+                        UPDATE caller_projections
+                           SET projection_id=?,grant_id=?,policy_sha256=?,
+                               projection_state='PENDING',
+                               credential_state='GENERATING',
+                               rehydration_state=?,
+                               issued_at=?,expires_at=?,updated_at=?
+                         WHERE caller=? AND scope=?
+                        """,
+                        (
+                            projection_id,
+                            grant_id,
+                            policy_sha256,
+                            "RECOVERING",
+                            issued_at,
+                            expires_at,
+                            issued_at,
+                            caller,
+                            projected_scope,
+                        ),
+                    )
+                con.execute(
+                    """
+                    UPDATE caller_projections
+                       SET projection_state='ACKED',
+                           credential_state='REHYDRATED',
+                           rehydration_state=?,
+                           updated_at=?
+                     WHERE caller=? AND scope=?
+                    """,
+                    (
+                        "RECOVERED" if recovery else "GENERATED",
+                        issued_at,
+                        caller,
+                        projected_scope,
+                    ),
+                )
+            con.commit()
+        return {
+            "schema": CALLER_PROJECTION_SCHEMA,
+            "projection_id": requested_projection_id,
+            "caller": caller,
+            "scope": scope,
+            "grant_id": grant_id,
+            "policy_sha256": policy_sha256,
+            "projection_state": "ACKED",
+            "credential_state": "REHYDRATED",
+            "rehydration_state": "RECOVERED" if requested_recovery else "GENERATED",
+            "effective_scopes": normalized_scopes,
+            "issued_at": issued_at,
+            "expires_at": expires_at,
+            "secret_exposed": False,
+        }
+
     def get(self, key: str) -> dict[str, Any] | None:
         with sqlite3.connect(self.path) as con:
             row = con.execute(
@@ -519,6 +804,37 @@ class ReceiptStore:
             )
             con.commit()
 
+    def put_gate_amendment(
+        self,
+        idempotency_key: str,
+        amendment: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        encoded = json.dumps(amendment, sort_keys=True, separators=(",", ":"))
+        lowered = encoded.lower()
+        if any(marker in lowered for marker in (
+            "bearer ", "access_token", "refresh_token", "private_key", "credential_file"
+        )):
+            raise AuthorizationError("GATE_AMENDMENT_SECRET_MATERIAL_DENY", 403)
+        digest = sha256(encoded.encode())
+        with self._lock, sqlite3.connect(self.path) as con:
+            con.execute("BEGIN IMMEDIATE")
+            row = con.execute(
+                "SELECT amendment_sha256,amendment_json FROM gate_amendments WHERE idempotency_key=?",
+                (idempotency_key,),
+            ).fetchone()
+            if row is not None:
+                if row[0] != digest or row[1] != encoded:
+                    con.rollback()
+                    raise AuthorizationError("GATE_AMENDMENT_IDEMPOTENCY_CONFLICT", 409)
+                con.commit()
+                return {"amendment_sha256": digest, "idempotent": True}
+            con.execute(
+                "INSERT INTO gate_amendments(idempotency_key,amendment_sha256,amendment_json,created_at) VALUES(?,?,?,?)",
+                (idempotency_key, digest, encoded, datetime.now(timezone.utc).isoformat()),
+            )
+            con.commit()
+        return {"amendment_sha256": digest, "idempotent": False}
+
     @staticmethod
     def projection_id(
         grant_id: str,
@@ -716,12 +1032,83 @@ class Broker:
         store: ReceiptStore,
         upstream: Upstream,
         expected_account: str,
+        expected_peer_user: str,
+        allowed_peer_users: frozenset[str] | None = None,
     ) -> None:
         self.policy_path = policy_path
         self.revocation_path = revocation_path
         self.store = store
         self.upstream = upstream
         self.expected_account = expected_account
+        self.expected_peer_user = expected_peer_user
+        self.allowed_peer_users = allowed_peer_users or frozenset({expected_peer_user})
+
+    @staticmethod
+    def validate_caller_projection(
+        value: Mapping[str, Any],
+        *,
+        expected_scopes: Any = None,
+    ) -> dict[str, Any]:
+        expected = sorted(
+            set(expected_scopes) if expected_scopes is not None else CALLER_PROJECTION_SCOPES
+        )
+        if value.get("caller") != "chatgpt-governed-implementer-singlehost":
+            raise AuthorizationError("CALLER_MISMATCH", 503)
+        if value.get("scope") not in expected:
+            raise AuthorizationError("SCOPE_MISMATCH", 503)
+        if value.get("projection_state") != "ACKED":
+            raise AuthorizationError("PROJECTION_STATE_INVALID", 503)
+        if value.get("credential_state") != "REHYDRATED":
+            raise AuthorizationError("CREDENTIAL_REHYDRATION_INVALID", 503)
+        if value.get("rehydration_state") not in {"GENERATED", "RECOVERED"}:
+            raise AuthorizationError("REHYDRATION_STATE_INVALID", 503)
+        if value.get("effective_scopes") != expected:
+            raise AuthorizationError("PROJECTION_SCOPE_INVALID", 503)
+        if value.get("secret_exposed") is not False:
+            raise AuthorizationError("PROJECTION_SECRET_EXPOSURE", 503)
+        return dict(value)
+
+    def project_caller(
+        self,
+        caller: str,
+        scope: str,
+        request: Mapping[str, Any],
+        *,
+        idempotency_key: str,
+    ) -> dict[str, Any]:
+        if not GRANT_RE.fullmatch(idempotency_key):
+            raise AuthorizationError("IDEMPOTENCY_KEY_INVALID")
+        if caller != self.expected_account:
+            raise AuthorizationError("CALLER_MISMATCH", 403)
+        if scope not in CALLER_PROJECTION_SCOPES:
+            raise AuthorizationError("SCOPE_MISMATCH", 403)
+        policy = self.policy()
+        policy.validate_caller_scope(caller, scope, request)
+        effective_scopes = sorted(policy.capabilities & CALLER_PROJECTION_SCOPES)
+        try:
+            ttl_seconds = int(
+                os.environ.get(
+                    "AIOS_OWNER_AUTH_CALLER_PROJECTION_TTL_SECONDS",
+                    str(CALLER_PROJECTION_TTL_DEFAULT),
+                )
+            )
+        except ValueError as exc:
+            raise AuthorizationError("CALLER_PROJECTION_TTL_INVALID", 503) from exc
+        credential = secrets.token_urlsafe(32)
+        value = self.store.upsert_caller_projection(
+            caller=caller,
+            scope=scope,
+            grant_id=policy.grant_id,
+            policy_sha256=policy.policy_sha256,
+            ttl_seconds=ttl_seconds,
+            effective_scopes=effective_scopes,
+        )
+        value = self.validate_caller_projection(
+            value,
+            expected_scopes=effective_scopes,
+        )
+        value["_credential"] = credential
+        return value
 
     def policy(self) -> DurablePolicy:
         policy = DurablePolicy.load(self.policy_path, self.revocation_path)
@@ -758,24 +1145,90 @@ class Broker:
             "separate_owner_release_approval_still_required": True,
         }
 
+    @staticmethod
+    def validate_gate_amendment(
+        amendment: Mapping[str, Any],
+        *,
+        policy: DurablePolicy,
+        request: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        expected = {
+            "schema": GATE_AMENDMENT_SCHEMA,
+            "classification": GATE_AMENDMENT_CLASSIFICATION,
+            "supervisor_task_id": "r42-release-preparation-autonomous-supervisor-20260820",
+            "supervisor_execution_digest": "491457595667a95809f6f3cd19e9e7c9e74181070fd8038b86f30eb41bdfc231",
+            "actor_type": "credentialed_host_coordinator",
+            "operation": "createGovernedTask",
+            "repository": "aios/aios",
+            "original_external_request_sha256": "08fc2f3d1b64a2ca11da9f22da48de834cff3442fa7fa2d3bc390f8dbf9b254a",
+            "external_connector_rule": "ACK_STILL_REQUIRED",
+            "host_coordinator_rule": "DURABLE_OWNER_GOVERNED_TASK_PROJECT",
+            "owner_grant_id": policy.grant_id,
+            "owner_policy_sha256": policy.policy_sha256,
+            "builder_task_id": "r42-trusted-candidate-builder-runtime-wiring-20260820",
+            "builder_base_oid": "d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
+            "builder_expected_branch_oid": "d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
+            "builder_target_branch": "agent/r42-trusted-candidate-builder-runtime-wiring-20260820",
+            "builder_generation": 1,
+            "builder_idempotency_key": "r42-trusted-candidate-builder-runtime-wiring-binding-v1",
+        }
+        value = dict(amendment)
+        if value != expected:
+            raise AuthorizationError("GATE_AMENDMENT_BINDING_INVALID", 403)
+        request_coordinates = {
+            "task_id": expected["builder_task_id"],
+            "base_oid": expected["builder_base_oid"],
+            "expected_branch_oid": expected["builder_expected_branch_oid"],
+            "target_branch": expected["builder_target_branch"],
+            "generation": expected["builder_generation"],
+            "idempotency_key": expected["builder_idempotency_key"],
+            "repository": expected["repository"],
+        }
+        if any(request.get(key) != item for key, item in request_coordinates.items()):
+            raise AuthorizationError("GATE_AMENDMENT_BUILDER_BINDING_INVALID", 403)
+        return value
+
     def execute(
         self,
         capability: str,
         request: Mapping[str, Any],
         *,
         idempotency_key: str,
+        amendment: Mapping[str, Any] | None = None,
     ) -> dict[str, Any]:
         if not GRANT_RE.fullmatch(idempotency_key):
             raise AuthorizationError("IDEMPOTENCY_KEY_INVALID")
         policy = self.policy()
         decision = policy.authorize(capability, request)
         request_sha = decision["request_sha256"]
+        amendment_value: dict[str, Any] | None = None
+        if amendment is not None:
+            if capability != "governed-task:project":
+                raise AuthorizationError("GATE_AMENDMENT_CAPABILITY_INVALID", 403)
+            amendment_value = self.validate_gate_amendment(
+                amendment,
+                policy=policy,
+                request=request,
+            )
 
         prior = self.store.get(idempotency_key)
         if prior is not None:
             if prior["request_sha256"] != request_sha:
                 raise AuthorizationError("IDEMPOTENCY_CONFLICT", 409)
             response = prior["response"]
+            if amendment_value is not None:
+                amendment_result = self.store.put_gate_amendment(
+                    idempotency_key,
+                    amendment_value,
+                )
+                response = {
+                    **response,
+                    "gate_amendment": {
+                        "classification": GATE_AMENDMENT_CLASSIFICATION,
+                        "amendment_sha256": amendment_result["amendment_sha256"],
+                    },
+                }
+                self.store.update_response(idempotency_key, request_sha, response)
             if (
                 capability == "release:candidate"
                 and not response.get("projection_id")
@@ -862,6 +1315,15 @@ class Broker:
             ),
             "separate_owner_release_approval_still_required": True,
         }
+        if amendment_value is not None:
+            amendment_result = self.store.put_gate_amendment(
+                idempotency_key,
+                amendment_value,
+            )
+            response["gate_amendment"] = {
+                "classification": GATE_AMENDMENT_CLASSIFICATION,
+                "amendment_sha256": amendment_result["amendment_sha256"],
+            }
         self.store.put(idempotency_key, capability, request_sha, response)
         return response
 
@@ -881,11 +1343,22 @@ class Handler(BaseHTTPRequestHandler):
     def log_message(self, *_args: Any) -> None:
         return
 
-    def send_json(self, value: Mapping[str, Any], status: int) -> None:
+    def send_json(
+        self,
+        value: Mapping[str, Any],
+        status: int,
+        *,
+        internal_credential: str | None = None,
+    ) -> None:
         raw = canonical(value)
         self.send_response(status)
         self.send_header("Content-Type", "application/json")
         self.send_header("Cache-Control", "no-store")
+        if internal_credential:
+            self.send_header(
+                "X-AIOS-Caller-Projection-Credential",
+                internal_credential,
+            )
         self.send_header("Content-Length", str(len(raw)))
         self.end_headers()
         self.wfile.write(raw)
@@ -893,6 +1366,16 @@ class Handler(BaseHTTPRequestHandler):
     def caller_account(self) -> str:
         return self.headers.get("X-AIOS-Account", "").strip()
 
+    def peer_user(self) -> str:
+        if self.broker is None:
+            return ""
+        return peer_user_from_connection(self.connection)
+
+    def peer_authorized(self) -> bool:
+        if self.broker is None:
+            return False
+        return authorize_broker_peer(self.connection)
+
     def account_authorized(self) -> bool:
         if self.broker is None:
             return False
@@ -924,6 +1407,14 @@ class Handler(BaseHTTPRequestHandler):
                 200,
             )
             return
+        try:
+            peer_ok = self.peer_authorized()
+        except AuthorizationError as exc:
+            self.send_json({"error": exc.code}, exc.status)
+            return
+        if not peer_ok:
+            self.send_json({"error": "SOCKGROUP_NOT_AUTHORIZED"}, 403)
+            return
         if not self.account_authorized():
             self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
             return
@@ -938,6 +1429,12 @@ class Handler(BaseHTTPRequestHandler):
                     "grant_id": policy.grant_id,
                     "account": policy.account,
                     "capabilities": sorted(policy.capabilities),
+                    "policy_sha256": policy.policy_sha256,
+                    "constraints": {
+                        "repository": policy.constraints.repository,
+                        "allowed_target_prefix": policy.constraints.allowed_target_prefix,
+                        "max_task_risk": policy.constraints.max_task_risk,
+                    },
                     "persistent_until_revoked": True,
                     "separate_owner_release_approval_still_required": True,
                 }
@@ -957,12 +1454,21 @@ class Handler(BaseHTTPRequestHandler):
             self.send_json({"error": exc.code}, exc.status)
 
     def do_POST(self) -> None:
+        try:
+            peer_ok = self.peer_authorized()
+        except AuthorizationError as exc:
+            self.send_json({"error": exc.code}, exc.status)
+            return
+        if not peer_ok:
+            self.send_json({"error": "SOCKGROUP_NOT_AUTHORIZED"}, 403)
+            return
         if not self.account_authorized():
             self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
             return
         routes = {
             "/v1/project-governed-task": "governed-task:project",
             "/v1/authorize-release-candidate": "release:candidate",
+            "/v1/project-caller": "caller:projection",
         }
         capability = routes.get(self.path)
         if capability is None or self.broker is None:
@@ -970,14 +1476,49 @@ class Handler(BaseHTTPRequestHandler):
             return
         try:
             body = self.body()
-            if set(body) != {"idempotency_key", "request"}:
+            if self.path == "/v1/project-caller":
+                if set(body) != {
+                    "caller",
+                    "scope",
+                    "idempotency_key",
+                    "request",
+                }:
+                    raise AuthorizationError("CALLER_PROJECTION_FIELDS_INVALID")
+                if not isinstance(body["request"], Mapping):
+                    raise AuthorizationError("CALLER_PROJECTION_REQUEST_INVALID")
+                value = self.broker.project_caller(
+                    str(body["caller"]),
+                    str(body["scope"]),
+                    dict(body["request"]),
+                    idempotency_key=str(body["idempotency_key"]),
+                )
+                credential = str(value.pop("_credential", ""))
+                if len(credential) < 32:
+                    raise AuthorizationError("CREDENTIAL_REHYDRATION_MISSING", 503)
+                self.send_json(
+                    value,
+                    200,
+                    internal_credential=credential,
+                )
+                return
+            if set(body) not in (
+                {"idempotency_key", "request"},
+                {"idempotency_key", "request", "amendment"},
+            ):
                 raise AuthorizationError("BROKER_FIELDS_INVALID")
             if not isinstance(body["request"], Mapping):
                 raise AuthorizationError("BROKER_REQUEST_INVALID")
+            if "amendment" in body and not isinstance(body["amendment"], Mapping):
+                raise AuthorizationError("GATE_AMENDMENT_INVALID")
             value = self.broker.execute(
                 capability,
                 dict(body["request"]),
                 idempotency_key=str(body["idempotency_key"]),
+                amendment=(
+                    dict(body["amendment"])
+                    if isinstance(body.get("amendment"), Mapping)
+                    else None
+                ),
             )
             self.send_json(value, 200)
         except AuthorizationError as exc:
@@ -989,6 +1530,41 @@ class UnixHTTPServer(socketserver.ThreadingUnixStreamServer):
 
 def serve() -> None:
     credentials = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+    raw_allowed_users = os.environ.get(
+        "AIOS_OWNER_AUTH_BROKER_ALLOWED_USERS",
+        "",
+    ).strip()
+    if raw_allowed_users:
+        allowed_peer_users = tuple(
+            dict.fromkeys(
+                item.strip()
+                for item in raw_allowed_users.split(",")
+                if item.strip()
+            )
+        )
+    else:
+        single_allowed_user = os.environ.get(
+            "AIOS_OWNER_AUTH_BROKER_ALLOWED_USER",
+            "aios-autonomous-agent-worker",
+        ).strip()
+        allowed_peer_users = (single_allowed_user,) if single_allowed_user else ()
+    if not allowed_peer_users:
+        raise RuntimeError("BROKER_ALLOWED_USER_MISSING")
+    allowed_peer_user = allowed_peer_users[0]
+    for username in allowed_peer_users:
+        pwd.getpwnam(username)
+    allowed_peer_uid = pwd.getpwnam(allowed_peer_user).pw_uid
+    global CANONICAL_BROKER_GROUP
+    CANONICAL_BROKER_GROUP = os.environ.get(
+        "AIOS_OWNER_AUTH_BROKER_ALLOWED_GROUP",
+        "aios-autonomous-agent-worker",
+    ).strip()
+    if not CANONICAL_BROKER_GROUP:
+        raise RuntimeError("BROKER_ALLOWED_GROUP_MISSING")
+    try:
+        allowed_peer_gid = grp.getgrnam(CANONICAL_BROKER_GROUP).gr_gid
+    except KeyError as exc:
+        raise RuntimeError("BROKER_ALLOWED_GROUP_INVALID") from exc
     policy_path = Path(
         os.environ.get(
             "AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY",
@@ -1029,6 +1605,8 @@ def serve() -> None:
             bridge_token=bridge_token,
         ),
         expected_account=account,
+        expected_peer_user=allowed_peer_user,
+        allowed_peer_users=frozenset(allowed_peer_users),
     )
     socket_path = Path(
         os.environ.get(
@@ -1044,6 +1622,7 @@ def serve() -> None:
             raise RuntimeError("BROKER_SOCKET_CONFLICT")
         socket_path.unlink()
     server = UnixHTTPServer(str(socket_path), Handler)
+    os.chown(socket_path, allowed_peer_uid, allowed_peer_gid)
     os.chmod(socket_path, 0o660)
     try:
         server.serve_forever()

active-runtime/app/aios_governed_task_bootstrap.py

@@ -2,13 +2,20 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, asdict
+from contextlib import contextmanager
+import fcntl
 import hashlib
 import json
 from pathlib import Path
 import re
 from typing import Any, Mapping
 
-from aios_gitops_job_controller import GitOpsJobController, JobStore
+from aios_gitops_job_controller import GitOpsJobController, JobStore, sha256_json
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+    project_online_task,
+)
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
@@ -36,35 +43,161 @@ class GovernedTaskBinding:
         return {"task_id": self.task_id, "repository": self.repository, "base_oid": self.base_oid, "target_branch": self.target_branch, "expected_branch_oid": self.expected_branch_oid, "generation": self.generation, "branch_mode": self.branch_mode, "create_target_branch": self.branch_mode == "CREATE_NEW", "worktree_mode": "detached_at_base_oid" if self.branch_mode == "CREATE_NEW" else "detached_at_expected_head", "cas_ref_update_required": True, "idempotent": idempotent, "state": "BOUND", "job_id": self.job_id}
 
 
-def _safe(value: str) -> str:
+def _legacy_safe(value: str) -> str:
     return re.sub(r"[^A-Za-z0-9_.-]+", "-", value).strip("-")[:120] or "task"
 
 
+def _task_key(value: str) -> str:
+    readable = _legacy_safe(value)[:72]
+    return f"{readable}-{hashlib.sha256(value.encode('utf-8')).hexdigest()[:24]}"
+
+
 class GovernedTaskBootstrap:
     def __init__(self, authority: Any, root: Path, *, remote_name: str, token_path: Path, ssh_key_path: Path, api_base: str = "https://g.cnnetmkt.top/git/api/v1") -> None:
-        self.authority = authority; self.root = Path(root); self.remote_name = remote_name; self.token_path = Path(token_path); self.ssh_key_path = Path(ssh_key_path); self.api_base = api_base; self.task_meta_root = self.root / "task-bindings"; self.job_root = self.root / "jobs"; self.task_meta_root.mkdir(parents=True, exist_ok=True); self.job_root.mkdir(parents=True, exist_ok=True)
+        self.authority = authority; self.root = Path(root); self.remote_name = remote_name; self.token_path = Path(token_path); self.ssh_key_path = Path(ssh_key_path); self.api_base = api_base; self.task_meta_root = self.root / "task-bindings"; self.job_root = self.root / "jobs"; self.lock_root = self.root / "task-locks"; self.task_meta_root.mkdir(parents=True, exist_ok=True); self.job_root.mkdir(parents=True, exist_ok=True); self.lock_root.mkdir(parents=True, exist_ok=True)
 
     def _store(self, task_id: str) -> tuple[JobStore, GitOpsJobController]:
-        root = self.job_root / _safe(task_id); store = JobStore(root, payload_root=root / "payloads"); return store, GitOpsJobController(store)
+        legacy = self.job_root / _legacy_safe(task_id)
+        keyed = self.job_root / _task_key(task_id)
+        root = legacy if legacy.is_dir() and not keyed.exists() else keyed
+        store = JobStore(root, payload_root=root / "payloads"); return store, GitOpsJobController(store)
 
     def _metadata(self, task_id: str) -> Path:
-        return self.task_meta_root / (_safe(task_id) + ".json")
+        return self.task_meta_root / (_task_key(task_id) + ".json")
+
+    def _legacy_metadata(self, task_id: str) -> Path:
+        return self.task_meta_root / (_legacy_safe(task_id) + ".json")
+
+    @contextmanager
+    def _task_lock(self, task_id: str):
+        path = self.lock_root / (_task_key(task_id) + ".lock")
+        with path.open("a+b") as handle:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+            try:
+                yield
+            finally:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+
+    @contextmanager
+    def _branch_lock(self, repository: str, target_branch: str):
+        key = hashlib.sha256((repository + "\0" + target_branch).encode("utf-8")).hexdigest()
+        path = self.lock_root / ("branch-" + key + ".lock")
+        with path.open("a+b") as handle:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+            try:
+                yield
+            finally:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
 
     def _read_existing(self, task_id: str) -> GovernedTaskBinding | None:
         path = self._metadata(task_id)
+        if not path.exists():
+            path = self._legacy_metadata(task_id)
         if not path.is_file() or path.is_symlink(): return None
-        return GovernedTaskBinding(**json.loads(path.read_text(encoding="utf-8")))
+        binding = GovernedTaskBinding(**json.loads(path.read_text(encoding="utf-8")))
+        if binding.task_id != task_id:
+            raise BootstrapError("TASK_BINDING_KEY_COLLISION")
+        return binding
+
+    def get_binding(self, task_id: str) -> GovernedTaskBinding:
+        """Return the exact durable binding; never synthesize missing lineage."""
+        existing = self._read_existing(str(task_id))
+        if existing is None or not existing.job_id:
+            raise BootstrapError("TASK_EXECUTION_PROJECTION_MISSING")
+        return existing
+
+    def execution_binding(self, binding: GovernedTaskBinding) -> GovernedExecutionBinding:
+        expected = binding.expected_branch_oid
+        mode = binding.branch_mode
+        if mode == "CREATE_NEW":
+            current = self.authority.read_ref("refs/heads/" + binding.target_branch)
+            if current is not None:
+                store, _ = self._store(binding.task_id)
+                with store._connect() as connection:
+                    rows = connection.execute(
+                        "SELECT current_head_oid,generation,state,lineage_anchor_oid "
+                        "FROM branch_leases WHERE job_id=? AND branch=? AND state='ACTIVE' "
+                        "ORDER BY revision DESC",
+                        (binding.job_id, binding.target_branch),
+                    ).fetchall()
+                if len(rows) != 1:
+                    raise BootstrapError("SAME_JOB_CONTINUATION_LEASE_INVALID")
+                row = rows[0]
+                if (
+                    str(row["current_head_oid"] or "") != current
+                    or int(row["generation"] or 0) != binding.generation
+                    or str(row["lineage_anchor_oid"] or "") != binding.base_oid
+                ):
+                    raise BootstrapError("SAME_JOB_CONTINUATION_BINDING_INVALID")
+                expected = current
+                mode = "CONTINUE_EXISTING"
+        return GovernedExecutionBinding.from_mapping(
+            {
+                "task_id": binding.task_id,
+                "job_id": binding.job_id,
+                "repository": binding.repository,
+                "base_oid": binding.base_oid,
+                "expected_branch_oid": expected,
+                "target_branch": binding.target_branch,
+                "branch_mode": mode,
+                "generation": binding.generation,
+                "idempotency_key": binding.idempotency_key,
+            }
+        )
+
+    def execution_envelope(
+        self,
+        task_id: str,
+        execution_request: Mapping[str, Any] | None = None,
+        *,
+        required_capabilities: tuple[str, ...] = (),
+        expected_job_id: str | None = None,
+    ) -> dict[str, Any]:
+        binding = self.get_binding(task_id)
+        if expected_job_id and binding.job_id != expected_job_id:
+            raise BootstrapError("JOB_BINDING_CONFLICT")
+        return project_execution_envelope(
+            self.execution_binding(binding),
+            execution_request,
+            required_capabilities=required_capabilities,
+        )
+
+    def online_task_projection(
+        self,
+        task_id: str,
+        task_request: Mapping[str, Any],
+        execution_request: Mapping[str, Any] | None = None,
+        *,
+        expected_job_id: str | None = None,
+    ) -> dict[str, Any]:
+        binding = self.get_binding(task_id)
+        if expected_job_id and binding.job_id != expected_job_id:
+            raise BootstrapError("JOB_BINDING_CONFLICT")
+        return project_online_task(self.execution_binding(binding), task_request, execution_request)
 
     def _write(self, binding: GovernedTaskBinding) -> None:
         path = self._metadata(binding.task_id); temporary = path.with_suffix(".tmp"); temporary.write_text(json.dumps(asdict(binding), sort_keys=True), encoding="utf-8"); temporary.replace(path)
 
     def create(self, request: Mapping[str, Any]) -> dict[str, Any]:
+        task_id = str(request.get("task_id") or "")
+        if not task_id:
+            raise BootstrapError("TASK_BINDING_INVALID")
+        repository = str(request.get("repository") or "")
+        target_branch = str(request.get("target_branch") or "")
+        with self._task_lock(task_id):
+            with self._branch_lock(repository, target_branch):
+                return self._create_locked(request)
+
+    def _create_locked(self, request: Mapping[str, Any]) -> dict[str, Any]:
         values = {key: str(request[key]) for key in ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")}
         if values["repository"] != "aios/aios" or not OID_RE.fullmatch(values["base_oid"]): raise BootstrapError("TASK_BINDING_INVALID")
         if not BRANCH_RE.fullmatch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
         existing = self._read_existing(values["task_id"])
         if existing is not None:
-            if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description")): raise BootstrapError("TASK_BINDING_CONFLICT")
+            if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")): raise BootstrapError("TASK_BINDING_CONFLICT")
+            optional = {"expected_branch_oid": existing.expected_branch_oid, "branch_mode": existing.branch_mode, "generation": existing.generation}
+            for key, durable in optional.items():
+                if key in request and str(request[key]) != str(durable): raise BootstrapError("TASK_BINDING_CONFLICT")
             return existing.response(idempotent=True)
         if not self.authority.commit_exists(values["base_oid"]): raise BootstrapError("BASE_COMMIT_NOT_FOUND")
         current = self.authority.read_ref("refs/heads/" + values["target_branch"])
@@ -74,8 +207,61 @@ class GovernedTaskBootstrap:
             if expected != values["base_oid"]: raise BootstrapError("EXPECTED_BASE_CONFLICT")
         else:
             if expected != current: raise BootstrapError("EXISTING_BRANCH_OID_CONFLICT")
-            mode, generation = "CONTINUE_EXISTING", int(request.get("generation", 2))
-            if generation < 1: raise BootstrapError("GENERATION_INVALID")
+            parent_task_id = str(request.get("parent_task_id") or "")
+            parent_job_id = str(request.get("parent_job_id") or "")
+            if not parent_task_id or not parent_job_id:
+                raise BootstrapError("EXISTING_BRANCH_OWNERSHIP_REQUIRED")
+            parent = self._read_existing(parent_task_id)
+            if parent is None or parent.job_id != parent_job_id or parent.repository != values["repository"] or parent.target_branch != values["target_branch"]:
+                raise BootstrapError("EXISTING_BRANCH_OWNERSHIP_CONFLICT")
+            store, _ = self._store(parent_task_id)
+            with store._connect() as connection:
+                rows = connection.execute(
+                    "SELECT current_head_oid,generation,state FROM branch_leases WHERE job_id=? AND branch=? AND state='ACTIVE' ORDER BY revision DESC",
+                    (parent_job_id, values["target_branch"]),
+                ).fetchall()
+            if len(rows) != 1 or str(rows[0]["current_head_oid"] or "") != current:
+                raise BootstrapError("EXISTING_BRANCH_OWNERSHIP_CONFLICT")
+            mode, generation = "CONTINUE_EXISTING", int(rows[0]["generation"]) + 1
+            if "generation" in request and int(request["generation"]) != generation:
+                raise BootstrapError("GENERATION_BINDING_CONFLICT")
+        online_request = request.get("online_task")
+        execution_request = request.get("execution")
+        projection_required = (
+            online_request is not None
+            or execution_request is not None
+            or request.get("require_online_task_projection") is True
+        )
+        if projection_required:
+            if not isinstance(online_request, Mapping):
+                raise BootstrapError("ONLINE_TASK_REQUEST_INVALID")
+            if execution_request is not None and not isinstance(execution_request, Mapping):
+                raise BootstrapError("EXECUTION_REQUEST_INVALID")
+            provisional_job_id = "job-" + sha256_json(
+                {"client_idempotency_key": values["idempotency_key"]}
+            )[:32]
+            try:
+                provisional_binding = GovernedExecutionBinding.from_mapping(
+                    {
+                        "task_id": values["task_id"],
+                        "job_id": provisional_job_id,
+                        "repository": values["repository"],
+                        "base_oid": values["base_oid"],
+                        "expected_branch_oid": expected,
+                        "target_branch": values["target_branch"],
+                        "branch_mode": mode,
+                        "generation": generation,
+                        "idempotency_key": values["idempotency_key"],
+                    }
+                )
+                project_online_task(
+                    provisional_binding,
+                    dict(online_request),
+                    dict(execution_request or {}),
+                )
+            except Exception as exc:
+                raise BootstrapError("EXECUTION_BINDING_INVALID") from exc
+
         payload = json.dumps(values, sort_keys=True, separators=(",", ":")).encode()
         _, controller = self._store(values["task_id"])
         receipt = controller.submit({"idempotency_key": values["idempotency_key"], "campaign_id": values["task_id"], "task_id": values["task_id"], "repository": values["repository"], "operation": "GOVERNED_TASK_CREATE", "branch": values["target_branch"], "expected_branch_oid": expected, "payload_digest": hashlib.sha256(payload).hexdigest(), "lineage_anchor_oid": values["base_oid"]}, payload=payload)

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
 
@@ -89,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 = [
@@ -112,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',
@@ -124,6 +141,25 @@ SILENT_RELEASE_POST_PATHS = frozenset({
     '/startDeployment',
 })
 
+GOVERNED_EXECUTION_CAPABILITIES = frozenset({
+    "governed.write.branch",
+    "governed.write.file",
+})
+
+
+def requires_governed_execution(value: Mapping[str, Any]) -> bool:
+    capabilities = value.get("required_capabilities")
+    if capabilities is None:
+        return False
+    if not isinstance(capabilities, list):
+        return True
+    return bool(
+        GOVERNED_EXECUTION_CAPABILITIES.intersection(
+            {str(item) for item in capabilities}
+        )
+    )
+
+
 def reject_non_read_http(handler, method: str, path: str) -> bool:
     if method == 'GET':
         return False
@@ -407,14 +443,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','')
+        try:
             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):
+        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 '*' not in a['scopes'] and scope not in a['scopes']: self.sendj({'error':'forbidden','required_scope':scope,'account':a['account']},403); return None
+        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 +581,183 @@ class H(BaseHTTPRequestHandler):
             return self.sendj(body, exc.code)
         except (urllib.error.URLError, TimeoutError) as exc:
             return self.sendj({'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__}, 503)
-    def do_POST(self):
-        path = urllib.parse.urlparse(self.path).path
-        if path in {
-            '/createTagApprovalRequest', '/requestOrCreateTag', '/getTagPublicationStatus',
-            '/buildProductionArtifact', '/executeApprovedProductionRelease',
-            '/getReleaseConsistency', '/runReleaseAcceptance',
-        }:
-            actor = self.need('tag:create')
-            if not actor:
-                return
-            return self.bridge_release(path, self.body())
-        if reject_non_read_http(self, 'POST', path):
-            return
-        return self.sendj({'error': 'not_found', 'path': path}, 404)
+    def bridge_v2_json(self, route, data=None, method='POST'):
+        base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
+        payload = None if data is None else json.dumps(
+            data, sort_keys=True, separators=(',', ':')
+        ).encode('utf-8')
+        request = urllib.request.Request(
+            base + route,
+            data=payload,
+            headers={'Content-Type': 'application/json'} if payload is not None else {},
+            method=method,
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                result = json.loads(response.read().decode('utf-8'))
+                return self.sendj(result, response.status)
+        except urllib.error.HTTPError as exc:
+            try:
+                body = json.loads(exc.read().decode('utf-8'))
+            except Exception:
+                body = {'error': 'bridge_http_error'}
+            return self.sendj(body, exc.code)
+        except (urllib.error.URLError, TimeoutError) as exc:
+            return self.sendj(
+                {'error': 'GOVERNED_BRIDGE_UNAVAILABLE', 'detail': type(exc).__name__},
+                503,
+            )
+
+    def control_plane_snapshot(self):
+        """Build a non-sensitive snapshot locally when Bridge v2 has no snapshot HTTP route."""
+        base = os.getenv('AIOS_BRIDGE_V2_URL', 'http://127.0.0.1:18892').rstrip('/')
+
+        def bridge_get(route):
+            request = urllib.request.Request(
+                base + route,
+                headers={'Accept': 'application/json'},
+                method='GET',
+            )
+            with urllib.request.urlopen(request, timeout=15) as response:
+                if response.status != 200:
+                    raise RuntimeError('bridge_read_http_' + str(response.status))
+                value = json.loads(response.read().decode('utf-8'))
+                if not isinstance(value, dict):
+                    raise RuntimeError('bridge_read_not_object')
+                return value
+
+        try:
+            runtime = bridge_get('/v2/runtime')
+            capabilities = bridge_get('/v2/capabilities')
+            repository = repository_snapshot(
+                git_dir=AUTHORITATIVE_GIT_DIR,
+                repository_identity=REPOSITORY_IDENTITY,
+            )
+            if not repository.get('repository_available'):
+                return self.sendj({
+                    'error': 'AUTHORITATIVE_REPOSITORY_UNAVAILABLE',
+                    'query_error': repository.get('query_error'),
+                }, 503)
+            baseline = dict(BASELINE_AUTHORITY.read())
+        except BaselineAuthorityError as exc:
+            return self.sendj({
+                'error': 'RECONCILIATION_REQUIRED',
+                'detail': str(exc),
+            }, 503)
+        except urllib.error.HTTPError as exc:
+            return self.sendj({
+                'error': 'GOVERNED_BRIDGE_UNAVAILABLE',
+                'detail': 'HTTP_' + str(exc.code),
+            }, 503)
+        except (urllib.error.URLError, TimeoutError, OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
+            return self.sendj({
+                'error': 'GOVERNED_BRIDGE_UNAVAILABLE',
+                'detail': type(exc).__name__,
+            }, 503)
+
+        legacy = capabilities.get('legacy_writes_enabled')
+        if legacy is None:
+            legacy = capabilities.get('legacy_global_writes_enabled')
+        payload = {
+            'schema': 'aios.control-plane-snapshot.aggregate.v1',
+            'snapshot_source': 'model-api-local-aggregate',
+            'version': {
+                'target_release': runtime.get('control_plane_release_version'),
+            },
+            'runtime': {
+                'bridge_schema_version': runtime.get('bridge_schema_version'),
+                'control_plane_release_version': runtime.get('control_plane_release_version'),
+                'runtime_release_commit': runtime.get('runtime_release_commit'),
+                'runtime_path': runtime.get('runtime_path'),
+                'artifact_sha256': runtime.get('artifact_sha256'),
+                'drift_status': runtime.get('drift_status'),
+                'deployment_state': runtime.get('deployment_state'),
+            },
+            'repository': {
+                'identity': runtime.get('repository_identity') or runtime.get('repository') or REPOSITORY_IDENTITY,
+                'authority_git_dir': runtime.get('authoritative_git_dir') or repository.get('authoritative_git_dir'),
+                'authority_main_oid': repository.get('authoritative_head_oid'),
+            },
+            'production': {
+                'baseline_version': baseline.get('current_baseline'),
+                'target_oid': baseline.get('target_oid'),
+                'deployment_state': baseline.get('status'),
+                'baseline_generation': baseline.get('generation'),
+                'baseline_sha256': baseline.get('authority_sha256'),
+            },
+            'task': {},
+            'capabilities': {
+                'legacy_writes_enabled': legacy,
+                'governed_writes_enabled': capabilities.get('governed_writes_enabled'),
+                'force_push': capabilities.get('force_push'),
+                'main_write': capabilities.get('main_write'),
+                'tag_write': capabilities.get('tag_write'),
+                'baseline_write': capabilities.get('baseline_write'),
+                'production_write': capabilities.get('production_write'),
+            },
+            'credential_broker': {
+                'status': 'SERVER_SIDE_ONLY',
+                'provider': 'BridgeV2RuntimeCredentialProjection',
+            },
+            'secret_material_exposed': False,
+        }
+        return self.sendj(payload)
+
+    def agent_task_control(self, method, route, data=None):
+        token_path = Path(os.getenv(
+            'AIOS_AGENT_TASK_CONTROL_TOKEN_FILE',
+            '/run/credentials/aios-gitops-model-api-v2r2c1-staging.service/agent_task_control_token',
+        ))
+        if token_path.is_symlink() or not token_path.is_file():
+            return self.sendj({'error': 'AGENT_TASK_CONTROL_CREDENTIAL_UNAVAILABLE'}, 503)
+        token = token_path.read_text(encoding='utf-8').strip()
+        if len(token) < 32:
+            return self.sendj({'error': 'AGENT_TASK_CONTROL_CREDENTIAL_INVALID'}, 503)
+        base = os.getenv('AIOS_AGENT_TASK_API_URL', 'http://127.0.0.1:18940').rstrip('/')
+        payload = None if data is None else json.dumps(
+            data, sort_keys=True, separators=(',', ':')
+        ).encode('utf-8')
+        headers = {'Authorization': 'Bearer ' + token, 'Accept': 'application/json'}
+        if payload is not None:
+            headers['Content-Type'] = 'application/json'
+        request = urllib.request.Request(
+            base + route, data=payload, headers=headers, method=method
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                result = json.loads(response.read().decode('utf-8'))
+                return self.sendj(result, response.status)
+        except urllib.error.HTTPError as exc:
+            try:
+                body = json.loads(exc.read().decode('utf-8'))
+            except Exception:
+                body = {'error': 'agent_task_http_error'}
+            return self.sendj(body, exc.code)
+        except (urllib.error.URLError, TimeoutError) as exc:
+            return self.sendj(
+                {'error': 'AGENT_TASK_API_UNAVAILABLE', 'detail': type(exc).__name__},
+                503,
+            )
+
     def do_GET(self):
         u=urllib.parse.urlparse(self.path); path=u.path.rstrip('/') if u.path != '/' else u.path
         if path not in READ_ONLY_GET_PATHS: return self.sendj({'error':'not_found','path':path},404)
         q=urllib.parse.parse_qs(u.query)
+        if path == '/api/governed-write-capabilities':
+            return self.bridge_v2_json('/v2/capabilities', method='GET')
+        if path == '/api/control-plane-snapshot':
+            return self.control_plane_snapshot()
+        if path == '/api/caller-capabilities':
+            return self.caller_capabilities()
+        if path in ('/api/online-agent-task', '/api/online-agent-task-events'):
+            actor = self.need('agent-task:control')
+            if not actor:
+                return
+            task_id = (q.get('task_id') or [''])[0]
+            if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{2,179}', task_id):
+                return self.sendj({'error': 'task_id_invalid'}, 400)
+            suffix = '/events' if path.endswith('-events') else ''
+            return self.agent_task_control('GET', '/v1/tasks/' + task_id + suffix)
         try:
             if path in ('/health','/api/health'):
                 return self.sendj({
@@ -700,12 +994,68 @@ 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':
+                if requires_governed_execution(data):
+                    return self.sendj(
+                        {
+                            'error': 'governed_execution_binding_required',
+                            'hint': 'use createGovernedTask',
+                        },
+                        409,
+                    )
+                return self.agent_task_control('POST', '/v1/tasks', data)
+            if path == '/registerOnlineAgent':
+                return self.agent_task_control('POST', '/v1/agents/register', data)
+            task_id = str(data.pop('task_id', '') or '')
+            if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:-]{2,179}', task_id):
+                return self.sendj({'error': 'task_id_invalid'}, 400)
+            suffix = '/verify' if path == '/verifyOnlineAgentTask' else '/cancel'
+            return self.agent_task_control('POST', '/v1/tasks/' + task_id + suffix, data)
         if reject_non_read_http(self, 'POST', path): return
         data=self.body()
         try:
             if path == '/createReleaseCandidate':
-                a=self.need('release:candidate')
-                if not a: return
+                a = self.need(
+                    'release:candidate',
+                    projection_request={
+                        'source_commit': str(data.get('source_commit') or ''),
+                        'version': str(data.get('version') or ''),
+                    },
+                )
+                if not a:
+                    return
                 return self.sendj(
                     silent_release_service().create_release_candidate(data)
                 )
@@ -996,7 +1346,7 @@ def api_spec():
     return {
         'version': 'V2R2C1-IMMUTABLE-STAGING',
         'openapi': '/api/openapi.yaml',
-        'read': ['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],
+        'read': ['GET /api/caller-capabilities','GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /getReleaseCandidate','GET /getApprovalEnvelope','GET /getReleaseSaga','GET /admin/ai-evolution'],
         'write': ['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl','POST /createReleaseCandidate','POST /submitOwnerApproval','POST /startDeployment'],
         'policy': {
             'bearer_required_for_write': True,
@@ -1011,6 +1361,9 @@ def api_spec():
             'silent_release_shadow_default': True,
             'runtime_mode': 'IMMUTABLE_STAGING',
             'governed_release_routing': 'bridge-v2',
+            'caller_admission_projection': 'durable-owner-policy-server-side-ack',
+            'caller_projection_scope': SCOPE,
+            'caller_projection_caller': CALLER,
         },
     }
 

active-runtime/app/aios_online_agent_execution.py

@@ -179,10 +179,7 @@ def _agent_runtime_input(value: object) -> dict[str, Any]:
     if not isinstance(value, Mapping):
         _error("AGENT_RUNTIME_INPUT_REQUIRED")
     raw = dict(value)
-    allowed = {
-        "goal", "instructions", "success_criteria", "constraints",
-        "scope", "trace_id",
-    }
+    allowed = {"goal", "instructions", "success_criteria", "constraints"}
     if set(raw) - allowed:
         _error("AGENT_RUNTIME_INPUT_FIELDS_INVALID")
     goal = _string(raw.get("goal"), "AGENT_RUNTIME_GOAL_REQUIRED")
@@ -211,34 +208,11 @@ def _agent_runtime_input(value: object) -> dict[str, Any]:
         lowered = str(key).lower()
         if any(marker in lowered for marker in sensitive_markers):
             _error("AGENT_RUNTIME_SECRET_FIELD_DENY")
-
-    scope = raw.get("scope", {})
-    if not isinstance(scope, Mapping):
-        _error("AGENT_RUNTIME_SCOPE_INVALID")
-    scope = {
-        "scope_type": str(scope.get("scope_type") or "legacy"),
-        "tenant_id": scope.get("tenant_id"),
-        "project_id": scope.get("project_id"),
-    }
-    if scope["scope_type"] not in {"legacy", "system", "tenant"}:
-        _error("AGENT_RUNTIME_SCOPE_INVALID")
-    if scope["scope_type"] == "tenant":
-        if not isinstance(scope["tenant_id"], str) or not scope["tenant_id"].strip():
-            _error("AGENT_RUNTIME_SCOPE_INVALID")
-    elif scope["tenant_id"] not in (None, "") or scope["project_id"] not in (None, ""):
-        _error("AGENT_RUNTIME_SCOPE_INVALID")
-
-    trace_id = str(raw.get("trace_id") or "").strip()
-    if not trace_id or len(trace_id) > 255:
-        _error("AGENT_RUNTIME_TRACE_ID_INVALID")
-
     normalized = {
         "goal": goal[:12000],
         "instructions": instructions[:32000],
         "success_criteria": normalized_criteria,
         "constraints": dict(constraints),
-        "scope": scope,
-        "trace_id": trace_id,
     }
     if len(_canonical(normalized).encode("utf-8")) > MAX_AGENT_RUNTIME_INPUT_BYTES:
         _error("AGENT_RUNTIME_INPUT_TOO_LARGE")
@@ -451,12 +425,6 @@ def project_online_task(
         if adapter not in AGENT_RUNTIME_ADAPTERS:
             _error("AGENT_RUNTIME_ADAPTER_INVALID")
         runtime_input = execution_value.get("runtime_input")
-        tenant_id = request.get("tenant_id")
-        project_id = request.get("project_id")
-        declared_scope = str(
-            request.get("scope_type")
-            or ("tenant" if isinstance(tenant_id, str) and tenant_id.strip() else "legacy")
-        )
         if runtime_input is None:
             runtime_input = {
                 "goal": request.get("goal"),
@@ -464,20 +432,6 @@ def project_online_task(
                 "success_criteria": request.get("success_criteria") or [],
                 "constraints": request.get("constraints") or {},
             }
-        elif not isinstance(runtime_input, Mapping):
-            _error("AGENT_RUNTIME_INPUT_REQUIRED")
-        runtime_input = dict(runtime_input)
-        # Scope/trace are authoritative Task identity, never caller-selected runtime hints.
-        runtime_input["scope"] = {
-            "scope_type": declared_scope,
-            "tenant_id": tenant_id,
-            "project_id": project_id,
-        }
-        runtime_input["trace_id"] = str(
-            request.get("trace_id")
-            or request.get("run_id")
-            or binding.task_id
-        )
         execution_value["runtime_input"] = runtime_input
     execution = project_execution_envelope(
         binding,

active-runtime/app/aios_online_agent_worker.py

@@ -1910,6 +1910,8 @@ class WorkerSupervisor:
             # engineering placeholder: never pass that stale payload back to the
             # resolver, and never claim it until a deterministic plan exists.
             pending_plan = source_capture_requires_deterministic_plan(persisted)
+            if not pending_plan:
+                return persisted, "persisted_verified"
             projection_input = (
                 execution
                 if persisted.action in {"source_capture", "agent_runtime"} and not pending_plan
@@ -1958,25 +1960,7 @@ class WorkerSupervisor:
                     phase="projection",
                 )
             return persisted, "persisted_verified"
-        status, projected = self.bridge.project_execution(task)
-        if status == 404 or projected.get("error") == "TASK_EXECUTION_PROJECTION_MISSING":
         raise WorkerError("TASK_SKIPPED_PROJECTION_MISSING", phase="projection")
-        if not 200 <= status < 300:
-            raise WorkerError(
-                "TASK_SKIPPED_POLICY_DENY",
-                phase="projection",
-                details={"api_status": status, "error": projected.get("error")},
-            )
-        envelope = ExecutionEnvelope.from_task(task, projected, WORKER_CAPABILITIES)
-        audit(
-            "TASK_EXECUTION_ENVELOPE_PROJECTED",
-            task_id=envelope.task_id,
-            job_id=envelope.job_id,
-            generation=envelope.generation,
-            action=envelope.action,
-            execution_digest=envelope.digest,
-        )
-        return envelope, "recovered_projection"
 
     def process_claim(
         self,
@@ -2448,23 +2432,21 @@ class WorkerSupervisor:
             write_worker_liveness("POLL_OK", visible_count=len(tasks))
             audit("TASK_POLL", visible_count=len(tasks))
         except WorkerError as exc:
-            write_worker_liveness(
-                "POLL_FAILED",
-                error=exc.code,
-            )
+            write_worker_liveness("POLL_FAILED", error=exc.code)
             audit("TASK_POLL_FAILED", error=exc.code, phase=exc.phase, details=exc.details)
             return 10
-        dynamic_tasks: list[tuple[dict[str, Any], ExecutionEnvelope]] = []
+        dynamic_tasks: list[tuple[dict[str, Any], ExecutionEnvelope, str]] = []
         for task in tasks:
             try:
-                envelope, source = self.resolve_envelope(task)
+                metadata = task.get("metadata")
+                execution = metadata.get("execution") if isinstance(metadata, Mapping) else None
+                if not isinstance(execution, Mapping):
+                    raise WorkerError("TASK_SKIPPED_PROJECTION_MISSING", phase="projection")
+                envelope = ExecutionEnvelope.from_task(task, worker_capabilities=WORKER_CAPABILITIES)
+                source = "pending_plan" if source_capture_requires_deterministic_plan(envelope) else "persisted_verified"
             except WorkerError as exc:
                 event = "TASK_SKIPPED_ENVELOPE_INVALID"
-                if exc.code in {
-                    "TASK_SKIPPED_PROJECTION_MISSING",
-                    "TASK_SKIPPED_POLICY_DENY",
-                    "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN",
-                }:
+                if exc.code in {"TASK_SKIPPED_PROJECTION_MISSING", "TASK_SKIPPED_POLICY_DENY", "TASK_SKIPPED_PENDING_DETERMINISTIC_PLAN"}:
                     event = exc.code
                 elif exc.code == "CAPABILITY_MISMATCH_DENY":
                     event = "TASK_SKIPPED_CAPABILITY_MISMATCH"
@@ -2472,16 +2454,11 @@ class WorkerSupervisor:
                 continue
             audit("TASK_VISIBLE_TO_WORKER_CLAIMABLE", task_id=task.get("task_id"), envelope_source=source, action=envelope.action)
             if envelope.adapter not in WORKER_ALLOWED_ADAPTERS:
-                audit(
-                    "TASK_SKIPPED_ADAPTER_MISMATCH",
-                    task_id=task.get("task_id"),
-                    adapter=envelope.adapter,
-                    allowed_adapters=sorted(WORKER_ALLOWED_ADAPTERS),
-                )
+                audit("TASK_SKIPPED_ADAPTER_MISMATCH", task_id=task.get("task_id"), adapter=envelope.adapter, allowed_adapters=sorted(WORKER_ALLOWED_ADAPTERS))
                 continue
-            dynamic_tasks.append((task, envelope))
-        dynamic_tasks.sort(key=_claim_order)
-        for task, envelope in dynamic_tasks:
+            dynamic_tasks.append((task, envelope, source))
+        dynamic_tasks.sort(key=lambda item: _claim_order(item[0]))
+        for task, envelope, source in dynamic_tasks:
             task_id = str(task.get("task_id") or "")
             status, claim = self.agent.claim(task_id)
             if status == 409:
@@ -2490,6 +2467,19 @@ class WorkerSupervisor:
             if status != 200:
                 audit("TASK_CLAIM_FAILED", task_id=task_id, api_status=status, error=claim.get("error"))
                 continue
+            if source == "pending_plan":
+                current_status, current_task = self.agent.get(task_id)
+                if current_status == 200:
+                    try:
+                        envelope, _ = self.resolve_envelope(current_task)
+                    except WorkerError as exc:
+                        audit("TASK_PLAN_RESOLUTION_FAILED", task_id=task_id, reason=exc.code, phase=exc.phase, details=exc.details)
+                        self.process_claim(task, claim, envelope)
+                        return 0
+                else:
+                    audit("TASK_READ_AFTER_CLAIM_FAILED", task_id=task_id, api_status=current_status)
+                    self.process_claim(task, claim, envelope)
+                    return 0
             self.process_claim(task, claim, envelope)
             return 0
         return 0

active-runtime/app/aios_release_preparation_supervisor_coordinator.py

@@ -0,0 +1,287 @@
+"""Credential-isolated durable coordinator for the release-preparation supervisor.
+
+The coordinator owns only its Online Task agent credential.  It never forwards
+that credential to Hermes/OpenClaw and never submits a Task result while the
+supervision goal is incomplete.  Its job is deliberately small: claim the one
+fixed supervisor Task after an exact coordinator projection and keep a durable
+heartbeat lease while later phases observe and repair through governed APIs.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import tempfile
+import time
+from typing import Any, Mapping
+import urllib.error
+import urllib.request
+
+
+API_BASE = os.environ.get(
+    "AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940"
+).rstrip("/")
+AGENT_ID = os.environ.get(
+    "AIOS_AGENT_ID",
+    "online-engineering-worker-cn-network-mkt-autonomous-20260815",
+)
+TASK_ID = os.environ.get(
+    "AIOS_RELEASE_PREPARATION_SUPERVISOR_TASK_ID",
+    "r42-release-preparation-autonomous-supervisor-20260820",
+)
+EXPECTED_EXECUTION_DIGEST = os.environ.get(
+    "AIOS_RELEASE_PREPARATION_SUPERVISOR_EXECUTION_DIGEST", ""
+)
+RUNTIME_CAPABILITY = "agent.runtime.openclaw"
+COORDINATOR_CAPABILITY = "engineering.autonomous"
+REQUIRED_CAPABILITIES = [RUNTIME_CAPABILITY, COORDINATOR_CAPABILITY]
+POLL_SECONDS = int(os.environ.get("AIOS_SUPERVISOR_COORDINATOR_POLL_SECONDS", "5"))
+HEARTBEAT_SECONDS = int(
+    os.environ.get("AIOS_SUPERVISOR_COORDINATOR_HEARTBEAT_SECONDS", "10")
+)
+LEASE_SECONDS = int(os.environ.get("AIOS_SUPERVISOR_COORDINATOR_LEASE_SECONDS", "300"))
+STATE_ROOT = Path(
+    os.environ.get(
+        "AIOS_SUPERVISOR_COORDINATOR_STATE_ROOT",
+        "/var/lib/aios-release-preparation-supervisor-coordinator",
+    )
+)
+
+
+class CoordinatorError(RuntimeError):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode("utf-8")
+
+
+def atomic_json(path: Path, value: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, mode=0o750, exist_ok=True)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise CoordinatorError("COORDINATOR_STATE_SYMLINK_REJECTED")
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
+    try:
+        os.fchmod(fd, 0o640)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(dict(value), handle, ensure_ascii=False, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def audit(event_type: str, **fields: Any) -> None:
+    record = {
+        "event_type": event_type,
+        "created_at": now(),
+        "agent_id": AGENT_ID,
+        "task_id": TASK_ID,
+        "credential_material_exposed": False,
+        "planner_credential_material_exposed": False,
+        "lease_material_exposed": False,
+        **fields,
+    }
+    STATE_ROOT.mkdir(parents=True, mode=0o750, exist_ok=True)
+    with (STATE_ROOT / "events.jsonl").open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
+    atomic_json(STATE_ROOT / "liveness.json", record)
+
+
+def api_call(
+    method: str,
+    route: str,
+    token: str,
+    body: Mapping[str, Any] | None = None,
+) -> tuple[int, dict[str, Any]]:
+    payload = None if body is None else canonical(dict(body))
+    headers = {"Authorization": "Bearer " + token, "Accept": "application/json"}
+    if payload is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(
+        API_BASE + route, data=payload, headers=headers, method=method
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=20) as response:
+            value = json.loads(response.read().decode("utf-8"))
+            return response.status, value if isinstance(value, dict) else {}
+    except urllib.error.HTTPError as exc:
+        try:
+            value = json.loads(exc.read(4096).decode("utf-8", "replace"))
+        except Exception:
+            value = {"error": "HTTP_ERROR"}
+        return exc.code, value if isinstance(value, dict) else {"error": "HTTP_ERROR"}
+    except (urllib.error.URLError, TimeoutError):
+        return 599, {"error": "AGENT_TASK_API_UNAVAILABLE"}
+
+
+def load_agent_token() -> str:
+    path = Path(os.environ.get("CREDENTIALS_DIRECTORY", "")) / "agent_token"
+    if path.is_symlink() or not path.is_file():
+        raise CoordinatorError("COORDINATOR_AGENT_CREDENTIAL_UNAVAILABLE")
+    token = path.read_text(encoding="utf-8").strip()
+    if len(token) < 24 or any(character.isspace() for character in token):
+        raise CoordinatorError("COORDINATOR_AGENT_CREDENTIAL_INVALID")
+    return token
+
+
+def validate_task(task: Mapping[str, Any]) -> dict[str, Any]:
+    if task.get("task_id") != TASK_ID or task.get("status") != "QUEUED":
+        raise CoordinatorError("SUPERVISOR_TASK_STATE_INVALID")
+    if task.get("required_capabilities") != REQUIRED_CAPABILITIES:
+        raise CoordinatorError("SUPERVISOR_COORDINATOR_CAPABILITIES_INVALID")
+    metadata = task.get("metadata")
+    if not isinstance(metadata, Mapping):
+        raise CoordinatorError("SUPERVISOR_EXECUTION_METADATA_MISSING")
+    execution = metadata.get("execution")
+    if not isinstance(execution, Mapping):
+        raise CoordinatorError("SUPERVISOR_EXECUTION_METADATA_MISSING")
+    governance = execution.get("governance")
+    if (
+        execution.get("action") != "agent_runtime"
+        or execution.get("adapter") != "openclaw"
+        or execution.get("capability_scope") != REQUIRED_CAPABILITIES
+        or execution.get("execution_digest") != EXPECTED_EXECUTION_DIGEST
+        or execution.get("change_set") != []
+        or execution.get("verification") != []
+        or not isinstance(governance, Mapping)
+        or governance.get("writer_lease") is not False
+        or any(
+            governance.get(field) is not False
+            for field in (
+                "main_write",
+                "production_write",
+                "tag_write",
+                "baseline_write",
+                "force_push",
+                "protected_ref_bypass",
+            )
+        )
+    ):
+        raise CoordinatorError("SUPERVISOR_COORDINATOR_ENVELOPE_INVALID")
+    return dict(execution)
+
+
+def heartbeat_progress(sequence: int) -> dict[str, Any]:
+    return {
+        "stage": "supervisor_coordinator_active",
+        "phase": "external_caller_projection",
+        "state": "PENDING_EXTERNAL_CONNECTOR_ACK",
+        "sequence": sequence,
+        "control_plane_access_granted": True,
+        "external_planner_role": "UNPRIVILEGED_PLANNER",
+        "credential_forwarded_to_planner": False,
+        "main_write": False,
+        "production_write": False,
+        "owner_approval_submitted": False,
+    }
+
+
+def coordinate_claim(
+    claim: Mapping[str, Any],
+    token: str,
+    *,
+    max_heartbeats: int | None = None,
+) -> int:
+    lease_token = claim.get("lease_token")
+    if not isinstance(lease_token, str) or len(lease_token) < 24:
+        raise CoordinatorError("COORDINATOR_LEASE_TOKEN_MISSING")
+    sequence = 0
+    while max_heartbeats is None or sequence < max_heartbeats:
+        sequence += 1
+        status, response = api_call(
+            "POST",
+            f"/v1/tasks/{TASK_ID}/heartbeat",
+            token,
+            {
+                "lease_token": lease_token,
+                "lease_seconds": LEASE_SECONDS,
+                "progress": heartbeat_progress(sequence),
+            },
+        )
+        if status != 200:
+            raise CoordinatorError(
+                "COORDINATOR_HEARTBEAT_FAILED_" + str(response.get("error") or status)
+            )
+        audit(
+            "SUPERVISOR_COORDINATOR_HEARTBEAT",
+            sequence=sequence,
+            task_status=response.get("status"),
+            lease_expires_at=response.get("lease_expires_at"),
+            control_plane_access_granted=True,
+            external_planner_role="UNPRIVILEGED_PLANNER",
+        )
+        if max_heartbeats is None or sequence < max_heartbeats:
+            time.sleep(HEARTBEAT_SECONDS)
+    return sequence
+
+
+def worker_once(token: str) -> bool:
+    status, poll = api_call("POST", "/v1/tasks/poll", token, {"limit": 100})
+    if status != 200:
+        audit("SUPERVISOR_COORDINATOR_POLL_FAILED", error=poll.get("error") or status)
+        return False
+    task = next(
+        (
+            item
+            for item in poll.get("tasks", [])
+            if isinstance(item, Mapping) and item.get("task_id") == TASK_ID
+        ),
+        None,
+    )
+    if task is None:
+        audit("SUPERVISOR_COORDINATOR_WAITING_FOR_TASK")
+        return False
+    validate_task(task)
+    status, claim = api_call(
+        "POST", f"/v1/tasks/{TASK_ID}/claim", token, {"lease_seconds": LEASE_SECONDS}
+    )
+    if status != 200:
+        audit("SUPERVISOR_COORDINATOR_CLAIM_FAILED", error=claim.get("error") or status)
+        return False
+    audit(
+        "SUPERVISOR_COORDINATOR_CLAIMED",
+        attempt=claim.get("attempt"),
+        revision=claim.get("revision"),
+        claim_mode="server_atomic",
+        control_plane_access_granted=True,
+    )
+    coordinate_claim(claim, token)
+    return True
+
+
+def main() -> int:
+    if not EXPECTED_EXECUTION_DIGEST or len(EXPECTED_EXECUTION_DIGEST) != 64:
+        raise CoordinatorError("SUPERVISOR_EXECUTION_DIGEST_CONFIG_INVALID")
+    token = load_agent_token()
+    audit(
+        "SUPERVISOR_COORDINATOR_READY",
+        required_capabilities=REQUIRED_CAPABILITIES,
+        control_plane_access_granted=True,
+        external_planner_role="UNPRIVILEGED_PLANNER",
+    )
+    while True:
+        try:
+            worker_once(token)
+        except Exception as exc:
+            audit(
+                "SUPERVISOR_COORDINATOR_LOOP_ERROR",
+                error=exc.code if isinstance(exc, CoordinatorError) else type(exc).__name__,
+            )
+        time.sleep(POLL_SECONDS)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

deploy/systemd/aios-agent-task-supervisor-coordinator-projection.service

@@ -0,0 +1,37 @@
+[Unit]
+Description=AIOS owner-controlled supervisor coordinator projection API
+After=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
+Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
+
+[Service]
+Type=simple
+User=aios-agent-task
+Group=aios-agent-task
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID=owner-supervisor-coordinator-projection-recovery-20260820
+Environment=AIOS_SUPERVISOR_COORDINATOR_PROJECTION_HOST=127.0.0.1
+Environment=AIOS_SUPERVISOR_COORDINATOR_PROJECTION_PORT=18944
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_supervisor_coordinator_projection.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+ReadWritePaths=/var/lib/aios-agent-task
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-release-preparation-supervisor-coordinator.service

@@ -0,0 +1,45 @@
+[Unit]
+Description=AIOS credential-isolated release-preparation supervisor coordinator
+After=network-online.target aios-agent-task-api.service aios-autonomous-agent-register.service
+Wants=network-online.target aios-agent-task-api.service
+Requires=aios-autonomous-agent-register.service
+
+[Service]
+Type=simple
+User=aios-autonomous-agent-worker
+Group=aios-autonomous-agent-worker
+SupplementaryGroups=aios-agent-runtime
+LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-autonomous-agent-worker/agent_token
+Environment=PYTHONUNBUFFERED=1
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_AGENT_ID=online-engineering-worker-cn-network-mkt-autonomous-20260815
+Environment=AIOS_RELEASE_PREPARATION_SUPERVISOR_TASK_ID=r42-release-preparation-autonomous-supervisor-20260820
+Environment=AIOS_SUPERVISOR_COORDINATOR_STATE_ROOT=/var/lib/aios-release-preparation-supervisor-coordinator
+Environment=AIOS_SUPERVISOR_COORDINATOR_POLL_SECONDS=5
+Environment=AIOS_SUPERVISOR_COORDINATOR_HEARTBEAT_SECONDS=10
+Environment=AIOS_SUPERVISOR_COORDINATOR_LEASE_SECONDS=300
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_release_preparation_supervisor_coordinator.py
+Restart=always
+RestartSec=5
+TimeoutStopSec=20
+UMask=0027
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_INET AF_UNIX
+ReadWritePaths=/var/lib/aios-release-preparation-supervisor-coordinator
+InaccessiblePaths=-/etc/credstore.encrypted
+CapabilityBoundingSet=
+AmbientCapabilities=
+
+[Install]
+WantedBy=multi-user.target

tests/placeholder.txt

@@ -0,0 +1 @@
+placeholder

tests/test_agent_runtime_capability_contract.py

@@ -10,9 +10,13 @@ ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / "active-runtime/app"))
 
 from aios_capability_contract import (  # noqa: E402
+    capability_contract,
     has_capabilities,
     normalize_required_capabilities,
     normalize_worker_capabilities,
+    normalized_capability_matrix,
+    registered_capability,
+    validate_capability_contract_registry,
 )
 
 
@@ -31,3 +35,32 @@ def test_runtime_capabilities_are_isolated_between_workers() -> None:
 
 def test_unknown_runtime_capability_remains_fail_closed() -> None:
     assert has_capabilities(["agent.runtime.hermes"], ["agent.runtime.unknown"]) is False
+
+
+def test_autonomous_coordinator_capability_is_registered_on_both_sides() -> None:
+    capability = "engineering.autonomous"
+    assert registered_capability(capability) is True
+    assert registered_capability(capability, worker=True) is True
+    assert normalize_required_capabilities([capability]) == frozenset({capability})
+    assert normalize_worker_capabilities([capability]) == frozenset({capability})
+    contract = capability_contract(capability)
+    assert contract is not None
+    assert contract.dependencies == ("agent_task_control_plane",)
+    assert validate_capability_contract_registry()["valid"] is True
+
+
+def test_supervisor_coordinator_requires_runtime_and_autonomous_capabilities() -> None:
+    required = ["agent.runtime.openclaw", "engineering.autonomous"]
+    assert normalized_capability_matrix(required, required) == {
+        "worker_effective": required,
+        "task_required": required,
+        "eligible": True,
+    }
+    assert has_capabilities(["agent.runtime.openclaw"], required) is False
+    assert has_capabilities(["engineering.autonomous"], required) is False
+
+
+def test_autonomous_capability_near_matches_remain_fail_closed() -> None:
+    worker = ["agent.runtime.openclaw", "engineering.autonomous"]
+    assert has_capabilities(worker, ["engineering.autonomous.extra"]) is False
+    assert has_capabilities(worker, ["engineering.*"]) is False

tests/test_agent_task_initial_execution_projection.py

@@ -15,6 +15,7 @@ from aios_agent_task_initial_execution_projection import (
     InitialExecutionProjectionError,
     InitialExecutionProjectionRequest,
     InitialExecutionProjectionStore,
+    SourceCaptureProjectionRecoveryRequest,
 )
 from aios_online_agent_execution import (
     GovernedExecutionBinding,
@@ -166,6 +167,87 @@ def resolver(_task_id,_job_id,_capability,_adapter,received_runtime_input):
     return envelope()
 
 
+def source_capture_result():
+    old=envelope(action="source_capture",capability="git.read")
+    return {
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "generation":1,
+        "status":"SUCCEEDED",
+        "classification":"NEEDS_REWORK_SOURCE_CAPTURE",
+        "execution_mode":"source_capture",
+        "execution_digest":old["execution_digest"],
+        "changed_files":[],
+        "writer_lease":"NONE",
+        "verification_state":"NEEDS_REWORK",
+        "direct_main_write":False,
+        "production_write":False,
+        "force_push":False,
+        "protected_ref_bypass":False,
+        "secret_material_exposed":False,
+        "source_snapshot":{"schema":"test.snapshot.v1","payload_sha256":"d"*64},
+    }
+
+
+def make_source_capture_rework_db(tmp_path, *, mutate=None, events=None):
+    path=make_db(tmp_path)
+    result=source_capture_result()
+    verification={
+        "decision":"NEEDS_REWORK",
+        "reason":"SOURCE_CAPTURE_COMPLETE_RUNTIME_PROJECTION_REQUIRED",
+    }
+    if mutate:
+        mutate(result,verification)
+    event_types=events or [
+        "TASK_CLAIMED","TASK_HEARTBEAT","RESULT_SUBMITTED","TASK_VERIFIED"
+    ]
+    con=sqlite3.connect(path)
+    con.execute(
+        "UPDATE tasks SET attempt=1,status='QUEUED',claimed_by=NULL,revision=5,"
+        "result_json=?,result_digest=?,result_idempotency_key=?,verification_json=?",
+        (
+            json.dumps(result,sort_keys=True,separators=(",",":")),
+            canon(result),"source-capture-result-v1",
+            json.dumps(verification,sort_keys=True,separators=(",",":")),
+        ),
+    )
+    for event_type in event_types:
+        payload={}
+        if event_type=="TASK_VERIFIED":
+            payload={"decision":"NEEDS_REWORK","next_state":"QUEUED"}
+        con.execute(
+            "INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) "
+            "VALUES (?,?,?,?,?)",
+            (TASK_ID,"generic-worker",event_type,json.dumps(payload),"2026-08-20T00:01:00Z"),
+        )
+    con.commit();con.close()
+    return path,result,verification
+
+
+def recovery_request(result,verification,*,digest=None,key="source-capture-recovery-a-v1"):
+    return SourceCaptureProjectionRecoveryRequest.from_mapping({
+        "expected_status":"QUEUED",
+        "expected_revision":5,
+        "expected_attempt":1,
+        "expected_metadata_sha256":canon({}),
+        "expected_required_capabilities_sha256":canon([]),
+        "previous_verification_sha256":canon(verification),
+        "expected_source_capture_result_sha256":canon(result),
+        "expected_source_capture_execution_digest":result["execution_digest"],
+        "expected_job_id":JOB_ID,
+        "expected_execution_digest":digest or envelope()["execution_digest"],
+        "required_capability":CAPABILITY,
+        "adapter":"openclaw",
+        "authorization_id":AUTHORIZATION,
+        "idempotency_key":key,
+        "reason":"first durable runtime projection after governed source capture",
+    })
+
+
 def test_initial_projection_cas_persists_envelope_capability_and_event(tmp_path):
     path=make_db(tmp_path)
     result=InitialExecutionProjectionStore(path,resolver).project(TASK_ID,request())
@@ -205,6 +287,110 @@ def test_initial_projection_idempotent_exact_replay(tmp_path):
     assert con.execute("SELECT COUNT(*) FROM events").fetchone()[0]==2
 
 
+def test_source_capture_recovery_projects_first_runtime_envelope(tmp_path):
+    path,result,verification=make_source_capture_rework_db(tmp_path)
+    response=InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+        TASK_ID,recovery_request(result,verification)
+    )
+    assert response["revision"]==6
+    assert response["attempt"]==1
+    assert response["recovery_from_source_capture"] is True
+    assert response["idempotent"] is False
+    con=sqlite3.connect(path)
+    row=con.execute(
+        "SELECT status,attempt,revision,claimed_by,metadata_json,"
+        "required_capabilities_json,result_json,verification_json FROM tasks"
+    ).fetchone()
+    assert row[:4]==("QUEUED",1,6,None)
+    metadata=json.loads(row[4])
+    assert metadata["execution"]["action"]=="agent_runtime"
+    assert metadata["execution"]["adapter"]=="openclaw"
+    assert metadata["execution"]["governance"]["writer_lease"] is False
+    assert json.loads(row[5])==[CAPABILITY]
+    assert json.loads(row[6])==result
+    assert json.loads(row[7])==verification
+    events=con.execute("SELECT event_type,payload_json FROM events ORDER BY event_id").fetchall()
+    assert [event[0] for event in events]==[
+        "TASK_CREATED","TASK_CLAIMED","TASK_HEARTBEAT","RESULT_SUBMITTED",
+        "TASK_VERIFIED","TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED",
+    ]
+    payload=json.loads(events[-1][1])
+    assert payload["recovery_from_source_capture"] is True
+    assert payload["source_capture_execution_digest"]==result["execution_digest"]
+    assert payload["main_write"] is False
+    assert payload["production_write"] is False
+
+
+def test_source_capture_recovery_idempotent_exact_replay(tmp_path):
+    path,result,verification=make_source_capture_rework_db(tmp_path)
+    store=InitialExecutionProjectionStore(path,resolver)
+    first=store.project_from_source_capture(TASK_ID,recovery_request(result,verification))
+    second=store.project_from_source_capture(TASK_ID,recovery_request(result,verification))
+    assert first["idempotent"] is False
+    assert second["idempotent"] is True
+    assert second["execution_digest"]==first["execution_digest"]
+    con=sqlite3.connect(path)
+    assert con.execute(
+        "SELECT COUNT(*) FROM task_source_capture_projection_recoveries"
+    ).fetchone()[0]==1
+    assert con.execute("SELECT COUNT(*) FROM events").fetchone()[0]==6
+
+
+@pytest.mark.parametrize(
+    "mutate,error",
+    [
+        (lambda result,verification: result.__setitem__("changed_files",["x"]),
+         "SOURCE_CAPTURE_RESULT_CONTRACT_INVALID"),
+        (lambda result,verification: result.__setitem__("job_id","job-other"),
+         "SOURCE_CAPTURE_RESULT_CONTRACT_INVALID"),
+        (lambda result,verification: verification.__setitem__("decision","SUCCEEDED"),
+         "SOURCE_CAPTURE_NEEDS_REWORK_VERIFICATION_REQUIRED"),
+    ],
+)
+def test_source_capture_recovery_rejects_unsafe_result_or_verification(
+    tmp_path,mutate,error
+):
+    path,result,verification=make_source_capture_rework_db(tmp_path,mutate=mutate)
+    with pytest.raises(InitialExecutionProjectionError,match=error):
+        InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+            TASK_ID,recovery_request(result,verification)
+        )
+
+
+@pytest.mark.parametrize(
+    "events",
+    [
+        ["TASK_CLAIMED","RESULT_SUBMITTED","TASK_VERIFIED"],
+        ["TASK_CLAIMED","TASK_HEARTBEAT","RESULT_SUBMITTED","TASK_VERIFIED","TASK_CLAIMED"],
+        ["TASK_CLAIMED","TASK_HEARTBEAT","TASK_VERIFIED","RESULT_SUBMITTED"],
+    ],
+)
+def test_source_capture_recovery_rejects_invalid_event_chain(tmp_path,events):
+    path,result,verification=make_source_capture_rework_db(tmp_path,events=events)
+    with pytest.raises(
+        InitialExecutionProjectionError,match="SOURCE_CAPTURE_RECOVERY_EVENT_CHAIN_INVALID"
+    ):
+        InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+            TASK_ID,recovery_request(result,verification)
+        )
+
+
+def test_source_capture_recovery_rejects_execution_digest_drift(tmp_path):
+    path,result,verification=make_source_capture_rework_db(tmp_path)
+    with pytest.raises(InitialExecutionProjectionError,match="EXECUTION_DIGEST_DRIFT"):
+        InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+            TASK_ID,recovery_request(result,verification,digest="f"*64)
+        )
+
+
+def test_source_capture_recovery_request_is_strict(tmp_path):
+    _path,result,verification=make_source_capture_rework_db(tmp_path)
+    data=recovery_request(result,verification).public()
+    data["metadata_patch"]={"execution":"forbidden"}
+    with pytest.raises(InitialExecutionProjectionError,match="FIELDS_INVALID"):
+        SourceCaptureProjectionRecoveryRequest.from_mapping(data)
+
+
 @pytest.mark.parametrize(
     "kwargs,error",
     [

tests/test_agent_task_supervisor_coordinator_projection.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+from pathlib import Path
+import hashlib
+import json
+import sqlite3
+import sys
+
+import pytest
+
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_agent_task_initial_execution_projection import InitialExecutionProjectionError
+from aios_agent_task_supervisor_coordinator_projection import (
+    COORDINATOR_CAPABILITIES,
+    RUNTIME_CAPABILITY,
+    SupervisorCoordinatorProjectionRequest,
+    SupervisorCoordinatorProjectionStore,
+)
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+
+TASK="task-supervisor-coordinator-a"
+JOB="job-supervisor-coordinator-a"
+BASE="a"*40
+BRANCH="agent/task-supervisor-coordinator-a"
+BINDING_KEY="task-supervisor-coordinator-binding-v1"
+AUTH="owner-supervisor-coordinator-projection-recovery-20260820"
+
+
+@pytest.fixture(autouse=True)
+def authorization(monkeypatch):
+    monkeypatch.setenv(
+        "AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID",AUTH
+    )
+
+
+def sha(value):
+    return hashlib.sha256(
+        json.dumps(value,sort_keys=True,separators=(",",":")).encode()
+    ).hexdigest()
+
+
+def binding():
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":TASK,"job_id":JOB,"repository":"aios/aios",
+        "base_oid":BASE,"expected_branch_oid":BASE,"target_branch":BRANCH,
+        "branch_mode":"CREATE_NEW","generation":1,"idempotency_key":BINDING_KEY,
+    })
+
+
+def runtime_input():
+    return {
+        "goal":"Durably supervise release preparation",
+        "instructions":"Wait for exact connector ACK and preserve the same lineage.",
+        "success_criteria":[{"id":"heartbeat","description":"Coordinator heartbeat"}],
+        "constraints":{},
+        "scope":{"scope_type":"legacy","tenant_id":None,"project_id":None},
+        "trace_id":TASK,
+    }
+
+
+def envelope(capabilities):
+    return project_execution_envelope(binding(),{
+        "action":"agent_runtime","adapter":"openclaw",
+        "capability_scope":capabilities,"runtime_input":runtime_input(),
+    })
+
+
+def planner_result(old_digest,*,access=False):
+    return {
+        "task_id":TASK,"job_id":JOB,"repository":"aios/aios",
+        "base_oid":BASE,"expected_branch_oid":BASE,"target_branch":BRANCH,
+        "generation":1,"status":"SUCCEEDED","execution_mode":"agent_runtime",
+        "agent_runtime":"openclaw","execution_digest":old_digest,
+        "control_plane_access_granted_to_runtime":access,
+        "credential_material_exposed":False,"secret_material_exposed":False,
+        "changed_files":[],"writer_lease":"NONE","direct_main_write":False,
+        "production_write":False,"force_push":False,"protected_ref_bypass":False,
+        "classification":"EXTERNAL_AGENT_RUNTIME_SUCCEEDED",
+    }
+
+
+def make_db(tmp_path,*,access=False,heartbeat=True):
+    tmp_path.mkdir(parents=True,exist_ok=True)
+    path=tmp_path/"state.sqlite3";con=sqlite3.connect(path)
+    con.executescript("""
+    CREATE TABLE 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,
+      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,created_at TEXT,updated_at TEXT
+    );
+    CREATE TABLE 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
+    );
+    """)
+    old=envelope([RUNTIME_CAPABILITY]);result=planner_result(old["execution_digest"],access=access)
+    verification={"decision":"NEEDS_REWORK","reason":"HOST_COORDINATOR_REQUIRED"}
+    metadata={"execution":old,"execution_projection":{
+        "schema":"aios.online-agent.task-projection.v1","task_id":TASK,
+        "job_id":JOB,"execution_digest":old["execution_digest"],
+    }}
+    con.execute("INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(
+        TASK,runtime_input()["goal"],runtime_input()["instructions"],
+        json.dumps(runtime_input()["success_criteria"]),json.dumps([RUNTIME_CAPABILITY]),
+        "{}",json.dumps(metadata,sort_keys=True,separators=(",",":")),None,100,20,2,
+        "QUEUED",None,None,None,json.dumps(result,sort_keys=True,separators=(",",":")),
+        sha(result),"planner-result-v1",json.dumps(verification,sort_keys=True,separators=(",",":")),
+        11,"2026-08-20T00:00:00Z","2026-08-20T00:00:00Z",
+    ))
+    events=["TASK_CREATED","TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED","TASK_CLAIMED"]
+    if heartbeat: events.append("TASK_HEARTBEAT")
+    events += ["RESULT_SUBMITTED","TASK_VERIFIED"]
+    for event in events:
+        payload={"decision":"NEEDS_REWORK","next_state":"QUEUED"} if event=="TASK_VERIFIED" else {}
+        con.execute("INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) VALUES (?,?,?,?,?)",(TASK,"worker",event,json.dumps(payload),"2026-08-20T00:00:00Z"))
+    con.commit();con.close()
+    return path,metadata,result,verification,old
+
+
+def request(metadata,result,verification,old,new):
+    return SupervisorCoordinatorProjectionRequest.from_mapping({
+        "expected_status":"QUEUED","expected_revision":11,"expected_attempt":2,
+        "expected_metadata_sha256":sha(metadata),
+        "expected_required_capabilities_sha256":sha([RUNTIME_CAPABILITY]),
+        "previous_verification_sha256":sha(verification),
+        "expected_planner_result_sha256":sha(result),
+        "expected_old_execution_digest":old["execution_digest"],
+        "expected_new_execution_digest":new["execution_digest"],
+        "expected_job_id":JOB,"authorization_id":AUTH,
+        "idempotency_key":"task-supervisor-coordinator-projection-v1",
+        "reason":"credential-isolated host coordinator required",
+    })
+
+
+def test_projects_exact_coordinator_capabilities_and_preserves_planner_evidence(tmp_path):
+    path,metadata,result,verification,old=make_db(tmp_path)
+    new=envelope(COORDINATOR_CAPABILITIES)
+    store=SupervisorCoordinatorProjectionStore(path,lambda task,job,runtime:new)
+    response=store.project(TASK,request(metadata,result,verification,old,new))
+    assert response["revision"]==12 and response["attempt"]==2
+    assert response["required_capabilities"]==COORDINATOR_CAPABILITIES
+    assert response["credential_forwarded_to_planner"] is False
+    con=sqlite3.connect(path)
+    row=con.execute("select status,revision,metadata_json,required_capabilities_json,result_json from tasks").fetchone()
+    assert row[:2]==("QUEUED",12)
+    assert json.loads(row[2])["execution"]["execution_digest"]==new["execution_digest"]
+    assert json.loads(row[3])==COORDINATOR_CAPABILITIES
+    assert json.loads(row[4])==result
+    assert con.execute("select event_type from events order by event_id desc limit 1").fetchone()[0]=="TASK_SUPERVISOR_COORDINATOR_PROJECTION_AUTHORIZED"
+
+
+def test_exact_replay_is_idempotent(tmp_path):
+    path,metadata,result,verification,old=make_db(tmp_path);new=envelope(COORDINATOR_CAPABILITIES)
+    store=SupervisorCoordinatorProjectionStore(path,lambda task,job,runtime:new)
+    req=request(metadata,result,verification,old,new)
+    assert store.project(TASK,req)["idempotent"] is False
+    assert store.project(TASK,req)["idempotent"] is True
+
+
+def test_rejects_planner_with_control_access_or_missing_heartbeat(tmp_path):
+    path,metadata,result,verification,old=make_db(tmp_path,access=True);new=envelope(COORDINATOR_CAPABILITIES)
+    with pytest.raises(InitialExecutionProjectionError,match="RESULT_CONTRACT_INVALID"):
+        SupervisorCoordinatorProjectionStore(path,lambda *args:new).project(
+            TASK,request(metadata,result,verification,old,new)
+        )
+    path,metadata,result,verification,old=make_db(tmp_path/"second",heartbeat=False)
+    with pytest.raises(InitialExecutionProjectionError,match="EVENT_CHAIN_INVALID"):
+        SupervisorCoordinatorProjectionStore(path,lambda *args:new).project(
+            TASK,request(metadata,result,verification,old,new)
+        )
+
+
+def test_request_rejects_arbitrary_metadata_patch(tmp_path):
+    path,metadata,result,verification,old=make_db(tmp_path);new=envelope(COORDINATOR_CAPABILITIES)
+    value=request(metadata,result,verification,old,new).public();value["metadata_patch"]={}
+    with pytest.raises(InitialExecutionProjectionError,match="FIELDS_INVALID"):
+        SupervisorCoordinatorProjectionRequest.from_mapping(value)

tests/test_model_api_continuation_openapi_contract.py

@@ -0,0 +1,112 @@
+from pathlib import Path
+
+import pytest
+import yaml
+
+from aios_chatgpt_bridge_v2 import OPERATION_REQUIRED_FIELDS
+from aios_chatgpt_bridge_v2_server import BridgeHandler, _HTTPError, ROUTES
+
+
+ROOT = Path(__file__).resolve().parents[1]
+OPENAPI = ROOT / "docs" / "gitops" / "openapi-chatgpt-actions.yaml"
+REQUIRED = {
+    "task_id",
+    "repository",
+    "base_oid",
+    "target_branch",
+    "risk",
+    "description",
+    "idempotency_key",
+}
+OPTIONAL_CONTINUATION = {
+    "expected_branch_oid",
+    "generation",
+    "parent_task_id",
+    "parent_job_id",
+    "priority",
+}
+
+
+def _document():
+    return yaml.safe_load(OPENAPI.read_text(encoding="utf-8"))
+
+
+def _create_request():
+    return {
+        "task_id": "continuation-contract-test",
+        "repository": "aios/aios",
+        "base_oid": "a" * 40,
+        "target_branch": "agent/continuation-contract-test",
+        "risk": "engineering",
+        "description": "contract test",
+        "idempotency_key": "continuation-contract-test-v1",
+    }
+
+
+def test_create_governed_task_openapi_contract_exposes_continuation_fields():
+    document = _document()
+    operation = document["paths"]["/createGovernedTask"]["post"]
+    schema_ref = operation["requestBody"]["content"]["application/json"]["schema"]
+    schema = document["components"]["schemas"][schema_ref["$ref"].rsplit("/", 1)[-1]]
+
+    assert operation["operationId"] == "createGovernedTask"
+    assert operation["security"] == [{"bearerAuth": []}]
+    assert operation["x-openai-isConsequential"] is True
+    assert set(schema["required"]) == REQUIRED
+    assert schema["additionalProperties"] is False
+    assert OPTIONAL_CONTINUATION <= set(schema["properties"])
+    assert not OPTIONAL_CONTINUATION.intersection(schema["required"])
+    assert schema["properties"]["repository"]["const"] == "aios/aios"
+    assert schema["properties"]["generation"]["minimum"] == 1
+    assert schema["properties"]["priority"] == {
+        "type": "integer",
+        "minimum": -100,
+        "maximum": 100,
+    }
+
+
+def test_create_governed_task_route_and_required_fields_match_bridge():
+    assert ROUTES["/v2/governed-task"] == "createGovernedTask"
+    assert tuple(OPERATION_REQUIRED_FIELDS["createGovernedTask"]) == (
+        "task_id",
+        "repository",
+        "base_oid",
+        "target_branch",
+        "risk",
+        "description",
+        "idempotency_key",
+    )
+
+
+def test_bridge_execution_plan_and_envelope_routes_remain_exposed():
+    assert ROUTES["/v2/execution-envelope"] == "projectOnlineAgentExecution"
+    assert ROUTES["/v2/execution-plan"] == "registerOnlineAgentExecutionPlan"
+
+
+def test_bridge_allowlist_accepts_parent_pair_and_priority_but_rejects_unknown():
+    handler = BridgeHandler.__new__(BridgeHandler)
+    body = _create_request()
+    body.update(
+        {
+            "expected_branch_oid": "a" * 40,
+            "generation": 2,
+            "parent_task_id": "continuation-parent-test",
+            "parent_job_id": "job-continuation-parent-test",
+            "priority": 100,
+        }
+    )
+    handler._allowed("createGovernedTask", body)
+
+    invalid = dict(body)
+    invalid["unexpected_field"] = True
+    with pytest.raises(_HTTPError, match="UNKNOWN_FIELDS"):
+        handler._allowed("createGovernedTask", invalid)
+
+
+def test_openapi_does_not_add_credential_fields_to_create_contract():
+    document = _document()
+    properties = set(
+        document["components"]["schemas"]["CreateGovernedTaskRequest"]["properties"]
+    )
+    forbidden = {"authorization", "bearer", "token", "credential", "secret", "password", "api_key"}
+    assert properties.isdisjoint(forbidden)

tests/test_release_preparation_supervisor_coordinator.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+import sys
+
+import pytest
+
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+import aios_release_preparation_supervisor_coordinator as coordinator
+
+
+def task():
+    return {
+        "task_id":coordinator.TASK_ID,
+        "status":"QUEUED",
+        "required_capabilities":coordinator.REQUIRED_CAPABILITIES,
+        "metadata":{
+            "execution":{
+                "action":"agent_runtime",
+                "adapter":"openclaw",
+                "capability_scope":coordinator.REQUIRED_CAPABILITIES,
+                "execution_digest":"e"*64,
+                "change_set":[],
+                "verification":[],
+                "governance":{
+                    "writer_lease":False,
+                    "main_write":False,
+                    "production_write":False,
+                    "tag_write":False,
+                    "baseline_write":False,
+                    "force_push":False,
+                    "protected_ref_bypass":False,
+                },
+            }
+        },
+    }
+
+
+@pytest.fixture(autouse=True)
+def expected_digest(monkeypatch):
+    monkeypatch.setattr(coordinator,"EXPECTED_EXECUTION_DIGEST","e"*64)
+
+
+def test_coordinator_accepts_only_exact_nonmutating_projection():
+    execution=coordinator.validate_task(task())
+    assert execution["adapter"]=="openclaw"
+    assert execution["governance"]["writer_lease"] is False
+
+
+@pytest.mark.parametrize(
+    "mutate,error",
+    [
+        (lambda value:value.__setitem__("required_capabilities",["agent.runtime.openclaw"]),
+         "CAPABILITIES_INVALID"),
+        (lambda value:value["metadata"]["execution"].__setitem__("execution_digest","f"*64),
+         "ENVELOPE_INVALID"),
+        (lambda value:value["metadata"]["execution"]["governance"].__setitem__("production_write",True),
+         "ENVELOPE_INVALID"),
+    ],
+)
+def test_coordinator_rejects_capability_or_governance_drift(mutate,error):
+    value=task();mutate(value)
+    with pytest.raises(coordinator.CoordinatorError,match=error):
+        coordinator.validate_task(value)
+
+
+def test_heartbeat_is_control_plane_host_only_and_never_forwards_credentials():
+    progress=coordinator.heartbeat_progress(7)
+    assert progress["control_plane_access_granted"] is True
+    assert progress["external_planner_role"]=="UNPRIVILEGED_PLANNER"
+    assert progress["credential_forwarded_to_planner"] is False
+    assert progress["main_write"] is False
+    assert progress["production_write"] is False
+    encoded=json.dumps(progress,sort_keys=True).lower()
+    assert "bearer " not in encoded
+    assert "lease_token" not in encoded
+
+
+def test_coordinate_claim_heartbeats_without_submitting_result(monkeypatch,tmp_path):
+    calls=[]
+    def fake(method,route,token,body=None):
+        calls.append((method,route,token,body))
+        return 200,{"status":"RUNNING","lease_expires_at":"2099-01-01T00:00:00Z"}
+    monkeypatch.setattr(coordinator,"api_call",fake)
+    monkeypatch.setattr(coordinator,"STATE_ROOT",tmp_path)
+    monkeypatch.setattr(coordinator,"HEARTBEAT_SECONDS",0)
+    count=coordinator.coordinate_claim(
+        {"lease_token":"l"*32},"agent-token-value-for-test",max_heartbeats=2
+    )
+    assert count==2
+    assert all(call[1].endswith("/heartbeat") for call in calls)
+    assert not any(call[1].endswith("/result") for call in calls)
+    assert all("agent-token" not in json.dumps(call[3]) for call in calls)
+    liveness=json.loads((tmp_path/"liveness.json").read_text())
+    assert liveness["event_type"]=="SUPERVISOR_COORDINATOR_HEARTBEAT"
+    assert liveness["credential_material_exposed"] is False
+
+
+def test_units_use_existing_encrypted_agent_credential_and_no_planner_credential():
+    coordinator_unit=(
+        ROOT/"deploy/systemd/aios-release-preparation-supervisor-coordinator.service"
+    ).read_text()
+    projection_unit=(
+        ROOT/"deploy/systemd/aios-agent-task-supervisor-coordinator-projection.service"
+    ).read_text()
+    assert "aios-autonomous-agent-register.service" in coordinator_unit
+    assert "LoadCredentialEncrypted=agent_token:" in coordinator_unit
+    assert "control_token" not in coordinator_unit
+    assert "SupplementaryGroups=aios-agent-runtime" in coordinator_unit
+    assert "LoadCredential=agent_task_control_token:" in projection_unit
+    assert "User=aios-agent-task" in projection_unit
+    assert "ReadWritePaths=/var/lib/aios-agent-task" in projection_unit

tests/test_unified_execution_admission.py

@@ -0,0 +1,215 @@
+import ast
+import json
+import os
+import struct
+from collections.abc import Mapping
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from aios_agent_task_protocol import TaskProtocolError, TaskStore
+from aios_chatgpt_bridge_v2_server import BridgeHandler
+import aios_durable_owner_authorization as owner_auth
+from aios_online_agent_execution import GovernedExecutionBinding, project_online_task
+from aios_online_agent_worker import WorkerError, WorkerSupervisor
+
+CAPABILITIES = ["git.read", "git.test", "governed.write.branch", "governed.write.file"]
+
+
+def binding(task_id="unified-test-task-20260824"):
+    return GovernedExecutionBinding.from_mapping({
+        "task_id": task_id,
+        "job_id": "job-" + "a" * 32,
+        "repository": "aios/aios",
+        "base_oid": "0" * 40,
+        "expected_branch_oid": "0" * 40,
+        "target_branch": "agent/unified-test-20260824",
+        "branch_mode": "CREATE_NEW",
+        "generation": 1,
+        "idempotency_key": "unified-test-task-20260824-v1",
+    })
+
+
+def projected(task_id="unified-test-task-20260824"):
+    return project_online_task(
+        binding(task_id),
+        {
+            "goal": "test governed task",
+            "instructions": "test only",
+            "success_criteria": [],
+            "required_capabilities": CAPABILITIES,
+            "priority": 50,
+            "max_attempts": 3,
+        },
+        {
+            "action": "source_capture",
+            "adapter": "source_capture",
+            "capability_scope": CAPABILITIES,
+            "source_signals": ["durable_policy"],
+        },
+    )
+
+
+def test_write_task_rejected_before_persistence(tmp_path):
+    store = TaskStore(tmp_path / "state.sqlite3")
+    with pytest.raises(TaskProtocolError) as error:
+        store.create_task(
+            task_id="bare-write-task-20260824",
+            goal="bare",
+            instructions="must reject",
+            success_criteria=[],
+            required_capabilities=CAPABILITIES,
+            metadata={},
+            idempotency_key="bare-write-task-20260824-v1",
+        )
+    assert error.value.code == "GOVERNED_EXECUTION_BINDING_REQUIRED"
+    with pytest.raises(TaskProtocolError) as missing:
+        store.get_task("bare-write-task-20260824")
+    assert missing.value.code == "TASK_NOT_FOUND"
+
+
+def test_complete_projection_persists_envelope(tmp_path):
+    store = TaskStore(tmp_path / "state.sqlite3")
+    task = projected()
+    created = store.create_task(
+        task_id=task["task_id"],
+        goal=task["goal"],
+        instructions=task["instructions"],
+        success_criteria=task["success_criteria"],
+        required_capabilities=task["required_capabilities"],
+        constraints={},
+        metadata=task["metadata"],
+        priority=task["priority"],
+        max_attempts=task["max_attempts"],
+        idempotency_key=task["idempotency_key"],
+    )
+    execution = created["metadata"]["execution"]
+    assert created["status"] == "QUEUED"
+    assert execution["schema"] == "aios.online-agent.execution.v2"
+    assert execution["job_id"].startswith("job-")
+    assert execution["execution_digest"]
+    assert created["metadata"]["execution_projection"]["execution_digest"] == execution["execution_digest"]
+
+
+def test_online_route_requires_governed_creation_for_write_caps():
+    source = Path(__file__).parents[1] / "active-runtime" / "app" / "aios_model_api_staging.py"
+    tree = ast.parse(source.read_text(encoding="utf-8"))
+    function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "requires_governed_execution")
+    namespace = {
+        "Mapping": Mapping,
+        "Any": Any,
+        "GOVERNED_EXECUTION_CAPABILITIES": {"governed.write.branch", "governed.write.file"},
+    }
+    exec(compile(ast.Module(body=[function], type_ignores=[]), str(source), "exec"), namespace)
+    guard = namespace["requires_governed_execution"]
+    assert guard({"required_capabilities": CAPABILITIES})
+    assert not guard({"required_capabilities": ["git.read", "git.test"]})
+
+
+def test_bridge_requires_claim_for_projection_and_job(monkeypatch):
+    handler = object.__new__(BridgeHandler)
+    handler.headers = {"Authorization": "Bearer agent-credential"}
+    handler.control_token = None
+    handler.agent_task_api_url = "http://task-api"
+
+    class Response:
+        def __init__(self, value):
+            self.value = value
+        def __enter__(self):
+            return self
+        def __exit__(self, *_args):
+            return False
+        def read(self):
+            return json.dumps(self.value).encode()
+
+    responses = iter([
+        {"agent_id": "worker-1"},
+        {"task_id": "task-1", "claimed_by": None, "status": "QUEUED"},
+    ])
+    monkeypatch.setattr("aios_chatgpt_bridge_v2_server.urllib.request.urlopen", lambda *_args, **_kwargs: Response(next(responses)))
+    assert not handler._authorized("projectOnlineAgentExecution", {"task_id": "task-1"})
+
+    responses = iter([
+        {"agent_id": "worker-1"},
+        {"task_id": "task-1", "claimed_by": "worker-1", "status": "CLAIMED"},
+    ])
+    assert handler._authorized("getGitOpsJob", {"task_id": "task-1"})
+
+
+def test_broker_uses_so_peercred_and_canonical_group(monkeypatch):
+    class Connection:
+        def getsockopt(self, *_args):
+            return struct.pack("3i", os.getpid(), os.getuid(), os.getgid())
+
+    monkeypatch.setattr(owner_auth, "_peer_process_status", lambda _pid: (os.getuid(), os.getgid(), {os.getgid(), 4242}))
+    monkeypatch.setattr(owner_auth, "CANONICAL_BROKER_GROUP", "aios-autonomous-agent-worker")
+    monkeypatch.setattr(owner_auth.grp, "getgrnam", lambda _name: SimpleNamespace(gr_gid=4242))
+    assert owner_auth.authorize_broker_peer(Connection())
+
+    monkeypatch.setattr(owner_auth, "_peer_process_status", lambda _pid: (os.getuid(), os.getgid(), {os.getgid()}))
+    with pytest.raises(owner_auth.AuthorizationError) as error:
+        owner_auth.authorize_broker_peer(Connection())
+    assert error.value.code == "SOCKGROUP_NOT_AUTHORIZED"
+
+
+def test_bootstrap_rejects_bad_projection_before_job(tmp_path):
+    from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
+
+    class Authority:
+        def commit_exists(self, _oid):
+            return True
+        def read_ref(self, _ref):
+            return None
+
+    bootstrap = GovernedTaskBootstrap(
+        Authority(), tmp_path, remote_name="gitea-http",
+        token_path=tmp_path / "token", ssh_key_path=tmp_path / "key",
+    )
+    with pytest.raises(BootstrapError) as error:
+        bootstrap.create({
+            "task_id": "atomic-projection-failure-20260824",
+            "repository": "aios/aios",
+            "base_oid": "0" * 40,
+            "expected_branch_oid": "0" * 40,
+            "target_branch": "agent/atomic-projection-failure-20260824",
+            "risk": "engineering",
+            "description": "invalid deterministic placeholder",
+            "idempotency_key": "atomic-projection-failure-20260824-v1",
+            "online_task": {
+                "goal": "invalid",
+                "instructions": "invalid",
+                "success_criteria": [],
+                "required_capabilities": CAPABILITIES,
+            },
+            "execution": {
+                "action": "deterministic",
+                "adapter": "deterministic",
+                "capability_scope": CAPABILITIES,
+                "change_set": [],
+                "verification": [],
+            },
+            "require_online_task_projection": True,
+        })
+    assert error.value.args[0] == "EXECUTION_BINDING_INVALID"
+    assert not list((tmp_path / "jobs").rglob("state.sqlite3"))
+    assert not list((tmp_path / "task-bindings").glob("*.json"))
+
+
+def test_worker_rejects_bare_write_task():
+    supervisor = object.__new__(WorkerSupervisor)
+    with pytest.raises(WorkerError) as error:
+        supervisor.resolve_envelope({
+            "task_id": "bare-worker-task-20260824",
+            "required_capabilities": CAPABILITIES,
+            "metadata": {},
+        })
+    assert error.value.code == "TASK_SKIPPED_PROJECTION_MISSING"
+
+
+def test_continuation_scope_allows_existing_subset():
+    source = Path(__file__).parents[1] / "active-runtime" / "app" / "aios_continuation_executor.py"
+    text = source.read_text(encoding="utf-8")
+    assert "if dirty and not dirty.issubset(expected_paths):" in text
+    assert "if dirty and dirty != expected_paths:" not in text