fix(task-api): normalize release candidate capability

active-runtime/app/aios_capability_contract.py

@@ -37,6 +37,7 @@ _TASK_ALIASES: dict[str, frozenset[str]] = {
     "gpu.train": frozenset({"legacy.gpu.train"}),
     "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
     "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+    "release:candidate": frozenset({"release:candidate"}),
 }
 
 _WORKER_ALIASES: dict[str, frozenset[str]] = {
@@ -71,6 +72,7 @@ _WORKER_ALIASES: dict[str, frozenset[str]] = {
     "gpu.train": frozenset({"legacy.gpu.train"}),
     "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
     "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+    "release:candidate": frozenset({"release:candidate"}),
 }
 
 _UNKNOWN_PREFIX = "exact:"

active-runtime/app/aios_durable_owner_authorization.py

@@ -32,6 +32,8 @@ from typing import Any, Mapping
 POLICY_SCHEMA = "aios.owner-durable-authorization.v1"
 RECEIPT_SCHEMA = "aios.owner-durable-authorization-receipt.v1"
 DECISION_SCHEMA = "aios.owner-durable-authorization-decision.v1"
+PROJECTION_SCHEMA = "aios.external-capability-projection.request.v1"
+PROJECTION_LIST_SCHEMA = "aios.external-capability-projection.list.v1"
 
 ALLOWED_PERSISTENT_CAPABILITIES = frozenset({
     "governed-task:project",
@@ -421,6 +423,30 @@ class ReceiptStore:
                 )
                 """
             )
+            con.execute(
+                """
+                CREATE TABLE IF NOT EXISTS projection_outbox(
+                  projection_id TEXT PRIMARY KEY,
+                  idempotency_key TEXT NOT NULL UNIQUE,
+                  grant_id TEXT NOT NULL,
+                  account TEXT NOT NULL,
+                  capability TEXT NOT NULL,
+                  policy_sha256 TEXT NOT NULL,
+                  request_sha256 TEXT NOT NULL,
+                  payload_sha256 TEXT NOT NULL,
+                  payload_json TEXT NOT NULL,
+                  state TEXT NOT NULL,
+                  created_at TEXT NOT NULL,
+                  updated_at TEXT NOT NULL
+                )
+                """
+            )
+            con.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_projection_outbox_state
+                    ON projection_outbox(state, capability, created_at)
+                """
+            )
             con.commit()
         self._lock = threading.RLock()
 
@@ -461,6 +487,183 @@ class ReceiptStore:
             )
             con.commit()
 
