fix(supervisor): make release admission actor aware

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_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_durable_owner_authorization.py

@@ -18,12 +18,15 @@ import hmac
 from http.server import BaseHTTPRequestHandler
 import json
 import os
+import pwd
 from pathlib import Path
 import re
 import sqlite3
+import socket
 import socketserver
 import stat
 import tempfile
+import struct
 import threading
 import urllib.error
 import urllib.request
@@ -34,6 +37,8 @@ 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"
+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 +69,28 @@ 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_user_from_connection(connection: object) -> str:
+    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
+    try:
+        _pid, uid, _gid = struct.unpack("3i", raw)
+    except struct.error as exc:
+        raise AuthorizationError("SO_PEERCRED_INVALID", 403) from exc
+    try:
+        return pwd.getpwuid(uid).pw_name
+    except (KeyError, OSError) as exc:
+        raise AuthorizationError("SO_PEERCRED_UID_UNKNOWN", 403) from exc
 
 
 class AuthorizationError(RuntimeError):
@@ -447,6 +474,16 @@ class ReceiptStore:
                     ON projection_outbox(state, capability, created_at)
                 """
             )
+            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()
 
@@ -519,6 +556,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 +784,14 @@ class Broker:
         store: ReceiptStore,
         upstream: Upstream,
         expected_account: str,
+        expected_peer_user: str,
     ) -> 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
 
     def policy(self) -> DurablePolicy:
         policy = DurablePolicy.load(self.policy_path, self.revocation_path)
@@ -758,24 +828,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 +998,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
 
@@ -893,6 +1038,19 @@ 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
+        try:
+            return self.peer_user() == self.broker.expected_peer_user
+        except AuthorizationError:
+            return False
+
     def account_authorized(self) -> bool:
         if self.broker is None:
             return False
@@ -924,6 +1082,9 @@ class Handler(BaseHTTPRequestHandler):
                 200,
             )
             return
+        if not self.peer_authorized():
+            self.send_json({"error": "SOCKUSER_NOT_AUTHORIZED"}, 403)
+            return
         if not self.account_authorized():
             self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
             return
@@ -938,6 +1099,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,6 +1124,9 @@ class Handler(BaseHTTPRequestHandler):
             self.send_json({"error": exc.code}, exc.status)
 
     def do_POST(self) -> None:
+        if not self.peer_authorized():
+            self.send_json({"error": "SOCKUSER_NOT_AUTHORIZED"}, 403)
+            return
         if not self.account_authorized():
             self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
             return
@@ -970,14 +1140,24 @@ class Handler(BaseHTTPRequestHandler):
             return
         try:
             body = self.body()
-            if set(body) != {"idempotency_key", "request"}:
+            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 +1169,14 @@ class UnixHTTPServer(socketserver.ThreadingUnixStreamServer):
 
 def serve() -> None:
     credentials = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+    allowed_peer_user = os.environ.get(
+        "AIOS_OWNER_AUTH_BROKER_ALLOWED_USER",
+        "aios-autonomous-agent-worker",
+    ).strip()
+    if not allowed_peer_user:
+        raise RuntimeError("BROKER_ALLOWED_USER_MISSING")
+    allowed_peer_uid = pwd.getpwnam(allowed_peer_user).pw_uid
+    allowed_peer_gid = pwd.getpwnam(allowed_peer_user).pw_gid
     policy_path = Path(
         os.environ.get(
             "AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY",
@@ -1029,6 +1217,7 @@ def serve() -> None:
             bridge_token=bridge_token,
         ),
         expected_account=account,
+        expected_peer_user=allowed_peer_user,
     )
     socket_path = Path(
         os.environ.get(
@@ -1044,6 +1233,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_release_preparation_supervisor_coordinator.py

@@ -0,0 +1,559 @@
+"""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 http.client
+import json
+import os
+from pathlib import Path
+import re
+import socket
+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",
+    )
+)
+BROKER_SOCKET = Path(os.environ.get(
+    "AIOS_OWNER_AUTH_BROKER_SOCKET",
+    "/run/aios-owner-authorization-broker/broker.sock",
+))
+BROKER_ACCOUNT = "chatgpt-governed-implementer-singlehost"
+EXTERNAL_REQUEST_SHA256 = "08fc2f3d1b64a2ca11da9f22da48de834cff3442fa7fa2d3bc390f8dbf9b254a"
+SUPERVISOR_EXECUTION_DIGEST = "491457595667a95809f6f3cd19e9e7c9e74181070fd8038b86f30eb41bdfc231"
+BUILDER_TASK_ID = "r42-trusted-candidate-builder-runtime-wiring-20260820"
+BUILDER_BASE_OID = "d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3"
+BUILDER_TARGET_BRANCH = "agent/r42-trusted-candidate-builder-runtime-wiring-20260820"
+BUILDER_IDEMPOTENCY_KEY = "r42-trusted-candidate-builder-runtime-wiring-binding-v1"
+GATE_AMENDMENT_CLASSIFICATION = "SUPERVISOR_GLOBAL_EXTERNAL_ACK_GATE_REMEDIATED"
+
+
+class CoordinatorError(RuntimeError):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
+class UnixHTTPConnection(http.client.HTTPConnection):
+    def __init__(self, socket_path: Path, *, timeout: int = 30) -> 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)
+        sock.connect(str(self.socket_path))
+        self.sock = sock
+
+
+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 broker_call(
+    method: str,
+    route: str,
+    body: Mapping[str, Any] | None = None,
+) -> tuple[int, dict[str, Any]]:
+    if not BROKER_SOCKET.is_absolute():
+        raise CoordinatorError("OWNER_BROKER_SOCKET_INVALID")
+    try:
+        if BROKER_SOCKET.is_symlink():
+            raise CoordinatorError("OWNER_BROKER_SOCKET_INVALID")
+    except OSError as exc:
+        raise CoordinatorError("OWNER_BROKER_UNAVAILABLE") from exc
+    connection = UnixHTTPConnection(BROKER_SOCKET)
+    payload = None if body is None else canonical(dict(body))
+    headers = {"Accept": "application/json", "X-AIOS-Account": BROKER_ACCOUNT}
+    if payload is not None:
+        headers["Content-Type"] = "application/json"
+    try:
+        connection.request(method, route, body=payload, headers=headers)
+        response = connection.getresponse()
+        raw = response.read()
+        value = json.loads(raw.decode("utf-8"))
+        if not isinstance(value, Mapping):
+            raise CoordinatorError("OWNER_BROKER_RESPONSE_INVALID")
+        return response.status, dict(value)
+    except (OSError, TimeoutError, http.client.HTTPException, ValueError) as exc:
+        raise CoordinatorError("OWNER_BROKER_UNAVAILABLE") from exc
+    finally:
+        connection.close()
+
+
+def authorize_actor(
+    actor_type: str,
+    *,
+    external_ack: Mapping[str, Any] | None = None,
+    host_authorization: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    if actor_type == "external_connector":
+        expected = {
+            "account": BROKER_ACCOUNT,
+            "capability": "governed-task:project",
+            "operation": "createGovernedTask",
+            "single_use": True,
+            "request_sha256": EXTERNAL_REQUEST_SHA256,
+        }
+        if external_ack is None:
+            raise CoordinatorError("WAIT_EXTERNAL_ACK")
+        if dict(external_ack) != expected:
+            raise CoordinatorError("EXTERNAL_ACK_INVALID")
+        return {
+            "actor_type": actor_type,
+            "decision": "ALLOW_EXTERNAL_CONNECTOR_EXACT_ACK",
+            "external_ack_required": True,
+            "request_sha256": EXTERNAL_REQUEST_SHA256,
+            "authorized_capabilities": ["governed-task:project"],
+        }
+    if actor_type == "credentialed_host_coordinator":
+        if not isinstance(host_authorization, Mapping):
+            raise CoordinatorError("HOST_COORDINATOR_DURABLE_OWNER_REQUIRED")
+        constraints = host_authorization.get("constraints")
+        capabilities = host_authorization.get("capabilities")
+        if (
+            host_authorization.get("account") != BROKER_ACCOUNT
+            or host_authorization.get("persistent_until_revoked") is not True
+            or host_authorization.get("separate_owner_release_approval_still_required") is not True
+            or not isinstance(capabilities, list)
+            or "governed-task:project" not in capabilities
+            or not isinstance(constraints, Mapping)
+            or constraints.get("repository") != "aios/aios"
+            or constraints.get("allowed_target_prefix") != "agent/"
+            or constraints.get("max_task_risk") not in {"low", "engineering"}
+            or re.fullmatch(r"[0-9a-f]{64}", str(host_authorization.get("policy_sha256") or "")) is None
+            or not str(host_authorization.get("grant_id") or "")
+        ):
+            raise CoordinatorError("HOST_COORDINATOR_DURABLE_OWNER_DENY")
+        return {
+            "actor_type": actor_type,
+            "decision": "ALLOW_HOST_COORDINATOR_GOVERNED_PROJECTION",
+            "external_ack_required": False,
+            "durable_owner_governed_task_project": "verified",
+            "grant_id": str(host_authorization["grant_id"]),
+            "policy_sha256": str(host_authorization["policy_sha256"]),
+            # Do not inherit release:candidate even if the durable policy also has it.
+            "authorized_capabilities": ["governed-task:project"],
+        }
+    raise CoordinatorError("UNKNOWN_ACTOR_DENY")
+
+
+def builder_request() -> dict[str, Any]:
+    capabilities = [
+        "git.read",
+        "git.test",
+        "governed.write.branch",
+        "governed.write.file",
+    ]
+    return {
+        "task_id": BUILDER_TASK_ID,
+        "repository": "aios/aios",
+        "base_oid": BUILDER_BASE_OID,
+        "expected_branch_oid": BUILDER_BASE_OID,
+        "target_branch": BUILDER_TARGET_BRANCH,
+        "generation": 1,
+        "branch_mode": "CREATE_NEW",
+        "risk": "engineering",
+        "description": "Wire TrustedCandidateBuilder through the canonical authenticated Release API and fixed candidate retry chain.",
+        "idempotency_key": BUILDER_IDEMPOTENCY_KEY,
+        "priority": 100,
+        "max_attempts": 3,
+        "require_online_task_projection": True,
+        "online_task": {
+            "goal": "Implement and verify TrustedCandidateBuilder runtime wiring from the fixed d957a216 lineage.",
+            "instructions": (
+                "Capture the real Builder and Release API contracts, implement only the fixed governed branch, "
+                "prove authenticated POST {} returns HTTP 400, and stop before Owner release approval or Production. "
+                "No main write, force push, tag, baseline, production deploy, runtime switch, credential forwarding, "
+                "or legacy candidate input."
+            ),
+            "success_criteria": [
+                {"id": "builder_wiring", "description": "TrustedCandidateBuilder wiring is verified."},
+                {"id": "release_api", "description": "Authenticated Release API POST {} returns HTTP 400."},
+                {"id": "governed_candidate", "description": "All modified files are committed only to the fixed governed branch."},
+            ],
+            "constraints": {
+                "main_write": False,
+                "production_write": False,
+                "owner_release_approval": False,
+                "tag_publish": False,
+                "baseline_finalize": False,
+                "runtime_switch": False,
+                "credential_forwarding": False,
+                "legacy_candidate_input": False,
+            },
+            "required_capabilities": capabilities,
+            "priority": 100,
+            "max_attempts": 3,
+        },
+        "execution": {
+            "action": "source_capture",
+            "adapter": "source_capture",
+            "capability_scope": capabilities,
+            "source_signals": [
+                "TrustedCandidateBuilder runtime wiring",
+                "Release API authenticated POST empty object HTTP 400",
+                "fixed V2R2C2 retry-extension attempt 2",
+                "V2R2C2 and V2R2C3 candidate verification",
+            ],
+        },
+    }
+
+
+def gate_amendment(host_decision: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "schema": "aios.release-preparation.gate-amendment.v1",
+        "classification": GATE_AMENDMENT_CLASSIFICATION,
+        "supervisor_task_id": TASK_ID,
+        "supervisor_execution_digest": SUPERVISOR_EXECUTION_DIGEST,
+        "actor_type": "credentialed_host_coordinator",
+        "operation": "createGovernedTask",
+        "repository": "aios/aios",
+        "original_external_request_sha256": EXTERNAL_REQUEST_SHA256,
+        "external_connector_rule": "ACK_STILL_REQUIRED",
+        "host_coordinator_rule": "DURABLE_OWNER_GOVERNED_TASK_PROJECT",
+        "owner_grant_id": host_decision["grant_id"],
+        "owner_policy_sha256": host_decision["policy_sha256"],
+        "builder_task_id": BUILDER_TASK_ID,
+        "builder_base_oid": BUILDER_BASE_OID,
+        "builder_expected_branch_oid": BUILDER_BASE_OID,
+        "builder_target_branch": BUILDER_TARGET_BRANCH,
+        "builder_generation": 1,
+        "builder_idempotency_key": BUILDER_IDEMPOTENCY_KEY,
+    }
+
+
+def ensure_builder_projection() -> dict[str, Any]:
+    status, authorization = broker_call("GET", "/v1/capabilities")
+    if status != 200:
+        raise CoordinatorError("HOST_COORDINATOR_DURABLE_OWNER_DENY")
+    decision = authorize_actor(
+        "credentialed_host_coordinator",
+        host_authorization=authorization,
+    )
+    request = builder_request()
+    status, receipt = broker_call(
+        "POST",
+        "/v1/project-governed-task",
+        {
+            "idempotency_key": BUILDER_IDEMPOTENCY_KEY,
+            "request": request,
+            "amendment": gate_amendment(decision),
+        },
+    )
+    upstream = receipt.get("upstream_response")
+    amendment = receipt.get("gate_amendment")
+    if (
+        status != 200
+        or receipt.get("operation_state") != "EXECUTED"
+        or receipt.get("capability") != "governed-task:project"
+        or receipt.get("grant_id") != decision["grant_id"]
+        or receipt.get("policy_sha256") != decision["policy_sha256"]
+        or receipt.get("external_capability_projection_required") is not False
+        or not isinstance(upstream, Mapping)
+        or upstream.get("task_id") != BUILDER_TASK_ID
+        or upstream.get("base_oid") != BUILDER_BASE_OID
+        or upstream.get("expected_branch_oid") != BUILDER_BASE_OID
+        or upstream.get("target_branch") != BUILDER_TARGET_BRANCH
+        or upstream.get("generation") != 1
+        or not isinstance(amendment, Mapping)
+        or amendment.get("classification") != GATE_AMENDMENT_CLASSIFICATION
+        or re.fullmatch(r"[0-9a-f]{64}", str(amendment.get("amendment_sha256") or "")) is None
+    ):
+        raise CoordinatorError("BUILDER_PROJECTION_RECEIPT_INVALID")
+    return {
+        **decision,
+        "builder_task_id": BUILDER_TASK_ID,
+        "builder_binding_verified": True,
+        "gate_amendment_sha256": amendment["amendment_sha256"],
+    }
+
+
+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,
+    gate: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    progress = {
+        "stage": "supervisor_coordinator_active",
+        "phase": "external_caller_projection",
+        "state": "PENDING_ACTOR_AWARE_ADMISSION",
+        "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,
+    }
+    if gate is not None:
+        progress.update({
+            "state": str(gate.get("decision") or "DENY"),
+            "actor_type": gate.get("actor_type"),
+            "external_ack_required": gate.get("external_ack_required"),
+            "durable_owner_governed_task_project": gate.get("durable_owner_governed_task_project"),
+            "builder_task_id": gate.get("builder_task_id"),
+            "builder_binding_verified": gate.get("builder_binding_verified", False),
+            "gate_amendment_sha256": gate.get("gate_amendment_sha256"),
+        })
+    return progress
+
+
+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
+    gate: dict[str, Any] | None = None
+    while max_heartbeats is None or sequence < max_heartbeats:
+        sequence += 1
+        if gate is None:
+            try:
+                gate = ensure_builder_projection()
+                audit("SUPERVISOR_ACTOR_AWARE_GATE_ALLOWED", **gate)
+            except CoordinatorError as exc:
+                audit(
+                    "SUPERVISOR_ACTOR_AWARE_GATE_DENIED",
+                    actor_type="credentialed_host_coordinator",
+                    decision="DENY",
+                    error=exc.code,
+                )
+        status, response = api_call(
+            "POST",
+            f"/v1/tasks/{TASK_ID}/heartbeat",
+            token,
+            {
+                "lease_token": lease_token,
+                "lease_seconds": LEASE_SECONDS,
+                "progress": heartbeat_progress(sequence, gate),
+            },
+        )
+        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-initial-execution-projection.service

@@ -0,0 +1,37 @@
+[Unit]
+Description=AIOS owner-controlled Task initial execution 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-initial-execution-projection-recovery-20260820
+Environment=AIOS_TASK_INITIAL_EXECUTION_PROJECTION_HOST=127.0.0.1
+Environment=AIOS_TASK_INITIAL_EXECUTION_PROJECTION_PORT=18943
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_initial_execution_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-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-owner-durable-authorization-broker.service

@@ -6,7 +6,7 @@ Wants=aios-chatgpt-bridge-v2-staging.service
 [Service]
 Type=simple
 User=root
-Group=aios-executor-exchange
+Group=aios-autonomous-agent-worker
 RuntimeDirectory=aios-owner-authorization-broker
 RuntimeDirectoryMode=0750
 StateDirectory=aios-owner-authorization-broker
@@ -16,6 +16,7 @@ Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
 Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY=/etc/aios-gitops/owner-durable-authorization.json
 Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_REVOCATIONS=/etc/aios-gitops/owner-durable-authorization-revocations.json
 Environment=AIOS_OWNER_AUTH_BROKER_ACCOUNT=chatgpt-governed-implementer-singlehost
+Environment=AIOS_OWNER_AUTH_BROKER_ALLOWED_USER=aios-autonomous-agent-worker
 Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
 ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_durable_owner_authorization.py
 Restart=on-failure

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

@@ -0,0 +1,46 @@
+[Unit]
+Description=AIOS credential-isolated release-preparation supervisor coordinator
+After=network-online.target aios-agent-task-api.service aios-autonomous-agent-register.service aios-owner-durable-authorization-broker.service
+Wants=network-online.target aios-agent-task-api.service aios-owner-durable-authorization-broker.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
+Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
+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/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

@@ -0,0 +1,452 @@
+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,
+    InitialExecutionProjectionRequest,
+    InitialExecutionProjectionStore,
+    SourceCaptureProjectionRecoveryRequest,
+)
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+AUTHORIZATION="owner-initial-execution-projection-recovery-20260820"
+TASK_ID="task-initial-projection-a"
+JOB_ID="job-initial-projection-a"
+BASE="a"*40
+BRANCH="agent/task-initial-projection-a"
+IDEMPOTENCY="task-initial-projection-binding-v1"
+CAPABILITY="agent.runtime.openclaw"
+
+
+@pytest.fixture(autouse=True)
+def _auth(monkeypatch):
+    monkeypatch.setenv(
+        "AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID",
+        AUTHORIZATION,
+    )
+
+
+def canon(value):
+    return hashlib.sha256(
+        json.dumps(value,sort_keys=True,separators=(",",":")).encode()
+    ).hexdigest()
+
+
+def binding():
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "branch_mode":"CREATE_NEW",
+        "generation":1,
+        "idempotency_key":IDEMPOTENCY,
+    })
+
+
+def runtime_input():
+    return {
+        "goal":"Supervise release preparation",
+        "instructions":"Observe and resume only the same lineage.",
+        "success_criteria":[{"id":"heartbeat","description":"Heartbeat observed"}],
+        "constraints":{"production_write":False},
+        "scope":{"scope_type":"legacy","tenant_id":None,"project_id":None},
+        "trace_id":TASK_ID,
+    }
+
+
+def envelope(*, job_id=JOB_ID, action="agent_runtime", capability=CAPABILITY):
+    bound=binding()
+    if job_id != JOB_ID:
+        bound=GovernedExecutionBinding.from_mapping({**bound.public(),"job_id":job_id})
+    return project_execution_envelope(
+        bound,
+        {
+            "action":action,
+            "adapter":"openclaw" if action=="agent_runtime" else "source_capture",
+            "capability_scope":[capability],
+            "runtime_input":runtime_input() if action=="agent_runtime" else None,
+        },
+    )
+
+
+def make_db(tmp_path, *, claimed_by=None, lease=False, metadata=None, caps=None, event="TASK_CREATED"):
+    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 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
+    );
+    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
+    );
+    """)
+    metadata={} if metadata is None else metadata
+    caps=[] if caps is None else caps
+    con.execute(
+        "INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        (
+            TASK_ID,"Supervise release preparation",
+            "Observe and resume only the same lineage.",
+            json.dumps([{"id":"heartbeat","description":"Heartbeat observed"}]),
+            json.dumps(caps),json.dumps({"production_write":False}),json.dumps(metadata),
+            None,100,20,0,"QUEUED",claimed_by,
+            "lease-hash" if lease else None,
+            "2099-01-01T00:00:00Z" if lease else None,
+            None,None,None,None,1,"2026-08-20T00:00:00Z","2026-08-20T00:00:00Z",
+        ),
+    )
+    con.execute(
+        "INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) VALUES (?,?,?,?,?)",
+        (TASK_ID,None,event,"{}","2026-08-20T00:00:00Z"),
+    )
+    con.commit();con.close()
+    return path
+
+
+def request(*, digest=None, key="initial-projection-a-v1"):
+    return InitialExecutionProjectionRequest.from_mapping({
+        "expected_status":"QUEUED",
+        "expected_revision":1,
+        "expected_attempt":0,
+        "expected_metadata_sha256":canon({}),
+        "expected_required_capabilities_sha256":canon([]),
+        "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":"initial governed supervisor runtime projection",
+    })
+
+
+def resolver(_task_id,_job_id,_capability,_adapter,received_runtime_input):
+    assert received_runtime_input==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())
+    assert result["revision"]==2
+    assert result["attempt"]==0
+    assert result["idempotent"] is False
+    con=sqlite3.connect(path)
+    row=con.execute(
+        "SELECT status,attempt,revision,claimed_by,metadata_json,required_capabilities_json FROM tasks"
+    ).fetchone()
+    assert row[:4]==("QUEUED",0,2,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]
+    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_INITIAL_EXECUTION_PROJECTION_AUTHORIZED"
+    ]
+    payload=json.loads(events[-1][1])
+    assert payload["job_id"]==JOB_ID
+    assert payload["main_write"] is False
+    assert payload["production_write"] is False
+
+
+def test_initial_projection_idempotent_exact_replay(tmp_path):
+    path=make_db(tmp_path)
+    store=InitialExecutionProjectionStore(path,resolver)
+    first=store.project(TASK_ID,request())
+    second=store.project(TASK_ID,request())
+    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_initial_execution_projections").fetchone()[0]==1
+    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",
+    [
+        ({"claimed_by":"worker-a"},"TASK_ALREADY_CLAIMED"),
+        ({"lease":True},"ACTIVE_LEASE_DENY"),
+        ({"metadata":{"execution":{}}},"TASK_EXECUTION_ALREADY_PROJECTED"),
+        ({"caps":["git.read"]},"TASK_INITIAL_CAPABILITIES_MUST_BE_EMPTY"),
+        ({"event":"TASK_CLAIMED"},"TASK_INITIAL_EVENT_CHAIN_REQUIRED"),
+    ],
+)
+def test_initial_projection_rejects_noninitial_state(tmp_path,kwargs,error):
+    path=make_db(tmp_path,**kwargs)
+    with pytest.raises(InitialExecutionProjectionError,match=error):
+        InitialExecutionProjectionStore(path,resolver).project(TASK_ID,request())
+
+
+def test_initial_projection_rejects_binding_drift(tmp_path):
+    path=make_db(tmp_path)
+    def wrong(*_args):
+        return envelope(job_id="job-other")
+    with pytest.raises(InitialExecutionProjectionError,match="GOVERNED_BINDING_CONFLICT"):
+        InitialExecutionProjectionStore(path,wrong).project(TASK_ID,request())
+
+
+def test_initial_projection_rejects_nonruntime_envelope(tmp_path):
+    path=make_db(tmp_path)
+    def wrong(*_args):
+        return envelope(action="source_capture",capability="git.read")
+    with pytest.raises(InitialExecutionProjectionError):
+        InitialExecutionProjectionStore(path,wrong).project(TASK_ID,request())
+
+
+def test_initial_projection_rejects_execution_digest_drift(tmp_path):
+    path=make_db(tmp_path)
+    with pytest.raises(InitialExecutionProjectionError,match="EXECUTION_DIGEST_DRIFT"):
+        InitialExecutionProjectionStore(path,resolver).project(
+            TASK_ID,request(digest="f"*64)
+        )
+
+
+def test_request_is_narrow_and_initial_state_bound():
+    data=request().public()
+    data["arbitrary_metadata"]={"x":1}
+    with pytest.raises(InitialExecutionProjectionError,match="FIELDS_INVALID"):
+        InitialExecutionProjectionRequest.from_mapping(data)
+    data=request().public();data["expected_attempt"]=1
+    with pytest.raises(InitialExecutionProjectionError,match="INITIAL_STATE_REQUIRED"):
+        InitialExecutionProjectionRequest.from_mapping(data)
+
+
+def test_initial_projection_unit_is_unprivileged_and_owner_credential_bound():
+    unit=(ROOT/"deploy/systemd/aios-agent-task-initial-execution-projection.service").read_text()
+    assert "User=aios-agent-task" in unit
+    assert "Group=aios-agent-task" in unit
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
+    assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
+    assert "ReadWritePaths=/var/lib/aios-agent-task" in unit
+    assert "User=root" not in unit

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

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
+import struct
 import json
 from pathlib import Path
 import sqlite3
 import sys
+from types import SimpleNamespace
 
 import pytest
 
@@ -134,6 +136,7 @@ def broker(tmp_path,monkeypatch):
         store=store,
         upstream=upstream,
         expected_account=ACCOUNT,
+        expected_peer_user="aios-autonomous-agent-worker",
     )
     monkeypatch.setattr(
         auth.DurablePolicy,
@@ -573,16 +576,110 @@ def test_receipt_store_contains_no_owner_policy_or_credentials(
     assert "policy_json" not in columns
 
 
+def gate_amendment(policy):
+    return {
+        "schema":auth.GATE_AMENDMENT_SCHEMA,
+        "classification":auth.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",
+    }
+
+
+def builder_governed_request():
+    value=governed_request()
+    value.update({
+        "task_id":"r42-trusted-candidate-builder-runtime-wiring-20260820",
+        "base_oid":"d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
+        "expected_branch_oid":"d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
+        "target_branch":"agent/r42-trusted-candidate-builder-runtime-wiring-20260820",
+        "generation":1,
+        "idempotency_key":"r42-trusted-candidate-builder-runtime-wiring-binding-v1",
+    })
+    return value
+
+
+def test_task_bound_gate_amendment_is_validated_and_persisted(tmp_path,monkeypatch):
+    instance,upstream,store,policy=broker(tmp_path,monkeypatch)
+    request=builder_governed_request()
+    response=instance.execute(
+        "governed-task:project",
+        request,
+        idempotency_key=request["idempotency_key"],
+        amendment=gate_amendment(policy),
+    )
+    assert response["operation_state"]=="EXECUTED"
+    assert response["gate_amendment"]["classification"]==auth.GATE_AMENDMENT_CLASSIFICATION
+    assert len(upstream.project_calls)==1
+    con=sqlite3.connect(store.path)
+    row=con.execute(
+        "select amendment_sha256,amendment_json from gate_amendments where idempotency_key=?",
+        (request["idempotency_key"],),
+    ).fetchone()
+    assert row[0]==response["gate_amendment"]["amendment_sha256"]
+    persisted=json.loads(row[1])
+    assert persisted["supervisor_task_id"]=="r42-release-preparation-autonomous-supervisor-20260820"
+    assert persisted["builder_task_id"]==request["task_id"]
+    for forbidden in ("bearer ","access_token","refresh_token","private_key","credential_file"):
+        assert forbidden not in row[1].lower()
+
+
+def test_gate_amendment_rejects_wrong_builder_or_policy_binding(tmp_path,monkeypatch):
+    instance,_,_,policy=broker(tmp_path,monkeypatch)
+    request=builder_governed_request()
+    for bad in (
+        {**gate_amendment(policy),"builder_base_oid":"0"*40},
+        {**gate_amendment(policy),"owner_policy_sha256":"0"*64},
+        {**gate_amendment(policy),"actor_type":"unknown"},
+    ):
+        with pytest.raises(auth.AuthorizationError,match="GATE_AMENDMENT_BINDING_INVALID"):
+            instance.execute(
+                "governed-task:project",
+                request,
+                idempotency_key=request["idempotency_key"],
+                amendment=bad,
+            )
+
+
+def test_gate_amendment_does_not_inherit_release_candidate_authority(tmp_path,monkeypatch):
+    instance,_,_,policy=broker(tmp_path,monkeypatch)
+    request=builder_governed_request()
+    response=instance.execute(
+        "governed-task:project",
+        request,
+        idempotency_key=request["idempotency_key"],
+        amendment=gate_amendment(policy),
+    )
+    assert response["capability"]=="governed-task:project"
+    assert response["external_capability_projection_required"] is False
+    assert "release:candidate" not in json.dumps(response["gate_amendment"])
+
+
 def test_systemd_broker_uses_unix_socket_and_only_host_control_credential():
     unit=(
         ROOT/"deploy/systemd/aios-owner-durable-authorization-broker.service"
     ).read_text()
     assert "LoadCredential=agent_task_control_token:" in unit
+    assert "AIOS_OWNER_AUTH_BROKER_ALLOWED_USER=aios-autonomous-agent-worker" in unit
     assert "release_candidate_token" not in unit
     assert "18947" not in unit
     assert "AIOS_RELEASE_CANDIDATE_API_URL" not in unit
     assert "AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock" in unit
-    assert "Group=aios-executor-exchange" in unit
+    assert "User=root" in unit
+    assert "Group=aios-autonomous-agent-worker" in unit
     assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
     assert "IPAddressDeny=any" in unit
     assert "IPAddressAllow=127.0.0.0/8" in unit
@@ -609,6 +706,74 @@ def test_source_contains_no_release_builder_endpoint_or_high_risk_execution():
     assert "INITIAL_EXECUTION_ACTION_NOT_AUTHORIZED" in source
 
 
+def test_peer_credential_mapping_extracts_expected_user(monkeypatch):
+    class FakeSocket:
+        def getsockopt(self, *_args, **_kwargs):
+            return struct.pack("3i", 0, 555, 0)
+
+    monkeypatch.setattr(
+        auth.pwd,
+        "getpwuid",
+        lambda uid: SimpleNamespace(pw_name="aios-autonomous-agent-worker"),
+    )
+    assert (
+        auth.peer_user_from_connection(FakeSocket())
+        == "aios-autonomous-agent-worker"
+    )
+
+
+def test_unknown_or_shared_aos_worker_user_is_rejected_before_business_logic(monkeypatch):
+    handler = object.__new__(auth.Handler)
+    observed = []
+    handler.path = "/v1/authorize-release-candidate"
+    handler.headers = {"X-AIOS-Account": ACCOUNT}
+    handler.send_json = lambda body, status: observed.append((body, status))
+    handler.connection = object()
+    handler.broker = SimpleNamespace(
+        expected_account=ACCOUNT,
+        expected_peer_user="aios-autonomous-agent-worker",
+        execute=lambda *args, **kwargs: observed.append("executed"),
+    )
+    handler.body = lambda: observed.append(("should_never_read_body",))
+
+    monkeypatch.setattr(auth, "peer_user_from_connection", lambda *_args: "aios-engineering-worker")
+    handler.do_POST()
+    assert observed == [({"error": "SOCKUSER_NOT_AUTHORIZED"}, 403)]
+
+
+def test_account_check_stays_secondary_after_peer_auth(monkeypatch):
+    handler = object.__new__(auth.Handler)
+    observed = []
+    handler.path = "/v1/authorize-release-candidate"
+    handler.headers = {"X-AIOS-Account": "wrong-account"}
+    handler.send_json = lambda body, status: observed.append((body, status))
+    handler.connection = object()
+    handler.broker = SimpleNamespace(
+        expected_account=ACCOUNT,
+        expected_peer_user="aios-autonomous-agent-worker",
+        execute=lambda *args, **kwargs: observed.append("executed"),
+    )
+    handler.body = lambda: observed.append(("should_never_read_body",))
+
+    monkeypatch.setattr(
+        auth,
+        "peer_user_from_connection",
+        lambda *_args: "aios-autonomous-agent-worker",
+    )
+    handler.do_POST()
+    assert observed == [({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)]
+
+
+def test_health_endpoint_remains_unauthenticated():
+    handler = object.__new__(auth.Handler)
+    observed = []
+    handler.path = "/health"
+    handler.send_json = lambda body, status: observed.append((body, status))
+    handler.do_GET()
+    assert observed[0][1] == 200
+    assert observed[0][0]["status"] == "ok"
+
+
 
 def test_docs_avoid_new_broker_bearer_secret():
     doc=(

tests/test_release_preparation_supervisor_coordinator.py

@@ -0,0 +1,248 @@
+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 exact_external_ack():
+    return {
+        "account":coordinator.BROKER_ACCOUNT,
+        "capability":"governed-task:project",
+        "operation":"createGovernedTask",
+        "single_use":True,
+        "request_sha256":coordinator.EXTERNAL_REQUEST_SHA256,
+    }
+
+
+def durable_host_authorization(**overrides):
+    value={
+        "grant_id":"owner-permanent-preparation-20260819",
+        "account":coordinator.BROKER_ACCOUNT,
+        "capabilities":["governed-task:project","release:candidate"],
+        "policy_sha256":"a"*64,
+        "constraints":{
+            "repository":"aios/aios",
+            "allowed_target_prefix":"agent/",
+            "max_task_risk":"engineering",
+        },
+        "persistent_until_revoked":True,
+        "separate_owner_release_approval_still_required":True,
+    }
+    value.update(overrides)
+    return value
+
+
+def test_external_connector_requires_exact_single_use_ack():
+    with pytest.raises(coordinator.CoordinatorError,match="WAIT_EXTERNAL_ACK"):
+        coordinator.authorize_actor("external_connector")
+    wrong=exact_external_ack();wrong["request_sha256"]="0"*64
+    with pytest.raises(coordinator.CoordinatorError,match="EXTERNAL_ACK_INVALID"):
+        coordinator.authorize_actor("external_connector",external_ack=wrong)
+    decision=coordinator.authorize_actor(
+        "external_connector",external_ack=exact_external_ack()
+    )
+    assert decision["decision"]=="ALLOW_EXTERNAL_CONNECTOR_EXACT_ACK"
+    assert decision["external_ack_required"] is True
+    assert decision["request_sha256"]==coordinator.EXTERNAL_REQUEST_SHA256
+
+
+def test_host_coordinator_requires_live_durable_owner_scope_without_external_ack():
+    with pytest.raises(coordinator.CoordinatorError,match="DURABLE_OWNER_REQUIRED"):
+        coordinator.authorize_actor("credentialed_host_coordinator")
+    for denied in (
+        durable_host_authorization(capabilities=["release:candidate"]),
+        durable_host_authorization(constraints={"repository":"other/repo","allowed_target_prefix":"agent/","max_task_risk":"engineering"}),
+        durable_host_authorization(constraints={"repository":"aios/aios","allowed_target_prefix":"release/","max_task_risk":"engineering"}),
+        durable_host_authorization(constraints={"repository":"aios/aios","allowed_target_prefix":"agent/","max_task_risk":"high"}),
+        durable_host_authorization(persistent_until_revoked=False),
+    ):
+        with pytest.raises(coordinator.CoordinatorError,match="DURABLE_OWNER_DENY"):
+            coordinator.authorize_actor(
+                "credentialed_host_coordinator",host_authorization=denied
+            )
+    decision=coordinator.authorize_actor(
+        "credentialed_host_coordinator",
+        host_authorization=durable_host_authorization(),
+    )
+    assert decision["decision"]=="ALLOW_HOST_COORDINATOR_GOVERNED_PROJECTION"
+    assert decision["external_ack_required"] is False
+    assert decision["authorized_capabilities"]==["governed-task:project"]
+    assert "release:candidate" not in decision["authorized_capabilities"]
+
+
+def test_unknown_actor_is_fail_closed():
+    with pytest.raises(coordinator.CoordinatorError,match="UNKNOWN_ACTOR_DENY"):
+        coordinator.authorize_actor("unknown")
+
+
+def test_builder_request_and_amendment_preserve_fixed_binding():
+    request=coordinator.builder_request()
+    assert request["task_id"]==coordinator.BUILDER_TASK_ID
+    assert request["base_oid"]==coordinator.BUILDER_BASE_OID
+    assert request["expected_branch_oid"]==coordinator.BUILDER_BASE_OID
+    assert request["target_branch"]==coordinator.BUILDER_TARGET_BRANCH
+    assert request["generation"]==1
+    assert request["idempotency_key"]==coordinator.BUILDER_IDEMPOTENCY_KEY
+    decision=coordinator.authorize_actor(
+        "credentialed_host_coordinator",
+        host_authorization=durable_host_authorization(),
+    )
+    amendment=coordinator.gate_amendment(decision)
+    assert amendment["supervisor_task_id"]==coordinator.TASK_ID
+    assert amendment["supervisor_execution_digest"]==coordinator.SUPERVISOR_EXECUTION_DIGEST
+    assert amendment["external_connector_rule"]=="ACK_STILL_REQUIRED"
+    assert amendment["host_coordinator_rule"]=="DURABLE_OWNER_GOVERNED_TASK_PROJECT"
+    assert amendment["builder_task_id"]==request["task_id"]
+    encoded=json.dumps(amendment,sort_keys=True).lower()
+    for forbidden in ("bearer ","access_token","refresh_token","private_key"):
+        assert forbidden not in encoded
+
+
+def test_host_coordinator_projects_builder_through_broker_and_verifies_receipt(monkeypatch):
+    calls=[]
+    authorization=durable_host_authorization()
+    def fake(method,route,body=None):
+        calls.append((method,route,body))
+        if method=="GET":
+            return 200,authorization
+        request=body["request"]
+        return 200,{
+            "operation_state":"EXECUTED",
+            "capability":"governed-task:project",
+            "grant_id":authorization["grant_id"],
+            "policy_sha256":authorization["policy_sha256"],
+            "external_capability_projection_required":False,
+            "upstream_response":{
+                "task_id":request["task_id"],
+                "base_oid":request["base_oid"],
+                "expected_branch_oid":request["expected_branch_oid"],
+                "target_branch":request["target_branch"],
+                "generation":request["generation"],
+            },
+            "gate_amendment":{
+                "classification":coordinator.GATE_AMENDMENT_CLASSIFICATION,
+                "amendment_sha256":"b"*64,
+            },
+        }
+    monkeypatch.setattr(coordinator,"broker_call",fake)
+    decision=coordinator.ensure_builder_projection()
+    assert decision["builder_binding_verified"] is True
+    assert decision["external_ack_required"] is False
+    assert [item[1] for item in calls]==[
+        "/v1/capabilities","/v1/project-governed-task"
+    ]
+    assert calls[1][2]["amendment"]["owner_grant_id"]==authorization["grant_id"]
+
+
+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 "aios-owner-durable-authorization-broker.service" in coordinator_unit
+    assert "AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock" 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