fix(task-plane): admit autonomous supervisor coordinator

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

@@ -0,0 +1,288 @@
+"""Unified autonomous control-plane primitives.
+
+This is deliberately a library, not another daemon. Task API, workers and host
+recovery code consume the same Admission -> Invariant -> Repair semantics.
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from enum import Enum
+import re
+from typing import Callable, Iterable, Mapping, Sequence
+
+from aios_capability_contract import (
+    capability_contract,
+    has_capabilities,
+    normalize_required_capabilities,
+    normalize_worker_capabilities,
+    registered_capability,
+    validate_capability_contract_registry,
+)
+
+
+class FailureClass(str, Enum):
+    CAPABILITY_CONTRACT_ERROR = "CAPABILITY_CONTRACT_ERROR"
+    CREDENTIAL_PROJECTION_ERROR = "CREDENTIAL_PROJECTION_ERROR"
+    DEPENDENCY_UNAVAILABLE = "DEPENDENCY_UNAVAILABLE"
+    POLICY_DENIED = "POLICY_DENIED"
+    RESOURCE_UNAVAILABLE = "RESOURCE_UNAVAILABLE"
+    WORKER_CRASH = "WORKER_CRASH"
+    LEASE_LOST = "LEASE_LOST"
+    BUSINESS_PERMANENT = "BUSINESS_PERMANENT"
+    UNKNOWN = "UNKNOWN"
+
+
+class RepairLevel(int, Enum):
+    AUTO_LOCAL = 0
+    AUTO_REPAIR_TASK = 1
+    OWNER_APPROVAL = 2
+    MANUAL = 3
+
+
+@dataclass(frozen=True)
+class InvariantResult:
+    name: str
+    passed: bool
+    detail: str = ""
+
+    def as_dict(self) -> dict[str, object]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class AdmissionDecision:
+    admitted: bool
+    state: str
+    invariants: tuple[InvariantResult, ...]
+    failure_class: FailureClass | None = None
+    repair_level: RepairLevel | None = None
+    repair_action: str | None = None
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "schema": "aios.runtime-admission.decision.v1",
+            "admitted": self.admitted,
+            "state": self.state,
+            "invariants": [item.as_dict() for item in self.invariants],
+            "failure_class": None if self.failure_class is None else self.failure_class.value,
+            "repair_level": None if self.repair_level is None else int(self.repair_level),
+            "repair_action": self.repair_action,
+        }
+
+
+@dataclass(frozen=True)
+class RepairDecision:
+    failure_class: FailureClass
+    level: RepairLevel
+    action: str
+    consumes_business_attempt: bool
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "failure_class": self.failure_class.value,
+            "level": int(self.level),
+            "action": self.action,
+            "consumes_business_attempt": self.consumes_business_attempt,
+        }
+
+
+def classify_failure(error: object, *, phase: str = "") -> FailureClass:
+    text = (str(error) + " " + str(phase)).lower()
+    if any(token in text for token in (
+        "credential", "permission denied", "token registry", "secret", "unauthorized"
+    )):
+        return FailureClass.CREDENTIAL_PROJECTION_ERROR
+    if any(token in text for token in (
+        "transport", "unavailable", "timeout", "connection", "bad gateway", "http_5"
+    )):
+        return FailureClass.DEPENDENCY_UNAVAILABLE
+    if any(token in text for token in (
+        "capability", "eligible", "matcher", "contract"
+    )):
+        return FailureClass.CAPABILITY_CONTRACT_ERROR
+    if any(token in text for token in (
+        "policy", "forbidden", "approval required", "not authorized"
+    )):
+        return FailureClass.POLICY_DENIED
+    if any(token in text for token in ("lease lost", "lease_expired", "heartbeat_failed")):
+        return FailureClass.LEASE_LOST
+    if any(token in text for token in ("out of memory", "disk full", "resource")):
+        return FailureClass.RESOURCE_UNAVAILABLE
+    if any(token in text for token in ("worker crash", "process exited", "signal")):
+        return FailureClass.WORKER_CRASH
+    if any(token in text for token in ("invalid release", "business permanent")):
+        return FailureClass.BUSINESS_PERMANENT
+    return FailureClass.UNKNOWN
+
+
+def repair_for_failure(failure: FailureClass) -> RepairDecision:
+    table = {
+        FailureClass.CAPABILITY_CONTRACT_ERROR: RepairDecision(
+            failure, RepairLevel.AUTO_REPAIR_TASK,
+            "create_same_lineage_capability_contract_repair", False,
+        ),
+        FailureClass.CREDENTIAL_PROJECTION_ERROR: RepairDecision(
+            failure, RepairLevel.AUTO_REPAIR_TASK,
+            "restore_service_private_credential_projection", False,
+        ),
+        FailureClass.DEPENDENCY_UNAVAILABLE: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "retry_dependency_health_and_resume", False,
+        ),
+        FailureClass.RESOURCE_UNAVAILABLE: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "heal_runtime_resource_and_resume", False,
+        ),
+        FailureClass.WORKER_CRASH: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "restart_worker_and_resume_same_lineage", False,
+        ),
+        FailureClass.LEASE_LOST: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "recover_lease_and_resume_same_lineage", False,
+        ),
+        FailureClass.POLICY_DENIED: RepairDecision(
+            failure, RepairLevel.OWNER_APPROVAL,
+            "request_bound_owner_authorization", False,
+        ),
+        FailureClass.BUSINESS_PERMANENT: RepairDecision(
+            failure, RepairLevel.MANUAL,
+            "terminal_business_failure", True,
+        ),
+        FailureClass.UNKNOWN: RepairDecision(
+            failure, RepairLevel.AUTO_REPAIR_TASK,
+            "create_diagnostic_repair_task", False,
+        ),
+    }
+    return table[failure]
+
+
+def _probe(
+    name: str,
+    probes: Mapping[str, Callable[[], bool]],
+) -> InvariantResult:
+    fn = probes.get(name)
+    if fn is None:
+        return InvariantResult(name, False, "PROBE_MISSING")
+    try:
+        passed = bool(fn())
+    except Exception as exc:
+        return InvariantResult(name, False, "PROBE_ERROR:" + type(exc).__name__)
+    return InvariantResult(name, passed, "PASS" if passed else "NOT_READY")
+
+
+def runtime_admission(
+    *,
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+    requested_operations: Sequence[str] = (),
+    probes: Mapping[str, Callable[[], bool]] | None = None,
+) -> AdmissionDecision:
+    """Admission is evaluated before claim; failures do not consume attempts."""
+
+    worker_values = tuple(worker_capabilities)
+    required_values = tuple(required_capabilities)
+    probe_map = probes or {}
+    invariants: list[InvariantResult] = []
+
+    registry = validate_capability_contract_registry()
+    invariants.append(InvariantResult(
+        "capability_contract_registry",
+        bool(registry.get("valid")),
+        "PASS" if registry.get("valid") else "REGISTRY_INVALID",
+    ))
+
+    task_known = all(registered_capability(item) for item in required_values)
+    worker_known = all(registered_capability(item, worker=True) for item in worker_values)
+    invariants.extend((
+        InvariantResult("task_capabilities_registered", task_known, "PASS" if task_known else "UNKNOWN_TASK_CAPABILITY"),
+        InvariantResult("worker_capabilities_registered", worker_known, "PASS" if worker_known else "UNKNOWN_WORKER_CAPABILITY"),
+    ))
+
+    capability_match = (
+        task_known
+        and worker_known
+        and has_capabilities(worker_values, required_values)
+    )
+    invariants.append(InvariantResult(
+        "capability_match",
+        capability_match,
+        "PASS" if capability_match else "WORKER_NOT_ELIGIBLE",
+    ))
+
+    normalized = normalize_required_capabilities(required_values)
+    contracts = [
+        capability_contract(item)
+        for item in normalized
+        if not str(item).startswith("exact:")
+    ]
+    contracts = [item for item in contracts if item is not None]
+
+    forbidden: set[str] = set()
+    dependencies: set[str] = set()
+    health: set[str] = set()
+    for contract in contracts:
+        forbidden.update(contract.forbidden)
+        dependencies.update(contract.dependencies)
+        health.update(contract.health_requirements)
+
+    requested = {str(item).strip().lower() for item in requested_operations if str(item).strip()}
+    forbidden_hit = sorted(requested & {item.lower() for item in forbidden})
+    invariants.append(InvariantResult(
+        "forbidden_operations_absent",
+        not forbidden_hit,
+        "PASS" if not forbidden_hit else "FORBIDDEN:" + ",".join(forbidden_hit),
+    ))
+
+    for name in sorted(dependencies | health):
+        invariants.append(_probe(name, probe_map))
+
+    failed = [item for item in invariants if not item.passed]
+    if not failed:
+        return AdmissionDecision(
+            True,
+            "EXECUTION_READY",
+            tuple(invariants),
+        )
+
+    details = " ".join(item.name + ":" + item.detail for item in failed)
+    failure = classify_failure(details, phase="admission")
+    if any(item.name == "forbidden_operations_absent" and not item.passed for item in failed):
+        failure = FailureClass.POLICY_DENIED
+    elif any("credential" in item.name for item in failed):
+        failure = FailureClass.CREDENTIAL_PROJECTION_ERROR
+    elif any(item.name in {"capability_contract_registry", "task_capabilities_registered", "worker_capabilities_registered", "capability_match"} for item in failed):
+        failure = FailureClass.CAPABILITY_CONTRACT_ERROR
+    elif any(item.name in dependencies or item.name in health for item in failed):
+        failure = FailureClass.DEPENDENCY_UNAVAILABLE
+
+    repair = repair_for_failure(failure)
+    return AdmissionDecision(
+        False,
+        "EXECUTION_READY_FAILED",
+        tuple(invariants),
+        failure,
+        repair.level,
+        repair.action,
+    )
+
+
+def system_invariant_snapshot(
+    *,
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+    requested_operations: Sequence[str] = (),
+    probes: Mapping[str, Callable[[], bool]] | None = None,
+) -> dict[str, object]:
+    decision = runtime_admission(
+        worker_capabilities=worker_capabilities,
+        required_capabilities=required_capabilities,
+        requested_operations=requested_operations,
+        probes=probes,
+    )
+    return {
+        "schema": "aios.system-invariants.v1",
+        "status": "HEALTHY" if decision.admitted else "DEGRADED",
+        "execution_ready": decision.admitted,
+        "admission": decision.as_dict(),
+    }

active-runtime/app/aios_capability_contract.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
+from dataclasses import asdict, dataclass
+from typing import Any
 
 
 _TASK_ALIASES: dict[str, frozenset[str]] = {
@@ -37,6 +39,8 @@ _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"}),
 }
 
 _WORKER_ALIASES: dict[str, frozenset[str]] = {
@@ -71,8 +75,178 @@ _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"}),
 }
 
+
+
+@dataclass(frozen=True)
+class CapabilityContract:
+    """Versioned runtime contract attached to one canonical semantic capability."""
+
+    name: str
+    min_version: int = 1
+    max_version: int = 1
+    permissions: tuple[str, ...] = ()
+    forbidden: tuple[str, ...] = ()
+    dependencies: tuple[str, ...] = ()
+    health_requirements: tuple[str, ...] = ()
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+# This registry is the semantic source of truth. Alias tables below are retained
+# as compatibility adapters; startup/tests validate every alias target against
+# this registry so a name cannot be accepted in one layer and be unknown in
+# another.
+_CAPABILITY_CONTRACTS: dict[str, CapabilityContract] = {
+    "repo.read": CapabilityContract("repo.read", permissions=("repository_read",)),
+    "test.execute": CapabilityContract("test.execute", permissions=("test_execution",)),
+    "governed.git.write.branch": CapabilityContract(
+        "governed.git.write.branch",
+        permissions=("governed_branch_write",),
+        dependencies=("writer_lease", "governed_gitops"),
+    ),
+    "governed.git.write.file": CapabilityContract(
+        "governed.git.write.file",
+        permissions=("governed_file_write",),
+        dependencies=("writer_lease", "governed_gitops"),
+    ),
+    "governed.git.write": CapabilityContract(
+        "governed.git.write",
+        permissions=("governed_git_write",),
+        dependencies=("writer_lease", "governed_gitops"),
+    ),
+    "runtime.shell": CapabilityContract("runtime.shell", permissions=("runtime_shell",)),
+    "runtime.git": CapabilityContract("runtime.git", permissions=("runtime_git",)),
+    "runtime.python": CapabilityContract("runtime.python", permissions=("runtime_python",)),
+    "legacy.code.read": CapabilityContract("legacy.code.read"),
+    "legacy.code.review": CapabilityContract("legacy.code.review"),
+    "legacy.code.write": CapabilityContract("legacy.code.write"),
+    "legacy.code.write.patch": CapabilityContract("legacy.code.write.patch"),
+    "legacy.test.python": CapabilityContract("legacy.test.python"),
+    "legacy.test.shell.allowlisted": CapabilityContract("legacy.test.shell.allowlisted"),
+    "legacy.git.commit": CapabilityContract("legacy.git.commit"),
+    "legacy.git.diff.read": CapabilityContract("legacy.git.diff.read"),
+    "legacy.git.force_push": CapabilityContract(
+        "legacy.git.force_push",
+        forbidden=("protected_ref_bypass",),
+    ),
+    "legacy.deploy.production": CapabilityContract(
+        "legacy.deploy.production",
+        permissions=("production_deploy",),
+        dependencies=("owner_approval", "deployment_controller"),
+        health_requirements=("approval_envelope_verified", "runtime_acceptance_available"),
+    ),
+    "legacy.secret.transform": CapabilityContract("legacy.secret.transform"),
+    "legacy.secure_file_transformer": CapabilityContract("legacy.secure_file_transformer"),
+    "legacy.runtime.shadow": CapabilityContract("legacy.runtime.shadow"),
+    "legacy.review.security": CapabilityContract("legacy.review.security"),
+    "legacy.review.api-contract": CapabilityContract("legacy.review.api-contract"),
+    "legacy.gpu.train": CapabilityContract("legacy.gpu.train"),
+    "agent.runtime.hermes": CapabilityContract(
+        "agent.runtime.hermes",
+        dependencies=("hermes_runtime",),
+        health_requirements=("worker_liveness",),
+    ),
+    "agent.runtime.openclaw": CapabilityContract(
+        "agent.runtime.openclaw",
+        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,
+        max_version=1,
+        permissions=("prepare_release",),
+        forbidden=(
+            "production_deploy",
+            "submit_owner_approval",
+            "tag_publish",
+            "baseline_registry_finalize",
+            "runtime_switch",
+        ),
+        dependencies=(
+            "credential_projection",
+            "trusted_builder",
+        ),
+        health_requirements=(
+            "authenticated_probe",
+            "release_api_health",
+            "worker_liveness",
+        ),
+    ),
+}
+
+
+def capability_contract(name: object) -> CapabilityContract | None:
+    """Return the contract for one canonical or registered capability name."""
+
+    text = _clean(name)
+    direct = _CAPABILITY_CONTRACTS.get(text)
+    if direct is not None:
+        return direct
+    mapped = _TASK_ALIASES.get(text) or _WORKER_ALIASES.get(text)
+    if mapped is None or len(mapped) != 1:
+        return None
+    return _CAPABILITY_CONTRACTS.get(next(iter(mapped)))
+
+
+def registered_capability(name: object, *, worker: bool = False) -> bool:
+    """Fail-closed validation used by registration/task admission."""
+
+    text = _clean(name)
+    if "*" in text:
+        return False
+    aliases = _WORKER_ALIASES if worker else _TASK_ALIASES
+    mapped = aliases.get(text)
+    if mapped is None:
+        return text in _CAPABILITY_CONTRACTS
+    return all(item in _CAPABILITY_CONTRACTS for item in mapped)
+
+
+def validate_capability_contract_registry() -> dict[str, object]:
+    """Prove alias/contract closure and contract safety invariants."""
+
+    alias_targets: set[str] = set()
+    for mapping in (_TASK_ALIASES, _WORKER_ALIASES):
+        for values in mapping.values():
+            alias_targets.update(values)
+    missing = sorted(alias_targets - set(_CAPABILITY_CONTRACTS))
+    invalid_versions = sorted(
+        name
+        for name, contract in _CAPABILITY_CONTRACTS.items()
+        if contract.min_version < 1 or contract.max_version < contract.min_version
+    )
+    permission_forbidden_overlap = sorted(
+        name
+        for name, contract in _CAPABILITY_CONTRACTS.items()
+        if set(contract.permissions) & set(contract.forbidden)
+    )
+    return {
+        "schema": "aios.capability-contract-registry.v1",
+        "contract_count": len(_CAPABILITY_CONTRACTS),
+        "missing_alias_targets": missing,
+        "invalid_versions": invalid_versions,
+        "permission_forbidden_overlap": permission_forbidden_overlap,
+        "valid": not missing and not invalid_versions and not permission_forbidden_overlap,
+    }
+
+
+def capability_contract_snapshot() -> dict[str, dict[str, Any]]:
+    return {
+        name: contract.as_dict()
+        for name, contract in sorted(_CAPABILITY_CONTRACTS.items())
+    }
+
+
 _UNKNOWN_PREFIX = "exact:"
 
 
@@ -98,7 +272,11 @@ def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
         if "*" in text:
             result.add(_UNKNOWN_PREFIX + text)
         else:
-            result.update(_TASK_ALIASES.get(text, frozenset({_UNKNOWN_PREFIX + text})))
+            mapped = _TASK_ALIASES.get(text)
+            if mapped is None or any(item not in _CAPABILITY_CONTRACTS for item in mapped):
+                result.add(_UNKNOWN_PREFIX + text)
+            else:
+                result.update(mapped)
     return frozenset(result)
 
 
@@ -111,7 +289,7 @@ def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
         if "*" in text:
             continue
         mapped = _WORKER_ALIASES.get(text)
-        if mapped is not None:
+        if mapped is not None and all(item in _CAPABILITY_CONTRACTS for item in mapped):
             result.update(mapped)
     if {
         "governed.git.write.branch",

active-runtime/app/aios_release_lineage_contract.py

@@ -0,0 +1,82 @@
+"""Immutable release-lineage contracts used by activation and recovery."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any, Mapping
+
+
+@dataclass(frozen=True)
+class ReleaseLineage:
+    task_id: str
+    candidate_id: str
+    source_commit: str
+    version: str
+    required_capabilities: tuple[str, ...]
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+FIXED_V2R2C2 = ReleaseLineage(
+    task_id="r42-release-candidate-v2r2c2-live-20260819",
+    candidate_id="r42-release-candidate-v2r2c2-20260819",
+    source_commit="71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+    version="V2R2C2",
+    required_capabilities=("release:candidate",),
+)
+
+
+class ReleaseLineageError(RuntimeError):
+    pass
+
+
+def validate_fixed_v2r2c2_task(task: Mapping[str, Any]) -> dict[str, Any]:
+    """Validate identity/source/version without mutating Task state.
+
+    This contract intentionally does not bind status/attempt/revision, because
+    those are lifecycle state and may change through governed retry/recovery.
+    """
+
+    if str(task.get("task_id") or "") != FIXED_V2R2C2.task_id:
+        raise ReleaseLineageError("FIXED_TASK_ID_DRIFT")
+
+    capabilities = task.get("required_capabilities")
+    if (
+        not isinstance(capabilities, list)
+        or tuple(capabilities) != FIXED_V2R2C2.required_capabilities
+    ):
+        raise ReleaseLineageError("FIXED_TASK_CAPABILITY_DRIFT")
+
+    constraints = task.get("constraints")
+    if not isinstance(constraints, Mapping):
+        raise ReleaseLineageError("FIXED_TASK_CONSTRAINTS_MISSING")
+    release = constraints.get("release_candidate")
+    if not isinstance(release, Mapping):
+        raise ReleaseLineageError("FIXED_RELEASE_BINDING_MISSING")
+
+    actual = {
+        "candidate_id": str(release.get("candidate_id") or ""),
+        "source_commit": str(release.get("source_commit") or ""),
+        "version": str(release.get("version") or ""),
+    }
+    expected = {
+        "candidate_id": FIXED_V2R2C2.candidate_id,
+        "source_commit": FIXED_V2R2C2.source_commit,
+        "version": FIXED_V2R2C2.version,
+    }
+    if actual != expected:
+        raise ReleaseLineageError(
+            "FIXED_RELEASE_LINEAGE_DRIFT:"
+            + repr({"expected": expected, "actual": actual})
+        )
+
+    return {
+        "schema": "aios.release-lineage.validation.v1",
+        "status": "VERIFIED",
+        "lineage": FIXED_V2R2C2.as_dict(),
+        "task_status": task.get("status"),
+        "attempt": task.get("attempt"),
+        "max_attempts": task.get("max_attempts"),
+        "revision": task.get("revision"),
+    }