+    def update_response(
+        self,
+        key: str,
+        request_sha256: str,
+        response: Mapping[str, Any],
+    ) -> None:
+        encoded = json.dumps(response, sort_keys=True, separators=(",", ":"))
+        with self._lock, sqlite3.connect(self.path) as con:
+            con.execute("BEGIN IMMEDIATE")
+            row = con.execute(
+                """
+                SELECT request_sha256 FROM receipts
+                 WHERE idempotency_key=?
+                """,
+                (key,),
+            ).fetchone()
+            if row is None:
+                con.rollback()
+                raise AuthorizationError("RECEIPT_NOT_FOUND", 409)
+            if row[0] != request_sha256:
+                con.rollback()
+                raise AuthorizationError("IDEMPOTENCY_CONFLICT", 409)
+            con.execute(
+                """
+                UPDATE receipts
+                   SET response_sha256=?,response_json=?
+                 WHERE idempotency_key=?
+                """,
+                (sha256(encoded.encode()), encoded, key),
+            )
+            con.commit()
+
+    @staticmethod
+    def projection_id(
+        grant_id: str,
+        account: str,
+        capability: str,
+        request_sha256: str,
+    ) -> str:
+        digest = sha256(
+            canonical(
+                {
+                    "grant_id": grant_id,
+                    "account": account,
+                    "capability": capability,
+                    "request_sha256": request_sha256,
+                }
+            )
+        )
+        return "projection-" + digest[:32]
+
+    def enqueue_projection(
+        self,
+        *,
+        idempotency_key: str,
+        grant_id: str,
+        account: str,
+        capability: str,
+        policy_sha256: str,
+        request_sha256: str,
+        request: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        projection_id = self.projection_id(
+            grant_id,
+            account,
+            capability,
+            request_sha256,
+        )
+        now = datetime.now(timezone.utc).isoformat()
+        payload = {
+            "schema": PROJECTION_SCHEMA,
+            "projection_id": projection_id,
+            "state": "PENDING_EXTERNAL_CONNECTOR",
+            "grant_id": grant_id,
+            "account": account,
+            "capability": capability,
+            "policy_sha256": policy_sha256,
+            "request_sha256": request_sha256,
+            "request": dict(request),
+            "created_at": now,
+            "separate_owner_release_approval_still_required": True,
+        }
+        encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        payload_sha = sha256(encoded.encode())
+
+        with self._lock, sqlite3.connect(self.path) as con:
+            con.row_factory = sqlite3.Row
+            con.execute("BEGIN IMMEDIATE")
+            existing = con.execute(
+                """
+                SELECT * FROM projection_outbox
+                 WHERE idempotency_key=?
+                """,
+                (idempotency_key,),
+            ).fetchone()
+            if existing is not None:
+                if (
+                    existing["projection_id"] != projection_id
+                    or existing["request_sha256"] != request_sha256
+                    or existing["grant_id"] != grant_id
+                    or existing["account"] != account
+                    or existing["capability"] != capability
+                    or existing["policy_sha256"] != policy_sha256
+                ):
+                    con.rollback()
+                    raise AuthorizationError(
+                        "PROJECTION_IDEMPOTENCY_CONFLICT",
+                        409,
+                    )
+                value = json.loads(existing["payload_json"])
+                con.commit()
+                return value
+
+            con.execute(
+                """
+                INSERT INTO projection_outbox(
+                  projection_id,idempotency_key,grant_id,account,capability,
+                  policy_sha256,request_sha256,payload_sha256,payload_json,
+                  state,created_at,updated_at
+                ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    projection_id,
+                    idempotency_key,
+                    grant_id,
+                    account,
+                    capability,
+                    policy_sha256,
+                    request_sha256,
+                    payload_sha,
+                    encoded,
+                    "PENDING_EXTERNAL_CONNECTOR",
+                    now,
+                    now,
+                ),
+            )
+            con.commit()
+        return payload
+
+    def get_projection(self, projection_id: str) -> dict[str, Any] | None:
+        if not re.fullmatch(r"projection-[0-9a-f]{32}", projection_id):
+            raise AuthorizationError("PROJECTION_ID_INVALID")
+        with sqlite3.connect(self.path) as con:
+            row = con.execute(
+                """
+                SELECT payload_json FROM projection_outbox
+                 WHERE projection_id=?
+                """,
+                (projection_id,),
+            ).fetchone()
+        if row is None:
+            return None
+        return json.loads(row[0])
+
+    def list_projections(
+        self,
+        *,
+        grant_id: str,
+        account: str,
+        capability: str = "release:candidate",
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        if not 1 <= limit <= 500:
+            raise AuthorizationError("PROJECTION_LIMIT_INVALID")
+        with sqlite3.connect(self.path) as con:
+            rows = con.execute(
+                """
+                SELECT payload_json
+                  FROM projection_outbox
+                 WHERE grant_id=? AND account=? AND capability=?
+                 ORDER BY created_at,idempotency_key
+                 LIMIT ?
+                """,
+                (grant_id, account, capability, limit),
+            ).fetchall()
+        return [json.loads(row[0]) for row in rows]
+
 
 @dataclass(frozen=True)
 class Upstream:
@@ -526,6 +729,35 @@ class Broker:
             raise AuthorizationError("POLICY_ACCOUNT_MISMATCH", 403)
         return policy
 
+    def projection(self, projection_id: str) -> dict[str, Any]:
+        policy = self.policy()
+        value = self.store.get_projection(projection_id)
+        if value is None:
+            raise AuthorizationError("PROJECTION_NOT_FOUND", 404)
+        if value.get("grant_id") != policy.grant_id:
+            raise AuthorizationError("PROJECTION_GRANT_MISMATCH", 403)
+        if value.get("account") != policy.account:
+            raise AuthorizationError("PROJECTION_ACCOUNT_MISMATCH", 403)
+        return value
+
+    def projections(self, *, limit: int = 100) -> dict[str, Any]:
+        policy = self.policy()
+        items = self.store.list_projections(
+            grant_id=policy.grant_id,
+            account=policy.account,
+            limit=limit,
+        )
+        return {
+            "schema": PROJECTION_LIST_SCHEMA,
+            "grant_id": policy.grant_id,
+            "account": policy.account,
+            "capability": "release:candidate",
+            "count": len(items),
+            "items": items,
+            "external_connector_ack_supported": False,
+            "separate_owner_release_approval_still_required": True,
+        }
+
     def execute(
         self,
         capability: str,
@@ -543,7 +775,33 @@ class Broker:
         if prior is not None:
             if prior["request_sha256"] != request_sha:
                 raise AuthorizationError("IDEMPOTENCY_CONFLICT", 409)
-            return prior["response"]
+            response = prior["response"]
+            if (
+                capability == "release:candidate"
+                and not response.get("projection_id")
+            ):
+                projection = self.store.enqueue_projection(
+                    idempotency_key=idempotency_key,
+                    grant_id=policy.grant_id,
+                    account=policy.account,
+                    capability=capability,
+                    policy_sha256=policy.policy_sha256,
+                    request_sha256=request_sha,
+                    request=request,
+                )
+                response = {
+                    **response,
+                    "projection_id": projection["projection_id"],
+                    "projection_state": projection["state"],
+                    "external_capability_projection_required": True,
+                    "separate_owner_release_approval_still_required": True,
+                }
+                self.store.update_response(
+                    idempotency_key,
+                    request_sha,
+                    response,
+                )
+            return response
 
         if capability == "governed-task:project":
             status, value = self.upstream.project_governed_task(request)
@@ -558,11 +816,19 @@ class Broker:
             upstream_status: int | None = status
             upstream_response: Mapping[str, Any] | None = value
         elif capability == "release:candidate":
-            # The trusted release builder lives outside the repository control
-            # plane. This broker persists/attests Owner intent but does not
-            # invent or proxy a release-builder endpoint. The external
-            # connector must project the account's release:candidate scope and
-            # may consume this grant/decision as authorization evidence.
+            # Persist a machine-readable projection request. The connector
+            # remains the only component that can turn this request into a real
+            # account scope; this broker never self-grants or calls a fake
+            # release-builder endpoint.
+            projection = self.store.enqueue_projection(
+                idempotency_key=idempotency_key,
+                grant_id=policy.grant_id,
+                account=policy.account,
+                capability=capability,
+                policy_sha256=policy.policy_sha256,
+                request_sha256=request_sha,
+                request=request,
+            )
             operation_state = "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION"
             upstream_status = None
             upstream_response = None
@@ -584,6 +850,16 @@ class Broker:
             "external_capability_projection_required": (
                 capability == "release:candidate"
             ),
+            "projection_id": (
+                projection["projection_id"]
+                if capability == "release:candidate"
+                else None
+            ),
+            "projection_state": (
+                projection["state"]
+                if capability == "release:candidate"
+                else None
+            ),
             "separate_owner_release_approval_still_required": True,
         }
         self.store.put(idempotency_key, capability, request_sha, response)
@@ -651,22 +927,32 @@ class Handler(BaseHTTPRequestHandler):
         if not self.account_authorized():
             self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
             return
-        if self.path != "/v1/capabilities" or self.broker is None:
+        if self.broker is None:
             self.send_json({"error": "NOT_FOUND"}, 404)
             return
         try:
-            policy = self.broker.policy()
-            self.send_json(
-                {
+            if self.path == "/v1/capabilities":
+                policy = self.broker.policy()
+                value = {
                     "schema": POLICY_SCHEMA,
                     "grant_id": policy.grant_id,
                     "account": policy.account,
                     "capabilities": sorted(policy.capabilities),
                     "persistent_until_revoked": True,
                     "separate_owner_release_approval_still_required": True,
-                },
-                200,
-            )
+                }
+            elif self.path == "/v1/projections":
+                value = self.broker.projections()
+            else:
+                match = re.fullmatch(
+                    r"/v1/projections/(projection-[0-9a-f]{32})",
+                    self.path,
+                )
+                if match is None:
+                    self.send_json({"error": "NOT_FOUND"}, 404)
+                    return
+                value = self.broker.projection(match.group(1))
+            self.send_json(value, 200)
         except AuthorizationError as exc:
             self.send_json({"error": exc.code}, exc.status)
 

active-runtime/app/aios_release_online_agent_worker.py

@@ -0,0 +1,636 @@
+#!/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
+
+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 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)
+    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:
+                    audit(
+                        "TASK_EXECUTION_FAILED",
+                        task_id=raw.get("task_id"),
+                        phase=exc.phase,
+                        error=exc.code,
+                    )
+            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())

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

docs/superpowers/specs/2026-08-19-durable-owner-authorization.md

@@ -135,6 +135,61 @@ therefore still requires the normal deterministic plan, WriterLease and governed
 This fixes the previous split-brain behavior where a bare Task could exist without the canonical
 governed execution projection.
 
+
+## External capability projection outbox
+
+A `release:candidate` authorization now creates a durable machine-readable outbox record:
+
+```text
+schema=aios.external-capability-projection.request.v1
+state=PENDING_EXTERNAL_CONNECTOR
+capability=release:candidate
+```
+
+The projection identifier is deterministic from:
+
+```text
+grant_id
+account
+capability
+request_sha256
+```
+
+The outbox stores no credentials or Owner signing material. It retains only the already-authorized
+release-candidate request, policy/request hashes, account, grant id and timestamps.
+
+The broker exposes read-only Unix-socket endpoints:
+
+```text
+GET /v1/projections
+GET /v1/projections/{projection_id}
+```
+
+These endpoints re-check the current Owner policy on every read. If the durable grant has been
+revoked, enumeration fails closed with `OWNER_GRANT_REVOKED`. The historical outbox record is
+retained for audit and is not silently deleted.
+
+The repository intentionally does **not** provide a connector acknowledgement/write endpoint yet.
+No in-repository code can truthfully assert that the external ChatGPT connector account has gained
+`release:candidate`. Until the platform consumes the outbox and performs its own identity-scope
+projection, the only valid state is:
+
+```text
+AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION
+PENDING_EXTERNAL_CONNECTOR
+```
+
+A future external projector must preserve:
+
+```text
+release:candidate
+!= submitOwnerApproval
+!= startDeployment
+!= tag_publish
+!= baseline_registry_finalize
+!= runtime_switch
+```
+
 ## Idempotency and audit
 
 Every broker operation requires an idempotency key. A receipt SQLite database stores only:

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

@@ -338,9 +338,190 @@ def test_release_candidate_authorization_does_not_invent_builder_upstream(
     assert response["upstream_status"] is None
     assert response["upstream_response"] is None
     assert response["external_capability_projection_required"] is True
+    assert response["projection_id"].startswith("projection-")
+    assert response["projection_state"]=="PENDING_EXTERNAL_CONNECTOR"
     assert response["separate_owner_release_approval_still_required"] is True
 
 
+
+def test_release_authorization_creates_durable_projection_outbox(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    request=release_request()
+    response=instance.execute(
+        "release:candidate",
+        request,
+        idempotency_key="owner-auth-release-outbox-0001",
+    )
+    projection=store.get_projection(response["projection_id"])
+    assert projection is not None
+    assert projection["schema"]==auth.PROJECTION_SCHEMA
+    assert projection["state"]=="PENDING_EXTERNAL_CONNECTOR"
+    assert projection["grant_id"]==policy.grant_id
+    assert projection["account"]==ACCOUNT
+    assert projection["capability"]=="release:candidate"
+    assert projection["request"]==request
+    assert projection["separate_owner_release_approval_still_required"] is True
+
+    listed=instance.projections()
+    assert listed["schema"]==auth.PROJECTION_LIST_SCHEMA
+    assert listed["count"]==1
+    assert listed["items"][0]["projection_id"]==response["projection_id"]
+    assert listed["external_connector_ack_supported"] is False
+
+
+
+def test_existing_pre_outbox_release_receipt_is_backfilled_on_exact_retry(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    request=release_request()
+    decision=policy.authorize("release:candidate",request)
+    key="owner-auth-release-pre-outbox-0001"
+    legacy_response={
+        "schema":auth.RECEIPT_SCHEMA,
+        "grant_id":policy.grant_id,
+        "owner_identity":policy.owner_identity,
+        "account":policy.account,
+        "capability":"release:candidate",
+        "policy_sha256":policy.policy_sha256,
+        "request_sha256":decision["request_sha256"],
+        "idempotency_key":key,
+        "operation_state":"AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION",
+        "upstream_status":None,
+        "upstream_response":None,
+        "external_capability_projection_required":True,
+        "separate_owner_release_approval_still_required":True,
+    }
+    store.put(
+        key,
+        "release:candidate",
+        decision["request_sha256"],
+        legacy_response,
+    )
+    response=instance.execute(
+        "release:candidate",
+        request,
+        idempotency_key=key,
+    )
+    assert response["projection_id"].startswith("projection-")
+    assert response["projection_state"]=="PENDING_EXTERNAL_CONNECTOR"
+    assert store.get_projection(response["projection_id"]) is not None
+    persisted=store.get(key)["response"]
+    assert persisted["projection_id"]==response["projection_id"]
+
+def test_projection_id_is_deterministic_and_exact_retry_does_not_duplicate(
+    tmp_path,monkeypatch
+):
+    instance,_,store,_=broker(tmp_path,monkeypatch)
+    key="owner-auth-release-outbox-0002"
+    first=instance.execute("release:candidate",release_request(),idempotency_key=key)
+    second=instance.execute("release:candidate",release_request(),idempotency_key=key)
+    assert second==first
+    con=sqlite3.connect(store.path)
+    rows=con.execute("select projection_id from projection_outbox").fetchall()
+    assert rows==[(first["projection_id"],)]
+
+
+def test_projection_enqueue_recovers_crash_between_outbox_and_receipt(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    request=release_request()
+    decision=policy.authorize("release:candidate",request)
+    key="owner-auth-release-outbox-crash-0001"
+    first=store.enqueue_projection(
+        idempotency_key=key,
+        grant_id=policy.grant_id,
+        account=policy.account,
+        capability="release:candidate",
+        policy_sha256=policy.policy_sha256,
+        request_sha256=decision["request_sha256"],
+        request=request,
+    )
+    # Simulate a later retry after the process crashed before receipt creation.
+    second=store.enqueue_projection(
+        idempotency_key=key,
+        grant_id=policy.grant_id,
+        account=policy.account,
+        capability="release:candidate",
+        policy_sha256=policy.policy_sha256,
+        request_sha256=decision["request_sha256"],
+        request=request,
+    )
+    assert second==first
+
+
+def test_projection_idempotency_conflict_is_fail_closed(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    first=release_request()
+    first_decision=policy.authorize("release:candidate",first)
+    key="owner-auth-release-outbox-conflict-0001"
+    store.enqueue_projection(
+        idempotency_key=key,
+        grant_id=policy.grant_id,
+        account=policy.account,
+        capability="release:candidate",
+        policy_sha256=policy.policy_sha256,
+        request_sha256=first_decision["request_sha256"],
+        request=first,
+    )
+    second=release_request(version="V2R2C3")
+    second_decision=policy.authorize("release:candidate",second)
+    with pytest.raises(auth.AuthorizationError,match="PROJECTION_IDEMPOTENCY_CONFLICT"):
+        store.enqueue_projection(
+            idempotency_key=key,
+            grant_id=policy.grant_id,
+            account=policy.account,
+            capability="release:candidate",
+            policy_sha256=policy.policy_sha256,
+            request_sha256=second_decision["request_sha256"],
+            request=second,
+        )
+
+
+def test_revocation_blocks_projection_enumeration(tmp_path,monkeypatch):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    response=instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-outbox-revoke-0001",
+    )
+    # Restore real policy loading and write a revocation for this grant.
+    monkeypatch.undo()
+    revocations=instance.revocation_path
+    revocations.write_text(
+        json.dumps({"revoked_grant_ids":[policy.grant_id]})+"\n",
+        encoding="utf-8",
+    )
+    revocations.chmod(0o600)
+    monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(auth.AuthorizationError,match="OWNER_GRANT_REVOKED"):
+        instance.projections()
+    # Evidence is retained; revocation does not erase history.
+    assert store.get_projection(response["projection_id"]) is not None
+
+
+def test_projection_outbox_contains_no_credentials(tmp_path,monkeypatch):
+    instance,_,store,_=broker(tmp_path,monkeypatch)
+    instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-outbox-schema-0001",
+    )
+    con=sqlite3.connect(store.path)
+    columns=[row[1] for row in con.execute("pragma table_info(projection_outbox)")]
+    joined=" ".join(columns).lower()
+    for forbidden in ("token","credential","private_key","secret"):
+        assert forbidden not in joined
+    payload=con.execute("select payload_json from projection_outbox").fetchone()[0]
+    assert "Bearer " not in payload
+    assert "submitOwnerApproval" not in payload
+    assert "startDeployment" not in payload
+
 def test_exact_retry_returns_same_receipt_without_second_projection(
     tmp_path,monkeypatch
 ):
@@ -421,6 +602,9 @@ def test_source_contains_no_release_builder_endpoint_or_high_risk_execution():
     assert "baseline_registry_finalize" in source
     assert "runtime_switch" in source
     assert "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION" in source
+    assert "PENDING_EXTERNAL_CONNECTOR" in source
+    assert "projection_outbox" in source
+    assert "external_connector_ack_supported" in source
     assert "SAFE_PROJECTED_TASK_CAPABILITIES" in source
     assert "INITIAL_EXECUTION_ACTION_NOT_AUTHORIZED" in source
 
@@ -440,3 +624,4 @@ def test_docs_preserve_separate_owner_release_approval():
     assert "persistent until explicitly revoked" in doc.lower()
     assert "Owner signature" in doc
     assert "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION" in doc
+    assert "PENDING_EXTERNAL_CONNECTOR" in doc

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

@@ -0,0 +1,216 @@
+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.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.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]
+    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