active-runtime/app/aios_release_online_agent_worker.py

@@ -0,0 +1,714 @@
+#!/usr/bin/env python3
+"""Persistent release-preparation Online Agent worker.
+
+The worker is intentionally narrower than the generic engineering worker:
+it can only consume tasks whose required capability set is exactly
+{"release:candidate"} and whose payload asks for immutable release-candidate
+creation. It never receives Owner signing, deployment, tag, baseline or Git
+write credentials.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import threading
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Mapping
+from aios_autonomous_control_plane import (
+    classify_failure,
+    repair_for_failure,
+    runtime_admission,
+)
+
+AGENT_ID = os.environ.get(
+    "AIOS_RELEASE_WORKER_AGENT_ID",
+    "online-release-worker-cn-network-mkt-20260819",
+).strip()
+AGENT_VERSION = os.environ.get(
+    "AIOS_RELEASE_WORKER_AGENT_VERSION",
+    "2026-08-19.1-release-candidate",
+).strip()
+TASK_API = os.environ.get(
+    "AIOS_AGENT_TASK_API_URL",
+    "http://127.0.0.1:18940",
+).rstrip("/")
+RELEASE_API = os.environ.get("AIOS_RELEASE_API_URL", "").rstrip("/")
+POLL_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_POLL_SECONDS", "3"))
+LEASE_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_LEASE_SECONDS", "60"))
+HEARTBEAT_SECONDS = int(
+    os.environ.get("AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS", "15")
+)
+STATE_ROOT = Path(
+    os.environ.get(
+        "AIOS_RELEASE_WORKER_STATE_ROOT",
+        "/var/lib/aios-online-agent-release",
+    )
+)
+EXPECTED_CAPABILITIES = frozenset({"release:candidate"})
+LIVENESS_PATH = Path(
+    os.environ.get(
+        "AIOS_WORKER_LIVENESS_PATH",
+        str(STATE_ROOT / "worker-liveness.json"),
+    )
+)
+CANDIDATE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,255}$")
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+VERSION_RE = re.compile(r"^V[0-9]+R[0-9]+C[0-9]+$")
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+SECRET_MARKERS = (
+    "authorization",
+    "token",
+    "secret",
+    "credential",
+    "password",
+    "private_key",
+    "signature",
+)
+
+
+class ReleaseWorkerError(RuntimeError):
+    def __init__(self, code: str, *, phase: str) -> None:
+        super().__init__(code)
+        self.code = code
+        self.phase = phase
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def sha256(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def sanitized(value: Any) -> Any:
+    if isinstance(value, Mapping):
+        result: dict[str, Any] = {}
+        for key, inner in value.items():
+            lowered = str(key).lower()
+            if any(marker in lowered for marker in SECRET_MARKERS):
+                result[str(key)] = "[REDACTED]"
+            else:
+                result[str(key)] = sanitized(inner)
+        return result
+    if isinstance(value, list):
+        return [sanitized(item) for item in value]
+    return value
+
+
+def audit(event: str, **fields: Any) -> None:
+    STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
+    path = STATE_ROOT / "events.jsonl"
+    record = {
+        "schema": "aios.release-worker.event.v1",
+        "event": event,
+        "created_at": now(),
+        "agent_id": AGENT_ID,
+        **sanitized(fields),
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, sort_keys=True) + "\n")
+    os.chmod(path, 0o640)
+
+
+def sd_notify(message: str) -> None:
+    target = os.environ.get("NOTIFY_SOCKET", "")
+    if not target:
+        return
+    if target.startswith("@"):
+        address = "\0" + target[1:]
+    else:
+        address = target
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    try:
+        sock.connect(address)
+        sock.sendall(message.encode("utf-8"))
+    except OSError:
+        # Liveness JSON remains the diagnostic source; systemd will enforce
+        # WatchdogSec if notify transport is persistently broken.
+        pass
+    finally:
+        sock.close()
+
+
+def write_liveness(
+    state: str,
+    *,
+    task_id: str | None = None,
+    stage: str | None = None,
+    error: str | None = None,
+    visible_count: int | None = None,
+) -> None:
+    STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
+    value = {
+        "schema": "aios.online-agent.worker-liveness.v1",
+        "agent_id": AGENT_ID,
+        "agent_type": "release-preparation-worker",
+        "agent_version": AGENT_VERSION,
+        "pid": os.getpid(),
+        "state": state,
+        "last_seen_epoch": time.time(),
+        "last_seen_at": now(),
+        "poll_seconds": POLL_SECONDS,
+        "heartbeat_seconds": HEARTBEAT_SECONDS,
+        "api_base": TASK_API,
+        "capabilities": sorted(EXPECTED_CAPABILITIES),
+        "task_id": task_id,
+        "stage": stage,
+        "error": error,
+        "visible_count": visible_count,
+    }
+    LIVENESS_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+    temporary = LIVENESS_PATH.with_name(
+        "." + LIVENESS_PATH.name + f".{os.getpid()}.tmp"
+    )
+    temporary.write_text(
+        json.dumps(value, ensure_ascii=False, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    os.chmod(temporary, 0o640)
+    os.replace(temporary, LIVENESS_PATH)
+
+
+def load_credential(name: str) -> str:
+    directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
+    path = directory / name
+    if path.is_symlink() or not path.is_file():
+        raise ReleaseWorkerError(
+            "CREDENTIAL_UNAVAILABLE:" + name,
+            phase="startup",
+        )
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 24 or any(ch.isspace() for ch in value):
+        raise ReleaseWorkerError(
+            "CREDENTIAL_INVALID:" + name,
+            phase="startup",
+        )
+    return value
+
+
+def http_json(
+    base: str,
+    method: str,
+    route: str,
+    token: str,
+    body: Mapping[str, Any] | None = None,
+    *,
+    timeout: int = 30,
+) -> tuple[int, dict[str, Any]]:
+    data = None if body is None else canonical(body)
+    headers = {"Accept": "application/json"}
+    if token:
+        headers["Authorization"] = "Bearer " + token
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(
+        base + route,
+        method=method,
+        data=data,
+        headers=headers,
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            raw, status = response.read(), response.status
+    except urllib.error.HTTPError as exc:
+        raw, status = exc.read(), exc.code
+    except (urllib.error.URLError, TimeoutError) as exc:
+        raise ReleaseWorkerError(
+            "HTTP_TRANSPORT_UNAVAILABLE",
+            phase="network",
+        ) from exc
+    try:
+        value = json.loads(raw.decode())
+    except Exception as exc:
+        raise ReleaseWorkerError(
+            "HTTP_RESPONSE_INVALID",
+            phase="network",
+        ) from exc
+    return status, value if isinstance(value, dict) else {}
+
+
+@dataclass(frozen=True)
+class ReleaseRequest:
+    candidate_id: str
+    source_commit: str
+    version: str
+    projection_id: str | None
+
+    @classmethod
+    def from_task(cls, task: Mapping[str, Any]) -> "ReleaseRequest":
+        if not TASK_RE.fullmatch(str(task.get("task_id") or "")):
+            raise ReleaseWorkerError("TASK_ID_INVALID", phase="validation")
+        caps = task.get("required_capabilities")
+        if (
+            not isinstance(caps, list)
+            or frozenset(caps) != EXPECTED_CAPABILITIES
+            or len(caps) != 1
+        ):
+            raise ReleaseWorkerError(
+                "TASK_CAPABILITY_SCOPE_INVALID",
+                phase="validation",
+            )
+        metadata = task.get("metadata")
+        if not isinstance(metadata, Mapping):
+            metadata = {}
+        constraints = task.get("constraints")
+        if not isinstance(constraints, Mapping):
+            constraints = {}
+        release = constraints.get("release_candidate")
+        if not isinstance(release, Mapping):
+            release = metadata.get("release_candidate")
+        if not isinstance(release, Mapping):
+            release = task.get("release_candidate")
+        if not isinstance(release, Mapping):
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_PAYLOAD_MISSING",
+                phase="validation",
+            )
+        allowed = {
+            "candidate_id",
+            "source_commit",
+            "version",
+            "projection_id",
+        }
+        if set(release) - allowed:
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_PAYLOAD_FIELDS_INVALID",
+                phase="validation",
+            )
+        candidate_id = str(release.get("candidate_id") or "")
+        source_commit = str(release.get("source_commit") or "")
+        version = str(release.get("version") or "")
+        projection_id = release.get("projection_id")
+        if CANDIDATE_RE.fullmatch(candidate_id) is None:
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_ID_INVALID",
+                phase="validation",
+            )
+        if OID_RE.fullmatch(source_commit) is None:
+            raise ReleaseWorkerError(
+                "RELEASE_SOURCE_COMMIT_INVALID",
+                phase="validation",
+            )
+        if VERSION_RE.fullmatch(version) is None:
+            raise ReleaseWorkerError(
+                "RELEASE_VERSION_INVALID",
+                phase="validation",
+            )
+        if projection_id is not None:
+            projection_id = str(projection_id)
+            if not re.fullmatch(r"projection-[0-9a-f]{32}", projection_id):
+                raise ReleaseWorkerError(
+                    "RELEASE_PROJECTION_ID_INVALID",
+                    phase="validation",
+                )
+        return cls(
+            candidate_id=candidate_id,
+            source_commit=source_commit,
+            version=version,
+            projection_id=projection_id,
+        )
+
+    def body(self) -> dict[str, str]:
+        return {
+            "candidate_id": self.candidate_id,
+            "source_commit": self.source_commit,
+            "version": self.version,
+        }
+
+
+class Heartbeat:
+    def __init__(
+        self,
+        agent_token: str,
+        task_id: str,
+        lease_token: str,
+    ) -> None:
+        self.agent_token = agent_token
+        self.task_id = task_id
+        self.lease_token = lease_token
+        self.stage = "claimed"
+        self.failure: str | None = None
+        self.stop = threading.Event()
+        self.thread = threading.Thread(
+            target=self._run,
+            name="release-worker-heartbeat",
+            daemon=True,
+        )
+
+    def first(self) -> None:
+        status, value = http_json(
+            TASK_API,
+            "POST",
+            f"/v1/tasks/{self.task_id}/heartbeat",
+            self.agent_token,
+            {
+                "lease_token": self.lease_token,
+                "lease_seconds": LEASE_SECONDS,
+                "progress": {
+                    "stage": self.stage,
+                    "worker_version": AGENT_VERSION,
+                },
+            },
+        )
+        if status != 200:
+            raise ReleaseWorkerError(
+                "TASK_HEARTBEAT_FAILED",
+                phase="heartbeat",
+            )
+        audit(
+            "TASK_HEARTBEAT",
+            task_id=self.task_id,
+            stage=self.stage,
+            api_status=status,
+        )
+        write_liveness(
+            "BUSY",
+            task_id=self.task_id,
+            stage=self.stage,
+        )
+        sd_notify("WATCHDOG=1")
+
+    def _run(self) -> None:
+        while not self.stop.wait(HEARTBEAT_SECONDS):
+            try:
+                status, value = http_json(
+                    TASK_API,
+                    "POST",
+                    f"/v1/tasks/{self.task_id}/heartbeat",
+                    self.agent_token,
+                    {
+                        "lease_token": self.lease_token,
+                        "lease_seconds": LEASE_SECONDS,
+                        "progress": {
+                            "stage": self.stage,
+                            "worker_version": AGENT_VERSION,
+                        },
+                    },
+                )
+                if status != 200:
+                    self.failure = str(value.get("error") or status)
+                    return
+                audit(
+                    "TASK_HEARTBEAT",
+                    task_id=self.task_id,
+                    stage=self.stage,
+                    api_status=status,
+                )
+                write_liveness(
+                    "BUSY",
+                    task_id=self.task_id,
+                    stage=self.stage,
+                )
+                sd_notify("WATCHDOG=1")
+            except Exception as exc:
+                self.failure = type(exc).__name__
+                return
+
+    def ensure(self) -> None:
+        if self.failure:
+            raise ReleaseWorkerError(
+                "TASK_LEASE_LOST",
+                phase="heartbeat",
+            )
+
+    def close(self) -> None:
+        self.stop.set()
+        if self.thread.ident is not None:
+            self.thread.join(timeout=2)
+
+
+def release_admission(
+    task: Mapping[str, Any],
+    *,
+    release_token: str,
+):
+    required = task.get("required_capabilities")
+    if not isinstance(required, list):
+        required = []
+
+    def release_api_health() -> bool:
+        status, value = http_json(
+            RELEASE_API,
+            "GET",
+            "/health",
+            "",
+            timeout=10,
+        )
+        return status == 200 and str(value.get("status") or "").lower() in {"ok", "healthy"}
+
+    def authenticated_probe() -> bool:
+        status, _ = http_json(
+            RELEASE_API,
+            "POST",
+            "/createReleaseCandidate",
+            release_token,
+            {},
+            timeout=10,
+        )
+        # Authentication + credential projection succeeded; the empty release
+        # request is intentionally rejected by request validation.
+        return status == 400
+
+    probes = {
+        "credential_projection": authenticated_probe,
+        "trusted_builder": release_api_health,
+        "authenticated_probe": authenticated_probe,
+        "release_api_health": release_api_health,
+        "worker_liveness": lambda: True,
+    }
+    return runtime_admission(
+        worker_capabilities=EXPECTED_CAPABILITIES,
+        required_capabilities=required,
+        requested_operations=("prepare_release",),
+        probes=probes,
+    )
+
+
+def result_payload(
+    task_id: str,
+    release: ReleaseRequest,
+    builder: Mapping[str, Any],
+) -> dict[str, Any]:
+    builder_candidate = builder.get("candidate_id")
+    if builder_candidate not in {None, release.candidate_id}:
+        raise ReleaseWorkerError(
+            "BUILDER_CANDIDATE_BINDING_INVALID",
+            phase="release",
+        )
+    immutable = {
+        "candidate_id": release.candidate_id,
+        "source_commit": release.source_commit,
+        "version": release.version,
+    }
+    return {
+        "status": "SUCCEEDED",
+        "classification": "RELEASE_CANDIDATE_CREATED",
+        "execution_mode": "release_candidate",
+        "adapter": "release_candidate",
+        "task_id": task_id,
+        "agent_id": AGENT_ID,
+        "release_candidate": immutable,
+        "release_candidate_request_sha256": sha256(canonical(immutable)),
+        "builder_response": sanitized(dict(builder)),
+        "projection_id": release.projection_id,
+        "writer_lease": "NONE",
+        "changed_files": [],
+        "direct_main_write": False,
+        "production_write": False,
+        "tag_write": False,
+        "baseline_write": False,
+        "runtime_switch": False,
+        "owner_approval_submitted": False,
+        "credential_material_exposed": False,
+    }
+
+
+def execute_task(
+    task: Mapping[str, Any],
+    *,
+    agent_token: str,
+    release_token: str,
+) -> bool:
+    task_id = str(task.get("task_id") or "")
+    release = ReleaseRequest.from_task(task)
+    admission = release_admission(
+        task,
+        release_token=release_token,
+    )
+    if not admission.admitted:
+        audit(
+            "TASK_ADMISSION_BLOCKED",
+            task_id=task_id,
+            admission=admission.as_dict(),
+        )
+        write_liveness(
+            "BLOCKED",
+            task_id=task_id,
+            stage="execution_ready",
+            error=admission.failure_class.value if admission.failure_class else "ADMISSION_FAILED",
+        )
+        return False
+    audit(
+        "TASK_ADMISSION_PASSED",
+        task_id=task_id,
+        admission=admission.as_dict(),
+    )
+    status, claimed = http_json(
+        TASK_API,
+        "POST",
+        f"/v1/tasks/{task_id}/claim",
+        agent_token,
+        {"lease_seconds": LEASE_SECONDS},
+    )
+    if status in {409, 423}:
+        return False
+    if status != 200:
+        raise ReleaseWorkerError(
+            "TASK_CLAIM_FAILED",
+            phase="claim",
+        )
+    lease_token = str(claimed.get("lease_token") or "")
+    if len(lease_token) < 16:
+        raise ReleaseWorkerError(
+            "TASK_LEASE_TOKEN_INVALID",
+            phase="claim",
+        )
+    attempt = int(claimed.get("attempt") or 0)
+    audit(
+        "TASK_CLAIMED",
+        task_id=task_id,
+        attempt=attempt,
+    )
+
+    heartbeat = Heartbeat(agent_token, task_id, lease_token)
+    heartbeat.first()
+    heartbeat.thread.start()
+    try:
+        heartbeat.stage = "release_candidate"
+        heartbeat.ensure()
+        status, builder = http_json(
+            RELEASE_API,
+            "POST",
+            "/createReleaseCandidate",
+            release_token,
+            release.body(),
+            timeout=120,
+        )
+        if status not in {200, 201, 202}:
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_CREATE_FAILED:"
+                + str(builder.get("error") or status),
+                phase="release",
+            )
+        heartbeat.ensure()
+        result = result_payload(task_id, release, builder)
+        idempotency_key = (
+            f"{task_id}-attempt-{attempt}-release-result-v1"
+        )
+        status, submitted = http_json(
+            TASK_API,
+            "POST",
+            f"/v1/tasks/{task_id}/result",
+            agent_token,
+            {
+                "lease_token": lease_token,
+                "idempotency_key": idempotency_key,
+                "result": result,
+            },
+        )
+        if status not in {200, 201, 202}:
+            raise ReleaseWorkerError(
+                "TASK_RESULT_SUBMIT_FAILED",
+                phase="result",
+            )
+        audit(
+            "RESULT_SUBMITTED",
+            task_id=task_id,
+            attempt=attempt,
+            release_candidate_id=release.candidate_id,
+        )
+        return True
+    finally:
+        heartbeat.close()
+
+
+def main() -> int:
+    if not RELEASE_API:
+        raise SystemExit("FATAL:AIOS_RELEASE_API_URL_REQUIRED")
+    agent_token = load_credential("agent_token")
+    release_token = load_credential("release_candidate_token")
+    STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
+    audit(
+        "WORKER_STARTED",
+        capabilities=sorted(EXPECTED_CAPABILITIES),
+        release_api=RELEASE_API,
+    )
+    write_liveness("STARTING", stage="startup")
+    sd_notify("READY=1\nSTATUS=release worker ready")
+    while True:
+        try:
+            status, response = http_json(
+                TASK_API,
+                "POST",
+                "/v1/tasks/poll",
+                agent_token,
+                {"limit": 100},
+            )
+            if status != 200:
+                raise ReleaseWorkerError(
+                    "TASK_POLL_FAILED",
+                    phase="poll",
+                )
+            tasks = response.get("tasks")
+            if not isinstance(tasks, list):
+                tasks = []
+            write_liveness(
+                "IDLE",
+                stage="poll",
+                visible_count=len(tasks),
+            )
+            sd_notify("WATCHDOG=1\nSTATUS=release worker poll ok")
+            handled = False
+            for raw in tasks:
+                if not isinstance(raw, Mapping):
+                    continue
+                caps = raw.get("required_capabilities")
+                if (
+                    not isinstance(caps, list)
+                    or frozenset(caps) != EXPECTED_CAPABILITIES
+                ):
+                    continue
+                try:
+                    handled = execute_task(
+                        raw,
+                        agent_token=agent_token,
+                        release_token=release_token,
+                    ) or handled
+                except ReleaseWorkerError as exc:
+                    failure = classify_failure(exc.code, phase=exc.phase)
+                    repair = repair_for_failure(failure)
+                    audit(
+                        "TASK_EXECUTION_FAILED",
+                        task_id=raw.get("task_id"),
+                        phase=exc.phase,
+                        error=exc.code,
+                        failure_class=failure.value,
+                        repair=repair.as_dict(),
+                    )
+            time.sleep(0 if handled else POLL_SECONDS)
+        except KeyboardInterrupt:
+            return 0
+        except Exception as exc:
+            audit(
+                "WORKER_LOOP_ERROR",
+                error=type(exc).__name__,
+            )
+            write_liveness(
+                "DEGRADED",
+                stage="loop",
+                error=type(exc).__name__,
+            )
+            time.sleep(POLL_SECONDS)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/aios_release_preparation_supervisor_coordinator.py

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

deploy/systemd/aios-agent-task-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-release-online-agent-worker.service

@@ -0,0 +1,50 @@
+[Unit]
+Description=AIOS release-candidate Online Agent worker
+After=network.target aios-agent-task-api.service
+Wants=aios-agent-task-api.service
+
+[Service]
+Type=notify
+NotifyAccess=main
+WatchdogSec=90
+User=root
+Group=root
+StateDirectory=aios-online-agent-release
+StateDirectoryMode=0750
+LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-release-online-agent-worker/agent_token
+LoadCredentialEncrypted=release_candidate_token:/etc/credstore.encrypted/aios-release-online-agent-worker/release_candidate_token
+Environment=AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819
+Environment=AIOS_RELEASE_WORKER_AGENT_VERSION=2026-08-19.1-release-candidate
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_RELEASE_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
+Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
+Environment=AIOS_RELEASE_WORKER_POLL_SECONDS=3
+Environment=AIOS_RELEASE_WORKER_LEASE_SECONDS=60
+Environment=AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS=15
+Environment=AIOS_RELEASE_API_URL=http://127.0.0.1:18890
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_release_online_agent_worker.py
+Restart=always
+RestartSec=2
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+LockPersonality=yes
+RestrictSUIDSGID=yes
+RestrictRealtime=yes
+RestrictNamespaces=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=127.0.0.0/8
+InaccessiblePaths=/data/aios-gitops /home
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

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

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

docs/superpowers/specs/2026-08-19-autonomous-control-plane.md

@@ -0,0 +1,210 @@
+# AIOS Autonomous Control Plane — Admission + Invariant + Repair
+
+## Why this exists
+
+Recent Agent registration, Task claim, release-worker, credential and retry failures shared one
+failure mode:
+
+> The control plane accepted an intent before the runtime had proved it could execute the intent.
+
+This design does not add another fleet of daemons. It adds one shared library contract that Task API,
+workers and governed host recovery consume.
+
+## Three layers
+
+```text
+Intent
+  ↓
+Runtime Admission
+  ↓
+System Invariants
+  ↓
+Repair Policy
+  ↓
+Execution
+```
+
+### Runtime Admission
+
+Admission occurs before Task claim.
+
+A capability such as `release:candidate` is not considered executable merely because its name
+matches. The capability contract also binds:
+
+- supported contract version;
+- allowed permission;
+- forbidden operations;
+- runtime dependencies;
+- health requirements.
+
+For `release:candidate`:
+
+```text
+permission:
+  prepare_release
+
+forbidden:
+  production_deploy
+  submit_owner_approval
+  tag_publish
+  baseline_registry_finalize
+  runtime_switch
+
+dependencies:
+  credential_projection
+  trusted_builder
+
+health:
+  authenticated_probe
+  release_api_health
+  worker_liveness
+```
+
+Only a fully passing decision reaches:
+
+```text
+EXECUTION_READY
+```
+
+A failed decision remains:
+
+```text
+EXECUTION_READY_FAILED
+```
+
+and the worker does not call `/claim`.
+
+Therefore an infrastructure defect discovered by admission does not consume a business attempt.
+
+### System Invariants
+
+The same admission checks are exposed as a machine-readable invariant snapshot.
+
+Examples:
+
+```text
+capability_contract_registry
+task_capabilities_registered
+worker_capabilities_registered
+capability_match
+forbidden_operations_absent
+credential_projection
+trusted_builder
+authenticated_probe
+release_api_health
+worker_liveness
+```
+
+A runtime can expose:
+
+```text
+HEALTHY
+DEGRADED
+```
+
+without relying on a human reconstructing state from logs.
+
+### Repair Policy
+
+Failure classification maps to bounded recovery:
+
+```text
+DEPENDENCY_UNAVAILABLE       → level 0 / auto local
+RESOURCE_UNAVAILABLE         → level 0 / auto local
+WORKER_CRASH                 → level 0 / auto local
+LEASE_LOST                   → level 0 / auto local
+
+CAPABILITY_CONTRACT_ERROR    → level 1 / same-lineage repair task
+CREDENTIAL_PROJECTION_ERROR  → level 1 / same-lineage repair task
+UNKNOWN                      → level 1 / diagnostic repair task
+
+POLICY_DENIED                → level 2 / Owner authorization
+BUSINESS_PERMANENT           → level 3 / terminal/manual decision
+```
+
+Only business-permanent failure consumes the business retry budget in this first contract.
+
+The current Task API does not yet have a first-class `EXECUTION_READY_FAILED` durable state. Until it
+does, the Release Worker enforces the most important invariant by refusing to claim a Task whose
+admission fails.
+
+## Capability Contract Registry
+
+`aios_capability_contract.py` remains backward-compatible with existing aliases but now validates
+every alias target against a versioned semantic contract registry.
+
+Unknown capabilities and wildcard capabilities remain fail-closed.
+
+Agent registration and Task creation should next be changed to call `registered_capability()` so a
+capability cannot be accepted at registration and rejected only later by the matcher.
+
+## Credential rule
+
+Admission probes credential usability, not merely credential existence.
+
+For release candidate creation the authenticated probe intentionally sends an invalid empty
+`/createReleaseCandidate` request.
+
+Expected:
+
+```text
+HTTP 400
+```
+
+This proves:
+
+```text
+credential readable
+→ authentication succeeded
+→ scope accepted
+→ request validation rejected empty business input
+```
+
+A permission failure/401/403/5xx is `EXECUTION_READY_FAILED`.
+
+## Lineage
+
+Repair actions are descriptive, not permission grants. Any repair Task must retain the existing
+root/parent/attempt/authority chain and use governed same-lineage recovery.
+
+No repair action may silently create a replacement business Task.
+
+## Authority separation
+
+Capability answers:
+
+```text
+can this runtime perform the operation?
+```
+
+Authority answers:
+
+```text
+is this operation currently authorized?
+```
+
+`release:candidate` never implies Owner approval, deployment, tag publication, baseline finalization
+or runtime switch.
+
+## Deployment sequence
+
+Phase 1 (this candidate):
+
+- Capability Contract Registry;
+- shared Autonomous Control Plane library;
+- Release Worker pre-claim admission;
+- failure classification + repair evidence.
+
+Phase 2:
+
+- Task API uses the same library before returning Tasks from poll and again during claim CAS;
+- Task schema gains durable `ADMITTED/EXECUTION_READY/EXECUTION_READY_FAILED` evidence without
+  counting pre-side-effect infrastructure failures against business attempts.
+
+Phase 3:
+
+- Agent registration validates against the same Capability Contract Registry;
+- external connector writes only durable Intent/Command Inbox records and leaves execution to the
+  local control plane.
+
+This sequencing avoids a second controller stack while converging every decision onto one contract.

docs/superpowers/specs/2026-08-19-release-candidate-capability-contract.md

@@ -0,0 +1,87 @@
+# Release candidate capability contract remediation
+
+## Problem
+
+The release-preparation worker can be correctly registered with the raw capability:
+
+```text
+release:candidate
+```
+
+while the canonical Task API matcher still rejects the worker.
+
+Before this repair, the capability matrix is:
+
+```text
+worker raw      = ["release:candidate"]
+worker effective= []
+task required   = ["exact:release:candidate"]
+eligible        = false
+```
+
+The Task therefore remains `QUEUED`, `attempt=0`, and cannot be claimed even though both the Task and
+worker use the same literal capability.
+
+## Root cause
+
+`aios_capability_contract.py` is deliberately fail-closed:
+
+- task-side unknown names become `exact:<name>`;
+- worker-side unknown names are discarded;
+- wildcard worker capabilities are ignored.
+
+`release:candidate` was absent from both canonical alias tables.
+
+## Repair
+
+Add exactly one semantic capability to both vocabularies:
+
+```text
+release:candidate → release:candidate
+```
+
+Do not add:
+
+```text
+release:*
+submitOwnerApproval
+startDeployment
+tag_publish
+baseline_registry_finalize
+runtime_switch
+```
+
+Unknown names and wildcards must continue to fail closed.
+
+## Runtime recovery
+
+After the governed remediation candidate is officially verified:
+
+1. install the candidate capability-contract module as a higher-precedence Task API overlay;
+2. restart only `aios-agent-task-api.service`;
+3. require `/health` success;
+4. require the fixed existing release Task to remain the same Task/idempotency;
+5. let the already-active dedicated release worker poll and claim it automatically;
+6. require heartbeat, trusted-builder result and official verification;
+7. do not manually claim or create a replacement live Task.
+
+The fixed V2R2C2 live Task may complete as the acceptance candidate it was originally bound to.
+Because this remediation is a later source commit, the first release candidate that includes the
+matcher repair should use the next immutable version (for example `V2R2C3`) rather than rewriting an
+already-created V2R2C2 candidate.
+
+## Governance boundary
+
+This repair affects only Task eligibility. It grants no credential and performs no release action by
+itself.
+
+The dedicated release worker still requires its independent `release:candidate` credential, and the
+following remain separate:
+
+```text
+Owner signed approval
+deployment
+tag publication
+baseline finalization
+runtime switch
+```

docs/superpowers/specs/2026-08-19-release-lineage-contract.md

@@ -0,0 +1,82 @@
+# Fixed V2R2C2 release lineage contract
+
+## Why this contract exists
+
+AIOS has multiple engineering commits in the same ancestry, and they serve different purposes.
+
+The fixed V2R2C2 live Task was created by the dedicated Release Worker activation. Its immutable
+release source is therefore the Release Worker implementation candidate:
+
+```text
+71e736b05fab13efc06af7a4fe2dc022c1cf9a85
+```
+
+Later engineering commits are descendants but are not retroactive V2R2C2 release sources:
+
+```text
+17deb91ca7a187dc81766945b1174e854f86bd25
+  capability-contract remediation
+
+b4b4416c93fedbe451d655d49e28f915b8fd38e6
+  Autonomous Control Plane engineering candidate
+```
+
+Conflating "current engineering base" with "immutable release source" caused ACP activation to
+fail closed with `FIXED_RELEASE_SOURCE_DRIFT`.
+
+## Canonical fixed lineage
+
+```text
+Task:
+r42-release-candidate-v2r2c2-live-20260819
+
+candidate_id:
+r42-release-candidate-v2r2c2-20260819
+
+source_commit:
+71e736b05fab13efc06af7a4fe2dc022c1cf9a85
+
+version:
+V2R2C2
+
+required_capabilities:
+["release:candidate"]
+```
+
+This contract is immutable for V2R2C2.
+
+## What may change
+
+The following are lifecycle state, not release identity:
+
+```text
+status
+attempt
+max_attempts
+revision
+claimed_by
+lease
+verification
+```
+
+They may change through official retry/claim/result/verification without changing the release
+lineage.
+
+## What must never be "repaired"
+
+Do not rewrite the fixed V2R2C2 Task source to a later engineering commit.
+
+A later engineering source that must itself be released receives a later immutable release version
+(for example V2R2C3), not a mutation of V2R2C2.
+
+## Activation invariant
+
+ACP activation must:
+
+1. verify its own governed engineering candidate separately;
+2. read the live fixed V2R2C2 Task;
+3. validate that Task through `validate_fixed_v2r2c2_task()`;
+4. never compare the fixed release source to the ACP engineering base/candidate;
+5. never mutate the fixed Task merely to satisfy activation.
+
+This makes engineering lineage and release lineage explicit and independently verifiable.

docs/superpowers/specs/2026-08-19-release-preparation-online-agent.md

@@ -0,0 +1,142 @@
+# Release Preparation Online Agent
+
+## Purpose
+
+This worker closes the release-preparation execution gap without widening the ChatGPT connector
+identity.
+
+The dedicated Agent identity is:
+
+```text
+online-release-worker-cn-network-mkt-20260819
+```
+
+Its only advertised task capability is:
+
+```text
+release:candidate
+```
+
+The worker may create an immutable release candidate, but it never submits Owner approval and never
+starts deployment, publishes a tag, finalizes a baseline, switches runtime or writes Git.
+
+## Credentials
+
+The worker uses two independently scoped credentials:
+
+```text
+agent_token
+release_candidate_token
+```
+
+`agent_token` is the Online Agent registration credential. It must be encrypted into systemd's
+credential store and must never be written to Git, Task metadata, evidence or logs.
+
+`release_candidate_token` is provisioned during Owner-authorized runtime activation. Only its SHA-256 record is stored in the release API token registry; the plaintext is immediately encrypted into systemd credstore and the temporary plaintext is removed. Its only allowed scope is:
+
+```text
+release:candidate
+```
+
+It is explicitly not an Owner signing key and must not grant:
+
+```text
+submitOwnerApproval
+startDeployment
+tag_publish
+baseline_registry_finalize
+runtime_switch
+```
+
+The unit reads both credentials through systemd `LoadCredentialEncrypted`; neither plaintext token is kept in the unit, repository, Task metadata, evidence, or logs.
+
+## Task contract
+
+The worker will only claim Tasks whose exact required capability set is:
+
+```json
+["release:candidate"]
+```
+
+The Task must provide a bounded `constraints.release_candidate` payload (metadata is accepted only for compatibility):
+
+```json
+{
+  "candidate_id": "...",
+  "source_commit": "<40 hex>",
+  "version": "V2R2C2",
+  "projection_id": "projection-..."
+}
+```
+
+No caller-supplied URL, method, headers, command, shell, environment, credential path, deployment
+operation or arbitrary release route is accepted.
+
+The release API base is host configuration, not Task input. The only release mutation invoked by the
+worker is:
+
+```text
+POST /createReleaseCandidate
+```
+
+## Lifecycle
+
+```text
+poll
+→ claim
+→ heartbeat
+→ createReleaseCandidate
+→ result submit
+→ official verifier
+```
+
+The result declares:
+
+```text
+writer_lease=NONE
+changed_files=[]
+direct_main_write=false
+production_write=false
+tag_write=false
+baseline_write=false
+runtime_switch=false
+owner_approval_submitted=false
+```
+
+## Remaining high-risk gate
+
+Successful release-candidate creation does not authorize Production.
+
+The next stage remains:
+
+```text
+release candidate
+→ immutable approval envelope
+→ Owner signature
+→ startDeployment
+```
+
+That separation is mandatory even when release preparation is fully unattended.
+
+
+## Unattended availability
+
+The release worker has its own systemd watchdog rather than modifying the shared worker-availability
+guard contract.
+
+Every successful Task poll and Task heartbeat emits:
+
+```text
+WATCHDOG=1
+```
+
+and the service uses:
+
+```text
+Type=notify
+WatchdogSec=90
+Restart=always
+```
+
+A process that is technically alive but no longer polling/heartbeating therefore gets restarted by
+systemd. The worker also writes `worker-liveness.json` for operator diagnostics.

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

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_autonomous_control_plane as acp
+import aios_capability_contract as contract
+import aios_release_online_agent_worker as worker
+
+
+def _ready_probes():
+    return {
+        "credential_projection": lambda: True,
+        "trusted_builder": lambda: True,
+        "authenticated_probe": lambda: True,
+        "release_api_health": lambda: True,
+        "worker_liveness": lambda: True,
+    }
+
+
+def _task():
+    return {
+        "task_id":"r42-autonomous-release-admission-test",
+        "required_capabilities":["release:candidate"],
+        "constraints":{
+            "release_candidate":{
+                "candidate_id":"r42-release-candidate-v2r2c9-test",
+                "source_commit":"1"*40,
+                "version":"V2R2C9",
+                "projection_id":"projection-"+"a"*32,
+            }
+        },
+    }
+
+
+def test_capability_contract_registry_is_closed_and_release_is_versioned():
+    status=contract.validate_capability_contract_registry()
+    assert status["valid"] is True
+    release=contract.capability_contract("release:candidate")
+    assert release is not None
+    assert release.min_version==1
+    assert release.max_version==1
+    assert release.permissions==("prepare_release",)
+    assert "credential_projection" in release.dependencies
+    assert "trusted_builder" in release.dependencies
+    assert "authenticated_probe" in release.health_requirements
+    assert "production_deploy" in release.forbidden
+    assert "runtime_switch" in release.forbidden
+
+
+def test_unknown_agent_capability_cannot_register_or_match():
+    assert contract.registered_capability("release:future",worker=True) is False
+    assert contract.registered_capability("release:*",worker=True) is False
+    assert contract.has_capabilities(["release:future"],["release:future"]) is False
+
+
+def test_runtime_admission_requires_all_release_execution_readiness():
+    decision=acp.runtime_admission(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        requested_operations=["prepare_release"],
+        probes=_ready_probes(),
+    )
+    assert decision.admitted is True
+    assert decision.state=="EXECUTION_READY"
+
+
+def test_runtime_admission_blocks_credential_failure_before_execution():
+    probes=_ready_probes()
+    probes["credential_projection"]=lambda:False
+    probes["authenticated_probe"]=lambda:False
+    decision=acp.runtime_admission(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        requested_operations=["prepare_release"],
+        probes=probes,
+    )
+    assert decision.admitted is False
+    assert decision.state=="EXECUTION_READY_FAILED"
+    assert decision.failure_class==acp.FailureClass.CREDENTIAL_PROJECTION_ERROR
+    assert decision.repair_level==acp.RepairLevel.AUTO_REPAIR_TASK
+
+
+def test_forbidden_release_operation_requires_owner_boundary():
+    decision=acp.runtime_admission(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        requested_operations=["production_deploy"],
+        probes=_ready_probes(),
+    )
+    assert decision.admitted is False
+    assert decision.failure_class==acp.FailureClass.POLICY_DENIED
+    assert decision.repair_level==acp.RepairLevel.OWNER_APPROVAL
+
+
+@pytest.mark.parametrize(
+    "error,phase,expected,level,consumes",
+    [
+        ("Permission denied: gitops.tokens","release",acp.FailureClass.CREDENTIAL_PROJECTION_ERROR,acp.RepairLevel.AUTO_REPAIR_TASK,False),
+        ("HTTP_TRANSPORT_UNAVAILABLE","network",acp.FailureClass.DEPENDENCY_UNAVAILABLE,acp.RepairLevel.AUTO_LOCAL,False),
+        ("TASK_CAPABILITY_SCOPE_INVALID","validation",acp.FailureClass.CAPABILITY_CONTRACT_ERROR,acp.RepairLevel.AUTO_REPAIR_TASK,False),
+        ("lease lost","heartbeat",acp.FailureClass.LEASE_LOST,acp.RepairLevel.AUTO_LOCAL,False),
+        ("business permanent invalid release","release",acp.FailureClass.BUSINESS_PERMANENT,acp.RepairLevel.MANUAL,True),
+    ],
+)
+def test_failure_classification_drives_repair_policy(error,phase,expected,level,consumes):
+    failure=acp.classify_failure(error,phase=phase)
+    assert failure==expected
+    repair=acp.repair_for_failure(failure)
+    assert repair.level==level
+    assert repair.consumes_business_attempt is consumes
+
+
+def test_release_worker_does_not_claim_when_admission_is_not_ready(monkeypatch):
+    calls=[]
+    def fake_http(base,method,route,token,body=None,timeout=30):
+        calls.append((base,method,route,body))
+        if route=="/health":
+            return 200,{"status":"ok"}
+        if route=="/createReleaseCandidate" and body=={}:
+            return 500,{"error":"Permission denied: gitops.tokens"}
+        if route.endswith("/claim"):
+            raise AssertionError("claim must not happen before EXECUTION_READY")
+        raise AssertionError(route)
+
+    events=[]
+    monkeypatch.setattr(worker,"http_json",fake_http)
+    monkeypatch.setattr(worker,"audit",lambda event,**fields:events.append((event,fields)))
+    monkeypatch.setattr(worker,"write_liveness",lambda *args,**kwargs:None)
+
+    assert worker.execute_task(
+        _task(),
+        agent_token="a"*32,
+        release_token="r"*32,
+    ) is False
+    assert not any(route.endswith("/claim") for _,_,route,_ in calls)
+    blocked=[item for item in events if item[0]=="TASK_ADMISSION_BLOCKED"]
+    assert len(blocked)==1
+    assert blocked[0][1]["admission"]["state"]=="EXECUTION_READY_FAILED"
+
+
+def test_system_invariant_snapshot_is_degraded_when_dependency_is_missing():
+    probes=_ready_probes()
+    probes["trusted_builder"]=lambda:False
+    snapshot=acp.system_invariant_snapshot(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        probes=probes,
+    )
+    assert snapshot["status"]=="DEGRADED"
+    assert snapshot["execution_ready"] is False

tests/test_release_candidate_capability_contract.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_capability_contract import (  # noqa: E402
+    has_capabilities,
+    normalize_required_capabilities,
+    normalize_worker_capabilities,
+    normalized_capability_matrix,
+    required_capabilities_subset,
+)
+
+
+def test_release_candidate_is_known_to_task_and_worker_vocabularies():
+    capability="release:candidate"
+    assert normalize_required_capabilities([capability])==frozenset({capability})
+    assert normalize_worker_capabilities([capability])==frozenset({capability})
+    assert has_capabilities([capability],[capability]) is True
+    assert normalized_capability_matrix([capability],[capability])=={
+        "worker_effective":["release:candidate"],
+        "task_required":["release:candidate"],
+        "eligible":True,
+    }
+
+
+def test_release_candidate_capability_does_not_imply_high_risk_release_operations():
+    worker=["release:candidate"]
+    for forbidden in (
+        "submitOwnerApproval",
+        "startDeployment",
+        "tag_publish",
+        "baseline_registry_finalize",
+        "runtime_switch",
+        "deploy.production",
+    ):
+        assert has_capabilities(worker,[forbidden]) is False
+
+
+def test_release_wildcards_and_nearby_unknown_names_remain_fail_closed():
+    assert has_capabilities(["release:candidate"],["release:*"]) is False
+    assert has_capabilities(["release:candidate"],["release:candidate.extra"]) is False
+    assert has_capabilities(["release:*"],["release:candidate"]) is False
+
+
+def test_release_candidate_delegation_subset_is_exact():
+    assert required_capabilities_subset(
+        ["release:candidate"],
+        ["release:candidate"],
+    ) is True
+    assert required_capabilities_subset(
+        ["release:candidate"],
+        ["startDeployment"],
+    ) is False
+
+
+@pytest.mark.parametrize(
+    "worker,required",
+    [
+        (["agent.runtime.hermes"],["agent.runtime.hermes"]),
+        (["agent.runtime.openclaw"],["agent.runtime.openclaw"]),
+        (["git.read"],["repository_read"]),
+        (["git.test"],["test_execution"]),
+        (
+            ["governed.write.branch","governed.write.file"],
+            ["governed.git.write"],
+        ),
+    ],
+)
+def test_existing_known_capability_semantics_are_non_regressed(worker,required):
+    assert has_capabilities(worker,required) is True
+
+
+def test_release_candidate_worker_does_not_match_empty_or_unrelated_task_by_escalation():
+    assert has_capabilities(["release:candidate"],[]) is True
+    assert has_capabilities(["release:candidate"],["git.read"]) is False
+    assert has_capabilities(["release:candidate"],["agent.runtime.hermes"]) is False

tests/test_release_lineage_contract.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_release_lineage_contract import (  # noqa: E402
+    FIXED_V2R2C2,
+    ReleaseLineageError,
+    validate_fixed_v2r2c2_task,
+)
+
+
+def fixed_task(source=FIXED_V2R2C2.source_commit, *, status="FAILED", attempt=1, max_attempts=1, revision=4):
+    return {
+        "task_id":FIXED_V2R2C2.task_id,
+        "status":status,
+        "attempt":attempt,
+        "max_attempts":max_attempts,
+        "revision":revision,
+        "required_capabilities":["release:candidate"],
+        "constraints":{
+            "release_candidate":{
+                "candidate_id":FIXED_V2R2C2.candidate_id,
+                "source_commit":source,
+                "version":FIXED_V2R2C2.version,
+                "projection_id":"projection-"+"a"*32,
+            }
+        },
+    }
+
+
+def test_fixed_v2r2c2_lineage_is_the_original_release_worker_candidate():
+    assert FIXED_V2R2C2.source_commit=="71e736b05fab13efc06af7a4fe2dc022c1cf9a85"
+    result=validate_fixed_v2r2c2_task(fixed_task())
+    assert result["status"]=="VERIFIED"
+    assert result["lineage"]["version"]=="V2R2C2"
+
+
+def test_later_capability_remediation_is_not_retroactively_v2r2c2_source():
+    with pytest.raises(ReleaseLineageError,match="FIXED_RELEASE_LINEAGE_DRIFT"):
+        validate_fixed_v2r2c2_task(
+            fixed_task("17deb91ca7a187dc81766945b1174e854f86bd25")
+        )
+
+
+def test_acp_engineering_candidate_is_not_retroactively_v2r2c2_source():
+    with pytest.raises(ReleaseLineageError,match="FIXED_RELEASE_LINEAGE_DRIFT"):
+        validate_fixed_v2r2c2_task(
+            fixed_task("b4b4416c93fedbe451d655d49e28f915b8fd38e6")
+        )
+
+
+@pytest.mark.parametrize(
+    "status,attempt,max_attempts,revision",
+    [
+        ("FAILED",1,1,4),
+        ("QUEUED",1,2,5),
+        ("CLAIMED",2,2,6),
+        ("VERIFYING",2,2,9),
+        ("SUCCEEDED",2,2,10),
+    ],
+)
+def test_lineage_contract_does_not_confuse_lifecycle_state_with_release_identity(
+    status,attempt,max_attempts,revision
+):
+    result=validate_fixed_v2r2c2_task(
+        fixed_task(
+            status=status,
+            attempt=attempt,
+            max_attempts=max_attempts,
+            revision=revision,
+        )
+    )
+    assert result["status"]=="VERIFIED"
+    assert result["task_status"]==status
+    assert result["attempt"]==attempt
+    assert result["max_attempts"]==max_attempts
+    assert result["revision"]==revision
+
+
+def test_capability_and_task_identity_are_bound():
+    value=fixed_task()
+    value["required_capabilities"]=[]
+    with pytest.raises(ReleaseLineageError,match="FIXED_TASK_CAPABILITY_DRIFT"):
+        validate_fixed_v2r2c2_task(value)
+
+    value=fixed_task()
+    value["task_id"]="replacement-v2r2c2"
+    with pytest.raises(ReleaseLineageError,match="FIXED_TASK_ID_DRIFT"):
+        validate_fixed_v2r2c2_task(value)

tests/test_release_online_agent_worker.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_release_online_agent_worker as worker
+
+
+def task(**overrides):
+    value={
+        "task_id":"r42-release-candidate-live-acceptance-20260819",
+        "required_capabilities":["release:candidate"],
+        "constraints":{
+            "release_candidate":{
+                "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+                "source_commit":"5"*40,
+                "version":"V2R2C2",
+                "projection_id":"projection-"+"a"*32,
+            }
+        }
+    }
+    value.update(overrides)
+    return value
+
+
+def test_release_request_accepts_only_exact_release_capability():
+    req=worker.ReleaseRequest.from_task(task())
+    assert req.candidate_id=="r42-release-candidate-v2r2c2-20260819"
+    assert req.source_commit=="5"*40
+    assert req.version=="V2R2C2"
+    assert req.projection_id=="projection-"+"a"*32
+
+    with pytest.raises(worker.ReleaseWorkerError,match="TASK_CAPABILITY_SCOPE_INVALID"):
+        worker.ReleaseRequest.from_task(task(required_capabilities=["release:candidate","git.read"]))
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"short",
+            "version":"V2R2C2",
+            "projection_id":"projection-"+"a"*32,
+        },
+        {
+            "candidate_id":"bad",
+            "source_commit":"5"*40,
+            "version":"V2R2C2",
+            "projection_id":"projection-"+"a"*32,
+        },
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"5"*40,
+            "version":"bad",
+            "projection_id":"projection-"+"a"*32,
+        },
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"5"*40,
+            "version":"V2R2C2",
+            "projection_id":"bad",
+        },
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"5"*40,
+            "version":"V2R2C2",
+            "projection_id":"projection-"+"a"*32,
+            "execute":True,
+        },
+    ],
+)
+def test_release_request_rejects_invalid_or_extra_fields(payload):
+    value=task()
+    value["constraints"]={"release_candidate":payload}
+    with pytest.raises(worker.ReleaseWorkerError):
+        worker.ReleaseRequest.from_task(value)
+
+
+def test_result_payload_preserves_high_risk_denials():
+    release=worker.ReleaseRequest.from_task(task())
+    result=worker.result_payload(
+        "r42-release-candidate-live-acceptance-20260819",
+        release,
+        {"candidate_id":release.candidate_id,"status":"BUILT"},
+    )
+    assert result["status"]=="SUCCEEDED"
+    assert result["classification"]=="RELEASE_CANDIDATE_CREATED"
+    assert result["writer_lease"]=="NONE"
+    assert result["changed_files"]==[]
+    assert result["direct_main_write"] is False
+    assert result["production_write"] is False
+    assert result["tag_write"] is False
+    assert result["baseline_write"] is False
+    assert result["runtime_switch"] is False
+    assert result["owner_approval_submitted"] is False
+    assert result["credential_material_exposed"] is False
+
+
+def test_result_payload_rejects_builder_candidate_mismatch():
+    release=worker.ReleaseRequest.from_task(task())
+    with pytest.raises(worker.ReleaseWorkerError,match="BUILDER_CANDIDATE_BINDING_INVALID"):
+        worker.result_payload(
+            "r42-release-candidate-live-acceptance-20260819",
+            release,
+            {"candidate_id":"other-release-candidate"},
+        )
+
+
+def test_worker_calls_only_fixed_release_candidate_route(monkeypatch):
+    release=worker.ReleaseRequest.from_task(task())
+    calls=[]
+
+    def fake_http(base,method,route,token,body=None,timeout=30):
+        calls.append((base,method,route,token,body))
+        if route=="/health":
+            return 200,{"status":"ok"}
+        if route=="/createReleaseCandidate" and body=={}:
+            return 400,{"error":"request_invalid"}
+        if route.endswith("/claim"):
+            return 200,{"lease_token":"l"*32,"attempt":1}
+        if route.endswith("/heartbeat"):
+            return 200,{}
+        if route=="/createReleaseCandidate":
+            return 200,{
+                "candidate_id":release.candidate_id,
+                "artifact_sha256":"f"*64,
+            }
+        if route.endswith("/result"):
+            return 200,{"status":"VERIFYING"}
+        raise AssertionError(route)
+
+    monkeypatch.setattr(worker,"http_json",fake_http)
+    monkeypatch.setattr(worker,"audit",lambda *args,**kwargs:None)
+    monkeypatch.setattr(worker,"write_liveness",lambda *args,**kwargs:None)
+    monkeypatch.setattr(worker.Heartbeat,"first",lambda self:None)
+    monkeypatch.setattr(worker.Heartbeat,"close",lambda self:None)
+    monkeypatch.setattr(worker.Heartbeat,"ensure",lambda self:None)
+    monkeypatch.setattr(worker.threading.Thread,"start",lambda self:None)
+
+    assert worker.execute_task(
+        task(),
+        agent_token="a"*32,
+        release_token="r"*32,
+    ) is True
+
+    release_calls=[
+        item for item in calls
+        if item[0]==worker.RELEASE_API
+        and item[2]=="/createReleaseCandidate"
+        and item[4]!={}
+    ]
+    assert len(release_calls)==1
+    assert release_calls[0][1]=="POST"
+    assert release_calls[0][2]=="/createReleaseCandidate"
+    assert release_calls[0][4]=={
+        "candidate_id":release.candidate_id,
+        "source_commit":release.source_commit,
+        "version":release.version,
+    }
+
+
+
+def test_write_liveness_records_release_capability(tmp_path,monkeypatch):
+    path=tmp_path/"worker-liveness.json"
+    monkeypatch.setattr(worker,"STATE_ROOT",tmp_path)
+    monkeypatch.setattr(worker,"LIVENESS_PATH",path)
+    worker.write_liveness(
+        "IDLE",
+        stage="poll",
+        visible_count=2,
+    )
+    value=json.loads(path.read_text())
+    assert value["schema"]=="aios.online-agent.worker-liveness.v1"
+    assert value["agent_type"]=="release-preparation-worker"
+    assert value["capabilities"]==["release:candidate"]
+    assert value["state"]=="IDLE"
+    assert value["visible_count"]==2
+
+
+def test_sd_notify_is_noop_without_systemd_socket(monkeypatch):
+    monkeypatch.delenv("NOTIFY_SOCKET",raising=False)
+    worker.sd_notify("WATCHDOG=1")
+
+def test_systemd_unit_uses_two_encrypted_credentials_and_loopback_release_api():
+    unit=(ROOT/"deploy/systemd/aios-release-online-agent-worker.service").read_text()
+    assert "LoadCredentialEncrypted=agent_token:" in unit
+    assert "LoadCredentialEncrypted=release_candidate_token:" in unit
+    assert "AIOS_RELEASE_API_URL=http://127.0.0.1:18890" in unit
+    assert "AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819" in unit
+    assert "Type=notify" in unit
+    assert "WatchdogSec=90" in unit
+    assert "Restart=always" in unit
+    assert "IPAddressDeny=any" in unit
+    assert "IPAddressAllow=127.0.0.0/8" in unit
+    assert "InaccessiblePaths=/data/aios-gitops /home" in unit
+    assert "submitOwnerApproval" not in unit
+    assert "startDeployment" not in unit
+
+
+def test_source_contains_no_high_risk_release_routes():
+    source=(ROOT/"active-runtime/app/aios_release_online_agent_worker.py").read_text()
+    assert '"/createReleaseCandidate"' in source
+    for forbidden in (
+        "/submitOwnerApproval",
+        "/startDeployment",
+        "/requestOrCreateTag",
+        "/buildProductionArtifact",
+        "/executeApprovedProductionRelease",
+    ):
+        assert forbidden not in source
+
+
+def test_docs_keep_release_candidate_distinct_from_production():
+    doc=(ROOT/"docs/superpowers/specs/2026-08-19-release-preparation-online-agent.md").read_text()
+    assert "release:candidate" in doc
+    assert "Owner signature" in doc
+    assert "never\nstarts deployment" in doc or "never starts deployment" in doc
+    assert "LoadCredentialEncrypted" in doc

tests/test_release_preparation_supervisor_coordinator.py

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