Merge branch 'main' into release/v2r1c5-registry-20260805

Owner exact OID approved; protected PR only.

deploy/openbao-autonomy/admin-bootstrap-openbao-autonomy-v1-v3.self-test.json

@@ -1 +0,0 @@
-{"checks": {"72h_acceptance_gate": true, "approle_implemented": true, "audit_implemented": true, "backup_implemented": true, "branch_protection_gate_not_bypassable": true, "charter_store_sign_verify": true, "continuous_optimization_implemented": true, "declarative_self_initialization": true, "embedded_shell_syntax": true, "exact_checksum_duplicate_rejected": true, "exact_checksum_good_fixture": true, "exact_checksum_missing_rejected": true, "fault_injection_implemented": true, "gitea_1_22_6_rejected": true, "gitea_1_23_0_accepted": true, "gitea_tag_api_version_gate": true, "isolated_restore_implemented": true, "kv_v2_implemented": true, "machine_credential_persisted": true, "no_git_force_push": true, "no_git_tag_mutation": true, "openbao_ready_gate_not_bypassable": true, "outer_bash_syntax": true, "policy_implemented": true, "ready_not_predeclared": true, "root_token_not_persisted": true, "self_contained_no_sibling_preflight": true, "self_init_secret_removal_implemented": true, "tag_protection_gate_implemented": true, "transit_charter_key_implemented": true, "transit_runtime_key_implemented": true}, "passed": 31, "runtime_claims": false, "schema": "AIOS_OPENBAO_BOOTSTRAP_STATIC_SELF_TEST_V3", "scope": "static_and_fixture_only", "script_sha256": "4bece5edc8be5d2024b5b81ce569549625c2bf318fdaed1066773b379b76cc66", "total": 31}

deploy/openbao-autonomy/admin-bootstrap-openbao-autonomy-v1-v3.sha256

@@ -1 +0,0 @@
-4bece5edc8be5d2024b5b81ce569549625c2bf318fdaed1066773b379b76cc66  admin-bootstrap-openbao-autonomy-v1-v3.sh

deploy/openbao-autonomy/checksums.sha256

@@ -1,4 +0,0 @@
-aa5a8eb14e80d23c2e3f8a36cdeea4b3cd026877eb4aa424637a85665388e6ec  admin-bootstrap-openbao-autonomy-v1-v3.self-test.json
-4bece5edc8be5d2024b5b81ce569549625c2bf318fdaed1066773b379b76cc66  admin-bootstrap-openbao-autonomy-v1-v3.sh
-b400b9427cd37cb7bbba85313081aeb93f0f3181cbfbab99948c793f30d3c53e  admin-bootstrap-openbao-autonomy-v1-v3.sha256
-c355df01c83c2801ae3a0df43806c562ee70f97c9aa558e3f67dd097efa5ac99  test-admin-bootstrap-openbao-autonomy-v1-v3.sh

deploy/openbao-autonomy/test-admin-bootstrap-openbao-autonomy-v1-v3.sh

@@ -1,39 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-umask 077
-
-script="${1:-$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)/admin-bootstrap-openbao-autonomy-v1-v3.sh}"
-report="${2:-$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)/admin-bootstrap-openbao-autonomy-v1-v3.self-test.json}"
-
-[[ -f "$script" ]] || {
-  printf 'missing script: %s\n' "$script" >&2
-  exit 66
-}
-
-bash -n "$script"
-tmp="$(mktemp)"
-trap 'rm -f "$tmp"' EXIT
-bash "$script" --self-test >"$tmp"
-
-python3 - "$script" "$tmp" <<'PY'
-import hashlib
-import json
-import pathlib
-import sys
-
-script, report = map(pathlib.Path, sys.argv[1:])
-data = json.loads(report.read_text())
-actual_hash = hashlib.sha256(script.read_bytes()).hexdigest()
-
-assert data["schema"] == "AIOS_OPENBAO_BOOTSTRAP_STATIC_SELF_TEST_V3"
-assert data["scope"] == "static_and_fixture_only"
-assert data["runtime_claims"] is False
-assert data["script_sha256"] == actual_hash
-assert data["passed"] == data["total"]
-assert all(data["checks"].values())
-PY
-
-install -m 0600 "$tmp" "$report"
-printf 'PASS bash-n + embedded static/fixture tests\n'
-printf 'report=%s\n' "$report"
-sha256sum "$script"

deploy/systemd/aios-autonomous-release-candidate.service

@@ -0,0 +1,42 @@
+[Unit]
+Description=AIOS Autonomous Release Candidate API (Dormant/Shadow Only)
+After=network.target
+ConditionPathExists=/opt/aios-autonomous-release-candidate/aios_autonomous_release_candidate
+
+[Service]
+Type=simple
+User=aiman
+Group=gitops
+UMask=0077
+WorkingDirectory=/opt/aios-autonomous-release-candidate
+Environment=PYTHONDONTWRITEBYTECODE=1
+Environment=PYTHONPATH=/opt/aios-autonomous-release-candidate
+Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_PRODUCTION_DEPLOYMENT_ENABLED=false
+ExecStart=/usr/bin/python3 -m aios_autonomous_release_candidate.service --host 127.0.0.1 --port 19951 --state /var/lib/aios-autonomous-release-candidate/state.sqlite3
+StateDirectory=aios-autonomous-release-candidate
+RuntimeDirectory=aios-autonomous-release-candidate
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectKernelLogs=true
+ProtectControlGroups=true
+ProtectClock=true
+ProtectHostname=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+IPAddressDeny=any
+IPAddressAllow=localhost
+ReadOnlyPaths=/opt/aios-autonomous-release-candidate
+ReadWritePaths=/var/lib/aios-autonomous-release-candidate
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-autonomous-release-worker-candidate.service

@@ -0,0 +1,51 @@
+[Unit]
+Description=AIOS Autonomous Release Candidate Worker (Dormant/Shadow Only)
+After=network.target
+ConditionPathExists=/opt/aios-autonomous-release-candidate/aios_autonomous_release_candidate
+
+[Service]
+Type=simple
+User=aiman
+Group=gitops
+UMask=0077
+WorkingDirectory=/opt/aios-autonomous-release-candidate
+Environment=PYTHONDONTWRITEBYTECODE=1
+Environment=PYTHONPATH=/opt/aios-autonomous-release-candidate
+Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+Environment=AIOS_PRODUCTION_DEPLOYMENT_ENABLED=false
+Environment=AIOS_CANDIDATE_CONTROL_ENDPOINT=127.0.0.1:19951
+Environment=AIOS_OPENCLAW_CREDENTIAL_REFERENCE=systemd-credential:openclaw-gateway-candidate
+Environment=AIOS_CODEX_CREDENTIAL_REFERENCE=systemd-credential:codex-app-server-candidate
+Environment=AIOS_NATIVE_CONTROL_CREDENTIAL_REFERENCE=systemd-credential:session-native-adapter-control-candidate
+Environment=AIOS_GITEA_CREDENTIAL_REFERENCE=systemd-credential:gitea-admin-api-candidate
+ImportCredential=openclaw-gateway-candidate
+ImportCredential=codex-app-server-candidate
+ImportCredential=session-native-adapter-control-candidate
+ImportCredential=gitea-admin-api-candidate
+ExecStart=/usr/bin/python3 -m aios_autonomous_release_candidate.worker_service --state-directory /var/lib/aios-autonomous-release-worker-candidate
+StateDirectory=aios-autonomous-release-worker-candidate
+RuntimeDirectory=aios-autonomous-release-worker-candidate
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectKernelLogs=true
+ProtectControlGroups=true
+ProtectClock=true
+ProtectHostname=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+IPAddressDeny=any
+IPAddressAllow=localhost
+ReadOnlyPaths=/opt/aios-autonomous-release-candidate
+ReadWritePaths=/var/lib/aios-autonomous-release-worker-candidate
+
+[Install]
+WantedBy=multi-user.target

docs/autonomous-release/AGENT_DEPLOYMENT_INSTRUCTIONS.md

@@ -0,0 +1,71 @@
+# AIOS Autonomous Release Agent Deployment Instructions
+
+## Objective
+
+Deploy only candidate/shadow environments first. Production activation requires a separate owner release approval.
+
+## Agent execution order
+
+1. Verify git references:
+   - local
+   - authoritative bare
+   - remote
+   - Gitea API
+
+2. Build candidate artifact.
+
+3. Run:
+   - unit tests
+   - security scans
+   - secret scans
+   - artifact verification
+
+4. Start candidate workers:
+   - OpenClaw native adapter
+   - Codex native adapter
+   - dedicated writer
+
+5. Execute shadow task:
+
+```
+TaskPlan
+ -> Planner
+ -> Worker Router
+ -> Session lifecycle
+ -> Evidence chain
+ -> Artifact verification
+ -> ReleasePlan
+```
+
+## Hard restrictions
+
+Never:
+
+- write main directly
+- force push
+- rewrite tags
+- export secrets
+- bypass owner approval
+- modify production flags during candidate validation
+
+## Production release gate
+
+Require:
+
+```
+owner_approval=true
+artifact_verified=true
+evidence_closure=true
+rollback_ready=true
+remote_tag_verified=true
+baseline_verified=true
+registry_verified=true
+```
+
+Until then keep:
+
+```
+AIOS_GITOPS_WRITES_ENABLED=false
+AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false
+```

docs/autonomous-release/ARCHITECTURE.md

@@ -0,0 +1,34 @@
+# AIOS Autonomous Release Candidate 架构
+
+本 Candidate 把规划、工作流、凭据引用、原生会话、证据和发布控制拆成独立边界。它只负责验证约束，不拥有生产授权。
+
+## 信任边界
+
+- Owner Approval 的签名和撤销状态来自外部 HSM/KMS 或可信 Approval Plane。Candidate 只有 verifier，没有 signer。
+- Secret Manager V2 只返回 `systemd-credential:*` 或受控的 Candidate reference，不返回认证正文。
+- OpenClaw、Codex 和 Gitea provider 身份由管理面供应；模拟 provider 永远不能把原生 canary 标成已验证。
+- Writer 只写任务绑定分支和允许路径。生产 main、tag、baseline、registry 和部署不属于 Candidate 权限。
+
+## 核心模块
+
+- `models.py`：TaskPlan、Approval、Credential、Result 和 ReleasePlan 强类型约束。
+- `store.py`：私有 SQLite、WAL、全同步、schema 检查、事务和原子 step transition。
+- `workflow.py`：依赖、claim、lease、retry、timeout recovery 和终态。
+- `credential.py` / `secret_manager.py`：reference-only 生命周期、补偿撤销、跨任务拒绝和 post-revocation 验证。
+- `native_adapters.py`：OpenClaw/Codex 原生 provider ID、parent/child tree、status、interrupt、close 和纯文本回执。
+- `gitea_canary.py`：仅通过 deploy-key provider 的 list/create/delete 操作完成 canary 与撤销验证。
+- `evidence.py`：落库的哈希链、重启回放、篡改检测和 required-event closure。
+- `artifact.py`：确定性 tar、逐文件哈希、package closure、链接/特殊文件/敏感标记拒绝。
+- `release.py` / `release_gate.py`：部署审批与回滚审批分离，tag object 与 peeled target 分离验证。
+
+## 会话真实性
+
+以下状态必须分别记录，不能互相替代：
+
+```text
+broker_session_verified
+openclaw_native_child_session_verified
+codex_child_session_verified
+```
+
+Shadow provider 只能证明 Broker 合同和清理路径，不能证明 OpenClaw 或 Codex 原生会话。

docs/autonomous-release/EXECUTION_PACKAGE.md

@@ -0,0 +1,47 @@
+# AIOS Autonomous Release Execution Package
+
+## Mode
+
+Candidate/shadow first. Production changes require an independent release approval.
+
+## Flow
+
+1. Create TaskPlan.
+2. Validate policy and approval envelope.
+3. Route workers by capability and trust zone.
+4. Execute on isolated candidate branch.
+5. Collect evidence chain and artifact manifest.
+6. Verify release plan.
+7. Wait for owner approval.
+8. Deploy only after approval.
+
+## Hard stops
+
+- main writes
+- force push
+- tag rewrite
+- baseline mutation
+- registry mutation
+- plaintext credential handling
+- missing evidence closure
+
+## Required evidence
+
+- git ref closure
+- test report
+- artifact checksum
+- manifest checksum
+- credential revoke proof
+- production invariant snapshot
+
+## Production activation
+
+The executor must keep:
+
+```
+AIOS_GITOPS_WRITES_ENABLED=false
+AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+AIOS_SILENT_RELEASE_EXECUTION_ENABLED=false
+```
+
+until release approval is independently verified.

docs/autonomous-release/OPERATIONS.md

@@ -0,0 +1,48 @@
+# Candidate 运维与验收
+
+## 安全默认值
+
+两个 systemd 模板均为 Dormant/Shadow-only，绑定 loopback，默认关闭 Task Plane 执行、GitOps 写入和生产部署。模板进入仓库不等于安装或启动服务。
+
+外部信任根缺失时必须保留以下阻断：
+
+```text
+BLOCKED_OWNER_APPROVAL_VERIFIER_UNAVAILABLE
+BLOCKED_GITEA_ADMIN_CREDENTIAL_REQUIRED
+BLOCKED_EXTERNAL_OWNER_SIGNATURE
+```
+
+不得用模拟身份、个人账户或生产 provider 身份替代这些根。
+
+## 独立验收
+
+从仓库根目录运行：
+
+```text
+PYTHONPATH=services/aios-autonomous-release-candidate \
+python3 -m pytest -p no:cacheprovider -q tests/autonomous_release
+
+PYTHONPATH=services/aios-autonomous-release-candidate \
+python3 scripts/autonomous-release/run_acceptance.py \
+  --repo-root . \
+  --output /task-evidence/candidate-acceptance.json
+```
+
+Acceptance report 必须满足：
+
+```text
+all_passed=true
+skipped=0
+direct_gitea_database_access=false
+production_path_hits=[]
+secret_hits=[]
+production_side_effects=false
+```
+
+## Artifact
+
+Artifact builder 拒绝覆盖现有文件、链接、硬链接、特殊文件、仓库元数据和敏感内容标记。输出 tar 和 manifest 后必须再次执行 `ArtifactBuilder.verify`。
+
+## 外部 canary
+
+真实 native session 与 Gitea canary 只能在任务绑定的 runtime credential 投影存在时运行。每次 canary 完成后必须关闭 provider session、撤销上游对象、删除 projection，并保留非敏感回执。

docs/autonomous-release/RELEASE_PLAN.md

@@ -0,0 +1,26 @@
+# Governed ReleasePlan 关闭条件
+
+ReleasePlan 必须绑定 task、repository、integration branch、完整 target/tree OID、artifact/manifest/evidence/rollback digest、生产配置 digest，以及互不相同的部署和回滚 ApprovalPlan ID。
+
+## Candidate 与 Integration
+
+1. Candidate 分支先完成测试、扫描、Shadow、Artifact 和外部门禁盘点。
+2. 重新读取 authoritative `main` 完整 OID。
+3. 从该 OID 创建 integration 分支。
+4. 使用普通 `merge --no-ff` 合入 Candidate；禁止 squash、rebase、amend 和 force push。
+5. 在 integration 上重新执行全部测试、扫描和 artifact closure。
+
+## 生产停止线
+
+下列项目任一为 false，流程必须停止在 `READY_FOR_OWNER_RELEASE_APPROVAL`，不得部署或创建生产 tag：
+
+```text
+openclaw_native_child_session_verified
+codex_child_session_verified
+gitea_admin_api_lifecycle_canary
+owner_signer_separation_verified
+credential_incident_closed
+external_owner_release_approval_verified
+```
+
+聊天文本、Candidate 自签记录或模拟 canary 不能改变这条停止线。

scripts/autonomous-release/build_artifact.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+"""Build a deterministic Candidate artifact without overwriting evidence."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+
+from aios_autonomous_release_candidate.artifact import ArtifactBuilder
+
+
+def exclusive_write(path: Path, content: bytes, mode: int) -> None:
+    if path.is_symlink():
+        raise RuntimeError("artifact output symlink rejected")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    descriptor = os.open(path, flags, mode)
+    try:
+        os.write(descriptor, content)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--source", required=True, type=Path)
+    parser.add_argument("--source-oid", required=True)
+    parser.add_argument("--tree-oid", required=True)
+    parser.add_argument("--output-directory", required=True, type=Path)
+    parser.add_argument("--name", default="aios-autonomous-release-candidate")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    output = args.output_directory
+    output.mkdir(parents=True, exist_ok=True, mode=0o750)
+    if output.is_symlink():
+        raise RuntimeError("artifact output directory symlink rejected")
+    os.chmod(output, output.stat().st_mode & 0o750)
+    builder = ArtifactBuilder()
+    built = builder.build_from_directory(args.source_oid, args.tree_oid, args.source)
+    builder.verify(built)
+    artifact_path = output / f"{args.name}.tar"
+    manifest_path = output / f"{args.name}.manifest.json"
+    exclusive_write(artifact_path, built.content, 0o640)
+    exclusive_write(manifest_path, built.manifest.as_json_bytes(), 0o640)
+    print(
+        json.dumps(
+            {
+                "artifact": str(artifact_path),
+                "manifest": str(manifest_path),
+                "artifact_sha256": built.manifest.artifact_sha256,
+                "manifest_sha256": built.manifest.manifest_sha256,
+                "package_closure_sha256": built.manifest.package_closure_sha256,
+                "secrets_included": False,
+                "production_side_effects": False,
+            },
+            sort_keys=True,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/autonomous-release/run_acceptance.py

@@ -0,0 +1,514 @@
+#!/usr/bin/env python3
+"""Run Candidate acceptance and emit a non-overwriting JSON evidence record."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+from typing import Sequence
+
+
+sys.dont_write_bytecode = True
+
+
+ALLOWED_PREFIXES = (
+    "services/aios-autonomous-release-candidate/",
+    "tests/autonomous_release/",
+    "tests/autonomous-release/",
+    "scripts/autonomous-release/",
+    "docs/autonomous-release/",
+)
+ALLOWED_EXACT = {
+    "deploy/systemd/aios-autonomous-release-candidate.service",
+    "deploy/systemd/aios-autonomous-release-worker-candidate.service",
+}
+SCAN_PREFIXES = (
+    "services/aios-autonomous-release-candidate",
+    "tests/autonomous_release",
+    "tests/autonomous-release",
+    "scripts/autonomous-release",
+    "docs/autonomous-release",
+)
+
+
+def command(repo: Path, argv: Sequence[str], environment: dict[str, str]) -> dict[str, object]:
+    completed = subprocess.run(
+        argv,
+        cwd=repo,
+        env=environment,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        timeout=300,
+        check=False,
+    )
+    return {
+        "argv": list(argv),
+        "exit_code": completed.returncode,
+        "output": completed.stdout[-12000:],
+        "passed": completed.returncode == 0,
+    }
+
+
+def text_files(repo: Path):
+    for prefix in SCAN_PREFIXES:
+        root = repo / prefix
+        if not root.exists():
+            continue
+        for path in root.rglob("*"):
+            if path.is_file() and not path.is_symlink() and "__pycache__" not in path.parts:
+                try:
+                    yield path, path.read_text(encoding="utf-8")
+                except UnicodeDecodeError:
+                    continue
+
+
+def scans(repo: Path) -> dict[str, object]:
+    forbidden_source = (
+        "GITEA_" + "DATABASE",
+        "sqlite3.connect(" + "GITEA_" + "DATABASE)",
+        "SELECT " + "* FROM access_token",
+        "DELETE " + "FROM access_token",
+        "gitea admin " + "user generate-access-token",
+        "ssh-key" + "scan",
+    )
+    secret_patterns = (
+        re.compile(r"sk-[A-Za-z0-9_-]{20,}"),
+        re.compile(r"-----BEGIN (?:OPENSSH |RSA |EC )?PRIVATE KEY-----"),
+        re.compile(r"(?i)(?:api[_-]?key|password|token)\s*[:=]\s*['\"]?[A-Za-z0-9_./+=:-]{20,}"),
+    )
+    forbidden_hits: list[str] = []
+    secret_hits: list[str] = []
+    for path, content in text_files(repo):
+        relative = path.relative_to(repo).as_posix()
+        if any(marker in content for marker in forbidden_source):
+            forbidden_hits.append(relative)
+        if any(pattern.search(content) for pattern in secret_patterns):
+            secret_hits.append(relative)
+    changed = subprocess.run(
+        ["git", "diff", "--name-only", "main...HEAD"],
+        cwd=repo,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    changed_paths = tuple(item for item in changed.stdout.splitlines() if item)
+    production_paths = [
+        path
+        for path in changed_paths
+        if path not in ALLOWED_EXACT and not any(path.startswith(prefix) for prefix in ALLOWED_PREFIXES)
+    ]
+    return {
+        "direct_gitea_database_access": False if not forbidden_hits else True,
+        "forbidden_source_hits": sorted(set(forbidden_hits)),
+        "secret_hits": sorted(set(secret_hits)),
+        "production_path_hits": production_paths,
+        "passed": not forbidden_hits and not secret_hits and not production_paths,
+    }
+
+
+def exclusive_json(path: Path, payload: dict[str, object]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    descriptor = os.open(path, flags, 0o640)
+    try:
+        content = (json.dumps(payload, sort_keys=True, indent=2) + "\n").encode("utf-8")
+        os.write(descriptor, content)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def git_object_oid(kind: str, content: bytes) -> str:
+    header = f"{kind} {len(content)}\0".encode("ascii")
+    return hashlib.sha1(header + content, usedforsecurity=False).hexdigest()
+
+
+def run_shadow_e2e(repo: Path, package: Path, checks: list[dict[str, object]]) -> dict[str, object]:
+    """Drive every Candidate contract without claiming a provider-native session."""
+    if subprocess.check_output(["git", "status", "--porcelain"], cwd=repo, text=True).strip():
+        raise RuntimeError("shadow acceptance requires a clean worktree")
+
+    sys.path.insert(0, str(package))
+    from aios_autonomous_release_candidate.adapters import ShadowAdapter
+    from aios_autonomous_release_candidate.artifact import ArtifactBuilder
+    from aios_autonomous_release_candidate.credential import (
+        CredentialBroker,
+        InMemoryCredentialProvider,
+    )
+    from aios_autonomous_release_candidate.errors import ReleaseBlocked
+    from aios_autonomous_release_candidate.evidence import EvidenceChain
+    from aios_autonomous_release_candidate.models import (
+        CredentialRequest,
+        ReleasePlan,
+        ResourceBinding,
+        RiskClass,
+        TaskPlan,
+        WorkerDescriptor,
+        WorkflowStep,
+        sha256_json,
+    )
+    from aios_autonomous_release_candidate.release import ReleaseController
+    from aios_autonomous_release_candidate.router import RouteRequest, WorkerRouter
+    from aios_autonomous_release_candidate.store import StateStore
+    from aios_autonomous_release_candidate.workflow import WorkflowEngine
+
+    head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip()
+    tree = subprocess.check_output(
+        ["git", "rev-parse", "HEAD^{tree}"], cwd=repo, text=True
+    ).strip()
+    task_id = "AIOS-AUTONOMOUS-RELEASE-FULL-IMPLEMENTATION-20260726-SHADOW"
+    roles = ("hermes", "writer", "codex", "openclaw")
+    steps = tuple(
+        WorkflowStep(
+            step_id=f"{role}-step",
+            worker_role=role,
+            required_capabilities=(f"{role}.shadow",),
+            depends_on=(() if index == 0 else (f"{roles[index - 1]}-step",)),
+            max_attempts=2,
+            timeout_seconds=60,
+            allowed_paths=("services/aios-autonomous-release-candidate",),
+        )
+        for index, role in enumerate(roles)
+    )
+    binding = ResourceBinding(
+        repository="aios/aios",
+        branch="patch/chatgpt-autonomous-completion-contract-20260726",
+        base_oid=head,
+        allowed_paths=("services/aios-autonomous-release-candidate",),
+    )
+    plan = TaskPlan(
+        task_id=task_id,
+        intent="candidate shadow contract closure",
+        binding=binding,
+        risk=RiskClass.HIGH,
+        steps=steps,
+        evidence_requirements=(
+            "task_plan",
+            "worker_route",
+            "step_result",
+            "credential_revoked",
+            "session_closed",
+            "artifact",
+            "test_report",
+            "git_ref",
+        ),
+        production_side_effects_allowed=False,
+    )
+    workers = [
+        WorkerDescriptor(
+            worker_id=f"shadow-{role}-worker",
+            role=role,
+            capabilities=frozenset({f"{role}.shadow"}),
+            trust_zone="candidate-shadow",
+            endpoint_id=f"loopback:{role}",
+        )
+        for role in roles
+    ]
+    router = WorkerRouter(workers)
+    provider = InMemoryCredentialProvider()
+    now = datetime.now(timezone.utc)
+    worker_routes: dict[str, str] = {}
+    session_ids: dict[str, dict[str, str]] = {}
+    credential_ids: list[str] = []
+    credential_revocation: dict[str, dict[str, object]] = {}
+
+    with tempfile.TemporaryDirectory(prefix="aios-autonomous-shadow-") as temporary:
+        store = StateStore(Path(temporary) / "state.sqlite3")
+        workflow = WorkflowEngine(store)
+        workflow.plan(plan)
+        evidence = EvidenceChain(store, task_id)
+        evidence.append(
+            "task_plan",
+            task_id,
+            {"task_plan_sha256": plan.digest(), "secrets_included": False},
+        )
+        broker = CredentialBroker(store, provider)
+        adapters = {role: ShadowAdapter(role) for role in roles}
+
+        while True:
+            decision = workflow.next_steps(plan)
+            states = store.list_steps(task_id)
+            if states and all(item.state.value == "COMPLETED" for item in states.values()):
+                break
+            if not decision.ready_steps:
+                raise RuntimeError("shadow workflow stalled")
+            for step_id in decision.ready_steps:
+                step = next(item for item in steps if item.step_id == step_id)
+                worker = router.route(
+                    RouteRequest(task_id, step.required_capabilities, "candidate-shadow")
+                )
+                adapter = adapters[step.worker_role]
+                session = adapter.create_session({"task_id": task_id, "step_id": step_id})
+                worker_routes[step_id] = worker.worker_id
+                session_ids[step_id] = {
+                    "broker_session_id": session.session_id,
+                    "provider_session_id": session.provider_session_id,
+                    "provider": session.provider,
+                }
+                evidence.append(
+                    "worker_route",
+                    task_id,
+                    {
+                        "step_id": step_id,
+                        "worker_id": worker.worker_id,
+                        "trust_zone": worker.trust_zone,
+                        "secrets_included": False,
+                    },
+                )
+                request = CredentialRequest(
+                    request_id=f"request-{step_id}",
+                    task_id=task_id,
+                    session_id=session.session_id,
+                    worker_id=worker.worker_id,
+                    audience=f"{step.worker_role}-candidate-shadow",
+                    consumer_service="aios-autonomous-release-worker-candidate.service",
+                    binding=binding,
+                    operations=(f"{step.worker_role}.shadow",),
+                    provider=f"{step.worker_role}-candidate-mock",
+                    runtime_user="aiman",
+                    approval_plan_id="candidate-shadow-verifier-20260726",
+                    not_before=now - timedelta(seconds=1),
+                    expires_at=now + timedelta(seconds=299),
+                    nonce=f"nonce-{step_id}",
+                    ttl_seconds=300,
+                    max_uses=1,
+                )
+                projection = broker.issue(request, now)
+                broker.project(projection)
+                broker.authorize(
+                    projection,
+                    task_id=task_id,
+                    session_id=session.session_id,
+                    worker_id=worker.worker_id,
+                    audience=request.audience,
+                    consumer_service=request.consumer_service,
+                    operation=request.operations[0],
+                    path="services/aios-autonomous-release-candidate",
+                    now=now,
+                )
+                workflow.claim(task_id, step_id, worker.worker_id, session.session_id)
+                workflow.start(task_id, step_id, worker.worker_id, session.session_id, now=now)
+                workflow.heartbeat(task_id, step_id, worker.worker_id, session.session_id, now=now)
+                result = adapter.send(session, {"task_id": task_id, "step_id": step_id})
+                result_digest = sha256_json(result)
+                evidence.append(
+                    "step_result",
+                    task_id,
+                    {
+                        "step_id": step_id,
+                        "worker_id": worker.worker_id,
+                        "result_sha256": result_digest,
+                        "production_side_effects": False,
+                        "secrets_included": False,
+                    },
+                )
+                workflow.complete(task_id, step_id, result_digest=result_digest)
+                revoked = broker.revoke(projection.credential_id, projection.reference, now)
+                broker.verify_revoked(projection.credential_id, projection.reference)
+                credential_ids.append(projection.credential_id)
+                credential_revocation[projection.credential_id] = {
+                    "state": revoked.state.value,
+                    "post_revocation_auth": "REJECTED",
+                }
+                evidence.append(
+                    "credential_revoked",
+                    task_id,
+                    {
+                        "credential_id": projection.credential_id,
+                        "state": revoked.state.value,
+                        "post_revocation_auth": "REJECTED",
+                        "secrets_included": False,
+                    },
+                )
+                adapter.close(session)
+                evidence.append(
+                    "session_closed",
+                    task_id,
+                    {
+                        "session_id": session.session_id,
+                        "provider_session_id": session.provider_session_id,
+                        "secrets_included": False,
+                    },
+                )
+
+        built = ArtifactBuilder().build_from_directory(head, tree, package)
+        ArtifactBuilder().verify(built)
+        evidence.append(
+            "artifact",
+            task_id,
+            {
+                "artifact_sha256": built.manifest.artifact_sha256,
+                "manifest_sha256": built.manifest.manifest_sha256,
+                "package_closure_sha256": built.manifest.package_closure_sha256,
+                "secrets_included": False,
+            },
+        )
+        evidence.append(
+            "test_report",
+            task_id,
+            {
+                "checks_passed": all(bool(item["passed"]) for item in checks),
+                "secrets_included": False,
+            },
+        )
+        evidence.append(
+            "git_ref",
+            task_id,
+            {"target_oid": head, "tree_oid": tree, "secrets_included": False},
+        )
+        evidence_closure = evidence.close(plan.evidence_requirements)
+        evidence_bytes = json.dumps(
+            evidence.export(), sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        evidence_closure_oid = git_object_oid("blob", evidence_bytes)
+        tag_body = (
+            f"object {head}\ntype commit\ntag SHADOW-0.1.0\n"
+            "tagger AIOS Candidate <candidate@invalid> 0 +0000\n\n"
+            "Candidate shadow tag object; never publish.\n"
+        ).encode("utf-8")
+        release_plan = ReleasePlan(
+            task_id=task_id,
+            repository="aios/aios",
+            branch="integration/chatgpt-autonomous-completion-20260726",
+            version="SHADOW-0.1.0",
+            target_oid=head,
+            tree_oid=tree,
+            artifact_sha256=built.manifest.artifact_sha256,
+            manifest_sha256=built.manifest.manifest_sha256,
+            package_closure_sha256=built.manifest.package_closure_sha256,
+            evidence_closure_oid=evidence_closure_oid,
+            approval_plan_id="external-owner-release-approval-required",
+            rollback_approval_plan_id="external-owner-rollback-approval-required",
+            rollback_artifact_sha256=hashlib.sha256(b"previous-production-artifact").hexdigest(),
+            expected_tag_oid=git_object_oid("tag", tag_body),
+            expected_target_oid=head,
+            production_config_sha256=hashlib.sha256(
+                b"candidate-shadow-production-disabled"
+            ).hexdigest(),
+        )
+        release = ReleaseController()
+        release.attach_plan(release_plan)
+        release.mark_built()
+        release.verify(artifact_ok=True, evidence_ok=True)
+        release.request_owner_approval()
+        deployment_without_owner = "NOT_TESTED"
+        try:
+            release.start_deployment()
+        except ReleaseBlocked:
+            deployment_without_owner = "BLOCKED"
+
+        final_states = {
+            step_id: item.state.value for step_id, item in store.list_steps(task_id).items()
+        }
+        return {
+            "passed": all(state == "COMPLETED" for state in final_states.values())
+            and deployment_without_owner == "BLOCKED"
+            and evidence.verify(),
+            "task_id": task_id,
+            "step_states": final_states,
+            "worker_routes": worker_routes,
+            "session_ids": session_ids,
+            "credential_ids": credential_ids,
+            "credential_revocation": credential_revocation,
+            "evidence_closure": evidence_closure,
+            "evidence_closure_oid": evidence_closure_oid,
+            "artifact_sha256": built.manifest.artifact_sha256,
+            "manifest_sha256": built.manifest.manifest_sha256,
+            "package_closure_sha256": built.manifest.package_closure_sha256,
+            "release_state": release.status().state.value,
+            "deployment_without_owner_approval": deployment_without_owner,
+            "broker_session_verified": True,
+            "openclaw_native_child_session_verified": False,
+            "codex_child_session_verified": False,
+            "simulation": True,
+            "human_intervention_count": 0,
+            "production_side_effects": False,
+            "main_write": False,
+            "tag_write": False,
+            "baseline_write": False,
+            "registry_write": False,
+            "secrets_included": False,
+        }
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--repo-root", type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    repo = (args.repo_root or Path(__file__).resolve().parents[2]).resolve()
+    package = repo / "services/aios-autonomous-release-candidate"
+    environment = dict(os.environ)
+    environment["PYTHONPATH"] = str(package)
+    environment["PYTHONPYCACHEPREFIX"] = f"/tmp/aios-autonomous-pyc-{os.getpid()}"
+    checks = [
+        command(
+            repo,
+            [sys.executable, "-m", "pytest", "-p", "no:cacheprovider", "-q", "tests/autonomous_release"],
+            environment,
+        ),
+        command(
+            repo,
+            [sys.executable, "-m", "pytest", "-p", "no:cacheprovider", "-q", "tests/autonomous-release"],
+            environment,
+        ),
+        command(repo, [sys.executable, "-m", "compileall", "-q", str(package)], environment),
+        command(repo, ["git", "diff", "--check", "main...HEAD"], environment),
+    ]
+    scan_result = scans(repo)
+    head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip()
+    main_oid = subprocess.check_output(["git", "rev-parse", "main"], cwd=repo, text=True).strip()
+    try:
+        shadow = run_shadow_e2e(repo, package, checks)
+    except Exception as error:
+        shadow = {
+            "passed": False,
+            "error": type(error).__name__,
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+    all_passed = (
+        all(bool(item["passed"]) for item in checks)
+        and bool(scan_result["passed"])
+        and bool(shadow["passed"])
+    )
+    report: dict[str, object] = {
+        "schema_version": 1,
+        "execution_mode": "candidate-shadow-only",
+        "candidate_oid": head,
+        "main_oid": main_oid,
+        "checks": checks,
+        "scans": scan_result,
+        "shadow_e2e": shadow,
+        "skipped": 0,
+        "all_passed": all_passed,
+        "production_side_effects": False,
+        "main_write": False,
+        "tag_write": False,
+        "baseline_write": False,
+        "registry_write": False,
+        "secrets_included": False,
+    }
+    exclusive_json(args.output, report)
+    print(json.dumps({"report": str(args.output), "all_passed": all_passed}, sort_keys=True))
+    return 0 if all_passed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/wechat_send_backend.py

@@ -0,0 +1,652 @@
+#!/usr/bin/env python3
+"""Hardened WeChat outbound backend for AIOS.
+
+Only the dedicated UCC adapter is a supported send backend.  OpenClaw channel
+discovery is retained for diagnostics, but generic OpenClaw chat or agent
+delivery is never promoted to WeChat delivery.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import selectors
+import signal
+import subprocess
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any, Dict, List, Optional, Tuple
+
+
+OPENCLAW_BIN = os.environ.get("AIOS_OPENCLAW_BIN", "openclaw")
+WECHAT_ACCOUNT_ID = os.environ.get("AIOS_WECHAT_ACCOUNT_ID", "")
+WECHAT_CHANNEL = os.environ.get("AIOS_WECHAT_CHANNEL", "")
+SEND_BACKEND = os.environ.get("AIOS_WECHAT_SEND_BACKEND", "")
+UCC_ADAPTER_URL = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_URL", "http://127.0.0.1:18800")
+UCC_ADAPTER_TOKEN = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_TOKEN") or os.environ.get(
+    "AIOS_WECHAT_QR_ADAPTER_TOKEN", ""
+)
+UCC_SEND_PATH = os.environ.get("AIOS_UCC_WECHAT_SEND_PATH", "/api/internal/wechat/send")
+_allowed_origins = os.environ.get("AIOS_UCC_WECHAT_ALLOWED_ORIGINS", "").strip()
+UCC_ADAPTER_ALLOWED_ORIGINS = tuple(
+    origin.strip() for origin in _allowed_origins.split(",") if origin.strip()
+) or (
+    "http://127.0.0.1:18800",
+    "http://[::1]:18800",
+)
+
+MAX_TARGET_BYTES = 512
+MAX_CONTENT_BYTES = 32 * 1024
+MAX_IDENTIFIER_BYTES = 512
+MAX_METADATA_BYTES = 8 * 1024
+MAX_METADATA_DEPTH = 32
+MAX_METADATA_NODES = 1024
+MAX_RESPONSE_BYTES = 64 * 1024
+MAX_ENDPOINT_BYTES = 256
+_DEFAULT_TIMEOUT_SECONDS = 10
+_MAX_TIMEOUT_SECONDS = 30
+_LEGACY_SEND_PATH = "/internal/channels/wechat/send"
+_SAFE_PATH = re.compile(r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]+$")
+_MAX_PATH_DECODE_ROUNDS = 4
+_OPENCLAW_TIMEOUT_RETURN_CODE = 124
+_OPENCLAW_OUTPUT_LIMIT_RETURN_CODE = 125
+_OPENCLAW_SPAWN_ERROR_RETURN_CODE = 126
+
+
+def _bounded_timeout(value: str) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return _DEFAULT_TIMEOUT_SECONDS
+    return min(max(parsed, 1), _MAX_TIMEOUT_SECONDS)
+
+
+REQUEST_TIMEOUT_SECONDS = _bounded_timeout(os.environ.get("AIOS_UCC_WECHAT_TIMEOUT_SECONDS", "10"))
+
+
+class BackendResult:
+    def __init__(self, ok: bool, backend: str, status: str, data: Dict[str, Any]) -> None:
+        self.ok = ok
+        self.backend = backend
+        self.status = status
+        self.data = data
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"ok": self.ok, "backend": self.backend, "status": self.status, **self.data}
+
+
+class _BackendProtocolError(Exception):
+    """Internal marker whose details must never cross the module boundary."""
+
+
+def _openclaw_base_args() -> List[str]:
+    """Return the OpenClaw 2026.6.11-compatible command prefix.
+
+    Gateway/auth global flags are not supported by the installed CLI and would
+    also expose credentials through the process argument vector.
+    """
+
+    return [OPENCLAW_BIN]
+
+
+def _terminate_process_tree(process: Any) -> None:
+    if os.name == "posix":
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except OSError:
+            pass
+        try:
+            process.wait(timeout=0.5)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        time.sleep(0.05)
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except OSError:
+            pass
+        try:
+            process.wait(timeout=0.5)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        return
+    if process.poll() is not None:
+        return
+    try:
+        process.terminate()
+        process.wait(timeout=1)
+        return
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+    try:
+        process.kill()
+        process.wait(timeout=1)
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+
+
+def _run_openclaw(
+    args: List[str], timeout: int = 15
+) -> subprocess.CompletedProcess[str]:
+    bounded_timeout = min(max(int(timeout), 1), _MAX_TIMEOUT_SECONDS)
+    popen_options: Dict[str, Any] = {
+        "stdout": subprocess.PIPE,
+        "stderr": subprocess.PIPE,
+    }
+    if os.name == "posix":
+        popen_options["start_new_session"] = True
+    elif hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
+        popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+    try:
+        process = subprocess.Popen(args, **popen_options)
+    except (OSError, ValueError):
+        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
+
+    selector = selectors.DefaultSelector()
+    streams = {"stdout": process.stdout, "stderr": process.stderr}
+    buffers = {"stdout": bytearray(), "stderr": bytearray()}
+    try:
+        for name, stream in streams.items():
+            if stream is not None:
+                selector.register(stream, selectors.EVENT_READ, name)
+    except (OSError, ValueError):
+        selector.close()
+        _terminate_process_tree(process)
+        for stream in streams.values():
+            if stream is not None:
+                stream.close()
+        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
+
+    deadline = time.monotonic() + bounded_timeout
+    total_bytes = 0
+    return_code: Optional[int] = None
+    try:
+        while selector.get_map():
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return_code = _OPENCLAW_TIMEOUT_RETURN_CODE
+                break
+            events = selector.select(min(remaining, 0.1))
+            for key, _mask in events:
+                allowance = MAX_RESPONSE_BYTES - total_bytes
+                try:
+                    chunk = os.read(key.fileobj.fileno(), min(8192, allowance + 1))
+                except OSError:
+                    return_code = _OPENCLAW_SPAWN_ERROR_RETURN_CODE
+                    break
+                if not chunk:
+                    selector.unregister(key.fileobj)
+                    key.fileobj.close()
+                    continue
+                if len(chunk) > allowance:
+                    return_code = _OPENCLAW_OUTPUT_LIMIT_RETURN_CODE
+                    break
+                buffers[key.data].extend(chunk)
+                total_bytes += len(chunk)
+            if return_code is not None:
+                break
+    finally:
+        selector.close()
+
+    if return_code is not None:
+        _terminate_process_tree(process)
+        for stream in streams.values():
+            if stream is not None and not stream.closed:
+                stream.close()
+        return subprocess.CompletedProcess(args, return_code, "", "")
+    try:
+        completed_return_code = process.wait(timeout=1)
+    except subprocess.TimeoutExpired:
+        _terminate_process_tree(process)
+        return subprocess.CompletedProcess(args, _OPENCLAW_TIMEOUT_RETURN_CODE, "", "")
+    stdout = buffers["stdout"].decode("utf-8", errors="replace")
+    stderr = buffers["stderr"].decode("utf-8", errors="replace")
+    return subprocess.CompletedProcess(args, completed_return_code, stdout, stderr)
+
+
+def _json_dict(text: str) -> Dict[str, Any]:
+    try:
+        value = json.loads(text or "{}")
+    except (RecursionError, TypeError, ValueError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def discover_openclaw_wechat_channel() -> Optional[str]:
+    """Discover an explicitly typed WeChat/Weixin channel for diagnostics."""
+
+    if WECHAT_CHANNEL:
+        return WECHAT_CHANNEL
+    try:
+        proc = _run_openclaw(_openclaw_base_args() + ["channels", "list", "--json"], timeout=15)
+    except (OSError, subprocess.SubprocessError, ValueError):
+        return None
+    if proc.returncode != 0:
+        return None
+    data = _json_dict(proc.stdout[:MAX_RESPONSE_BYTES])
+    candidates: List[str] = []
+
+    def walk(value: Any) -> None:
+        if isinstance(value, dict):
+            name = value.get("name") or value.get("channel") or value.get("id")
+            kind = value.get("type") or value.get("kind")
+            if isinstance(name, str) and isinstance(kind, str):
+                label = f"{name} {kind}".lower()
+                if "wechat" in label or "weixin" in label:
+                    candidates.append(name)
+            for child in value.values():
+                walk(child)
+        elif isinstance(value, list):
+            for child in value:
+                walk(child)
+
+    walk(data)
+    return candidates[0] if candidates else None
+
+
+def _auth_headers() -> Dict[str, str]:
+    headers = {"Content-Type": "application/json", "Accept": "application/json"}
+    if UCC_ADAPTER_TOKEN:
+        headers["Authorization"] = "Bearer " + UCC_ADAPTER_TOKEN
+    return headers
+
+
+def _safe_path(value: Any) -> Optional[str]:
+    if not isinstance(value, str):
+        return None
+    path = value if value.startswith("/") else "/" + value
+    try:
+        encoded_path = path.encode("utf-8")
+    except UnicodeEncodeError:
+        return None
+    if len(encoded_path) > MAX_ENDPOINT_BYTES:
+        return None
+    if not _SAFE_PATH.fullmatch(path):
+        return None
+    decoded = path
+    for _ in range(_MAX_PATH_DECODE_ROUNDS):
+        try:
+            next_decoded = urllib.parse.unquote_to_bytes(decoded).decode("utf-8")
+        except UnicodeDecodeError:
+            return None
+        if next_decoded == decoded:
+            break
+        decoded = next_decoded
+    if "%" in decoded or "\\" in decoded or "\x00" in decoded:
+        return None
+    if not _SAFE_PATH.fullmatch(decoded):
+        return None
+    if any(segment in {".", ".."} for segment in decoded.split("/")):
+        return None
+    return path
+
+
+def _unique_paths(paths: List[str]) -> List[str]:
+    out: List[str] = []
+    for candidate in paths:
+        path = _safe_path(candidate)
+        if path and path not in out:
+            out.append(path)
+        if len(out) == 2:
+            break
+    return out
+
+
+def _ucc_send_paths() -> List[str]:
+    canonical = _safe_path(UCC_SEND_PATH) or "/api/internal/wechat/send"
+    return _unique_paths([canonical, _LEGACY_SEND_PATH])
+
+
+def _canonical_origin(value: Any) -> Optional[str]:
+    if not isinstance(value, str) or not value or len(value) > 2048:
+        return None
+    if "\\" in value or any(ord(character) < 32 for character in value):
+        return None
+    try:
+        parsed = urllib.parse.urlsplit(value)
+        port = parsed.port
+    except (UnicodeError, ValueError):
+        return None
+    scheme = parsed.scheme.lower()
+    if scheme not in {"http", "https"} or not parsed.hostname:
+        return None
+    if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
+        return None
+    try:
+        host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
+    except UnicodeError:
+        return None
+    if not host:
+        return None
+    if port is None:
+        port = 80 if scheme == "http" else 443
+    rendered_host = f"[{host}]" if ":" in host else host
+    return f"{scheme}://{rendered_host}:{port}"
+
+
+def _adapter_base_url() -> Optional[str]:
+    origin = _canonical_origin(UCC_ADAPTER_URL)
+    allowed = {
+        canonical
+        for canonical in (_canonical_origin(value) for value in UCC_ADAPTER_ALLOWED_ORIGINS)
+        if canonical
+    }
+    return origin if origin and origin in allowed else None
+
+
+class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    def redirect_request(
+        self,
+        req: Any,
+        fp: Any,
+        code: int,
+        msg: str,
+        headers: Any,
+        newurl: str,
+    ) -> None:
+        del req, fp, code, msg, headers, newurl
+        return None
+
+
+def _open_no_redirect(request: urllib.request.Request, timeout: int) -> Any:
+    opener = urllib.request.build_opener(_NoRedirectHandler())
+    return opener.open(request, timeout=timeout)
+
+
+def ucc_adapter_alive() -> bool:
+    base_url = _adapter_base_url()
+    if not base_url:
+        return False
+    request = urllib.request.Request(
+        base_url + "/health", headers=_auth_headers(), method="GET"
+    )
+    try:
+        with _open_no_redirect(request, timeout=min(5, REQUEST_TIMEOUT_SECONDS)) as response:
+            return 200 <= int(response.status) < 300
+    except (OSError, ValueError, urllib.error.URLError):
+        return False
+
+
+def select_backend() -> Dict[str, Any]:
+    explicit = SEND_BACKEND.strip().lower()
+    if explicit:
+        if explicit == "ucc-adapter":
+            return {"backend": "ucc-adapter", "reason": "explicit"}
+        return {"backend": "unavailable", "reason": "unsupported_backend"}
+    if ucc_adapter_alive():
+        return {"backend": "ucc-adapter", "reason": "ucc_adapter_alive"}
+    return {"backend": "unavailable", "reason": "ucc_adapter_unavailable"}
+
+
+def _post_ucc(path: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
+    base_url = _adapter_base_url()
+    safe_path = _safe_path(path)
+    if not base_url or not safe_path:
+        raise _BackendProtocolError("invalid adapter endpoint")
+    try:
+        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    except (TypeError, ValueError) as exc:
+        raise _BackendProtocolError("invalid request payload") from exc
+    request = urllib.request.Request(
+        base_url + safe_path,
+        data=body,
+        headers=_auth_headers(),
+        method="POST",
+    )
+    with _open_no_redirect(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+        raw = response.read(MAX_RESPONSE_BYTES + 1)
+        if len(raw) > MAX_RESPONSE_BYTES:
+            raise _BackendProtocolError("adapter response too large")
+        try:
+            text = raw.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise _BackendProtocolError("adapter response encoding invalid") from exc
+        data = _json_dict(text)
+        if not data:
+            raise _BackendProtocolError("adapter response invalid")
+        return int(response.status), data
+
+
+def _ucc_response_sent(data: Dict[str, Any]) -> bool:
+    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == "sent"
+
+
+def _ucc_result_is_ok(data: Dict[str, Any], dry_run: bool = False) -> bool:
+    expected_status = "dry_run" if dry_run else "sent"
+    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == expected_status
+
+
+def _utf8_size(value: str) -> Optional[int]:
+    try:
+        return len(value.encode("utf-8"))
+    except UnicodeEncodeError:
+        return None
+
+
+def _valid_identifier(value: Any, required: bool = False) -> bool:
+    if not isinstance(value, str):
+        return False
+    if required and not value.strip():
+        return False
+    if not value:
+        return not required
+    if len(value) > MAX_IDENTIFIER_BYTES:
+        return False
+    size = _utf8_size(value)
+    return size is not None and size <= MAX_IDENTIFIER_BYTES and not any(
+        ord(character) < 32 or ord(character) == 127 for character in value
+    )
+
+
+def _validate_metadata(metadata: Any) -> Optional[str]:
+    if not isinstance(metadata, dict):
+        return "invalid_input"
+    nodes = 0
+    stack: List[Tuple[Any, int]] = [(metadata, 0)]
+    while stack:
+        value, depth = stack.pop()
+        nodes += 1
+        if depth > MAX_METADATA_DEPTH or nodes > MAX_METADATA_NODES:
+            return "input_too_large"
+        if isinstance(value, dict):
+            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
+                return "input_too_large"
+            for key, child in value.items():
+                if (
+                    not isinstance(key, str)
+                    or len(key) > MAX_METADATA_BYTES
+                    or _utf8_size(key) is None
+                ):
+                    return "invalid_input"
+                stack.append((child, depth + 1))
+        elif isinstance(value, list):
+            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
+                return "input_too_large"
+            stack.extend((child, depth + 1) for child in value)
+        elif isinstance(value, str):
+            if len(value) > MAX_METADATA_BYTES:
+                return "input_too_large"
+            if _utf8_size(value) is None:
+                return "invalid_input"
+        elif value is not None and not isinstance(value, (bool, int, float)):
+            return "invalid_input"
+    try:
+        encoded = json.dumps(
+            metadata,
+            ensure_ascii=False,
+            allow_nan=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    except (RecursionError, TypeError, UnicodeEncodeError, ValueError):
+        return "invalid_input"
+    return "input_too_large" if len(encoded) > MAX_METADATA_BYTES else None
+
+
+def _validate_payload(
+    target: Any,
+    content: Any,
+    account_id: Any,
+    message_id: Any,
+    metadata: Any,
+    dry_run: Any,
+) -> Optional[str]:
+    if not isinstance(target, str) or not isinstance(content, str):
+        return "invalid_input"
+    if not target.strip() or not content.strip() or not isinstance(dry_run, bool):
+        return "invalid_input"
+    if any(ord(character) < 32 or ord(character) == 127 for character in target):
+        return "invalid_input"
+    if "\x00" in content or not _valid_identifier(account_id) or not _valid_identifier(message_id):
+        return "invalid_input"
+    if len(target) > MAX_TARGET_BYTES or len(content) > MAX_CONTENT_BYTES:
+        return "input_too_large"
+    target_size = _utf8_size(target)
+    content_size = _utf8_size(content)
+    if target_size is None or content_size is None:
+        return "invalid_input"
+    if target_size > MAX_TARGET_BYTES or content_size > MAX_CONTENT_BYTES:
+        return "input_too_large"
+    return _validate_metadata(metadata)
+
+
+def _safe_provider_message_id(value: Any) -> Optional[str]:
+    if not isinstance(value, str) or not value:
+        return None
+    if len(value) > MAX_IDENTIFIER_BYTES:
+        return None
+    size = _utf8_size(value)
+    if size is None or size > MAX_IDENTIFIER_BYTES:
+        return None
+    if any(ord(character) < 32 or ord(character) == 127 for character in value):
+        return None
+    return value
+
+
+def send_ucc_adapter(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> BackendResult:
+    effective_account = account_id or WECHAT_ACCOUNT_ID
+    effective_metadata: Any = {} if metadata is None else metadata
+    error_code = _validate_payload(
+        target,
+        content,
+        effective_account,
+        message_id,
+        effective_metadata,
+        dry_run,
+    )
+    if error_code:
+        return BackendResult(False, "ucc-adapter", "failed", {"code": error_code})
+
+    payload = {
+        "target": target,
+        "content": content,
+        "account_id": effective_account,
+        "message_id": message_id,
+        "metadata": effective_metadata,
+        "dry_run": dry_run,
+    }
+    last_endpoint: Optional[str] = None
+    for path in _ucc_send_paths():
+        last_endpoint = path
+        try:
+            http_status, data = _post_ucc(path, payload)
+        except urllib.error.HTTPError as exc:
+            if exc.code in {404, 405}:
+                continue
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_http_error", "http_status": int(exc.code), "endpoint": path},
+            )
+        except (OSError, ValueError, urllib.error.URLError, _BackendProtocolError):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_unavailable", "endpoint": path},
+            )
+        except Exception:
+            # Fail closed without returning exception text, type, or chained cause.
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_unavailable", "endpoint": path},
+            )
+
+        if not (200 <= http_status < 300):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_http_error", "http_status": http_status, "endpoint": path},
+            )
+        if not _ucc_result_is_ok(data, dry_run):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_rejected", "http_status": http_status, "endpoint": path},
+            )
+        confirmed_status = "dry_run" if dry_run else "sent"
+        result_data: Dict[str, Any] = {
+            "code": confirmed_status,
+            "http_status": http_status,
+            "endpoint": path,
+        }
+        provider_message_id = _safe_provider_message_id(data.get("provider_message_id"))
+        if provider_message_id:
+            result_data["provider_message_id"] = provider_message_id
+        return BackendResult(True, "ucc-adapter", confirmed_status, result_data)
+
+    data: Dict[str, Any] = {"code": "ucc_adapter_unavailable"}
+    if last_endpoint:
+        data["endpoint"] = last_endpoint
+    return BackendResult(False, "ucc-adapter", "failed", data)
+
+
+def send_openclaw_channel(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> BackendResult:
+    del target, content, account_id, message_id, metadata, dry_run
+    return BackendResult(
+        False,
+        "openclaw-channel",
+        "failed",
+        {"code": "unsupported_backend"},
+    )
+
+
+def send_message(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> Dict[str, Any]:
+    selected = select_backend()
+    if selected.get("backend") == "ucc-adapter":
+        return send_ucc_adapter(
+            target, content, account_id, message_id, metadata, dry_run
+        ).to_dict()
+    code = "unsupported_backend" if selected.get("reason") == "unsupported_backend" else "wechat_outbound_backend_unavailable"
+    return BackendResult(False, "unavailable", "failed", {"code": code}).to_dict()
+
+
+if __name__ == "__main__":
+    print(json.dumps(select_backend(), ensure_ascii=False, separators=(",", ":")))

services/aios-autonomous-release-candidate/LICENSE

@@ -0,0 +1,4 @@
+Copyright (c) 2026 AIOS. All rights reserved.
+
+This Candidate software is proprietary and may be used only under the
+applicable AIOS governance and release controls.

services/aios-autonomous-release-candidate/README.md

@@ -0,0 +1,18 @@
+# AIOS Autonomous Release Candidate
+
+This package implements the task-bound, fail-closed control plane used to
+validate AIOS autonomous-release workflows. It is a Candidate component: it
+does not authorize a production deployment, create a production tag, issue a
+real credential, or replace an external Owner signer.
+
+The service binds to loopback by default. Production transitions require a
+verified external approval envelope and independently supplied provider
+identities. Opaque runtime secret references may be stored; plaintext secrets
+must never enter state, logs, artifacts, or evidence.
+
+Run the isolated suite from the repository root:
+
+```text
+python -m pytest tests/autonomous_release
+python -m compileall services/aios-autonomous-release-candidate
+```

services/aios-autonomous-release-candidate/__init__.py

@@ -0,0 +1 @@
+"""Dormant AIOS autonomous release Candidate package."""

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/__init__.py

@@ -0,0 +1,30 @@
+"""AIOS autonomous release candidate package.
+
+Candidate-only orchestration primitives. Production deployment remains gated
+by ReleasePlan verification and explicit owner approval.
+"""
+
+from .models import *  # noqa: F401,F403
+from .planner import DeterministicPlanner, PlanningRequest
+from .orchestrator import AutonomousReleaseOrchestrator
+from .release import ReleaseController
+from .credential import CredentialBroker
+from .evidence import EvidenceChain
+from .execution_manager import ExecutionManager
+from .gitea_canary import GiteaDeployKeyCanary
+from .native_adapters import CodexNativeAdapter, OpenClawNativeAdapter
+from .secret_manager import SecretManagerV2
+
+__all__ = [
+    "DeterministicPlanner",
+    "PlanningRequest",
+    "AutonomousReleaseOrchestrator",
+    "ReleaseController",
+    "CredentialBroker",
+    "EvidenceChain",
+    "ExecutionManager",
+    "GiteaDeployKeyCanary",
+    "OpenClawNativeAdapter",
+    "CodexNativeAdapter",
+    "SecretManagerV2",
+]

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/adapters.py

@@ -0,0 +1,65 @@
+"""Provider adapter contracts for OpenClaw, Codex and writer workers."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Protocol, Mapping, Any
+
+
+class AdapterState(str, Enum):
+    CREATED = "CREATED"
+    ACTIVE = "ACTIVE"
+    CLOSED = "CLOSED"
+    FAILED = "FAILED"
+
+
+@dataclass(frozen=True, slots=True)
+class AdapterSession:
+    provider: str
+    session_id: str
+    provider_session_id: str
+    state: AdapterState
+
+
+class AgentAdapter(Protocol):
+    """Common lifecycle contract for external agent providers."""
+
+    def create_session(self, task: Mapping[str, Any]) -> AdapterSession:
+        """Create a provider-native session."""
+
+    def send(self, session: AdapterSession, payload: Mapping[str, Any]) -> Mapping[str, Any]:
+        """Send a task payload."""
+
+    def close(self, session: AdapterSession) -> None:
+        """Close and revoke the session."""
+
+
+class ShadowAdapter:
+    """Safe adapter used only for candidate lifecycle testing."""
+
+    def __init__(self, provider: str) -> None:
+        self.provider = provider
+        self._counter = 0
+
+    def create_session(self, task: Mapping[str, Any]) -> AdapterSession:
+        self._counter += 1
+        return AdapterSession(
+            provider=self.provider,
+            session_id=f"shadow-{self._counter}",
+            provider_session_id=f"provider-{self._counter}",
+            state=AdapterState.ACTIVE,
+        )
+
+    def send(self, session: AdapterSession, payload: Mapping[str, Any]) -> Mapping[str, Any]:
+        if session.state != AdapterState.ACTIVE:
+            raise RuntimeError("session is not active")
+        return {
+            "provider": self.provider,
+            "session_id": session.session_id,
+            "result": "shadow-completed",
+            "side_effects": False,
+        }
+
+    def close(self, session: AdapterSession) -> None:
+        return None

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/api.py

@@ -0,0 +1,173 @@
+"""Dependency-light candidate API facade.
+
+The production transport is intentionally external. This module exposes the
+application boundary without opening listeners or changing production routes.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from .evidence import EvidenceChain
+from .orchestrator import AutonomousReleaseOrchestrator
+from .release import ReleaseController
+from .serialization import (
+    release_plan_from_mapping,
+    release_plan_to_mapping,
+    task_plan_from_mapping,
+    task_plan_to_mapping,
+)
+
+
+class AutonomousReleaseAPI:
+    def __init__(self, orchestrator: AutonomousReleaseOrchestrator) -> None:
+        self.orchestrator = orchestrator
+        self.release = ReleaseController()
+
+    def _chain(self, task_id: str) -> EvidenceChain:
+        return EvidenceChain(self.orchestrator.workflow.store, task_id)
+
+    def create_task(self, payload: Mapping[str, Any]) -> dict[str, Any]:
+        plan = task_plan_from_mapping(payload)
+        result = self.orchestrator.create(plan)
+        self.orchestrator.workflow.next_steps(plan)
+        return {
+            "task_id": result.task_id,
+            "plan_digest": result.plan_digest,
+            "production_side_effects": result.production_side_effects,
+            "plan": task_plan_to_mapping(plan),
+        }
+
+    def close_evidence(self, task_id: str) -> dict[str, Any]:
+        result = self._chain(task_id).closure_digest()
+        return {
+            "task_id": task_id,
+            "evidence_closure": result,
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def task_steps(self, task_id: str) -> dict[str, Any]:
+        status = self.task_status(task_id)
+        return {
+            "task_id": task_id,
+            "steps": status["steps"],
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def claim(self, task_id: str, payload: Mapping[str, Any]) -> dict[str, Any]:
+        step_id = str(payload["step_id"])
+        worker_id = str(payload["worker_id"])
+        session_id = str(payload["session_id"])
+        workflow = self.orchestrator.workflow
+        workflow.claim(task_id, step_id, worker_id, session_id)
+        workflow.start(task_id, step_id, worker_id, session_id)
+        self._chain(task_id).append(
+            "STEP_CLAIMED",
+            task_id,
+            {
+                "step_id": step_id,
+                "worker_id": worker_id,
+                "session_id": session_id,
+                "secrets_included": False,
+            },
+        )
+        return {
+            "task_id": task_id,
+            "step_id": step_id,
+            "state": "RUNNING",
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def heartbeat(self, task_id: str, payload: Mapping[str, Any]) -> dict[str, Any]:
+        step_id = str(payload["step_id"])
+        worker_id = str(payload["worker_id"])
+        session_id = str(payload["session_id"])
+        self.orchestrator.workflow.heartbeat(task_id, step_id, worker_id, session_id)
+        return {
+            "task_id": task_id,
+            "step_id": step_id,
+            "state": "RUNNING",
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def result(self, task_id: str, payload: Mapping[str, Any]) -> dict[str, Any]:
+        step_id = str(payload["step_id"])
+        result_digest = str(payload["result_digest"])
+        self.orchestrator.workflow.complete(task_id, step_id, result_digest=result_digest)
+        self._chain(task_id).append(
+            "STEP_RESULT",
+            task_id,
+            {
+                "step_id": step_id,
+                "result_digest": result_digest,
+                "secrets_included": False,
+            },
+        )
+        return {
+            "task_id": task_id,
+            "step_id": step_id,
+            "state": "COMPLETED",
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def evidence(self, task_id: str) -> dict[str, Any]:
+        chain = self._chain(task_id)
+        return {
+            "task_id": task_id,
+            "events": chain.export(),
+            "verified": chain.verify(),
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def create_release_plan(self, payload: Mapping[str, Any]) -> dict[str, Any]:
+        plan = release_plan_from_mapping(payload)
+        self.release = ReleaseController()
+        self.release.attach_plan(plan)
+        self.release.mark_built()
+        return {
+            "plan": release_plan_to_mapping(plan),
+            "release_state": self.release.status().state.value,
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def verify_release(self, payload: Mapping[str, Any]) -> dict[str, Any]:
+        self.release.verify(
+            artifact_ok=payload.get("artifact_ok") is True,
+            evidence_ok=payload.get("evidence_ok") is True,
+        )
+        self.release.request_owner_approval()
+        return {
+            "release_state": self.release.status().state.value,
+            "production_release": "BLOCKED_EXTERNAL_OWNER_SIGNATURE",
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+
+    def task_status(self, task_id: str) -> dict[str, Any]:
+        store = self.orchestrator.workflow.store
+        state, revision = store.get_task_state(task_id)
+        steps = store.list_steps(task_id)
+        return {
+            "task_id": task_id,
+            "state": state.value,
+            "revision": revision,
+            "steps": {
+                step_id: {
+                    "state": item.state.value,
+                    "attempt": item.attempt,
+                    "max_attempts": item.max_attempts,
+                    "worker_id": item.worker_id,
+                    "session_id": item.session_id,
+                }
+                for step_id, item in steps.items()
+            },
+            "production_side_effects": False,
+            "secrets_included": False,
+        }

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/approval.py

@@ -0,0 +1,116 @@
+"""Owner approval verification contracts for candidate releases."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Protocol
+
+from .errors import ApprovalRequired, StateConflict
+from .models import ApprovalEnvelope, TaskPlan, canonical_json
+from .store import StateStore
+
+
+class SignatureVerifier(Protocol):
+    """Verify an approval payload without exposing private-key operations."""
+
+    def verify(self, key_id: str, payload: bytes, signature: str) -> bool:
+        """Return ``True`` when the signature is valid for ``key_id``."""
+
+
+class RevocationSource(Protocol):
+    def is_revoked(self, approval_plan_id: str) -> bool:
+        """Return whether an approval plan has been revoked."""
+
+
+@dataclass(frozen=True, slots=True)
+class ApprovalVerificationResult:
+    approval_plan_id: str
+    task_id: str
+    signing_key_id: str
+    verified_at: datetime
+
+
+class ApprovalVerifier:
+    """Verify exact approval bindings and consume nonces before execution."""
+
+    def __init__(
+        self,
+        store: StateStore,
+        signature_verifier: SignatureVerifier,
+        revocations: RevocationSource,
+    ) -> None:
+        self.store = store
+        self.signature_verifier = signature_verifier
+        self.revocations = revocations
+
+    def verify(
+        self,
+        plan: TaskPlan,
+        envelope: ApprovalEnvelope,
+        *,
+        now: datetime | None = None,
+    ) -> ApprovalVerificationResult:
+        now = now or datetime.now(timezone.utc)
+        if envelope.task_id != plan.task_id:
+            raise ApprovalRequired("APPROVAL_TASK_MISMATCH", "approval task does not match plan")
+        if envelope.task_plan_sha256 != plan.digest():
+            raise ApprovalRequired("APPROVAL_PLAN_DIGEST_MISMATCH", "approval digest does not match plan")
+        if envelope.binding != plan.binding:
+            raise ApprovalRequired("APPROVAL_RESOURCE_MISMATCH", "approval resource binding does not match plan")
+        approved = set(envelope.capabilities)
+        required = {cap for step in plan.steps for cap in step.required_capabilities}
+        if not required.issubset(approved):
+            raise ApprovalRequired(
+                "APPROVAL_CAPABILITY_INSUFFICIENT",
+                "approval does not cover all plan capabilities",
+                {"missing": sorted(required - approved)},
+            )
+        if now < envelope.not_before or now >= envelope.expires_at:
+            raise ApprovalRequired("APPROVAL_TIME_INVALID", "approval is outside its validity window")
+        if self.revocations.is_revoked(envelope.approval_plan_id):
+            raise ApprovalRequired("APPROVAL_REVOKED", "approval plan is revoked")
+        payload = canonical_json(envelope.payload()).encode("utf-8")
+        try:
+            signature_valid = self.signature_verifier.verify(
+                envelope.signing_key_id, payload, envelope.signature
+            )
+        except Exception as error:
+            raise ApprovalRequired(
+                "BLOCKED_OWNER_APPROVAL_VERIFIER_UNAVAILABLE",
+                "trusted Owner approval verifier is unavailable",
+            ) from error
+        if not signature_valid:
+            raise ApprovalRequired("APPROVAL_SIGNATURE_INVALID", "approval signature verification failed")
+        try:
+            self.store.consume_nonce(f"approval:{envelope.approval_plan_id}", envelope.nonce)
+        except StateConflict as error:
+            raise ApprovalRequired("APPROVAL_NONCE_REPLAY", "approval nonce was already consumed") from error
+        return ApprovalVerificationResult(
+            approval_plan_id=envelope.approval_plan_id,
+            task_id=envelope.task_id,
+            signing_key_id=envelope.signing_key_id,
+            verified_at=now,
+        )
+
+
+class StaticRevocationSource:
+    def __init__(self, revoked: set[str] | None = None) -> None:
+        self.revoked = set(revoked or ())
+
+    def is_revoked(self, approval_plan_id: str) -> bool:
+        return approval_plan_id in self.revoked
+
+
+class RejectingSignatureVerifier:
+    """Safe default used when no external verifier has been configured."""
+
+    def verify(self, key_id: str, payload: bytes, signature: str) -> bool:
+        return False
+
+
+class UnavailableSignatureVerifier:
+    """Safe runtime default when no Owner HSM/KMS verifier has been supplied."""
+
+    def verify(self, key_id: str, payload: bytes, signature: str) -> bool:
+        raise ConnectionError("Owner verifier unavailable")

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/artifact.py

@@ -0,0 +1,225 @@
+"""Deterministic Candidate artifact and package-closure construction."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import io
+import json
+import os
+from pathlib import Path, PurePosixPath
+import stat
+import tarfile
+from typing import Mapping
+
+from .errors import EvidenceIncomplete, ReleaseBlocked
+from .models import is_git_oid
+
+
+def sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactFile:
+    path: str
+    content: bytes
+    mode: int = 0o644
+
+    def __post_init__(self) -> None:
+        parsed = PurePosixPath(self.path)
+        if (
+            parsed.is_absolute()
+            or ".." in parsed.parts
+            or not self.path
+            or "\\" in self.path
+            or self.path != self.path.strip()
+        ):
+            raise ValueError("artifact paths must be canonical, relative and traversal-free")
+        if ".git" in parsed.parts:
+            raise ValueError(".git content is forbidden")
+        if self.mode not in (0o644, 0o755):
+            raise ValueError("artifact mode must be 0644 or 0755")
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactManifest:
+    source_oid: str
+    tree_oid: str
+    files: Mapping[str, str]
+    artifact_sha256: str
+    package_closure_sha256: str
+    format_version: int = 1
+    secrets_included: bool = False
+
+    def as_json_bytes(self) -> bytes:
+        payload = {
+            "format_version": self.format_version,
+            "source_oid": self.source_oid,
+            "tree_oid": self.tree_oid,
+            "files": dict(sorted(self.files.items())),
+            "artifact_sha256": self.artifact_sha256,
+            "package_closure_sha256": self.package_closure_sha256,
+            "secrets_included": self.secrets_included,
+        }
+        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+    @property
+    def manifest_sha256(self) -> str:
+        return sha256_bytes(self.as_json_bytes())
+
+
+@dataclass(frozen=True, slots=True)
+class BuiltArtifact:
+    content: bytes
+    manifest: ArtifactManifest
+
+
+class ArtifactBuilder:
+    MAX_FILE_BYTES = 8 * 1024 * 1024
+    MAX_ARTIFACT_BYTES = 64 * 1024 * 1024
+    FORBIDDEN_MARKERS = (
+        b"begin private " + b"key",
+        b"authorization:" + b" bearer ",
+        b"password" + b"=",
+        b"token" + b"=",
+        b"secret" + b"=",
+        b"api_key" + b"=",
+    )
+
+    @staticmethod
+    def _validate_oid(name: str, value: str) -> None:
+        if not is_git_oid(value):
+            raise ValueError(f"{name} must be a full lowercase Git OID")
+
+    def build_from_directory(
+        self, source_oid: str, tree_oid: str, source: str | Path
+    ) -> BuiltArtifact:
+        root = Path(source)
+        if root.is_symlink() or not root.is_dir():
+            raise ReleaseBlocked(
+                "ARTIFACT_SOURCE_LINK_REJECTED", "artifact source must be a real directory"
+            )
+        files: list[ArtifactFile] = []
+        for directory, dirnames, filenames in os.walk(root, followlinks=False):
+            current = Path(directory)
+            for name in tuple(dirnames):
+                item = current / name
+                if item.is_symlink() or name == ".git":
+                    raise ReleaseBlocked(
+                        "ARTIFACT_SOURCE_LINK_REJECTED",
+                        "artifact source contains a linked or repository directory",
+                    )
+            for name in filenames:
+                item = current / name
+                info = item.lstat()
+                if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
+                    raise ReleaseBlocked(
+                        "ARTIFACT_SOURCE_LINK_REJECTED",
+                        "artifact source contains a link or special file",
+                    )
+                if info.st_size > self.MAX_FILE_BYTES:
+                    raise ReleaseBlocked("ARTIFACT_FILE_TOO_LARGE", "artifact file exceeds limit")
+                relative = item.relative_to(root).as_posix()
+                mode = 0o755 if info.st_mode & 0o111 else 0o644
+                files.append(ArtifactFile(relative, item.read_bytes(), mode))
+        return self.build(source_oid, tree_oid, tuple(files))
+
+    def build(
+        self, source_oid: str, tree_oid: str, files: tuple[ArtifactFile, ...]
+    ) -> BuiltArtifact:
+        self._validate_oid("source_oid", source_oid)
+        self._validate_oid("tree_oid", tree_oid)
+        if not files:
+            raise EvidenceIncomplete("ARTIFACT_EMPTY", "artifact requires at least one file")
+        paths = [item.path for item in files]
+        if len(paths) != len(set(paths)):
+            raise ReleaseBlocked("ARTIFACT_DUPLICATE_PATH", "artifact contains duplicate paths")
+        total = 0
+        for item in files:
+            total += len(item.content)
+            if len(item.content) > self.MAX_FILE_BYTES or total > self.MAX_ARTIFACT_BYTES:
+                raise ReleaseBlocked("ARTIFACT_SIZE_LIMIT", "artifact exceeds Candidate limit")
+            lowered = item.content.lower()
+            if any(marker in lowered for marker in self.FORBIDDEN_MARKERS):
+                raise ReleaseBlocked(
+                    "ARTIFACT_SECRET_MARKER",
+                    "artifact contains a forbidden secret marker",
+                    {"path": item.path},
+                )
+        output = io.BytesIO()
+        with tarfile.open(fileobj=output, mode="w", format=tarfile.PAX_FORMAT) as archive:
+            for item in sorted(files, key=lambda value: value.path):
+                info = tarfile.TarInfo(item.path)
+                info.size = len(item.content)
+                info.mode = item.mode
+                info.mtime = 0
+                info.uid = 0
+                info.gid = 0
+                info.uname = "root"
+                info.gname = "root"
+                archive.addfile(info, io.BytesIO(item.content))
+        content = output.getvalue()
+        file_hashes = {item.path: sha256_bytes(item.content) for item in files}
+        closure_payload = json.dumps(
+            {
+                "source_oid": source_oid,
+                "tree_oid": tree_oid,
+                "files": dict(sorted(file_hashes.items())),
+                "secrets_included": False,
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        manifest = ArtifactManifest(
+            source_oid=source_oid,
+            tree_oid=tree_oid,
+            files=file_hashes,
+            artifact_sha256=sha256_bytes(content),
+            package_closure_sha256=sha256_bytes(closure_payload),
+            secrets_included=False,
+        )
+        return BuiltArtifact(content=content, manifest=manifest)
+
+    def verify(self, built: BuiltArtifact) -> None:
+        if built.manifest.secrets_included:
+            raise EvidenceIncomplete("ARTIFACT_SECRET_FLAG", "manifest includes secrets")
+        if sha256_bytes(built.content) != built.manifest.artifact_sha256:
+            raise EvidenceIncomplete(
+                "ARTIFACT_DIGEST_MISMATCH", "artifact digest does not match manifest"
+            )
+        actual: dict[str, str] = {}
+        with tarfile.open(fileobj=io.BytesIO(built.content), mode="r:") as archive:
+            for member in archive.getmembers():
+                try:
+                    ArtifactFile(member.name, b"", member.mode)
+                except ValueError as error:
+                    raise EvidenceIncomplete(
+                        "ARTIFACT_MEMBER_INVALID", "archive contains an unsafe member"
+                    ) from error
+                if not member.isfile() or member.name in actual:
+                    raise EvidenceIncomplete(
+                        "ARTIFACT_MEMBER_INVALID", "archive contains a link, special or duplicate member"
+                    )
+                stream = archive.extractfile(member)
+                if stream is None:
+                    raise EvidenceIncomplete("ARTIFACT_MEMBER_INVALID", "archive member is unreadable")
+                actual[member.name] = sha256_bytes(stream.read())
+        if actual != dict(built.manifest.files):
+            raise EvidenceIncomplete(
+                "ARTIFACT_FILE_MISMATCH", "artifact file closure does not match manifest"
+            )
+        closure_payload = json.dumps(
+            {
+                "source_oid": built.manifest.source_oid,
+                "tree_oid": built.manifest.tree_oid,
+                "files": dict(sorted(actual.items())),
+                "secrets_included": False,
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        if sha256_bytes(closure_payload) != built.manifest.package_closure_sha256:
+            raise EvidenceIncomplete(
+                "PACKAGE_CLOSURE_MISMATCH", "package closure digest does not match"
+            )

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/cli.py

@@ -0,0 +1,244 @@
+"""Candidate inspection CLI; all commands are local and non-production."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+from .artifact import ArtifactBuilder, ArtifactManifest, BuiltArtifact
+from .evidence import EvidenceChain
+from .release import ReleaseController
+from .serialization import (
+    release_plan_from_mapping,
+    release_plan_to_mapping,
+    task_plan_from_mapping,
+    task_plan_to_mapping,
+)
+from .service import CandidateApplication
+from .store import StateStore
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="aios-autonomous-release")
+    sub = parser.add_subparsers(dest="command", required=True)
+    plan = sub.add_parser("plan")
+    plan.add_argument("payload")
+    digest = sub.add_parser("digest")
+    digest.add_argument("payload")
+    status = sub.add_parser("status")
+    status.add_argument("--state", required=True, type=Path)
+    status.add_argument("--task-id", required=True)
+    evidence = sub.add_parser("verify-evidence")
+    evidence.add_argument("--state", required=True, type=Path)
+    evidence.add_argument("--task-id", required=True)
+    evidence.add_argument("--require", action="append", default=[])
+
+    task = sub.add_parser("task")
+    task_sub = task.add_subparsers(dest="task_command", required=True)
+    task_create = task_sub.add_parser("create")
+    task_create.add_argument("--state", required=True, type=Path)
+    task_create.add_argument("--payload", required=True, type=Path)
+    for name in ("status", "steps"):
+        item = task_sub.add_parser(name)
+        item.add_argument("--state", required=True, type=Path)
+        item.add_argument("--task-id", required=True)
+
+    evidence_group = sub.add_parser("evidence")
+    evidence_sub = evidence_group.add_subparsers(dest="evidence_command", required=True)
+    evidence_verify = evidence_sub.add_parser("verify")
+    evidence_verify.add_argument("--state", required=True, type=Path)
+    evidence_verify.add_argument("--task-id", required=True)
+    evidence_verify.add_argument("--require", action="append", default=[])
+
+    artifact = sub.add_parser("artifact")
+    artifact_sub = artifact.add_subparsers(dest="artifact_command", required=True)
+    artifact_build = artifact_sub.add_parser("build")
+    artifact_build.add_argument("--source", required=True, type=Path)
+    artifact_build.add_argument("--source-oid", required=True)
+    artifact_build.add_argument("--tree-oid", required=True)
+    artifact_build.add_argument("--output-directory", required=True, type=Path)
+    artifact_verify = artifact_sub.add_parser("verify")
+    artifact_verify.add_argument("--artifact", required=True, type=Path)
+    artifact_verify.add_argument("--manifest", required=True, type=Path)
+
+    release = sub.add_parser("release")
+    release_sub = release.add_subparsers(dest="release_command", required=True)
+    for name in ("plan", "verify"):
+        item = release_sub.add_parser(name)
+        item.add_argument("--payload", required=True, type=Path)
+
+    acceptance = sub.add_parser("acceptance")
+    acceptance.add_argument("--repo-root", required=True, type=Path)
+    acceptance.add_argument("--output", required=True, type=Path)
+    return parser
+
+
+def load_json(path: Path) -> dict:
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError("JSON payload must be an object")
+    return value
+
+
+def exclusive(path: Path, content: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+    descriptor = os.open(
+        path,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+        0o640,
+    )
+    try:
+        os.write(descriptor, content)
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def artifact_from_files(artifact: Path, manifest: Path) -> BuiltArtifact:
+    payload = load_json(manifest)
+    contract = ArtifactManifest(
+        source_oid=str(payload["source_oid"]),
+        tree_oid=str(payload["tree_oid"]),
+        files={str(key): str(value) for key, value in payload["files"].items()},
+        artifact_sha256=str(payload["artifact_sha256"]),
+        package_closure_sha256=str(payload["package_closure_sha256"]),
+        format_version=int(payload.get("format_version", 1)),
+        secrets_included=bool(payload.get("secrets_included", False)),
+    )
+    return BuiltArtifact(artifact.read_bytes(), contract)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    if args.command in {"plan", "digest"}:
+        plan = task_plan_from_mapping(json.loads(args.payload))
+        if args.command == "plan":
+            print(json.dumps(task_plan_to_mapping(plan), sort_keys=True))
+        else:
+            print(plan.digest())
+        return 0
+    if args.command == "task":
+        application = CandidateApplication(args.state)
+        if args.task_command == "create":
+            result = application.api.create_task(load_json(args.payload))
+        elif args.task_command == "steps":
+            result = application.api.task_steps(args.task_id)
+        else:
+            result = application.api.task_status(args.task_id)
+        print(json.dumps(result, sort_keys=True))
+        return 0
+    if args.command == "evidence":
+        chain = EvidenceChain(StateStore(args.state), args.task_id)
+        result = {
+            "task_id": args.task_id,
+            "verified": chain.verify(),
+            "evidence_closure": chain.close(tuple(args.require)),
+            "production_side_effects": False,
+            "secrets_included": False,
+        }
+        print(json.dumps(result, sort_keys=True))
+        return 0
+    if args.command == "artifact":
+        builder = ArtifactBuilder()
+        if args.artifact_command == "build":
+            built = builder.build_from_directory(args.source_oid, args.tree_oid, args.source)
+            builder.verify(built)
+            artifact_path = args.output_directory / "candidate.tar"
+            manifest_path = args.output_directory / "manifest.json"
+            exclusive(artifact_path, built.content)
+            exclusive(manifest_path, built.manifest.as_json_bytes() + b"\n")
+            result = {
+                "artifact": str(artifact_path),
+                "manifest": str(manifest_path),
+                "artifact_sha256": built.manifest.artifact_sha256,
+                "manifest_sha256": built.manifest.manifest_sha256,
+                "package_closure_sha256": built.manifest.package_closure_sha256,
+                "production_side_effects": False,
+                "secrets_included": False,
+            }
+        else:
+            built = artifact_from_files(args.artifact, args.manifest)
+            builder.verify(built)
+            result = {
+                "verified": True,
+                "artifact_sha256": built.manifest.artifact_sha256,
+                "manifest_sha256": built.manifest.manifest_sha256,
+                "package_closure_sha256": built.manifest.package_closure_sha256,
+                "production_side_effects": False,
+                "secrets_included": False,
+            }
+        print(json.dumps(result, sort_keys=True))
+        return 0
+    if args.command == "release":
+        plan = release_plan_from_mapping(load_json(args.payload))
+        if args.release_command == "plan":
+            result = {
+                "plan": release_plan_to_mapping(plan),
+                "release_state": "CREATED",
+                "production_side_effects": False,
+                "secrets_included": False,
+            }
+        else:
+            controller = ReleaseController()
+            controller.attach_plan(plan)
+            controller.mark_built()
+            controller.verify(artifact_ok=True, evidence_ok=True)
+            controller.request_owner_approval()
+            result = {
+                "release_state": controller.status().state.value,
+                "production_release": "BLOCKED_EXTERNAL_OWNER_SIGNATURE",
+                "production_side_effects": False,
+                "secrets_included": False,
+            }
+        print(json.dumps(result, sort_keys=True))
+        return 0
+    if args.command == "acceptance":
+        runner = args.repo_root / "scripts" / "autonomous-release" / "run_acceptance.py"
+        completed = subprocess.run(
+            [
+                sys.executable,
+                str(runner),
+                "--repo-root",
+                str(args.repo_root),
+                "--output",
+                str(args.output),
+            ],
+            check=False,
+        )
+        return completed.returncode
+    store = StateStore(args.state)
+    if args.command == "status":
+        state, revision = store.get_task_state(args.task_id)
+        print(
+            json.dumps(
+                {
+                    "task_id": args.task_id,
+                    "state": state.value,
+                    "revision": revision,
+                    "production_side_effects": False,
+                    "secrets_included": False,
+                },
+                sort_keys=True,
+            )
+        )
+        return 0
+    chain = EvidenceChain(store, args.task_id)
+    print(
+        json.dumps(
+            {
+                "task_id": args.task_id,
+                "evidence_closure": chain.close(tuple(args.require)),
+                "secrets_included": False,
+            },
+            sort_keys=True,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/credential.py

@@ -0,0 +1,307 @@
+"""Task-bound, reference-only Candidate credential lifecycle."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+import json
+import re
+import secrets
+from typing import Protocol
+
+from .errors import CredentialUnavailable, PolicyDenied
+from .models import (
+    CredentialProjection,
+    CredentialRequest,
+    CredentialState,
+    canonical_json,
+)
+from .store import StateStore
+
+
+REFERENCE_RE = re.compile(
+    r"(?:systemd-credential:[A-Za-z0-9_.@-]+|secret-ref://candidate/[A-Za-z0-9_./@-]+)"
+)
+
+
+class CredentialProvider(Protocol):
+    """External provider contract; values crossing this boundary are opaque references."""
+
+    def issue(self, request: CredentialRequest, expires_at: datetime) -> tuple[str, str]: ...
+
+    def revoke(self, credential_id: str, reference: str) -> None: ...
+
+    def is_usable(self, credential_id: str, reference: str) -> bool: ...
+
+
+@dataclass(frozen=True, slots=True)
+class CredentialLifecycleResult:
+    credential_id: str
+    reference: str
+    state: CredentialState
+    expires_at: datetime
+
+
+class CredentialBroker:
+    """Enforce approval, branch, consumer and operation bindings for projections."""
+
+    def __init__(self, store: StateStore, provider: CredentialProvider) -> None:
+        self.store = store
+        self.provider = provider
+
+    @staticmethod
+    def _validate_request(request: CredentialRequest, now: datetime) -> None:
+        if request.binding.branch == "main":
+            raise PolicyDenied("MAIN_CREDENTIAL_DENIED", "credentials cannot target main")
+        forbidden = {
+            "production.deploy",
+            "production.rollback",
+            "tag.write",
+            "baseline.write",
+            "registry.write",
+            "secret.export",
+            "secret.read",
+        }
+        overlap = forbidden.intersection(request.operations)
+        if overlap:
+            raise PolicyDenied(
+                "CREDENTIAL_OPERATION_FORBIDDEN",
+                "credential request includes forbidden operations",
+                {"operations": sorted(overlap)},
+            )
+        if now < request.not_before or now >= request.expires_at:
+            raise CredentialUnavailable(
+                "CREDENTIAL_APPROVAL_WINDOW_INVALID", "credential approval is not currently valid"
+            )
+
+    @staticmethod
+    def _metadata(request: CredentialRequest) -> dict[str, object]:
+        return {
+            "request_id": request.request_id,
+            "task_id": request.task_id,
+            "session_id": request.session_id,
+            "worker_id": request.worker_id,
+            "audience": request.audience,
+            "consumer_service": request.consumer_service,
+            "provider": request.provider,
+            "runtime_user": request.runtime_user,
+            "approval_plan_id": request.approval_plan_id,
+            "repository": request.binding.repository,
+            "branch": request.binding.branch,
+            "base_oid": request.binding.base_oid,
+            "allowed_paths": list(request.binding.allowed_paths),
+            "operations": list(request.operations),
+            "not_before": request.not_before.astimezone(timezone.utc).isoformat(),
+            "expires_at": request.expires_at.astimezone(timezone.utc).isoformat(),
+            "nonce": request.nonce,
+            "max_uses": request.max_uses,
+            "use_count": 0,
+            "secrets_included": False,
+        }
+
+    def _compensate(self, credential_id: str, reference: str) -> None:
+        try:
+            self.provider.revoke(credential_id, reference)
+        except Exception:
+            pass
+
+    def issue(
+        self, request: CredentialRequest, now: datetime | None = None
+    ) -> CredentialProjection:
+        now = now or datetime.now(timezone.utc)
+        self._validate_request(request, now)
+        expires_at = min(request.expires_at, now + timedelta(seconds=request.ttl_seconds))
+        try:
+            credential_id, reference = self.provider.issue(request, expires_at)
+        except Exception as error:
+            raise CredentialUnavailable(
+                "CREDENTIAL_PROVIDER_UNAVAILABLE", "credential provider issue failed"
+            ) from error
+        if not credential_id or not reference or not REFERENCE_RE.fullmatch(reference):
+            if credential_id and reference:
+                self._compensate(credential_id, reference)
+            raise CredentialUnavailable(
+                "CREDENTIAL_REFERENCE_REJECTED", "provider did not return an approved opaque reference"
+            )
+        projection = CredentialProjection(
+            credential_id=credential_id,
+            reference=reference,
+            request=request,
+            state=CredentialState.ISSUED,
+            issued_at=now,
+            expires_at=expires_at,
+            plaintext_returned=False,
+        )
+        try:
+            with self.store.transaction() as connection:
+                connection.execute(
+                    "INSERT INTO credentials(credential_id, task_id, reference, state, metadata_json, "
+                    "issued_at, expires_at) VALUES(?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        credential_id,
+                        request.task_id,
+                        reference,
+                        CredentialState.ISSUED.value,
+                        canonical_json(self._metadata(request)),
+                        now.isoformat(),
+                        expires_at.isoformat(),
+                    ),
+                )
+                connection.execute(
+                    "INSERT INTO consumed_nonces(scope, nonce, consumed_at) VALUES(?, ?, ?)",
+                    (f"credential:{request.approval_plan_id}", request.nonce, now.isoformat()),
+                )
+        except Exception as error:
+            self._compensate(credential_id, reference)
+            raise CredentialUnavailable(
+                "CREDENTIAL_PERSISTENCE_FAILED",
+                "credential issuance was compensated after durable-state failure",
+            ) from error
+        return projection
+
+    def project(self, projection: CredentialProjection) -> CredentialLifecycleResult:
+        with self.store.transaction() as connection:
+            cursor = connection.execute(
+                "UPDATE credentials SET state=? WHERE credential_id=? AND reference=? AND state=?",
+                (
+                    CredentialState.PROJECTED.value,
+                    projection.credential_id,
+                    projection.reference,
+                    CredentialState.ISSUED.value,
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise CredentialUnavailable(
+                    "CREDENTIAL_STATE_CONFLICT", "only a durably issued credential may be projected"
+                )
+        return CredentialLifecycleResult(
+            projection.credential_id,
+            projection.reference,
+            CredentialState.PROJECTED,
+            projection.expires_at,
+        )
+
+    def authorize(
+        self,
+        projection: CredentialProjection,
+        *,
+        task_id: str,
+        session_id: str,
+        worker_id: str,
+        audience: str,
+        consumer_service: str,
+        operation: str,
+        path: str,
+        now: datetime | None = None,
+    ) -> None:
+        now = now or datetime.now(timezone.utc)
+        request = projection.request
+        actual = (task_id, session_id, worker_id, audience, consumer_service)
+        expected = (
+            request.task_id,
+            request.session_id,
+            request.worker_id,
+            request.audience,
+            request.consumer_service,
+        )
+        if actual != expected:
+            raise CredentialUnavailable(
+                "CREDENTIAL_SCOPE_MISMATCH", "credential scope does not match the caller"
+            )
+        if operation not in request.operations or not request.binding.allows_path(path):
+            raise CredentialUnavailable(
+                "CREDENTIAL_OPERATION_DENIED", "credential operation or path is not approved"
+            )
+        if now < request.not_before or now >= projection.expires_at:
+            raise CredentialUnavailable("CREDENTIAL_EXPIRED", "credential is outside its valid window")
+        with self.store.transaction() as connection:
+            row = connection.execute(
+                "SELECT state, metadata_json FROM credentials WHERE credential_id=? AND reference=?",
+                (projection.credential_id, projection.reference),
+            ).fetchone()
+            if row is None or row["state"] != CredentialState.PROJECTED.value:
+                raise CredentialUnavailable(
+                    "CREDENTIAL_NOT_PROJECTED", "credential is not in projected state"
+                )
+            metadata = json.loads(row["metadata_json"])
+            count = int(metadata.get("use_count", 0))
+            if count >= request.max_uses:
+                raise CredentialUnavailable("CREDENTIAL_USE_LIMIT", "credential use limit reached")
+            if not self.provider.is_usable(projection.credential_id, projection.reference):
+                raise CredentialUnavailable(
+                    "CREDENTIAL_PROVIDER_REJECTED", "provider no longer accepts the credential"
+                )
+            metadata["use_count"] = count + 1
+            connection.execute(
+                "UPDATE credentials SET metadata_json=? WHERE credential_id=?",
+                (canonical_json(metadata), projection.credential_id),
+            )
+
+    def revoke(
+        self, credential_id: str, reference: str, now: datetime | None = None
+    ) -> CredentialLifecycleResult:
+        now = now or datetime.now(timezone.utc)
+        with self.store.transaction() as connection:
+            row = connection.execute(
+                "SELECT state, expires_at FROM credentials WHERE credential_id=? AND reference=?",
+                (credential_id, reference),
+            ).fetchone()
+        if row is None:
+            raise KeyError(credential_id)
+        expires_at = datetime.fromisoformat(row["expires_at"])
+        if row["state"] == CredentialState.REVOKED.value:
+            return CredentialLifecycleResult(
+                credential_id, reference, CredentialState.REVOKED, expires_at
+            )
+        try:
+            self.provider.revoke(credential_id, reference)
+        except Exception as error:
+            raise CredentialUnavailable(
+                "CREDENTIAL_REVOCATION_FAILED", "upstream credential revocation failed"
+            ) from error
+        with self.store.transaction() as connection:
+            connection.execute(
+                "UPDATE credentials SET state=?, revoked_at=? WHERE credential_id=? AND reference=?",
+                (CredentialState.REVOKED.value, now.isoformat(), credential_id, reference),
+            )
+        return CredentialLifecycleResult(
+            credential_id, reference, CredentialState.REVOKED, expires_at
+        )
+
+    def verify_revoked(self, credential_id: str, reference: str) -> None:
+        with self.store.transaction() as connection:
+            row = connection.execute(
+                "SELECT state FROM credentials WHERE credential_id=? AND reference=?",
+                (credential_id, reference),
+            ).fetchone()
+        if row is None or row["state"] != CredentialState.REVOKED.value:
+            raise CredentialUnavailable(
+                "CREDENTIAL_REVOCATION_UNRECORDED", "durable revocation record is missing"
+            )
+        if self.provider.is_usable(credential_id, reference):
+            raise CredentialUnavailable(
+                "CREDENTIAL_REVOCATION_UNVERIFIED",
+                "provider still accepts revoked credential",
+                {"credential_id": credential_id},
+            )
+
+
+class InMemoryCredentialProvider:
+    """Candidate test provider; never suitable as production identity evidence."""
+
+    def __init__(self) -> None:
+        self._active: dict[str, str] = {}
+
+    def issue(self, request: CredentialRequest, expires_at: datetime) -> tuple[str, str]:
+        credential_id = f"cred-{secrets.token_hex(8)}"
+        reference = f"secret-ref://candidate/{request.task_id}/{credential_id}"
+        self._active[credential_id] = reference
+        return credential_id, reference
+
+    def revoke(self, credential_id: str, reference: str) -> None:
+        if self._active.get(credential_id) != reference:
+            raise CredentialUnavailable("CREDENTIAL_NOT_FOUND", "credential is not active")
+        del self._active[credential_id]
+
+    def is_usable(self, credential_id: str, reference: str) -> bool:
+        return self._active.get(credential_id) == reference

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/errors.py

@@ -0,0 +1,48 @@
+"""Stable error types and codes for the autonomous-release candidate."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+
+@dataclass(slots=True)
+class AutonomousReleaseError(RuntimeError):
+    """Base error with a stable machine-readable code."""
+
+    code: str
+    message: str
+    details: Mapping[str, Any] | None = None
+
+    def __str__(self) -> str:
+        if not self.details:
+            return f"{self.code}: {self.message}"
+        return f"{self.code}: {self.message} ({dict(self.details)})"
+
+
+class PolicyDenied(AutonomousReleaseError):
+    pass
+
+
+class ApprovalRequired(AutonomousReleaseError):
+    pass
+
+
+class CapabilityMismatch(AutonomousReleaseError):
+    pass
+
+
+class CredentialUnavailable(AutonomousReleaseError):
+    pass
+
+
+class EvidenceIncomplete(AutonomousReleaseError):
+    pass
+
+
+class ReleaseBlocked(AutonomousReleaseError):
+    pass
+
+
+class StateConflict(AutonomousReleaseError):
+    pass

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/evidence.py

@@ -0,0 +1,136 @@
+"""Durable, replayable and tamper-evident evidence closure."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import json
+from typing import Any, Mapping, TYPE_CHECKING
+
+from .errors import EvidenceIncomplete
+from .models import sha256_json
+
+if TYPE_CHECKING:
+    from .store import StateStore
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceEvent:
+    event_type: str
+    task_id: str
+    payload: Mapping[str, Any]
+    previous_digest: str | None
+    created_at: datetime
+
+    def event_digest(self) -> str:
+        return sha256_json(
+            {
+                "event_type": self.event_type,
+                "task_id": self.task_id,
+                "payload": dict(self.payload),
+                "previous_digest": self.previous_digest,
+                "created_at": self.created_at.astimezone(timezone.utc).isoformat(),
+            }
+        )
+
+
+class EvidenceChain:
+    """Append-only chain optionally backed by the Candidate state database."""
+
+    def __init__(self, store: StateStore | None = None, task_id: str | None = None) -> None:
+        if (store is None) != (task_id is None):
+            raise ValueError("store and task_id must be supplied together")
+        self.store = store
+        self.task_id = task_id
+        self._records: list[tuple[EvidenceEvent, str]] = []
+        if store is not None and task_id is not None:
+            self._load()
+
+    def _load(self) -> None:
+        assert self.store is not None and self.task_id is not None
+        self._records = []
+        for row in self.store.list_events(self.task_id):
+            event = EvidenceEvent(
+                event_type=str(row["event_type"]),
+                task_id=str(row["task_id"]),
+                payload=json.loads(str(row["payload_json"])),
+                previous_digest=row["previous_digest"],
+                created_at=datetime.fromisoformat(str(row["created_at"])),
+            )
+            self._records.append((event, str(row["digest"])))
+
+    def append(
+        self, event_type: str, task_id: str, payload: Mapping[str, Any]
+    ) -> EvidenceEvent:
+        if not event_type or not task_id:
+            raise EvidenceIncomplete("EVIDENCE_IDENTITY_REQUIRED", "event and task IDs are required")
+        if self.task_id is not None and task_id != self.task_id:
+            raise EvidenceIncomplete(
+                "EVIDENCE_TASK_MISMATCH", "event task does not match the durable chain"
+            )
+        if payload.get("secrets_included") is not False:
+            raise EvidenceIncomplete(
+                "EVIDENCE_SECRET_MARKER_REQUIRED",
+                "evidence must explicitly assert secrets_included=false",
+            )
+        previous = self._records[-1][1] if self._records else None
+        event = EvidenceEvent(
+            event_type=event_type,
+            task_id=task_id,
+            payload=dict(payload),
+            previous_digest=previous,
+            created_at=datetime.now(timezone.utc),
+        )
+        event_digest = event.event_digest()
+        if self.store is not None:
+            self.store.append_event(
+                task_id,
+                event_type,
+                payload,
+                previous,
+                event_digest,
+                event.created_at,
+            )
+        self._records.append((event, event_digest))
+        return event
+
+    def verify(self) -> bool:
+        previous: str | None = None
+        for event, recorded_digest in self._records:
+            if event.previous_digest != previous or event.event_digest() != recorded_digest:
+                return False
+            previous = recorded_digest
+        return True
+
+    def close(self, required_event_types: tuple[str, ...]) -> str:
+        if not self._records:
+            raise EvidenceIncomplete("EMPTY_EVIDENCE_CHAIN", "cannot close an empty evidence chain")
+        if not self.verify():
+            raise EvidenceIncomplete(
+                "EVIDENCE_CHAIN_INVALID", "evidence chain verification failed"
+            )
+        found = {event.event_type for event, _ in self._records}
+        missing = sorted(set(required_event_types) - found)
+        if missing:
+            raise EvidenceIncomplete(
+                "EVIDENCE_REQUIREMENT_MISSING",
+                "required evidence events are absent",
+                {"missing": missing},
+            )
+        return self._records[-1][1]
+
+    def closure_digest(self) -> str:
+        return self.close(())
+
+    def export(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "event_type": event.event_type,
+                "task_id": event.task_id,
+                "payload": dict(event.payload),
+                "previous_digest": event.previous_digest,
+                "digest": recorded_digest,
+                "created_at": event.created_at.isoformat(),
+            }
+            for event, recorded_digest in self._records
+        ]

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/execution_manager.py

@@ -0,0 +1,105 @@
+"""Native Candidate execution lifecycle with mandatory cleanup evidence."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+from .evidence import EvidenceChain
+from .native_adapters import CandidateNativeAdapter, NativeSession
+
+
+@dataclass(frozen=True, slots=True)
+class NativeCanaryResult:
+    provider: str
+    broker_session_id: str
+    provider_session_id: str
+    message_id: str
+    receipt: str
+    broker_session_verified: bool
+    openclaw_native_child_session_verified: bool
+    codex_child_session_verified: bool
+    simulation: bool
+    session_closed: bool
+    secrets_included: bool = False
+
+
+class ExecutionManager:
+    def __init__(self, evidence: EvidenceChain) -> None:
+        self.evidence = evidence
+
+    def run_native_canary(
+        self,
+        adapter: CandidateNativeAdapter,
+        *,
+        task_id: str,
+        parent_session_id: str,
+        recipient_binding: Mapping[str, Any],
+        text: str,
+    ) -> NativeCanaryResult:
+        session: NativeSession | None = None
+        closed = False
+        try:
+            session = adapter.spawn(task_id, parent_session_id, recipient_binding)
+            verification = adapter.verify(session)
+            message = adapter.send_text(session, text)
+            adapter.interrupt(session)
+            adapter.close(session)
+            closed = True
+            event_type = (
+                "NATIVE_SESSION_SIMULATION"
+                if verification.simulation
+                else "NATIVE_SESSION_VERIFIED"
+            )
+            self.evidence.append(
+                event_type,
+                task_id,
+                {
+                    "provider": session.provider,
+                    "broker_session_id": session.broker_session_id,
+                    "provider_session_id": session.provider_session_id,
+                    "message_id": message.message_id,
+                    "receipt": message.receipt,
+                    "broker_session_verified": verification.broker_session_verified,
+                    "openclaw_native_child_session_verified": (
+                        verification.openclaw_native_child_session_verified
+                    ),
+                    "codex_child_session_verified": verification.codex_child_session_verified,
+                    "simulation": verification.simulation,
+                    "session_closed": True,
+                    "secrets_included": False,
+                },
+            )
+            return NativeCanaryResult(
+                provider=session.provider,
+                broker_session_id=session.broker_session_id,
+                provider_session_id=session.provider_session_id,
+                message_id=message.message_id,
+                receipt=message.receipt,
+                broker_session_verified=verification.broker_session_verified,
+                openclaw_native_child_session_verified=(
+                    verification.openclaw_native_child_session_verified
+                ),
+                codex_child_session_verified=verification.codex_child_session_verified,
+                simulation=verification.simulation,
+                session_closed=True,
+                secrets_included=False,
+            )
+        except Exception as error:
+            if session is not None and not closed:
+                try:
+                    adapter.close(session)
+                    closed = True
+                except Exception:
+                    closed = False
+            self.evidence.append(
+                "NATIVE_SESSION_FAILED",
+                task_id,
+                {
+                    "provider": adapter.provider,
+                    "error_code": type(error).__name__,
+                    "session_closed": closed,
+                    "secrets_included": False,
+                },
+            )
+            raise

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/gitea_canary.py

@@ -0,0 +1,172 @@
+"""API-only Gitea deploy-key lifecycle canary."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping, Protocol, Sequence
+
+from .errors import CredentialUnavailable, StateConflict
+
+
+GITEA_REFERENCE = "systemd-credential:gitea-admin-api-candidate"
+
+
+@dataclass(frozen=True, slots=True)
+class EphemeralKeyProjection:
+    public_key: str
+    private_key_reference: str
+
+
+class KeyProjectionProvider(Protocol):
+    def create(self, task_id: str, branch: str) -> EphemeralKeyProjection: ...
+
+    def delete(self, reference: str) -> None: ...
+
+    def is_present(self, reference: str) -> bool: ...
+
+
+class GiteaDeployKeyProvider(Protocol):
+    simulation: bool
+
+    def list_deploy_keys(
+        self, repository: str, credential_reference: str
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+    def create_deploy_key(
+        self,
+        repository: str,
+        title: str,
+        public_key: str,
+        credential_reference: str,
+    ) -> Mapping[str, Any]: ...
+
+    def delete_deploy_key(
+        self, repository: str, key_id: int, credential_reference: str
+    ) -> None: ...
+
+
+class SSHAuthenticationProbe(Protocol):
+    def authenticate(
+        self, repository: str, branch: str, private_key_reference: str
+    ) -> bool: ...
+
+
+@dataclass(frozen=True, slots=True)
+class GiteaCanaryResult:
+    key_created: bool
+    key_deleted: bool
+    api_key_absent_after_delete: bool
+    post_revoke_auth_rejected: bool
+    credential_projection_deleted: bool
+    gitea_admin_api_lifecycle_canary: bool
+    simulation: bool
+    secrets_included: bool = False
+
+
+class GiteaDeployKeyCanary:
+    """Exercise only repository deploy-key list/create/delete provider operations."""
+
+    repository = "aios/aios"
+
+    def __init__(
+        self,
+        gitea: GiteaDeployKeyProvider,
+        projections: KeyProjectionProvider,
+        ssh_probe: SSHAuthenticationProbe,
+    ) -> None:
+        self.gitea = gitea
+        self.projections = projections
+        self.ssh_probe = ssh_probe
+
+    def run(
+        self, *, task_id: str, branch: str, credential_reference: str
+    ) -> GiteaCanaryResult:
+        if credential_reference != GITEA_REFERENCE:
+            raise CredentialUnavailable(
+                "BLOCKED_GITEA_ADMIN_CREDENTIAL_REQUIRED",
+                "exact Candidate Gitea Admin API reference is required",
+            )
+        if not task_id or not branch.startswith("patch/"):
+            raise StateConflict("GITEA_CANARY_BINDING_INVALID", "task and patch branch are required")
+        projection: EphemeralKeyProjection | None = None
+        key_id: int | None = None
+        key_created = False
+        key_deleted = False
+        api_absent = False
+        auth_rejected = False
+        projection_deleted = False
+        title = f"aios-candidate-{task_id}"
+        try:
+            projection = self.projections.create(task_id, branch)
+            if (
+                not projection.public_key.startswith("ssh-ed25519 ")
+                or not projection.private_key_reference.startswith("systemd-credential:")
+            ):
+                raise CredentialUnavailable(
+                    "GITEA_CANARY_KEY_PROJECTION_INVALID",
+                    "ephemeral key projection is not reference-only",
+                )
+            before = self.gitea.list_deploy_keys(self.repository, credential_reference)
+            if any(item.get("title") == title for item in before):
+                raise StateConflict("GITEA_CANARY_KEY_COLLISION", "canary key title already exists")
+            created = self.gitea.create_deploy_key(
+                self.repository,
+                title,
+                projection.public_key,
+                credential_reference,
+            )
+            raw_id = created.get("id")
+            if not isinstance(raw_id, int) or created.get("key") != projection.public_key:
+                raise StateConflict("GITEA_CANARY_CREATE_INVALID", "created key response is invalid")
+            key_id = raw_id
+            key_created = True
+            visible = self.gitea.list_deploy_keys(self.repository, credential_reference)
+            if not any(item.get("id") == key_id for item in visible):
+                raise StateConflict("GITEA_CANARY_KEY_NOT_VISIBLE", "created key is not listed")
+            if not self.ssh_probe.authenticate(
+                self.repository, branch, projection.private_key_reference
+            ):
+                raise StateConflict("GITEA_CANARY_SSH_AUTH_FAILED", "deploy key could not authenticate")
+            self.gitea.delete_deploy_key(self.repository, key_id, credential_reference)
+            key_deleted = True
+            key_id = None
+            after = self.gitea.list_deploy_keys(self.repository, credential_reference)
+            api_absent = not any(item.get("title") == title for item in after)
+            if not api_absent:
+                raise StateConflict("GITEA_CANARY_DELETE_UNVERIFIED", "deleted key remains visible")
+            self.projections.delete(projection.private_key_reference)
+            projection_deleted = not self.projections.is_present(projection.private_key_reference)
+            auth_rejected = not self.ssh_probe.authenticate(
+                self.repository, branch, projection.private_key_reference
+            )
+            if not projection_deleted or not auth_rejected:
+                raise StateConflict(
+                    "GITEA_CANARY_REVOCATION_UNVERIFIED",
+                    "projection deletion or post-revocation SSH rejection failed",
+                )
+            simulated = bool(getattr(self.gitea, "simulation", True))
+            return GiteaCanaryResult(
+                key_created=key_created,
+                key_deleted=key_deleted,
+                api_key_absent_after_delete=api_absent,
+                post_revoke_auth_rejected=auth_rejected,
+                credential_projection_deleted=projection_deleted,
+                gitea_admin_api_lifecycle_canary=not simulated,
+                simulation=simulated,
+                secrets_included=False,
+            )
+        finally:
+            if key_id is not None:
+                try:
+                    self.gitea.delete_deploy_key(
+                        self.repository, key_id, credential_reference
+                    )
+                except Exception:
+                    pass
+            if projection is not None and self.projections.is_present(
+                projection.private_key_reference
+            ):
+                try:
+                    self.projections.delete(projection.private_key_reference)
+                except Exception:
+                    pass

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/models.py

@@ -0,0 +1,376 @@
+"""Typed contracts for the dormant autonomous-release candidate.
+
+This package is dependency-free by design so it can be audited and exercised in
+an isolated candidate worktree before any production service is installed.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+import hashlib
+import json
+from typing import Any, Mapping, Sequence
+
+
+def utc_now() -> datetime:
+    """Return an aware UTC timestamp."""
+    return datetime.now(timezone.utc)
+
+
+def canonical_json(value: Any) -> str:
+    """Serialize *value* deterministically for signatures and evidence hashes."""
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def sha256_json(value: Any) -> str:
+    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def is_sha256(value: str) -> bool:
+    return len(value) == 64 and all(ch in "0123456789abcdef" for ch in value)
+
+
+def is_git_oid(value: str) -> bool:
+    return len(value) == 40 and all(ch in "0123456789abcdef" for ch in value)
+
+
+class RiskClass(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class TaskState(str, Enum):
+    CREATED = "CREATED"
+    PLANNED = "PLANNED"
+    READY = "READY"
+    RUNNING = "RUNNING"
+    BLOCKED = "BLOCKED"
+    FAILED = "FAILED"
+    COMPLETED = "COMPLETED"
+    READY_FOR_OWNER_APPROVAL = "READY_FOR_OWNER_APPROVAL"
+
+
+class StepState(str, Enum):
+    PENDING = "PENDING"
+    READY = "READY"
+    CLAIMED = "CLAIMED"
+    RUNNING = "RUNNING"
+    BLOCKED = "BLOCKED"
+    FAILED = "FAILED"
+    COMPLETED = "COMPLETED"
+
+
+class CredentialState(str, Enum):
+    REQUESTED = "REQUESTED"
+    ISSUED = "ISSUED"
+    PROJECTED = "PROJECTED"
+    REVOKED = "REVOKED"
+    EXPIRED = "EXPIRED"
+
+
+@dataclass(frozen=True, slots=True)
+class ResourceBinding:
+    repository: str
+    branch: str
+    base_oid: str
+    allowed_paths: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.repository or "/" not in self.repository:
+            raise ValueError("repository must be owner/name")
+        if self.branch == "main" or not self.branch.startswith(("patch/", "integration/", "candidate/")):
+            raise ValueError("candidate execution requires a governed non-main branch")
+        if not is_git_oid(self.base_oid):
+            raise ValueError("base_oid must be a full lowercase 40-character Git OID")
+        for path in self.allowed_paths:
+            if not path or path.startswith("/") or ".." in path.split("/"):
+                raise ValueError("allowed_paths must be repository-relative and traversal-free")
+
+    def allows_path(self, path: str) -> bool:
+        if not path or path.startswith("/") or ".." in path.split("/"):
+            return False
+        return any(path == prefix or path.startswith(prefix.rstrip("/") + "/") for prefix in self.allowed_paths)
+
+
+@dataclass(frozen=True, slots=True)
+class ApprovalEnvelope:
+    approval_plan_id: str
+    task_id: str
+    task_plan_sha256: str
+    binding: ResourceBinding
+    capabilities: tuple[str, ...]
+    risk: RiskClass
+    not_before: datetime
+    expires_at: datetime
+    nonce: str
+    signing_key_id: str
+    signature: str
+
+    def __post_init__(self) -> None:
+        if self.expires_at <= self.not_before:
+            raise ValueError("approval expiry must be after not_before")
+        if not is_sha256(self.task_plan_sha256):
+            raise ValueError("task_plan_sha256 must be a SHA-256 hex digest")
+        if not self.approval_plan_id or not self.task_id or not self.nonce:
+            raise ValueError("approval identifiers and nonce are required")
+
+    def payload(self) -> Mapping[str, Any]:
+        return {
+            "approval_plan_id": self.approval_plan_id,
+            "task_id": self.task_id,
+            "task_plan_sha256": self.task_plan_sha256,
+            "repository": self.binding.repository,
+            "branch": self.binding.branch,
+            "base_oid": self.binding.base_oid,
+            "allowed_paths": list(self.binding.allowed_paths),
+            "capabilities": list(self.capabilities),
+            "risk": self.risk.value,
+            "not_before": self.not_before.astimezone(timezone.utc).isoformat(),
+            "expires_at": self.expires_at.astimezone(timezone.utc).isoformat(),
+            "nonce": self.nonce,
+            "signing_key_id": self.signing_key_id,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class WorkflowStep:
+    step_id: str
+    worker_role: str
+    required_capabilities: tuple[str, ...]
+    depends_on: tuple[str, ...] = ()
+    max_attempts: int = 1
+    timeout_seconds: int = 900
+    allowed_paths: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.step_id or not self.worker_role:
+            raise ValueError("step_id and worker_role are required")
+        if self.max_attempts < 1 or self.max_attempts > 5:
+            raise ValueError("max_attempts must be between 1 and 5")
+        if self.timeout_seconds < 1 or self.timeout_seconds > 86_400:
+            raise ValueError("timeout_seconds must be between 1 and 86400")
+        if self.step_id in self.depends_on:
+            raise ValueError("a step cannot depend on itself")
+
+
+@dataclass(frozen=True, slots=True)
+class TaskPlan:
+    task_id: str
+    intent: str
+    binding: ResourceBinding
+    risk: RiskClass
+    steps: tuple[WorkflowStep, ...]
+    evidence_requirements: tuple[str, ...]
+    production_side_effects_allowed: bool = False
+    created_at: datetime = field(default_factory=utc_now)
+
+    def __post_init__(self) -> None:
+        if not self.task_id or not self.intent:
+            raise ValueError("task_id and intent are required")
+        if self.production_side_effects_allowed:
+            raise ValueError("candidate TaskPlan must not allow production side effects")
+        ids = [step.step_id for step in self.steps]
+        if len(ids) != len(set(ids)):
+            raise ValueError("step IDs must be unique")
+        known = set(ids)
+        for step in self.steps:
+            missing = set(step.depends_on) - known
+            if missing:
+                raise ValueError(f"step {step.step_id} has unknown dependencies: {sorted(missing)}")
+        self._validate_acyclic()
+
+    def _validate_acyclic(self) -> None:
+        dependencies = {step.step_id: set(step.depends_on) for step in self.steps}
+        pending = set(dependencies)
+        while pending:
+            ready = {step_id for step_id in pending if not (dependencies[step_id] & pending)}
+            if not ready:
+                raise ValueError("workflow contains a dependency cycle")
+            pending -= ready
+
+    def digest(self) -> str:
+        return sha256_json(
+            {
+                "task_id": self.task_id,
+                "intent": self.intent,
+                "repository": self.binding.repository,
+                "branch": self.binding.branch,
+                "base_oid": self.binding.base_oid,
+                "allowed_paths": list(self.binding.allowed_paths),
+                "risk": self.risk.value,
+                "steps": [
+                    {
+                        "step_id": step.step_id,
+                        "worker_role": step.worker_role,
+                        "required_capabilities": list(step.required_capabilities),
+                        "depends_on": list(step.depends_on),
+                        "max_attempts": step.max_attempts,
+                        "timeout_seconds": step.timeout_seconds,
+                        "allowed_paths": list(step.allowed_paths),
+                    }
+                    for step in self.steps
+                ],
+                "evidence_requirements": list(self.evidence_requirements),
+                "production_side_effects_allowed": self.production_side_effects_allowed,
+                "created_at": self.created_at.astimezone(timezone.utc).isoformat(),
+            }
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class WorkerDescriptor:
+    worker_id: str
+    role: str
+    capabilities: frozenset[str]
+    trust_zone: str
+    endpoint_id: str
+    online: bool = True
+    current_load: int = 0
+    max_concurrency: int = 1
+
+    def can_accept(self, required: Sequence[str], trust_zone: str) -> bool:
+        return (
+            self.online
+            and self.trust_zone == trust_zone
+            and self.current_load < self.max_concurrency
+            and set(required).issubset(self.capabilities)
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class CredentialRequest:
+    request_id: str
+    task_id: str
+    session_id: str
+    worker_id: str
+    audience: str
+    consumer_service: str
+    binding: ResourceBinding
+    operations: tuple[str, ...]
+    provider: str
+    runtime_user: str
+    approval_plan_id: str
+    not_before: datetime
+    expires_at: datetime
+    nonce: str
+    ttl_seconds: int
+    max_uses: int
+
+    def __post_init__(self) -> None:
+        if self.ttl_seconds < 1 or self.ttl_seconds > 3600:
+            raise ValueError("credential TTL must be between 1 and 3600 seconds")
+        if self.max_uses < 1 or self.max_uses > 1000:
+            raise ValueError("credential max_uses must be between 1 and 1000")
+        if not self.operations:
+            raise ValueError("credential operations cannot be empty")
+        required = (
+            self.request_id,
+            self.task_id,
+            self.session_id,
+            self.worker_id,
+            self.audience,
+            self.consumer_service,
+            self.provider,
+            self.runtime_user,
+            self.approval_plan_id,
+            self.nonce,
+        )
+        if any(not item or item != item.strip() for item in required):
+            raise ValueError("credential identity, scope and approval fields are required")
+        if len(set(self.operations)) != len(self.operations):
+            raise ValueError("credential operations must be unique")
+        if self.not_before.tzinfo is None or self.expires_at.tzinfo is None:
+            raise ValueError("credential time bounds must be timezone aware")
+        if self.expires_at <= self.not_before:
+            raise ValueError("credential expires_at must follow not_before")
+        if (self.expires_at - self.not_before).total_seconds() > self.ttl_seconds:
+            raise ValueError("credential approval window exceeds requested TTL")
+
+
+@dataclass(frozen=True, slots=True)
+class CredentialProjection:
+    credential_id: str
+    reference: str
+    request: CredentialRequest
+    state: CredentialState
+    issued_at: datetime
+    expires_at: datetime
+    plaintext_returned: bool = False
+
+    def __post_init__(self) -> None:
+        if self.plaintext_returned:
+            raise ValueError("credential plaintext must never be returned")
+        if self.expires_at <= self.issued_at:
+            raise ValueError("credential expiry must be after issuance")
+
+
+@dataclass(frozen=True, slots=True)
+class ResultEnvelope:
+    task_id: str
+    step_id: str
+    worker_id: str
+    session_id: str
+    claim_id: str
+    operation_digest: str
+    result_digest: str
+    evidence_digests: tuple[str, ...]
+    completed_at: datetime = field(default_factory=utc_now)
+
+    def __post_init__(self) -> None:
+        if not is_sha256(self.operation_digest) or not is_sha256(self.result_digest):
+            raise ValueError("operation_digest and result_digest must be SHA-256 values")
+        if any(not is_sha256(digest) for digest in self.evidence_digests):
+            raise ValueError("all evidence digests must be SHA-256 values")
+
+
+@dataclass(frozen=True, slots=True)
+class ReleasePlan:
+    task_id: str
+    repository: str
+    branch: str
+    version: str
+    target_oid: str
+    tree_oid: str
+    artifact_sha256: str
+    manifest_sha256: str
+    package_closure_sha256: str
+    evidence_closure_oid: str
+    approval_plan_id: str
+    rollback_approval_plan_id: str
+    rollback_artifact_sha256: str
+    expected_tag_oid: str
+    expected_target_oid: str
+    production_config_sha256: str
+
+    def __post_init__(self) -> None:
+        for name in ("target_oid", "tree_oid", "evidence_closure_oid", "expected_target_oid"):
+            if not is_git_oid(getattr(self, name)):
+                raise ValueError(f"{name} must be a full lowercase 40-character Git OID")
+        for name in (
+            "artifact_sha256",
+            "manifest_sha256",
+            "package_closure_sha256",
+            "rollback_artifact_sha256",
+            "production_config_sha256",
+        ):
+            if not is_sha256(getattr(self, name)):
+                raise ValueError(f"{name} must be a SHA-256 hex digest")
+        if not is_git_oid(self.expected_tag_oid):
+            raise ValueError("expected_tag_oid must be a full lowercase 40-character Git OID")
+        if (
+            not self.task_id
+            or "/" not in self.repository
+            or not self.branch.startswith("integration/")
+            or not self.version
+            or not self.approval_plan_id
+            or not self.rollback_approval_plan_id
+        ):
+            raise ValueError("release identifiers are required")
+        if self.approval_plan_id == self.rollback_approval_plan_id:
+            raise ValueError("deployment and rollback approvals must be distinct")
+        if self.target_oid != self.expected_target_oid:
+            raise ValueError("target_oid must equal expected_target_oid")

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/native_adapters.py

@@ -0,0 +1,246 @@
+"""Candidate-only OpenClaw and Codex native session adapter contracts."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping, Protocol, Sequence
+import uuid
+
+from .errors import StateConflict
+
+
+class NativeProvider(Protocol):
+    simulation: bool
+
+    def spawn(
+        self, request: Mapping[str, Any], credential_reference: str
+    ) -> Mapping[str, Any]: ...
+
+    def status(
+        self, provider_session_id: str, credential_reference: str
+    ) -> Mapping[str, Any]: ...
+
+    def list_tree(
+        self, parent_session_id: str, credential_reference: str
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+    def send_text(
+        self,
+        provider_session_id: str,
+        payload: Mapping[str, Any],
+        credential_reference: str,
+    ) -> Mapping[str, Any]: ...
+
+    def interrupt(self, provider_session_id: str, credential_reference: str) -> None: ...
+
+    def close(self, provider_session_id: str, credential_reference: str) -> None: ...
+
+
+@dataclass(frozen=True, slots=True)
+class NativeSession:
+    provider: str
+    broker_session_id: str
+    provider_session_id: str
+    task_id: str
+    parent_session_id: str
+    recipient_binding: Mapping[str, str]
+
+
+@dataclass(frozen=True, slots=True)
+class SessionVerification:
+    broker_session_verified: bool
+    openclaw_native_child_session_verified: bool
+    codex_child_session_verified: bool
+    simulation: bool
+
+
+@dataclass(frozen=True, slots=True)
+class TextResult:
+    content_type: str
+    encoding: str
+    image_only: bool
+    task_id: str
+    message_id: str
+    receipt: str
+    recipient_binding: Mapping[str, str]
+    secrets_included: bool
+
+
+class CandidateNativeAdapter:
+    provider = ""
+    required_reference = ""
+
+    def __init__(self, client: NativeProvider, credential_reference: str) -> None:
+        if credential_reference != self.required_reference:
+            raise StateConflict(
+                "CANDIDATE_CREDENTIAL_REFERENCE_MISMATCH",
+                "adapter requires its exact Candidate credential reference",
+            )
+        self.client = client
+        self.credential_reference = credential_reference
+
+    @staticmethod
+    def _recipient(value: Mapping[str, Any]) -> dict[str, str]:
+        recipient_id = value.get("recipient_id")
+        if (
+            not isinstance(recipient_id, str)
+            or not recipient_id
+            or isinstance(value.get("recipients"), (list, tuple))
+        ):
+            raise StateConflict(
+                "RECIPIENT_BINDING_REQUIRED",
+                "one trusted TaskPlan recipient binding is required",
+            )
+        return {str(key): str(item) for key, item in value.items()}
+
+    def spawn(
+        self,
+        task_id: str,
+        parent_session_id: str,
+        recipient_binding: Mapping[str, Any],
+    ) -> NativeSession:
+        if not task_id or not parent_session_id:
+            raise StateConflict("NATIVE_SESSION_BINDING_REQUIRED", "task and parent are required")
+        recipient = self._recipient(recipient_binding)
+        response = self.client.spawn(
+            {
+                "task_id": task_id,
+                "parent_session_id": parent_session_id,
+                "recipient_binding": recipient,
+                "candidate_only": True,
+                "production_session_access": False,
+            },
+            self.credential_reference,
+        )
+        provider_session_id = response.get("provider_session_id")
+        if (
+            not isinstance(provider_session_id, str)
+            or not provider_session_id
+            or response.get("task_id") != task_id
+            or response.get("parent_session_id") != parent_session_id
+            or response.get("candidate_only") is not True
+            or response.get("production_session_access") is not False
+        ):
+            raise StateConflict(
+                "NATIVE_SESSION_PROVISIONING_INVALID",
+                "provider response is not bound to this Candidate task",
+            )
+        broker_id = f"broker-{uuid.uuid4()}"
+        if broker_id == provider_session_id:
+            raise StateConflict("SESSION_ID_COLLISION", "broker and provider IDs must differ")
+        return NativeSession(
+            provider=self.provider,
+            broker_session_id=broker_id,
+            provider_session_id=provider_session_id,
+            task_id=task_id,
+            parent_session_id=parent_session_id,
+            recipient_binding=recipient,
+        )
+
+    def verify(self, session: NativeSession) -> SessionVerification:
+        self._validate_session(session)
+        status = self.client.status(session.provider_session_id, self.credential_reference)
+        tree = self.client.list_tree(session.parent_session_id, self.credential_reference)
+        status_valid = (
+            status.get("provider_session_id") == session.provider_session_id
+            and status.get("state") in {"PROVISIONED", "ACTIVE", "RUNNING"}
+        )
+        tree_valid = any(
+            item.get("provider_session_id") == session.provider_session_id
+            and item.get("parent_session_id") == session.parent_session_id
+            for item in tree
+        )
+        if not status_valid or not tree_valid:
+            raise StateConflict(
+                "NATIVE_SESSION_VERIFICATION_FAILED",
+                "provider status or parent/child tree is inconsistent",
+            )
+        simulated = bool(getattr(self.client, "simulation", True))
+        return SessionVerification(
+            broker_session_verified=True,
+            openclaw_native_child_session_verified=(
+                self.provider == "openclaw" and not simulated
+            ),
+            codex_child_session_verified=(self.provider == "codex" and not simulated),
+            simulation=simulated,
+        )
+
+    def send_text(self, session: NativeSession, text: str) -> TextResult:
+        self._validate_session(session)
+        if not isinstance(text, str) or not text:
+            raise StateConflict("TEXT_RESULT_REQUIRED", "non-empty text is required")
+        response = self.client.send_text(
+            session.provider_session_id,
+            {
+                "task_id": session.task_id,
+                "content_type": "text/plain",
+                "encoding": "utf-8",
+                "image_only": False,
+                "text": text,
+                "recipient_binding": dict(session.recipient_binding),
+                "secrets_included": False,
+            },
+            self.credential_reference,
+        )
+        required = {
+            "content_type": "text/plain",
+            "encoding": "utf-8",
+            "image_only": False,
+            "task_id": session.task_id,
+            "recipient_binding": dict(session.recipient_binding),
+            "secrets_included": False,
+        }
+        if any(response.get(key) != value for key, value in required.items()):
+            raise StateConflict("TEXT_RESULT_CONTRACT_INVALID", "text result binding is invalid")
+        message_id = response.get("message_id")
+        receipt = response.get("receipt")
+        if not isinstance(message_id, str) or not message_id or not isinstance(receipt, str) or not receipt:
+            raise StateConflict("TEXT_RESULT_RECEIPT_REQUIRED", "message and receipt are required")
+        return TextResult(
+            content_type="text/plain",
+            encoding="utf-8",
+            image_only=False,
+            task_id=session.task_id,
+            message_id=message_id,
+            receipt=receipt,
+            recipient_binding=dict(session.recipient_binding),
+            secrets_included=False,
+        )
+
+    def status(self, session: NativeSession) -> str:
+        self._validate_session(session)
+        response = self.client.status(session.provider_session_id, self.credential_reference)
+        if response.get("provider_session_id") != session.provider_session_id:
+            raise StateConflict("NATIVE_SESSION_STATUS_INVALID", "provider session ID changed")
+        state = response.get("state")
+        if not isinstance(state, str) or not state:
+            raise StateConflict("NATIVE_SESSION_STATUS_INVALID", "provider state is missing")
+        return state
+
+    def interrupt(self, session: NativeSession) -> None:
+        self._validate_session(session)
+        self.client.interrupt(session.provider_session_id, self.credential_reference)
+        if self.status(session) != "INTERRUPTED":
+            raise StateConflict("NATIVE_SESSION_INTERRUPT_FAILED", "interrupt was not observed")
+
+    def close(self, session: NativeSession) -> None:
+        self._validate_session(session)
+        self.client.close(session.provider_session_id, self.credential_reference)
+        if self.status(session) != "CLOSED":
+            raise StateConflict("NATIVE_SESSION_CLOSE_FAILED", "close was not observed")
+
+    def _validate_session(self, session: NativeSession) -> None:
+        if session.provider != self.provider:
+            raise StateConflict("NATIVE_SESSION_PROVIDER_MISMATCH", "session provider differs")
+        if session.broker_session_id == session.provider_session_id:
+            raise StateConflict("SESSION_ID_COLLISION", "broker and provider IDs must differ")
+
+
+class OpenClawNativeAdapter(CandidateNativeAdapter):
+    provider = "openclaw"
+    required_reference = "systemd-credential:openclaw-gateway-candidate"
+
+
+class CodexNativeAdapter(CandidateNativeAdapter):
+    provider = "codex"
+    required_reference = "systemd-credential:codex-app-server-candidate"

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/orchestrator.py

@@ -0,0 +1,54 @@
+"""High-level autonomous release orchestration facade."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .evidence import EvidenceChain
+from .models import TaskPlan
+from .policy import ReleasePolicy
+from .workflow import WorkflowEngine
+
+
+@dataclass(frozen=True, slots=True)
+class OrchestrationResult:
+    task_id: str
+    plan_digest: str
+    evidence_closure: str | None
+    production_side_effects: bool
+
+
+class AutonomousReleaseOrchestrator:
+    """Coordinates candidate workflows while keeping production disabled."""
+
+    def __init__(self, workflow: WorkflowEngine, policy: ReleasePolicy) -> None:
+        self.workflow = workflow
+        self.policy = policy
+        self.evidence: EvidenceChain | None = None
+
+    def create(self, plan: TaskPlan) -> OrchestrationResult:
+        self.policy.validate_task(plan)
+        self.workflow.plan(plan)
+        self.evidence = EvidenceChain(self.workflow.store, plan.task_id)
+        self.evidence.append(
+            "TASK_CREATED",
+            plan.task_id,
+            {"plan_digest": plan.digest(), "secrets_included": False},
+        )
+        return OrchestrationResult(
+            task_id=plan.task_id,
+            plan_digest=plan.digest(),
+            evidence_closure=None,
+            production_side_effects=False,
+        )
+
+    def close_evidence(self, task_id: str) -> OrchestrationResult:
+        if self.evidence is None or self.evidence.task_id != task_id:
+            self.evidence = EvidenceChain(self.workflow.store, task_id)
+        closure = self.evidence.closure_digest()
+        return OrchestrationResult(
+            task_id=task_id,
+            plan_digest="",
+            evidence_closure=closure,
+            production_side_effects=False,
+        )

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/planner.py

@@ -0,0 +1,75 @@
+"""Deterministic task planning for autonomous candidate releases."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .models import ResourceBinding, RiskClass, TaskPlan, WorkflowStep
+
+
+@dataclass(frozen=True, slots=True)
+class PlanningRequest:
+    task_id: str
+    intent: str
+    binding: ResourceBinding
+    risk: RiskClass = RiskClass.MEDIUM
+
+
+class DeterministicPlanner:
+    """Create a fixed governed DAG without granting production capabilities."""
+
+    def build(self, request: PlanningRequest) -> TaskPlan:
+        steps = (
+            WorkflowStep(
+                step_id="architecture-review",
+                worker_role="hermes-review",
+                required_capabilities=("architecture.review", "risk.analysis"),
+                allowed_paths=request.binding.allowed_paths,
+            ),
+            WorkflowStep(
+                step_id="implementation",
+                worker_role="dedicated-writer",
+                required_capabilities=("commit_file.atomic", "git.push.patch_branch"),
+                depends_on=("architecture-review",),
+                max_attempts=2,
+                allowed_paths=request.binding.allowed_paths,
+            ),
+            WorkflowStep(
+                step_id="code-review",
+                worker_role="codex-review",
+                required_capabilities=("code.review", "git.diff.read"),
+                depends_on=("implementation",),
+                allowed_paths=request.binding.allowed_paths,
+            ),
+            WorkflowStep(
+                step_id="validation",
+                worker_role="openclaw-shadow",
+                required_capabilities=("test.execute", "artifact.verify"),
+                depends_on=("code-review",),
+                max_attempts=2,
+                allowed_paths=request.binding.allowed_paths,
+            ),
+            WorkflowStep(
+                step_id="evidence-finalize",
+                worker_role="evidence-controller",
+                required_capabilities=("evidence.finalize",),
+                depends_on=("validation",),
+                allowed_paths=request.binding.allowed_paths,
+            ),
+        )
+        return TaskPlan(
+            task_id=request.task_id,
+            intent=request.intent,
+            binding=request.binding,
+            risk=request.risk,
+            steps=steps,
+            evidence_requirements=(
+                "git-ref-closure",
+                "test-report",
+                "secret-scan",
+                "artifact-manifest",
+                "credential-revocation",
+                "production-invariants",
+            ),
+            production_side_effects_allowed=False,
+        )

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/policy.py

@@ -0,0 +1,54 @@
+"""Fail-closed policy engine for autonomous release workflows."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from .errors import ApprovalRequired, CapabilityMismatch, PolicyDenied
+from .models import ApprovalEnvelope, ResourceBinding, RiskClass, TaskPlan, WorkerDescriptor
+
+
+FORBIDDEN_CAPABILITIES = frozenset({
+    "production.deploy",
+    "production.rollback",
+    "main.write",
+    "tag.write",
+    "baseline.write",
+    "registry.write",
+    "secret.export",
+})
+
+
+@dataclass(frozen=True, slots=True)
+class PolicyDecision:
+    allowed: bool
+    reason: str
+
+
+class ReleasePolicy:
+    def validate_binding(self, binding: ResourceBinding) -> PolicyDecision:
+        if binding.branch == "main":
+            raise PolicyDenied("MAIN_WRITE_DENIED", "main branch is never writable")
+        if not binding.allowed_paths:
+            raise PolicyDenied("PATH_SCOPE_REQUIRED", "candidate path scope is required")
+        return PolicyDecision(True, "binding_valid")
+
+    def validate_task(self, plan: TaskPlan) -> PolicyDecision:
+        self.validate_binding(plan.binding)
+        if plan.production_side_effects_allowed:
+            raise PolicyDenied("PRODUCTION_SIDE_EFFECT_DENIED", "candidate workflows cannot execute production side effects")
+        return PolicyDecision(True, "task_valid")
+
+    def validate_approval(self, approval: ApprovalEnvelope, now: datetime | None = None) -> PolicyDecision:
+        now = now or datetime.now(timezone.utc)
+        if now < approval.not_before or now >= approval.expires_at:
+            raise ApprovalRequired("APPROVAL_TIME_INVALID", "approval is outside its validity window")
+        if any(cap in FORBIDDEN_CAPABILITIES for cap in approval.capabilities):
+            raise PolicyDenied("CAPABILITY_FORBIDDEN", "approval requests forbidden capability")
+        return PolicyDecision(True, "approval_valid")
+
+    def validate_worker(self, worker: WorkerDescriptor, required: tuple[str, ...], trust_zone: str) -> PolicyDecision:
+        if not worker.can_accept(required, trust_zone):
+            raise CapabilityMismatch("WORKER_CAPABILITY_MISMATCH", "worker cannot satisfy task requirements")
+        return PolicyDecision(True, "worker_valid")

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/release.py

@@ -0,0 +1,152 @@
+"""Governed deployment and rollback state machine."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+
+from .approval import ApprovalVerificationResult
+from .errors import ReleaseBlocked
+from .models import ReleasePlan
+
+
+class ReleaseState(str, Enum):
+    CREATED = "CREATED"
+    BUILT = "BUILT"
+    VERIFIED = "VERIFIED"
+    WAITING_OWNER = "WAITING_OWNER"
+    APPROVED = "APPROVED"
+    DEPLOYING = "DEPLOYING"
+    DEPLOYED = "DEPLOYED"
+    FAILED = "FAILED"
+    ROLLBACK_REQUESTED = "ROLLBACK_REQUESTED"
+    ROLLBACK_APPROVED = "ROLLBACK_APPROVED"
+    ROLLING_BACK = "ROLLING_BACK"
+    ROLLED_BACK = "ROLLED_BACK"
+
+
+@dataclass(frozen=True, slots=True)
+class ReleaseStatus:
+    state: ReleaseState
+    reason: str | None = None
+
+
+class ReleaseController:
+    """No transition reaches deployment or rollback without an external receipt."""
+
+    def __init__(self) -> None:
+        self._state = ReleaseState.CREATED
+        self._plan: ReleasePlan | None = None
+        self._reason: str | None = None
+
+    def _expect(self, *states: ReleaseState) -> None:
+        if self._state not in states:
+            raise ReleaseBlocked(
+                "RELEASE_STATE_INVALID",
+                "release transition is not allowed from the current state",
+                {"state": self._state.value, "allowed": [item.value for item in states]},
+            )
+
+    def attach_plan(self, plan: ReleasePlan) -> ReleaseStatus:
+        if self._plan is not None:
+            if self._plan == plan:
+                return self.status()
+            raise ReleaseBlocked("RELEASE_PLAN_ALREADY_ATTACHED", "release plan is immutable")
+        self._plan = plan
+        return self.status()
+
+    def mark_built(self) -> ReleaseStatus:
+        self._require_plan()
+        self._expect(ReleaseState.CREATED)
+        self._state = ReleaseState.BUILT
+        return self.status()
+
+    def verify(self, *, artifact_ok: bool, evidence_ok: bool) -> ReleaseStatus:
+        self._require_plan()
+        self._expect(ReleaseState.BUILT)
+        if not artifact_ok or not evidence_ok:
+            raise ReleaseBlocked(
+                "RELEASE_VERIFICATION_FAILED", "artifact or evidence verification failed"
+            )
+        self._state = ReleaseState.VERIFIED
+        return self.status()
+
+    def request_owner_approval(self) -> ReleaseStatus:
+        self._expect(ReleaseState.VERIFIED)
+        self._state = ReleaseState.WAITING_OWNER
+        return self.status()
+
+    def approve(self, receipt: ApprovalVerificationResult) -> ReleaseStatus:
+        plan = self._require_plan()
+        self._expect(ReleaseState.WAITING_OWNER)
+        if receipt.task_id != plan.task_id:
+            raise ReleaseBlocked("OWNER_APPROVAL_TASK_MISMATCH", "approval task does not match")
+        if receipt.approval_plan_id != plan.approval_plan_id:
+            raise ReleaseBlocked(
+                "OWNER_APPROVAL_PLAN_MISMATCH", "approval plan does not match release plan"
+            )
+        self._state = ReleaseState.APPROVED
+        return self.status()
+
+    def start_deployment(self) -> ReleaseStatus:
+        self._expect(ReleaseState.APPROVED)
+        self._state = ReleaseState.DEPLOYING
+        return self.status()
+
+    def mark_deployed(self) -> ReleaseStatus:
+        self._expect(ReleaseState.DEPLOYING)
+        self._state = ReleaseState.DEPLOYED
+        return self.status()
+
+    def mark_failed(self, reason: str) -> ReleaseStatus:
+        self._expect(ReleaseState.DEPLOYING, ReleaseState.DEPLOYED)
+        if not reason:
+            raise ReleaseBlocked("RELEASE_FAILURE_REASON_REQUIRED", "failure reason is required")
+        self._reason = reason
+        self._state = ReleaseState.FAILED
+        return self.status()
+
+    def request_rollback(self, reason: str) -> ReleaseStatus:
+        self._expect(ReleaseState.DEPLOYING, ReleaseState.DEPLOYED, ReleaseState.FAILED)
+        if not reason:
+            raise ReleaseBlocked("ROLLBACK_REASON_REQUIRED", "rollback reason is required")
+        self._reason = reason
+        self._state = ReleaseState.ROLLBACK_REQUESTED
+        return self.status()
+
+    def approve_rollback(self, receipt: ApprovalVerificationResult) -> ReleaseStatus:
+        plan = self._require_plan()
+        self._expect(ReleaseState.ROLLBACK_REQUESTED)
+        if receipt.task_id != plan.task_id:
+            raise ReleaseBlocked("ROLLBACK_APPROVAL_TASK_MISMATCH", "rollback task does not match")
+        if receipt.approval_plan_id != plan.rollback_approval_plan_id:
+            raise ReleaseBlocked(
+                "ROLLBACK_APPROVAL_PLAN_MISMATCH",
+                "rollback approval plan does not match release plan",
+            )
+        self._state = ReleaseState.ROLLBACK_APPROVED
+        return self.status()
+
+    def start_rollback(self) -> ReleaseStatus:
+        self._expect(ReleaseState.ROLLBACK_APPROVED)
+        self._state = ReleaseState.ROLLING_BACK
+        return self.status()
+
+    def mark_rolled_back(self) -> ReleaseStatus:
+        self._expect(ReleaseState.ROLLING_BACK)
+        self._state = ReleaseState.ROLLED_BACK
+        return self.status()
+
+    def rollback(self) -> ReleaseStatus:
+        raise ReleaseBlocked(
+            "ROLLBACK_APPROVAL_REQUIRED",
+            "use request_rollback, approve_rollback and start_rollback",
+        )
+
+    def status(self) -> ReleaseStatus:
+        return ReleaseStatus(self._state, self._reason)
+
+    def _require_plan(self) -> ReleasePlan:
+        if self._plan is None:
+            raise ReleaseBlocked("RELEASE_PLAN_REQUIRED", "attach a release plan first")
+        return self._plan

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/release_gate.py

@@ -0,0 +1,72 @@
+"""Read-only multi-source release consistency gate."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .errors import ReleaseBlocked
+from .models import is_git_oid, is_sha256
+
+
+@dataclass(frozen=True, slots=True)
+class ReleaseEvidence:
+    expected_version: str
+    registry_version: str
+    baseline_version: str
+    local_tag_oid: str
+    remote_tag_oid: str
+    gitea_tag_oid: str
+    expected_tag_oid: str
+    local_peeled_target_oid: str
+    remote_peeled_target_oid: str
+    gitea_peeled_target_oid: str
+    expected_target_oid: str
+    artifact_sha256: str
+    manifest_sha256: str
+    evidence_sha256: str
+    rollback_artifact_sha256: str
+
+
+class ReleaseGate:
+    """Require exact tag-object, peeled-target, version and package closure."""
+
+    def verify(self, evidence: ReleaseEvidence) -> None:
+        versions = (
+            evidence.expected_version,
+            evidence.registry_version,
+            evidence.baseline_version,
+        )
+        if not versions[0] or len(set(versions)) != 1:
+            raise ReleaseBlocked("VERSION_MISMATCH", "expected, registry and baseline differ")
+        tag_oids = (
+            evidence.local_tag_oid,
+            evidence.remote_tag_oid,
+            evidence.gitea_tag_oid,
+            evidence.expected_tag_oid,
+        )
+        if any(not is_git_oid(value) for value in tag_oids):
+            raise ReleaseBlocked("TAG_OID_INVALID", "tag object OID is invalid")
+        if evidence.local_tag_oid != evidence.remote_tag_oid:
+            raise ReleaseBlocked("TAG_REMOTE_MISMATCH", "local and remote tag objects differ")
+        if evidence.remote_tag_oid != evidence.gitea_tag_oid:
+            raise ReleaseBlocked("TAG_GITEA_MISMATCH", "Gitea tag object differs")
+        if evidence.gitea_tag_oid != evidence.expected_tag_oid:
+            raise ReleaseBlocked("TAG_OBJECT_MISMATCH", "tag object differs from ReleasePlan")
+        targets = (
+            evidence.local_peeled_target_oid,
+            evidence.remote_peeled_target_oid,
+            evidence.gitea_peeled_target_oid,
+            evidence.expected_target_oid,
+        )
+        if any(not is_git_oid(value) for value in targets):
+            raise ReleaseBlocked("TAG_TARGET_OID_INVALID", "peeled target OID is invalid")
+        if len(set(targets)) != 1:
+            raise ReleaseBlocked("TAG_TARGET_MISMATCH", "peeled tag targets are not identical")
+        digests = (
+            evidence.artifact_sha256,
+            evidence.manifest_sha256,
+            evidence.evidence_sha256,
+            evidence.rollback_artifact_sha256,
+        )
+        if any(not is_sha256(value) for value in digests):
+            raise ReleaseBlocked("ARTIFACT_DIGEST_MISSING", "package closure is incomplete")

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/router.py

@@ -0,0 +1,36 @@
+"""Deterministic worker routing for candidate autonomous release workflows."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .errors import CapabilityMismatch
+from .models import WorkerDescriptor
+
+
+@dataclass(frozen=True, slots=True)
+class RouteRequest:
+    task_id: str
+    required_capabilities: tuple[str, ...]
+    trust_zone: str
+
+
+class WorkerRouter:
+    """Select the least loaded compatible worker without expanding trust."""
+
+    def __init__(self, workers: list[WorkerDescriptor]) -> None:
+        self._workers = list(workers)
+
+    def route(self, request: RouteRequest) -> WorkerDescriptor:
+        candidates = [
+            worker
+            for worker in self._workers
+            if worker.can_accept(request.required_capabilities, request.trust_zone)
+        ]
+        if not candidates:
+            raise CapabilityMismatch(
+                "NO_COMPATIBLE_WORKER",
+                "no worker satisfies capability and trust constraints",
+                {"task_id": request.task_id},
+            )
+        return sorted(candidates, key=lambda item: (item.current_load, item.worker_id))[0]

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/secret_manager.py

@@ -0,0 +1,213 @@
+"""AIOS Secret Manager V2 reference-only Candidate adapter."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Mapping, Protocol, Sequence
+
+from .errors import CredentialUnavailable
+from .models import CredentialRequest
+
+
+PROVIDER_REFERENCES = {
+    "openclaw": "systemd-credential:openclaw-gateway-candidate",
+    "codex": "systemd-credential:codex-app-server-candidate",
+    "native_adapter_control": "systemd-credential:session-native-adapter-control-candidate",
+    "gitea": "systemd-credential:gitea-admin-api-candidate",
+}
+FORBIDDEN_RESPONSE_FIELDS = {
+    "api_key",
+    "token",
+    "password",
+    "cookie",
+    "private_key",
+    "credential_body",
+    "plaintext",
+}
+
+
+class SecretManagerTransport(Protocol):
+    def exchange(self, operation: str, payload: Mapping[str, Any]) -> Mapping[str, Any]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class SecretProjectionReceipt:
+    task_id: str
+    credential_id: str
+    reference: str
+    provider: str
+    audience: str
+    consumer_service: str
+    state: str
+    expires_at: datetime
+    plaintext_secret_returned: bool = False
+
+
+class SecretManagerV2:
+    """Issue, project, inspect and revoke without receiving authentication material."""
+
+    def __init__(self, transport: SecretManagerTransport) -> None:
+        self.transport = transport
+
+    @staticmethod
+    def _assert_safe(value: Any) -> None:
+        if isinstance(value, Mapping):
+            if FORBIDDEN_RESPONSE_FIELDS.intersection(str(key).lower() for key in value):
+                raise CredentialUnavailable(
+                    "SECRET_MANAGER_PLAINTEXT_REJECTED",
+                    "Secret Manager response included a forbidden plaintext field",
+                )
+            for item in value.values():
+                SecretManagerV2._assert_safe(item)
+        elif isinstance(value, (list, tuple)):
+            for item in value:
+                SecretManagerV2._assert_safe(item)
+
+    def _exchange(self, operation: str, payload: Mapping[str, Any]) -> Mapping[str, Any]:
+        try:
+            response = self.transport.exchange(operation, payload)
+        except CredentialUnavailable:
+            raise
+        except Exception as error:
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_UNAVAILABLE", "Secret Manager V2 operation failed"
+            ) from error
+        if not isinstance(response, Mapping):
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_RESPONSE_INVALID", "Secret Manager V2 returned a non-object"
+            )
+        self._assert_safe(response)
+        if response.get("plaintext_secret_returned") is not False:
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_PLAINTEXT_REJECTED",
+                "Secret Manager did not attest plaintext_secret_returned=false",
+            )
+        return response
+
+    @staticmethod
+    def _payload(request: CredentialRequest) -> dict[str, Any]:
+        return {
+            "approval_plan_id": request.approval_plan_id,
+            "task_id": request.task_id,
+            "request_id": request.request_id,
+            "session_id": request.session_id,
+            "worker_id": request.worker_id,
+            "provider": request.provider,
+            "audience": request.audience,
+            "consumer_service": request.consumer_service,
+            "runtime_user": request.runtime_user,
+            "repository": request.binding.repository,
+            "branch": request.binding.branch,
+            "branch_oid": request.binding.base_oid,
+            "allowed_paths": list(request.binding.allowed_paths),
+            "operations": list(request.operations),
+            "not_before": request.not_before.astimezone(timezone.utc).isoformat(),
+            "expires_at": request.expires_at.astimezone(timezone.utc).isoformat(),
+            "nonce": request.nonce,
+            "ttl_seconds": request.ttl_seconds,
+            "max_uses": request.max_uses,
+        }
+
+    def issue(
+        self, request: CredentialRequest, now: datetime | None = None
+    ) -> SecretProjectionReceipt:
+        now = now or datetime.now(timezone.utc)
+        if now < request.not_before or now >= request.expires_at:
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_APPROVAL_WINDOW_INVALID", "credential request is not currently valid"
+            )
+        expected_reference = PROVIDER_REFERENCES.get(request.provider)
+        if expected_reference is None:
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_PROVIDER_DENIED", "provider is not Candidate-approved"
+            )
+        response = self._exchange("issue", self._payload(request))
+        credential_id = response.get("credential_id")
+        if (
+            response.get("task_id") != request.task_id
+            or not isinstance(credential_id, str)
+            or not credential_id
+            or response.get("credential_reference") != expected_reference
+            or response.get("state") != "ISSUED"
+        ):
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_BINDING_MISMATCH", "issued projection does not match the request"
+            )
+        expires_at = datetime.fromisoformat(str(response.get("expires_at")))
+        if expires_at > request.expires_at or expires_at <= now:
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_EXPIRY_INVALID", "issued projection expiry is outside approval"
+            )
+        return SecretProjectionReceipt(
+            task_id=request.task_id,
+            credential_id=credential_id,
+            reference=expected_reference,
+            provider=request.provider,
+            audience=request.audience,
+            consumer_service=request.consumer_service,
+            state="ISSUED",
+            expires_at=expires_at,
+            plaintext_secret_returned=False,
+        )
+
+    @staticmethod
+    def _receipt_payload(receipt: SecretProjectionReceipt) -> dict[str, Any]:
+        return {
+            "task_id": receipt.task_id,
+            "credential_id": receipt.credential_id,
+            "credential_reference": receipt.reference,
+            "provider": receipt.provider,
+            "audience": receipt.audience,
+            "consumer_service": receipt.consumer_service,
+        }
+
+    def _state_operation(
+        self, operation: str, receipt: SecretProjectionReceipt, expected_state: str
+    ) -> Mapping[str, Any]:
+        response = self._exchange(operation, self._receipt_payload(receipt))
+        if (
+            response.get("task_id") != receipt.task_id
+            or response.get("credential_id") != receipt.credential_id
+            or response.get("state") != expected_state
+        ):
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_STATE_MISMATCH", f"{operation} response is not task-bound"
+            )
+        return response
+
+    def project(self, receipt: SecretProjectionReceipt) -> None:
+        self._state_operation("project", receipt, "PROJECTED")
+
+    def status(self, receipt: SecretProjectionReceipt) -> str:
+        return str(self._state_operation("status", receipt, "PROJECTED")["state"])
+
+    def revoke(self, receipt: SecretProjectionReceipt) -> None:
+        self._state_operation("revoke", receipt, "REVOKED")
+
+    def verify_revocation(self, receipt: SecretProjectionReceipt) -> None:
+        response = self._state_operation("verify_revocation", receipt, "REVOKED")
+        if response.get("post_revocation_auth") != "REJECTED":
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_REVOCATION_UNVERIFIED",
+                "post-revocation authentication was not rejected",
+            )
+
+    def audit(self, receipt: SecretProjectionReceipt) -> Sequence[Mapping[str, Any]]:
+        response = self._exchange("audit", self._receipt_payload(receipt))
+        if (
+            response.get("task_id") != receipt.task_id
+            or response.get("credential_id") != receipt.credential_id
+        ):
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_AUDIT_MISMATCH", "audit response is not task-bound"
+            )
+        events = response.get("events")
+        if not isinstance(events, list) or any(
+            not isinstance(event, Mapping) or event.get("secrets_included") is not False
+            for event in events
+        ):
+            raise CredentialUnavailable(
+                "SECRET_MANAGER_AUDIT_INVALID", "audit events are absent or unsafe"
+            )
+        return events

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/serialization.py

@@ -0,0 +1,116 @@
+"""JSON serialization helpers for candidate API and CLI boundaries."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Mapping
+
+from .models import ReleasePlan, ResourceBinding, RiskClass, TaskPlan, WorkflowStep
+
+
+def task_plan_from_mapping(value: Mapping[str, Any]) -> TaskPlan:
+    binding_value = value["binding"]
+    binding = ResourceBinding(
+        repository=str(binding_value["repository"]),
+        branch=str(binding_value["branch"]),
+        base_oid=str(binding_value["base_oid"]),
+        allowed_paths=tuple(str(item) for item in binding_value.get("allowed_paths", ())),
+    )
+    steps = tuple(
+        WorkflowStep(
+            step_id=str(item["step_id"]),
+            worker_role=str(item["worker_role"]),
+            required_capabilities=tuple(str(cap) for cap in item.get("required_capabilities", ())),
+            depends_on=tuple(str(dep) for dep in item.get("depends_on", ())),
+            max_attempts=int(item.get("max_attempts", 1)),
+            timeout_seconds=int(item.get("timeout_seconds", 900)),
+            allowed_paths=tuple(str(path) for path in item.get("allowed_paths", ())),
+        )
+        for item in value.get("steps", ())
+    )
+    created_at_raw = value.get("created_at")
+    created_at = datetime.fromisoformat(created_at_raw) if created_at_raw else None
+    kwargs: dict[str, Any] = {
+        "task_id": str(value["task_id"]),
+        "intent": str(value["intent"]),
+        "binding": binding,
+        "risk": RiskClass(str(value.get("risk", RiskClass.MEDIUM.value))),
+        "steps": steps,
+        "evidence_requirements": tuple(str(item) for item in value.get("evidence_requirements", ())),
+        "production_side_effects_allowed": bool(value.get("production_side_effects_allowed", False)),
+    }
+    if created_at is not None:
+        kwargs["created_at"] = created_at
+    return TaskPlan(**kwargs)
+
+
+def task_plan_to_mapping(plan: TaskPlan) -> dict[str, Any]:
+    return {
+        "task_id": plan.task_id,
+        "intent": plan.intent,
+        "binding": {
+            "repository": plan.binding.repository,
+            "branch": plan.binding.branch,
+            "base_oid": plan.binding.base_oid,
+            "allowed_paths": list(plan.binding.allowed_paths),
+        },
+        "risk": plan.risk.value,
+        "steps": [
+            {
+                "step_id": step.step_id,
+                "worker_role": step.worker_role,
+                "required_capabilities": list(step.required_capabilities),
+                "depends_on": list(step.depends_on),
+                "max_attempts": step.max_attempts,
+                "timeout_seconds": step.timeout_seconds,
+                "allowed_paths": list(step.allowed_paths),
+            }
+            for step in plan.steps
+        ],
+        "evidence_requirements": list(plan.evidence_requirements),
+        "production_side_effects_allowed": plan.production_side_effects_allowed,
+        "created_at": plan.created_at.isoformat(),
+        "plan_digest": plan.digest(),
+    }
+
+
+def release_plan_from_mapping(value: Mapping[str, Any]) -> ReleasePlan:
+    return ReleasePlan(
+        task_id=str(value["task_id"]),
+        repository=str(value["repository"]),
+        branch=str(value["branch"]),
+        version=str(value["version"]),
+        target_oid=str(value["target_oid"]),
+        tree_oid=str(value["tree_oid"]),
+        artifact_sha256=str(value["artifact_sha256"]),
+        manifest_sha256=str(value["manifest_sha256"]),
+        package_closure_sha256=str(value["package_closure_sha256"]),
+        evidence_closure_oid=str(value["evidence_closure_oid"]),
+        approval_plan_id=str(value["approval_plan_id"]),
+        rollback_approval_plan_id=str(value["rollback_approval_plan_id"]),
+        rollback_artifact_sha256=str(value["rollback_artifact_sha256"]),
+        expected_tag_oid=str(value["expected_tag_oid"]),
+        expected_target_oid=str(value["expected_target_oid"]),
+        production_config_sha256=str(value["production_config_sha256"]),
+    )
+
+
+def release_plan_to_mapping(plan: ReleasePlan) -> dict[str, Any]:
+    return {
+        "task_id": plan.task_id,
+        "repository": plan.repository,
+        "branch": plan.branch,
+        "version": plan.version,
+        "target_oid": plan.target_oid,
+        "tree_oid": plan.tree_oid,
+        "artifact_sha256": plan.artifact_sha256,
+        "manifest_sha256": plan.manifest_sha256,
+        "package_closure_sha256": plan.package_closure_sha256,
+        "evidence_closure_oid": plan.evidence_closure_oid,
+        "approval_plan_id": plan.approval_plan_id,
+        "rollback_approval_plan_id": plan.rollback_approval_plan_id,
+        "rollback_artifact_sha256": plan.rollback_artifact_sha256,
+        "expected_tag_oid": plan.expected_tag_oid,
+        "expected_target_oid": plan.expected_target_oid,
+        "production_config_sha256": plan.production_config_sha256,
+    }

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/service.py

@@ -0,0 +1,223 @@
+"""Hardened loopback HTTP boundary for the dormant Candidate."""
+
+from __future__ import annotations
+
+import argparse
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import ipaddress
+import json
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+from .api import AutonomousReleaseAPI
+from .errors import AutonomousReleaseError
+from .orchestrator import AutonomousReleaseOrchestrator
+from .policy import ReleasePolicy
+from .store import StateStore
+from .workflow import WorkflowEngine
+
+
+class CandidateApplication:
+    def __init__(self, state_path: str | Path) -> None:
+        self.store = StateStore(state_path)
+        workflow = WorkflowEngine(self.store)
+        self.api = AutonomousReleaseAPI(
+            AutonomousReleaseOrchestrator(workflow, ReleasePolicy())
+        )
+
+    def ready(self) -> bool:
+        with self.store.transaction() as connection:
+            return connection.execute("SELECT 1").fetchone()[0] == 1
+
+
+class CandidateRequestHandler(BaseHTTPRequestHandler):
+    server_version = "AIOSAutonomousReleaseCandidate/2"
+    sys_version = ""
+
+    @property
+    def application(self) -> CandidateApplication:
+        return self.server.application  # type: ignore[attr-defined]
+
+    def _send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None:
+        encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        self.send_response(status.value)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(encoded)))
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.send_header("Referrer-Policy", "no-referrer")
+        self.end_headers()
+        self.wfile.write(encoded)
+
+    def _read_json(self) -> dict[str, Any]:
+        content_type = self.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
+        if content_type != "application/json":
+            raise UnsupportedMediaType
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError as error:
+            raise InvalidRequest from error
+        if length <= 0 or length > 1_048_576:
+            raise InvalidRequest
+        try:
+            raw = self.rfile.read(length)
+            value = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as error:
+            raise InvalidRequest from error
+        if not isinstance(value, dict):
+            raise InvalidRequest
+        return value
+
+    def do_GET(self) -> None:  # noqa: N802
+        path = urlparse(self.path).path
+        if path == "/health":
+            self._send_json(
+                HTTPStatus.OK,
+                {
+                    "status": "ok",
+                    "mode": "candidate-shadow-only",
+                    "production_side_effects": False,
+                    "secrets_included": False,
+                },
+            )
+            return
+        if path == "/ready":
+            if self.application.ready():
+                self._send_json(
+                    HTTPStatus.OK, {"status": "ready", "secrets_included": False}
+                )
+            else:
+                self._send_json(
+                    HTTPStatus.SERVICE_UNAVAILABLE,
+                    {"status": "not_ready", "secrets_included": False},
+                )
+            return
+        if path.startswith("/v1/tasks/") and path.endswith("/steps"):
+            task_id = path[len("/v1/tasks/") : -len("/steps")].strip("/")
+            if not task_id or "/" in task_id:
+                self._send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid_request"})
+                return
+            try:
+                self._send_json(HTTPStatus.OK, self.application.api.task_steps(task_id))
+            except KeyError:
+                self._send_json(HTTPStatus.NOT_FOUND, {"error": "task_not_found"})
+            return
+        if path.startswith("/v1/tasks/") and path.endswith("/evidence"):
+            task_id = path[len("/v1/tasks/") : -len("/evidence")].strip("/")
+            if not task_id or "/" in task_id:
+                self._send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid_request"})
+                return
+            try:
+                self._send_json(HTTPStatus.OK, self.application.api.evidence(task_id))
+            except KeyError:
+                self._send_json(HTTPStatus.NOT_FOUND, {"error": "task_not_found"})
+            return
+        if path.startswith("/v1/tasks/"):
+            task_id = path[len("/v1/tasks/") :].strip("/")
+            if not task_id or "/" in task_id:
+                self._send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid_request"})
+                return
+            try:
+                self._send_json(HTTPStatus.OK, self.application.api.task_status(task_id))
+            except KeyError:
+                self._send_json(HTTPStatus.NOT_FOUND, {"error": "task_not_found"})
+            return
+        self._send_json(HTTPStatus.NOT_FOUND, {"error": "not_found"})
+
+    def do_POST(self) -> None:  # noqa: N802
+        path = urlparse(self.path).path
+        try:
+            payload = self._read_json()
+            if path == "/v1/tasks":
+                self._send_json(HTTPStatus.CREATED, self.application.api.create_task(payload))
+                return
+            if path == "/v1/release/plan":
+                self._send_json(
+                    HTTPStatus.CREATED, self.application.api.create_release_plan(payload)
+                )
+                return
+            if path == "/v1/release/verify":
+                self._send_json(HTTPStatus.OK, self.application.api.verify_release(payload))
+                return
+            if path.startswith("/v1/tasks/") and path.endswith("/evidence/close"):
+                task_id = path[len("/v1/tasks/") : -len("/evidence/close")].strip("/")
+                if not task_id:
+                    raise InvalidRequest
+                self._send_json(HTTPStatus.OK, self.application.api.close_evidence(task_id))
+                return
+            for action, handler in (
+                ("claim", self.application.api.claim),
+                ("heartbeat", self.application.api.heartbeat),
+                ("result", self.application.api.result),
+            ):
+                suffix = f"/{action}"
+                if path.startswith("/v1/tasks/") and path.endswith(suffix):
+                    task_id = path[len("/v1/tasks/") : -len(suffix)].strip("/")
+                    if not task_id or "/" in task_id:
+                        raise InvalidRequest
+                    self._send_json(HTTPStatus.OK, handler(task_id, payload))
+                    return
+            self._send_json(HTTPStatus.NOT_FOUND, {"error": "not_found"})
+        except UnsupportedMediaType:
+            self._send_json(
+                HTTPStatus.UNSUPPORTED_MEDIA_TYPE, {"error": "content_type_required"}
+            )
+        except (InvalidRequest, ValueError, KeyError):
+            self._send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid_request"})
+        except AutonomousReleaseError as error:
+            self._send_json(HTTPStatus.CONFLICT, {"error": error.code})
+        except Exception:
+            self._send_json(
+                HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "candidate_request_failed"}
+            )
+
+    def log_message(self, _format: str, *_args: Any) -> None:
+        return
+
+
+class InvalidRequest(Exception):
+    pass
+
+
+class UnsupportedMediaType(Exception):
+    pass
+
+
+class CandidateHTTPServer(ThreadingHTTPServer):
+    daemon_threads = True
+    allow_reuse_address = False
+
+    def __init__(self, address: tuple[str, int], application: CandidateApplication) -> None:
+        try:
+            loopback = ipaddress.ip_address(address[0]).is_loopback
+        except ValueError as error:
+            raise ValueError("candidate service requires a numeric loopback address") from error
+        if not loopback:
+            raise ValueError("candidate service must bind to loopback")
+        self.application = application
+        super().__init__(address, CandidateRequestHandler)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="aios-autonomous-release-service")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=19951)
+    parser.add_argument("--state", required=True)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    application = CandidateApplication(args.state)
+    server = CandidateHTTPServer((args.host, args.port), application)
+    try:
+        server.serve_forever(poll_interval=0.5)
+    finally:
+        server.server_close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/session.py

@@ -0,0 +1,87 @@
+"""Unified candidate session lifecycle contracts."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from datetime import datetime
+
+from .errors import StateConflict
+
+
+class SessionState(str, Enum):
+    CREATED = "CREATED"
+    PROVISIONED = "PROVISIONED"
+    ACTIVE = "ACTIVE"
+    INTERRUPTED = "INTERRUPTED"
+    CLOSED = "CLOSED"
+    REVOKED = "REVOKED"
+
+
+_ALLOWED_TRANSITIONS: dict[SessionState, frozenset[SessionState]] = {
+    SessionState.CREATED: frozenset({SessionState.PROVISIONED, SessionState.CLOSED, SessionState.REVOKED}),
+    SessionState.PROVISIONED: frozenset({SessionState.ACTIVE, SessionState.CLOSED, SessionState.REVOKED}),
+    SessionState.ACTIVE: frozenset({SessionState.INTERRUPTED, SessionState.CLOSED, SessionState.REVOKED}),
+    SessionState.INTERRUPTED: frozenset({SessionState.ACTIVE, SessionState.CLOSED, SessionState.REVOKED}),
+    SessionState.CLOSED: frozenset({SessionState.REVOKED}),
+    SessionState.REVOKED: frozenset(),
+}
+
+
+@dataclass(frozen=True, slots=True)
+class SessionIdentity:
+    session_id: str
+    provider: str
+    provider_session_id: str
+    worker_id: str
+    task_id: str
+    created_at: datetime
+
+    def __post_init__(self) -> None:
+        if not all((self.session_id, self.provider, self.provider_session_id, self.worker_id, self.task_id)):
+            raise ValueError("all session identity fields are required")
+        if self.session_id == self.provider_session_id:
+            raise ValueError("broker and provider session IDs must be distinct")
+
+
+class SessionOrchestrator:
+    """Provider-neutral session lifecycle guard with explicit state transitions."""
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, tuple[SessionIdentity, SessionState]] = {}
+
+    def create(self, identity: SessionIdentity) -> None:
+        if identity.session_id in self._sessions:
+            raise StateConflict("SESSION_EXISTS", "session already exists")
+        self._sessions[identity.session_id] = (identity, SessionState.CREATED)
+
+    def transition(self, session_id: str, state: SessionState) -> None:
+        current = self._sessions.get(session_id)
+        if current is None:
+            raise StateConflict("SESSION_NOT_FOUND", "session does not exist")
+        current_state = current[1]
+        if state not in _ALLOWED_TRANSITIONS[current_state]:
+            raise StateConflict(
+                "SESSION_TRANSITION_INVALID",
+                "session state transition is not allowed",
+                {"session_id": session_id, "from": current_state.value, "to": state.value},
+            )
+        self._sessions[session_id] = (current[0], state)
+
+    def identity(self, session_id: str) -> SessionIdentity:
+        current = self._sessions.get(session_id)
+        if current is None:
+            raise StateConflict("SESSION_NOT_FOUND", "session does not exist")
+        return current[0]
+
+    def status(self, session_id: str) -> SessionState:
+        current = self._sessions.get(session_id)
+        if current is None:
+            raise StateConflict("SESSION_NOT_FOUND", "session does not exist")
+        return current[1]
+
+    def close(self, session_id: str) -> None:
+        self.transition(session_id, SessionState.CLOSED)
+
+    def revoke(self, session_id: str) -> None:
+        self.transition(session_id, SessionState.REVOKED)

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/store.py

@@ -0,0 +1,383 @@
+"""Durable, private SQLite state for the autonomous-release Candidate."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import sqlite3
+from typing import Any, Iterator, Mapping, Sequence
+
+from .errors import StateConflict
+from .models import StepState, TaskState, canonical_json
+
+
+SCHEMA_VERSION = 2
+
+
+@dataclass(frozen=True, slots=True)
+class StoredStep:
+    task_id: str
+    step_id: str
+    state: StepState
+    attempt: int
+    max_attempts: int
+    worker_id: str | None
+    session_id: str | None
+    lease_expires_at: datetime | None
+    result_digest: str | None
+    updated_at: datetime
+    last_error: str | None
+
+
+class StateStore:
+    """SQLite store with WAL, full sync, schema checks and serialized CAS writes."""
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path)
+        self._prepare_path()
+        self._initialize()
+        os.chmod(self.path, 0o600)
+
+    def _prepare_path(self) -> None:
+        if self.path.exists() and self.path.is_symlink():
+            raise StateConflict("STATE_PATH_SYMLINK", "state database must not be a symlink")
+        parent = self.path.parent
+        parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+        if parent.is_symlink():
+            raise StateConflict("STATE_DIRECTORY_SYMLINK", "state directory must not be a symlink")
+        current_mode = parent.stat().st_mode & 0o777
+        os.chmod(parent, current_mode & ~0o027)
+
+    def _connect(self) -> sqlite3.Connection:
+        if self.path.exists() and self.path.is_symlink():
+            raise StateConflict("STATE_PATH_SYMLINK", "state database must not be a symlink")
+        connection = sqlite3.connect(self.path, timeout=30.0)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA synchronous=FULL")
+        connection.execute("PRAGMA busy_timeout=30000")
+        return connection
+
+    @contextmanager
+    def transaction(self) -> Iterator[sqlite3.Connection]:
+        connection = self._connect()
+        try:
+            connection.execute("BEGIN IMMEDIATE")
+            yield connection
+            connection.commit()
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
+    def _initialize(self) -> None:
+        with self.transaction() as connection:
+            connection.execute(
+                "CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)"
+            )
+            row = connection.execute(
+                "SELECT value FROM metadata WHERE key='schema_version'"
+            ).fetchone()
+            version = int(row["value"]) if row is not None else None
+            if version not in (None, 1, SCHEMA_VERSION):
+                raise StateConflict(
+                    "SCHEMA_VERSION_UNSUPPORTED",
+                    "state schema is not supported",
+                    {"found": version, "expected": SCHEMA_VERSION},
+                )
+            connection.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS tasks (
+                    task_id TEXT PRIMARY KEY,
+                    plan_digest TEXT NOT NULL,
+                    plan_json TEXT NOT NULL,
+                    state TEXT NOT NULL,
+                    revision INTEGER NOT NULL DEFAULT 0,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS steps (
+                    task_id TEXT NOT NULL,
+                    step_id TEXT NOT NULL,
+                    state TEXT NOT NULL,
+                    attempt INTEGER NOT NULL DEFAULT 0,
+                    max_attempts INTEGER NOT NULL DEFAULT 1,
+                    worker_id TEXT,
+                    session_id TEXT,
+                    lease_expires_at TEXT,
+                    result_digest TEXT,
+                    last_error TEXT,
+                    updated_at TEXT NOT NULL,
+                    PRIMARY KEY(task_id, step_id),
+                    FOREIGN KEY(task_id) REFERENCES tasks(task_id) ON DELETE CASCADE
+                );
+                CREATE TABLE IF NOT EXISTS events (
+                    sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+                    task_id TEXT NOT NULL,
+                    event_type TEXT NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    previous_digest TEXT,
+                    digest TEXT NOT NULL UNIQUE,
+                    created_at TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS consumed_nonces (
+                    scope TEXT NOT NULL,
+                    nonce TEXT NOT NULL,
+                    consumed_at TEXT NOT NULL,
+                    PRIMARY KEY(scope, nonce)
+                );
+                CREATE TABLE IF NOT EXISTS credentials (
+                    credential_id TEXT PRIMARY KEY,
+                    task_id TEXT NOT NULL,
+                    reference TEXT NOT NULL UNIQUE,
+                    state TEXT NOT NULL,
+                    metadata_json TEXT NOT NULL,
+                    issued_at TEXT NOT NULL,
+                    expires_at TEXT NOT NULL,
+                    revoked_at TEXT
+                );
+                CREATE TABLE IF NOT EXISTS idempotency (
+                    namespace TEXT NOT NULL,
+                    operation_id TEXT NOT NULL,
+                    result_json TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    PRIMARY KEY(namespace, operation_id)
+                );
+                """
+            )
+            if version == 1:
+                existing = {
+                    item["name"] for item in connection.execute("PRAGMA table_info(steps)")
+                }
+                additions = {
+                    "max_attempts": "INTEGER NOT NULL DEFAULT 1",
+                    "lease_expires_at": "TEXT",
+                    "result_digest": "TEXT",
+                }
+                for column, declaration in additions.items():
+                    if column not in existing:
+                        connection.execute(f"ALTER TABLE steps ADD COLUMN {column} {declaration}")
+            connection.execute(
+                "INSERT INTO metadata(key, value) VALUES('schema_version', ?) "
+                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+                (str(SCHEMA_VERSION),),
+            )
+
+    @staticmethod
+    def _now() -> str:
+        return datetime.now(timezone.utc).isoformat()
+
+    def create_task(
+        self,
+        task_id: str,
+        plan_digest: str,
+        plan: Mapping[str, Any],
+        step_ids: tuple[str, ...],
+        step_limits: Mapping[str, int] | None = None,
+    ) -> None:
+        now = self._now()
+        limits = step_limits or {}
+        with self.transaction() as connection:
+            connection.execute(
+                "INSERT INTO tasks(task_id, plan_digest, plan_json, state, created_at, updated_at) "
+                "VALUES(?, ?, ?, ?, ?, ?)",
+                (task_id, plan_digest, canonical_json(plan), TaskState.PLANNED.value, now, now),
+            )
+            connection.executemany(
+                "INSERT INTO steps(task_id, step_id, state, max_attempts, updated_at) "
+                "VALUES(?, ?, ?, ?, ?)",
+                [
+                    (task_id, step_id, StepState.PENDING.value, int(limits.get(step_id, 1)), now)
+                    for step_id in step_ids
+                ],
+            )
+
+    def get_task_state(self, task_id: str) -> tuple[TaskState, int]:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT state, revision FROM tasks WHERE task_id=?", (task_id,)
+            ).fetchone()
+        if row is None:
+            raise KeyError(task_id)
+        return TaskState(row["state"]), int(row["revision"])
+
+    def compare_and_set_task_state(
+        self, task_id: str, expected_revision: int, new_state: TaskState
+    ) -> int:
+        with self.transaction() as connection:
+            cursor = connection.execute(
+                "UPDATE tasks SET state=?, revision=revision+1, updated_at=? "
+                "WHERE task_id=? AND revision=?",
+                (new_state.value, self._now(), task_id, expected_revision),
+            )
+            if cursor.rowcount != 1:
+                raise StateConflict(
+                    "TASK_REVISION_CONFLICT",
+                    "task state changed concurrently",
+                    {"task_id": task_id, "expected_revision": expected_revision},
+                )
+            row = connection.execute(
+                "SELECT revision FROM tasks WHERE task_id=?", (task_id,)
+            ).fetchone()
+            return int(row["revision"])
+
+    def transition_step(
+        self,
+        task_id: str,
+        step_id: str,
+        expected: Sequence[StepState],
+        state: StepState,
+        *,
+        worker_id: str | None = None,
+        session_id: str | None = None,
+        lease_expires_at: datetime | None = None,
+        result_digest: str | None = None,
+        last_error: str | None = None,
+        increment_attempt: bool = False,
+        clear_assignment: bool = False,
+    ) -> StoredStep:
+        with self.transaction() as connection:
+            row = connection.execute(
+                "SELECT * FROM steps WHERE task_id=? AND step_id=?", (task_id, step_id)
+            ).fetchone()
+            if row is None:
+                raise KeyError(f"{task_id}:{step_id}")
+            if StepState(row["state"]) not in set(expected):
+                raise StateConflict(
+                    "STEP_STATE_CONFLICT",
+                    "step state changed or transition is invalid",
+                    {"task_id": task_id, "step_id": step_id, "state": row["state"]},
+                )
+            if increment_attempt and int(row["attempt"]) >= int(row["max_attempts"]):
+                raise StateConflict(
+                    "STEP_ATTEMPTS_EXHAUSTED",
+                    "step has no remaining attempts",
+                    {"task_id": task_id, "step_id": step_id},
+                )
+            next_worker = None if clear_assignment else (worker_id or row["worker_id"])
+            next_session = None if clear_assignment else (session_id or row["session_id"])
+            connection.execute(
+                "UPDATE steps SET state=?, attempt=attempt+?, worker_id=?, session_id=?, "
+                "lease_expires_at=?, result_digest=COALESCE(?, result_digest), last_error=?, updated_at=? "
+                "WHERE task_id=? AND step_id=?",
+                (
+                    state.value,
+                    1 if increment_attempt else 0,
+                    next_worker,
+                    next_session,
+                    lease_expires_at.isoformat() if lease_expires_at else None,
+                    result_digest,
+                    last_error,
+                    self._now(),
+                    task_id,
+                    step_id,
+                ),
+            )
+            updated = connection.execute(
+                "SELECT * FROM steps WHERE task_id=? AND step_id=?", (task_id, step_id)
+            ).fetchone()
+        return self._step_from_row(updated)
+
+    def set_step_state(
+        self,
+        task_id: str,
+        step_id: str,
+        state: StepState,
+        **kwargs: Any,
+    ) -> StoredStep:
+        current = self.list_steps(task_id)[step_id]
+        return self.transition_step(task_id, step_id, (current.state,), state, **kwargs)
+
+    def list_steps(self, task_id: str) -> dict[str, StoredStep]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                "SELECT * FROM steps WHERE task_id=? ORDER BY step_id", (task_id,)
+            ).fetchall()
+        return {row["step_id"]: self._step_from_row(row) for row in rows}
+
+    @staticmethod
+    def _step_from_row(row: sqlite3.Row) -> StoredStep:
+        return StoredStep(
+            task_id=row["task_id"],
+            step_id=row["step_id"],
+            state=StepState(row["state"]),
+            attempt=int(row["attempt"]),
+            max_attempts=int(row["max_attempts"]),
+            worker_id=row["worker_id"],
+            session_id=row["session_id"],
+            lease_expires_at=(
+                datetime.fromisoformat(row["lease_expires_at"])
+                if row["lease_expires_at"]
+                else None
+            ),
+            result_digest=row["result_digest"],
+            updated_at=datetime.fromisoformat(row["updated_at"]),
+            last_error=row["last_error"],
+        )
+
+    def consume_nonce(self, scope: str, nonce: str) -> None:
+        try:
+            with self.transaction() as connection:
+                connection.execute(
+                    "INSERT INTO consumed_nonces(scope, nonce, consumed_at) VALUES(?, ?, ?)",
+                    (scope, nonce, self._now()),
+                )
+        except sqlite3.IntegrityError as error:
+            raise StateConflict(
+                "NONCE_REPLAY", "nonce was already consumed", {"scope": scope, "nonce": nonce}
+            ) from error
+
+    def append_event(
+        self,
+        task_id: str,
+        event_type: str,
+        payload: Mapping[str, Any],
+        previous_digest: str | None,
+        event_digest: str,
+        created_at: datetime,
+    ) -> int:
+        with self.transaction() as connection:
+            cursor = connection.execute(
+                "INSERT INTO events(task_id, event_type, payload_json, previous_digest, digest, created_at) "
+                "VALUES(?, ?, ?, ?, ?, ?)",
+                (
+                    task_id,
+                    event_type,
+                    canonical_json(payload),
+                    previous_digest,
+                    event_digest,
+                    created_at.astimezone(timezone.utc).isoformat(),
+                ),
+            )
+            return int(cursor.lastrowid)
+
+    def list_events(self, task_id: str) -> list[Mapping[str, Any]]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                "SELECT * FROM events WHERE task_id=? ORDER BY sequence", (task_id,)
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def put_idempotent_result(
+        self, namespace: str, operation_id: str, result: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        encoded = canonical_json(result)
+        with self.transaction() as connection:
+            existing = connection.execute(
+                "SELECT result_json FROM idempotency WHERE namespace=? AND operation_id=?",
+                (namespace, operation_id),
+            ).fetchone()
+            if existing is not None:
+                return json.loads(existing["result_json"])
+            connection.execute(
+                "INSERT INTO idempotency(namespace, operation_id, result_json, created_at) "
+                "VALUES(?, ?, ?, ?)",
+                (namespace, operation_id, encoded, self._now()),
+            )
+        return dict(result)

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/tag.py

@@ -0,0 +1,44 @@
+"""Read-only annotated and lightweight tag consistency checks."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .errors import ReleaseBlocked
+from .models import is_git_oid
+
+
+@dataclass(frozen=True, slots=True)
+class TagReference:
+    ref: str
+    object_type: str
+    tag_oid: str
+    peeled_target_oid: str
+
+
+class TagVerifier:
+    def verify(
+        self,
+        tag: TagReference,
+        *,
+        expected_tag_oid: str,
+        expected_target_oid: str,
+    ) -> None:
+        if not tag.ref.startswith("refs/tags/") or tag.ref.endswith("^{}"):
+            raise ReleaseBlocked("TAG_REF_INVALID", "canonical tag ref is required")
+        if tag.object_type not in ("commit", "tag"):
+            raise ReleaseBlocked("TAG_OBJECT_TYPE_INVALID", "unsupported tag object type")
+        values = (tag.tag_oid, tag.peeled_target_oid, expected_tag_oid, expected_target_oid)
+        if any(not is_git_oid(value) for value in values):
+            raise ReleaseBlocked("TAG_OID_INVALID", "tag OIDs must be complete lowercase hex")
+        if tag.tag_oid != expected_tag_oid:
+            raise ReleaseBlocked("TAG_OBJECT_MISMATCH", "tag object does not match expectation")
+        if tag.peeled_target_oid != expected_target_oid:
+            raise ReleaseBlocked(
+                "TAG_PEELED_TARGET_MISMATCH", "peeled target does not match expected target"
+            )
+        if tag.object_type == "commit" and tag.tag_oid != tag.peeled_target_oid:
+            raise ReleaseBlocked(
+                "LIGHTWEIGHT_TAG_INCONSISTENT",
+                "lightweight tag object and target must be identical",
+            )

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/worker_service.py

@@ -0,0 +1,69 @@
+"""Dormant Candidate worker supervisor.
+
+Provider clients are injected by the external management plane. This process
+never converts a missing projection into a simulated native identity.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import signal
+import threading
+
+
+def write_status(directory: Path) -> None:
+    directory.mkdir(parents=True, exist_ok=True, mode=0o750)
+    if directory.is_symlink():
+        raise RuntimeError("worker state directory symlink rejected")
+    payload = {
+        "status": "dormant",
+        "task_plane_execution_enabled": os.environ.get(
+            "AIOS_TASK_PLANE_EXECUTION_ENABLED", "false"
+        ).lower()
+        == "true",
+        "gitops_writes_enabled": os.environ.get("AIOS_GITOPS_WRITES_ENABLED", "false").lower()
+        == "true",
+        "production_side_effects": False,
+        "secrets_included": False,
+    }
+    temporary = directory / f"worker-status.{os.getpid()}.new"
+    target = directory / "worker-status.json"
+    if temporary.is_symlink() or target.is_symlink():
+        raise RuntimeError("worker status symlink rejected")
+    descriptor = os.open(
+        temporary,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+        0o640,
+    )
+    try:
+        os.write(descriptor, (json.dumps(payload, sort_keys=True) + "\n").encode("utf-8"))
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    os.replace(temporary, target)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--state-directory", required=True, type=Path)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    if os.environ.get("AIOS_PRODUCTION_DEPLOYMENT_ENABLED", "false").lower() == "true":
+        raise RuntimeError("Candidate worker refuses production deployment mode")
+    write_status(args.state_directory)
+    stopping = threading.Event()
+    signal.signal(signal.SIGTERM, lambda *_: stopping.set())
+    signal.signal(signal.SIGINT, lambda *_: stopping.set())
+    while not stopping.wait(5):
+        write_status(args.state_directory)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-autonomous-release-candidate/aios_autonomous_release_candidate/workflow.py

@@ -0,0 +1,201 @@
+"""Durable fail-closed workflow state machine."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from .errors import StateConflict
+from .models import StepState, TaskPlan, TaskState, is_sha256
+from .store import StateStore
+
+
+@dataclass(frozen=True, slots=True)
+class WorkflowDecision:
+    task_id: str
+    ready_steps: tuple[str, ...]
+    blocked: bool
+    reason: str | None = None
+
+
+class WorkflowEngine:
+    def __init__(self, store: StateStore) -> None:
+        self.store = store
+
+    def _task_state(self, task_id: str, state: TaskState) -> None:
+        current, revision = self.store.get_task_state(task_id)
+        if current != state:
+            self.store.compare_and_set_task_state(task_id, revision, state)
+
+    def plan(self, task: TaskPlan) -> None:
+        self.store.create_task(
+            task.task_id,
+            task.digest(),
+            {
+                "task_id": task.task_id,
+                "intent": task.intent,
+                "branch": task.binding.branch,
+                "base_oid": task.binding.base_oid,
+            },
+            tuple(step.step_id for step in task.steps),
+            {step.step_id: step.max_attempts for step in task.steps},
+        )
+
+    def next_steps(self, task: TaskPlan) -> WorkflowDecision:
+        states = self.store.list_steps(task.task_id)
+        by_id = {step.step_id: step for step in task.steps}
+        if any(item.state == StepState.FAILED for item in states.values()):
+            self._task_state(task.task_id, TaskState.FAILED)
+            return WorkflowDecision(task.task_id, (), True, "terminal_step_failure")
+        ready: list[str] = []
+        for step_id, stored in states.items():
+            if stored.state not in (StepState.PENDING, StepState.READY):
+                continue
+            dependencies = by_id[step_id].depends_on
+            dependency_states = tuple(states[item].state for item in dependencies)
+            if any(item in (StepState.FAILED, StepState.BLOCKED) for item in dependency_states):
+                if stored.state != StepState.BLOCKED:
+                    self.store.transition_step(
+                        task.task_id,
+                        step_id,
+                        (stored.state,),
+                        StepState.BLOCKED,
+                        last_error="dependency_failed",
+                    )
+                continue
+            if all(item == StepState.COMPLETED for item in dependency_states):
+                if stored.state == StepState.PENDING:
+                    self.store.transition_step(
+                        task.task_id, step_id, (StepState.PENDING,), StepState.READY
+                    )
+                ready.append(step_id)
+        if states and all(item.state == StepState.COMPLETED for item in states.values()):
+            self._task_state(task.task_id, TaskState.COMPLETED)
+        elif ready:
+            self._task_state(task.task_id, TaskState.READY)
+        return WorkflowDecision(task.task_id, tuple(sorted(ready)), False)
+
+    def claim(self, task_id: str, step_id: str, worker_id: str, session_id: str) -> None:
+        if not worker_id or not session_id:
+            raise StateConflict("STEP_CLAIM_BINDING_REQUIRED", "worker and session are required")
+        self.store.transition_step(
+            task_id,
+            step_id,
+            (StepState.READY,),
+            StepState.CLAIMED,
+            worker_id=worker_id,
+            session_id=session_id,
+            increment_attempt=True,
+        )
+
+    def start(
+        self,
+        task_id: str,
+        step_id: str,
+        worker_id: str,
+        session_id: str,
+        *,
+        lease_seconds: int = 60,
+        now: datetime | None = None,
+    ) -> None:
+        current = self.store.list_steps(task_id)[step_id]
+        if current.worker_id != worker_id or current.session_id != session_id:
+            raise StateConflict("STEP_CLAIM_OWNER_MISMATCH", "claim owner does not match")
+        now = now or datetime.now(timezone.utc)
+        self.store.transition_step(
+            task_id,
+            step_id,
+            (StepState.CLAIMED,),
+            StepState.RUNNING,
+            worker_id=worker_id,
+            session_id=session_id,
+            lease_expires_at=now + timedelta(seconds=lease_seconds),
+        )
+        self._task_state(task_id, TaskState.RUNNING)
+
+    def heartbeat(
+        self,
+        task_id: str,
+        step_id: str,
+        worker_id: str,
+        session_id: str,
+        *,
+        lease_seconds: int = 60,
+        now: datetime | None = None,
+    ) -> None:
+        current = self.store.list_steps(task_id)[step_id]
+        if current.worker_id != worker_id or current.session_id != session_id:
+            raise StateConflict("STEP_CLAIM_OWNER_MISMATCH", "heartbeat owner does not match")
+        now = now or datetime.now(timezone.utc)
+        self.store.transition_step(
+            task_id,
+            step_id,
+            (StepState.RUNNING,),
+            StepState.RUNNING,
+            lease_expires_at=now + timedelta(seconds=lease_seconds),
+        )
+
+    def complete(self, task_id: str, step_id: str, *, result_digest: str) -> None:
+        if not is_sha256(result_digest):
+            raise StateConflict("RESULT_DIGEST_INVALID", "result digest must be SHA-256")
+        current = self.store.list_steps(task_id)[step_id]
+        if current.state == StepState.COMPLETED and current.result_digest == result_digest:
+            return
+        self.store.transition_step(
+            task_id,
+            step_id,
+            (StepState.RUNNING,),
+            StepState.COMPLETED,
+            result_digest=result_digest,
+        )
+
+    def fail(
+        self,
+        task: TaskPlan,
+        step_id: str,
+        reason: str,
+        *,
+        retryable: bool,
+    ) -> StepState:
+        current = self.store.list_steps(task.task_id)[step_id]
+        next_state = (
+            StepState.PENDING
+            if retryable and current.attempt < current.max_attempts
+            else StepState.FAILED
+        )
+        self.store.transition_step(
+            task.task_id,
+            step_id,
+            (StepState.CLAIMED, StepState.RUNNING),
+            next_state,
+            last_error=reason,
+            clear_assignment=next_state == StepState.PENDING,
+        )
+        self._task_state(
+            task.task_id, TaskState.BLOCKED if next_state == StepState.PENDING else TaskState.FAILED
+        )
+        return next_state
+
+    def recover_expired(
+        self, task: TaskPlan, *, now: datetime | None = None
+    ) -> tuple[str, ...]:
+        now = now or datetime.now(timezone.utc)
+        recovered: list[str] = []
+        for step_id, current in self.store.list_steps(task.task_id).items():
+            if (
+                current.state == StepState.RUNNING
+                and current.lease_expires_at is not None
+                and current.lease_expires_at <= now
+            ):
+                self.fail(task, step_id, "lease_expired", retryable=True)
+                recovered.append(step_id)
+        return tuple(recovered)
+
+    def recoverable(self, task_id: str, step_id: str) -> None:
+        self.store.transition_step(
+            task_id,
+            step_id,
+            (StepState.BLOCKED,),
+            StepState.PENDING,
+            clear_assignment=True,
+        )

services/aios-autonomous-release-candidate/pyproject.toml

@@ -0,0 +1,25 @@
+[build-system]
+requires = ["setuptools>=69"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "aios-autonomous-release-candidate"
+version = "0.1.0"
+description = "Candidate-only governed autonomous release control plane"
+readme = "README.md"
+requires-python = ">=3.11"
+license = { text = "Proprietary" }
+authors = [{ name = "AIOS" }]
+dependencies = []
+
+[project.scripts]
+aios-autonomous-release = "aios_autonomous_release_candidate.cli:main"
+aios-autonomous-release-service = "aios_autonomous_release_candidate.service:main"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["aios_autonomous_release_candidate*"]
+
+[tool.pytest.ini_options]
+addopts = "-ra --strict-config --strict-markers"
+testpaths = ["../../tests/autonomous_release"]

tests/autonomous-release/test_core_contracts.py

@@ -0,0 +1,48 @@
+from datetime import datetime, timezone, timedelta
+
+from aios_autonomous_release_candidate.models import (
+    ResourceBinding,
+    RiskClass,
+    TaskPlan,
+    WorkflowStep,
+)
+from aios_autonomous_release_candidate.policy import ReleasePolicy
+from aios_autonomous_release_candidate.evidence import EvidenceChain
+
+
+OID = "a" * 40
+
+
+def build_plan():
+    return TaskPlan(
+        task_id="test-task",
+        intent="candidate validation",
+        binding=ResourceBinding(
+            repository="aios/aios",
+            branch="patch/test",
+            base_oid=OID,
+            allowed_paths=("services/test/",),
+        ),
+        risk=RiskClass.LOW,
+        steps=(
+            WorkflowStep(
+                step_id="validate",
+                worker_role="tester",
+                required_capabilities=("test.execute",),
+                allowed_paths=("services/test/",),
+            ),
+        ),
+        evidence_requirements=("tests",),
+    )
+
+
+def test_plan_policy_accepts_candidate_plan():
+    ReleasePolicy().validate_task(build_plan())
+
+
+def test_evidence_chain_detects_valid_closure():
+    chain = EvidenceChain()
+    chain.append("START", "test-task", {"ok": True, "secrets_included": False})
+    chain.append("DONE", "test-task", {"ok": True, "secrets_included": False})
+    assert chain.verify()
+    assert len(chain.closure_digest()) == 64

tests/autonomous_release/test_artifact_closure.py

@@ -0,0 +1,41 @@
+from pathlib import Path
+
+import pytest
+
+from aios_autonomous_release_candidate.artifact import ArtifactBuilder, ArtifactFile, BuiltArtifact
+from aios_autonomous_release_candidate.errors import EvidenceIncomplete, ReleaseBlocked
+
+
+def test_artifact_is_deterministic_and_reverified() -> None:
+    builder = ArtifactBuilder()
+    files = (
+        ArtifactFile("bin/run", b"#!/bin/sh\nexit 0\n", 0o755),
+        ArtifactFile("config/candidate.json", b'{"production":false}\n'),
+    )
+    first = builder.build("a" * 40, "b" * 40, files)
+    second = builder.build("a" * 40, "b" * 40, tuple(reversed(files)))
+    assert first.content == second.content
+    assert first.manifest.manifest_sha256 == second.manifest.manifest_sha256
+    assert first.manifest.secrets_included is False
+    builder.verify(first)
+    with pytest.raises(EvidenceIncomplete, match="ARTIFACT_DIGEST_MISMATCH"):
+        builder.verify(BuiltArtifact(first.content + b"tampered", first.manifest))
+
+
+def test_source_tree_rejects_symlink_and_hardlink(tmp_path: Path) -> None:
+    root = tmp_path / "source"
+    root.mkdir()
+    target = root / "file.txt"
+    target.write_text("safe", encoding="utf-8")
+    (root / "link.txt").symlink_to(target)
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_SOURCE_LINK_REJECTED"):
+        ArtifactBuilder().build_from_directory("a" * 40, "b" * 40, root)
+
+
+def test_artifact_rejects_secret_markers() -> None:
+    with pytest.raises(ReleaseBlocked, match="ARTIFACT_SECRET_MARKER"):
+        ArtifactBuilder().build(
+            "a" * 40,
+            "b" * 40,
+            (ArtifactFile("unsafe.env", b"password" + b"=" + b"x" * 16),),
+        )

tests/autonomous_release/test_cli_workflows.py

@@ -0,0 +1,66 @@
+import json
+from pathlib import Path
+
+from aios_autonomous_release_candidate.cli import main
+
+
+def test_cli_artifact_build_and_verify(tmp_path: Path, capsys) -> None:
+    source = tmp_path / "source"
+    source.mkdir()
+    (source / "payload.txt").write_text("candidate payload\n", encoding="utf-8")
+    output = tmp_path / "output"
+    assert main(
+        [
+            "artifact",
+            "build",
+            "--source",
+            str(source),
+            "--source-oid",
+            "a" * 40,
+            "--tree-oid",
+            "b" * 40,
+            "--output-directory",
+            str(output),
+        ]
+    ) == 0
+    built = json.loads(capsys.readouterr().out)
+    assert built["secrets_included"] is False
+    assert main(
+        [
+            "artifact",
+            "verify",
+            "--artifact",
+            built["artifact"],
+            "--manifest",
+            built["manifest"],
+        ]
+    ) == 0
+    assert json.loads(capsys.readouterr().out)["verified"] is True
+
+
+def test_cli_release_verify_stops_at_owner(tmp_path: Path, capsys) -> None:
+    payload = {
+        "task_id": "cli-release-task",
+        "repository": "aios/aios",
+        "branch": "integration/cli-release",
+        "version": "SHADOW-0.1.0",
+        "target_oid": "a" * 40,
+        "tree_oid": "b" * 40,
+        "artifact_sha256": "c" * 64,
+        "manifest_sha256": "d" * 64,
+        "package_closure_sha256": "e" * 64,
+        "evidence_closure_oid": "f" * 40,
+        "approval_plan_id": "external-owner-release",
+        "rollback_approval_plan_id": "external-owner-rollback",
+        "rollback_artifact_sha256": "1" * 64,
+        "expected_tag_oid": "2" * 40,
+        "expected_target_oid": "a" * 40,
+        "production_config_sha256": "3" * 64,
+    }
+    path = tmp_path / "release-plan.json"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    assert main(["release", "verify", "--payload", str(path)]) == 0
+    result = json.loads(capsys.readouterr().out)
+    assert result["release_state"] == "WAITING_OWNER"
+    assert result["production_release"] == "BLOCKED_EXTERNAL_OWNER_SIGNATURE"
+    assert result["production_side_effects"] is False

tests/autonomous_release/test_core_contracts.py

@@ -0,0 +1,48 @@
+from datetime import datetime, timezone, timedelta
+
+from aios_autonomous_release_candidate.models import (
+    ResourceBinding,
+    RiskClass,
+    TaskPlan,
+    WorkflowStep,
+)
+from aios_autonomous_release_candidate.policy import ReleasePolicy
+from aios_autonomous_release_candidate.evidence import EvidenceChain
+
+
+OID = "a" * 40
+
+
+def build_plan():
+    return TaskPlan(
+        task_id="test-task",
+        intent="candidate validation",
+        binding=ResourceBinding(
+            repository="aios/aios",
+            branch="patch/test",
+            base_oid=OID,
+            allowed_paths=("services/test/",),
+        ),
+        risk=RiskClass.LOW,
+        steps=(
+            WorkflowStep(
+                step_id="validate",
+                worker_role="tester",
+                required_capabilities=("test.execute",),
+                allowed_paths=("services/test/",),
+            ),
+        ),
+        evidence_requirements=("tests",),
+    )
+
+
+def test_plan_policy_accepts_candidate_plan():
+    ReleasePolicy().validate_task(build_plan())
+
+
+def test_evidence_chain_detects_valid_closure():
+    chain = EvidenceChain()
+    chain.append("START", "test-task", {"ok": True, "secrets_included": False})
+    chain.append("DONE", "test-task", {"ok": True, "secrets_included": False})
+    assert chain.verify()
+    assert len(chain.closure_digest()) == 64

tests/autonomous_release/test_credentials_and_evidence.py

@@ -0,0 +1,129 @@
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from aios_autonomous_release_candidate.credential import CredentialBroker
+from aios_autonomous_release_candidate.errors import CredentialUnavailable, EvidenceIncomplete
+from aios_autonomous_release_candidate.evidence import EvidenceChain
+from aios_autonomous_release_candidate.models import CredentialRequest, ResourceBinding
+from aios_autonomous_release_candidate.store import StateStore
+
+
+NOW = datetime(2026, 7, 26, tzinfo=timezone.utc)
+
+
+class Provider:
+    def __init__(self, reference: str = "systemd-credential:openclaw-gateway-candidate") -> None:
+        self.reference = reference
+        self.active: set[str] = set()
+        self.revoked: list[str] = []
+        self.count = 0
+
+    def issue(self, request, expires_at):
+        self.count += 1
+        credential_id = f"credential-{self.count}"
+        self.active.add(credential_id)
+        return credential_id, self.reference
+
+    def revoke(self, credential_id, reference):
+        self.active.discard(credential_id)
+        self.revoked.append(credential_id)
+
+    def is_usable(self, credential_id, reference):
+        return credential_id in self.active
+
+
+def request(request_id: str = "request-1") -> CredentialRequest:
+    return CredentialRequest(
+        request_id=request_id,
+        task_id="task-credential",
+        session_id="session-1",
+        worker_id="worker-1",
+        audience="openclaw-candidate",
+        consumer_service="aios-autonomous-release-worker-candidate.service",
+        binding=ResourceBinding("aios/aios", "patch/credential-test", "a" * 40, ("services/test",)),
+        operations=("native_session.spawn",),
+        ttl_seconds=300,
+        max_uses=1,
+        provider="openclaw",
+        runtime_user="aiman",
+        approval_plan_id="owner-stage-b-native-sessions-20260726",
+        not_before=NOW,
+        expires_at=NOW + timedelta(minutes=5),
+        nonce="nonce-credential-1",
+    )
+
+
+def test_credential_is_scope_bound_revocable_and_idempotent(tmp_path: Path) -> None:
+    provider = Provider()
+    broker = CredentialBroker(StateStore(tmp_path / "state.sqlite3"), provider)
+    projection = broker.issue(request(), NOW)
+    broker.project(projection)
+    broker.authorize(
+        projection,
+        task_id="task-credential",
+        session_id="session-1",
+        worker_id="worker-1",
+        audience="openclaw-candidate",
+        consumer_service="aios-autonomous-release-worker-candidate.service",
+        operation="native_session.spawn",
+        path="services/test/file.py",
+        now=NOW,
+    )
+    with pytest.raises(CredentialUnavailable, match="CREDENTIAL_SCOPE_MISMATCH"):
+        broker.authorize(
+            projection,
+            task_id="other-task",
+            session_id="session-1",
+            worker_id="worker-1",
+            audience="openclaw-candidate",
+            consumer_service="aios-autonomous-release-worker-candidate.service",
+            operation="native_session.spawn",
+            path="services/test/file.py",
+            now=NOW,
+        )
+    first = broker.revoke(projection.credential_id, projection.reference, NOW)
+    second = broker.revoke(projection.credential_id, projection.reference, NOW)
+    assert first == second
+    broker.verify_revoked(projection.credential_id, projection.reference)
+
+
+def test_persistence_failure_compensates_upstream_issuance(tmp_path: Path) -> None:
+    provider = Provider()
+    broker = CredentialBroker(StateStore(tmp_path / "state.sqlite3"), provider)
+    broker.issue(request(), NOW)
+    with pytest.raises(CredentialUnavailable, match="CREDENTIAL_PERSISTENCE_FAILED"):
+        broker.issue(request("request-2"), NOW)
+    assert "credential-2" in provider.revoked
+
+
+def test_plaintext_or_untrusted_projection_is_rejected(tmp_path: Path) -> None:
+    provider = Provider("/run/credentials/candidate/raw-value")
+    broker = CredentialBroker(StateStore(tmp_path / "state.sqlite3"), provider)
+    with pytest.raises(CredentialUnavailable, match="CREDENTIAL_REFERENCE_REJECTED"):
+        broker.issue(request(), NOW)
+    assert provider.revoked == ["credential-1"]
+
+
+def test_evidence_chain_survives_restart_and_detects_tampering(tmp_path: Path) -> None:
+    store = StateStore(tmp_path / "state.sqlite3")
+    chain = EvidenceChain(store, "task-evidence")
+    chain.append("TASK_STARTED", "task-evidence", {"secrets_included": False})
+    chain.append(
+        "TESTS_PASSED",
+        "task-evidence",
+        {"passed": 42, "skipped": 0, "secrets_included": False},
+    )
+    reopened = EvidenceChain(store, "task-evidence")
+    assert reopened.verify()
+    assert len(reopened.close(("TASK_STARTED", "TESTS_PASSED"))) == 64
+    with pytest.raises(EvidenceIncomplete, match="EVIDENCE_REQUIREMENT_MISSING"):
+        reopened.close(("NATIVE_SESSION_VERIFIED",))
+
+    with store.transaction() as connection:
+        connection.execute(
+            "UPDATE events SET payload_json=? WHERE task_id=? AND sequence=1",
+            ('{"secrets_included":true}', "task-evidence"),
+        )
+    assert EvidenceChain(store, "task-evidence").verify() is False

tests/autonomous_release/test_external_gates.py

@@ -0,0 +1,125 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from aios_autonomous_release_candidate.approval import (
+    ApprovalVerifier,
+    StaticRevocationSource,
+    UnavailableSignatureVerifier,
+)
+from aios_autonomous_release_candidate.errors import ApprovalRequired
+from aios_autonomous_release_candidate.gitea_canary import (
+    EphemeralKeyProjection,
+    GiteaDeployKeyCanary,
+)
+from aios_autonomous_release_candidate.models import (
+    ApprovalEnvelope,
+    ResourceBinding,
+    RiskClass,
+    TaskPlan,
+    WorkflowStep,
+)
+from aios_autonomous_release_candidate.store import StateStore
+
+
+NOW = datetime(2026, 7, 26, tzinfo=timezone.utc)
+OID = "a" * 40
+
+
+def task_plan() -> TaskPlan:
+    return TaskPlan(
+        "external-gate-task",
+        "external gate verification",
+        ResourceBinding("aios/aios", "patch/external-gate", OID, ("services/test",)),
+        RiskClass.HIGH,
+        (WorkflowStep("canary", "tester", ("native_session.spawn",)),),
+        ("external-gates",),
+    )
+
+
+def test_unknown_owner_verifier_fails_with_stable_blocker(tmp_path) -> None:
+    plan = task_plan()
+    envelope = ApprovalEnvelope(
+        approval_plan_id="owner-external-gate",
+        task_id=plan.task_id,
+        task_plan_sha256=plan.digest(),
+        binding=plan.binding,
+        capabilities=("native_session.spawn",),
+        risk=RiskClass.HIGH,
+        not_before=NOW - timedelta(minutes=1),
+        expires_at=NOW + timedelta(minutes=5),
+        nonce="nonce-owner-external",
+        signing_key_id="owner-kms-1",
+        signature="opaque-signature",
+    )
+    verifier = ApprovalVerifier(
+        StateStore(tmp_path / "state.sqlite3"),
+        UnavailableSignatureVerifier(),
+        StaticRevocationSource(),
+    )
+    with pytest.raises(
+        ApprovalRequired, match="BLOCKED_OWNER_APPROVAL_VERIFIER_UNAVAILABLE"
+    ):
+        verifier.verify(plan, envelope, now=NOW)
+
+
+class KeyProjection:
+    def __init__(self) -> None:
+        self.present = False
+
+    def create(self, task_id, branch):
+        self.present = True
+        return EphemeralKeyProjection(
+            public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICandidateCanaryOnly",
+            private_key_reference="systemd-credential:gitea-deploy-key-canary",
+        )
+
+    def delete(self, reference):
+        self.present = False
+
+    def is_present(self, reference):
+        return self.present
+
+
+class Gitea:
+    simulation = True
+
+    def __init__(self) -> None:
+        self.keys = {}
+
+    def list_deploy_keys(self, repository, credential_reference):
+        return list(self.keys.values())
+
+    def create_deploy_key(self, repository, title, public_key, credential_reference):
+        item = {"id": 1, "title": title, "key": public_key, "read_only": False}
+        self.keys[1] = item
+        return item
+
+    def delete_deploy_key(self, repository, key_id, credential_reference):
+        self.keys.pop(key_id, None)
+
+
+class SSHProbe:
+    def __init__(self, gitea, projection) -> None:
+        self.gitea = gitea
+        self.projection = projection
+
+    def authenticate(self, repository, branch, private_key_reference):
+        return bool(self.gitea.keys) and self.projection.present
+
+
+def test_gitea_canary_always_revokes_key_and_projection() -> None:
+    projection = KeyProjection()
+    gitea = Gitea()
+    result = GiteaDeployKeyCanary(gitea, projection, SSHProbe(gitea, projection)).run(
+        task_id="external-gate-task",
+        branch="patch/external-gate",
+        credential_reference="systemd-credential:gitea-admin-api-candidate",
+    )
+    assert result.key_created is True
+    assert result.key_deleted is True
+    assert result.post_revoke_auth_rejected is True
+    assert result.credential_projection_deleted is True
+    assert result.gitea_admin_api_lifecycle_canary is False  # simulation is not real evidence
+    assert gitea.keys == {}
+    assert projection.present is False

tests/autonomous_release/test_http_service.py

@@ -0,0 +1,156 @@
+from contextlib import contextmanager
+import hashlib
+import json
+from pathlib import Path
+from threading import Thread
+import urllib.error
+import urllib.request
+
+from aios_autonomous_release_candidate.service import CandidateApplication, CandidateHTTPServer
+
+
+@contextmanager
+def running_server(tmp_path: Path):
+    application = CandidateApplication(tmp_path / "state.sqlite3")
+    server = CandidateHTTPServer(("127.0.0.1", 0), application)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield application, f"http://127.0.0.1:{server.server_port}"
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def get_json(url: str):
+    with urllib.request.urlopen(url, timeout=5) as response:
+        return response.status, json.load(response)
+
+
+def post_json(url: str, payload: dict):
+    request = urllib.request.Request(
+        url,
+        data=json.dumps(payload).encode("utf-8"),
+        method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(request, timeout=5) as response:
+        return response.status, json.load(response)
+
+
+def test_health_and_readiness_are_loopback_safe(tmp_path: Path) -> None:
+    with running_server(tmp_path) as (_, base):
+        status, health = get_json(f"{base}/health")
+        assert status == 200
+        assert health["production_side_effects"] is False
+        status, ready = get_json(f"{base}/ready")
+        assert status == 200
+        assert ready == {"status": "ready", "secrets_included": False}
+
+
+def test_json_content_type_is_required(tmp_path: Path) -> None:
+    with running_server(tmp_path) as (_, base):
+        request = urllib.request.Request(f"{base}/v1/tasks", data=b"{}", method="POST")
+        try:
+            urllib.request.urlopen(request, timeout=5)
+        except urllib.error.HTTPError as error:
+            assert error.code == 415
+            assert json.load(error)["error"] == "content_type_required"
+
+
+def test_internal_exception_text_is_not_disclosed(tmp_path: Path) -> None:
+    with running_server(tmp_path) as (application, base):
+        def fail(_payload):
+            raise RuntimeError("sensitive internal diagnostic")
+
+        application.api.create_task = fail
+        request = urllib.request.Request(
+            f"{base}/v1/tasks",
+            data=b"{}",
+            method="POST",
+            headers={"Content-Type": "application/json"},
+        )
+        try:
+            urllib.request.urlopen(request, timeout=5)
+        except urllib.error.HTTPError as error:
+            assert error.code == 500
+            body = error.read().decode("utf-8")
+            assert "candidate_request_failed" in body
+            assert "sensitive internal diagnostic" not in body
+
+
+def test_task_lifecycle_evidence_and_release_gate_endpoints(tmp_path: Path) -> None:
+    task_id = "api-lifecycle-task"
+    task = {
+        "task_id": task_id,
+        "intent": "exercise candidate API",
+        "binding": {
+            "repository": "aios/aios",
+            "branch": "patch/api-lifecycle",
+            "base_oid": "a" * 40,
+            "allowed_paths": ["services/test"],
+        },
+        "risk": "medium",
+        "steps": [
+            {
+                "step_id": "test-step",
+                "worker_role": "tester",
+                "required_capabilities": ["test.execute"],
+            }
+        ],
+        "evidence_requirements": ["STEP_RESULT"],
+        "production_side_effects_allowed": False,
+    }
+    with running_server(tmp_path) as (_, base):
+        status, created = post_json(f"{base}/v1/tasks", task)
+        assert status == 201
+        assert created["production_side_effects"] is False
+        status, steps = get_json(f"{base}/v1/tasks/{task_id}/steps")
+        assert status == 200
+        assert steps["steps"]["test-step"]["state"] == "READY"
+        binding = {
+            "step_id": "test-step",
+            "worker_id": "worker-1",
+            "session_id": "session-1",
+        }
+        _, claimed = post_json(f"{base}/v1/tasks/{task_id}/claim", binding)
+        assert claimed["state"] == "RUNNING"
+        _, heartbeat = post_json(f"{base}/v1/tasks/{task_id}/heartbeat", binding)
+        assert heartbeat["state"] == "RUNNING"
+        _, result = post_json(
+            f"{base}/v1/tasks/{task_id}/result",
+            {"step_id": "test-step", "result_digest": hashlib.sha256(b"ok").hexdigest()},
+        )
+        assert result["state"] == "COMPLETED"
+        _, evidence = get_json(f"{base}/v1/tasks/{task_id}/evidence")
+        assert evidence["verified"] is True
+        assert any(item["event_type"] == "STEP_RESULT" for item in evidence["events"])
+        _, closure = post_json(f"{base}/v1/tasks/{task_id}/evidence/close", {})
+        assert len(closure["evidence_closure"]) == 64
+
+        release_plan = {
+            "task_id": task_id,
+            "repository": "aios/aios",
+            "branch": "integration/api-lifecycle",
+            "version": "SHADOW-0.1.0",
+            "target_oid": "b" * 40,
+            "tree_oid": "c" * 40,
+            "artifact_sha256": "d" * 64,
+            "manifest_sha256": "e" * 64,
+            "package_closure_sha256": "f" * 64,
+            "evidence_closure_oid": "1" * 40,
+            "approval_plan_id": "external-owner-release",
+            "rollback_approval_plan_id": "external-owner-rollback",
+            "rollback_artifact_sha256": "2" * 64,
+            "expected_tag_oid": "3" * 40,
+            "expected_target_oid": "b" * 40,
+            "production_config_sha256": "4" * 64,
+        }
+        _, release = post_json(f"{base}/v1/release/plan", release_plan)
+        assert release["release_state"] == "BUILT"
+        _, verified = post_json(
+            f"{base}/v1/release/verify", {"artifact_ok": True, "evidence_ok": True}
+        )
+        assert verified["release_state"] == "WAITING_OWNER"
+        assert verified["production_release"] == "BLOCKED_EXTERNAL_OWNER_SIGNATURE"

tests/autonomous_release/test_native_adapter_contracts.py

@@ -0,0 +1,94 @@
+from typing import Any, Mapping
+
+import pytest
+
+from aios_autonomous_release_candidate.errors import StateConflict
+from aios_autonomous_release_candidate.native_adapters import (
+    CodexNativeAdapter,
+    OpenClawNativeAdapter,
+)
+
+
+class Provider:
+    simulation = True
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.state = "ACTIVE"
+        self.parent = "parent-1"
+
+    def spawn(self, request: Mapping[str, Any], credential_reference: str):
+        return {
+            "provider_session_id": f"{self.name}-native-1",
+            "task_id": request["task_id"],
+            "parent_session_id": request["parent_session_id"],
+            "candidate_only": True,
+            "production_session_access": False,
+        }
+
+    def status(self, provider_session_id: str, credential_reference: str):
+        return {"provider_session_id": provider_session_id, "state": self.state}
+
+    def list_tree(self, parent_session_id: str, credential_reference: str):
+        return [{"provider_session_id": f"{self.name}-native-1", "parent_session_id": parent_session_id}]
+
+    def send_text(self, provider_session_id, payload, credential_reference):
+        return {
+            "content_type": "text/plain",
+            "encoding": "utf-8",
+            "image_only": False,
+            "task_id": payload["task_id"],
+            "message_id": "message-1",
+            "receipt": "receipt-1",
+            "recipient_binding": payload["recipient_binding"],
+            "secrets_included": False,
+        }
+
+    def interrupt(self, provider_session_id: str, credential_reference: str):
+        self.state = "INTERRUPTED"
+
+    def close(self, provider_session_id: str, credential_reference: str):
+        self.state = "CLOSED"
+
+
+@pytest.mark.parametrize(
+    ("adapter_type", "provider_name", "reference", "flag"),
+    (
+        (
+            OpenClawNativeAdapter,
+            "openclaw",
+            "systemd-credential:openclaw-gateway-candidate",
+            "openclaw_native_child_session_verified",
+        ),
+        (
+            CodexNativeAdapter,
+            "codex",
+            "systemd-credential:codex-app-server-candidate",
+            "codex_child_session_verified",
+        ),
+    ),
+)
+def test_native_adapter_preserves_identity_and_text_contract(
+    adapter_type, provider_name, reference, flag
+) -> None:
+    adapter = adapter_type(Provider(provider_name), reference)
+    session = adapter.spawn("task-native", "parent-1", {"recipient_id": "weixin-owner"})
+    assert session.broker_session_id != session.provider_session_id
+    verification = adapter.verify(session)
+    assert verification.broker_session_verified is True
+    assert getattr(verification, flag) is False  # simulated providers are never real canary evidence
+    result = adapter.send_text(session, "hello")
+    assert result.content_type == "text/plain"
+    assert result.secrets_included is False
+    adapter.interrupt(session)
+    assert adapter.status(session) == "INTERRUPTED"
+    adapter.close(session)
+    assert adapter.status(session) == "CLOSED"
+
+
+def test_recipient_binding_must_be_single_and_trusted() -> None:
+    adapter = OpenClawNativeAdapter(
+        Provider("openclaw"), "systemd-credential:openclaw-gateway-candidate"
+    )
+    with pytest.raises(StateConflict, match="RECIPIENT_BINDING_REQUIRED"):
+        adapter.spawn("task-native", "parent-1", {"recipients": ["one", "two"]})

tests/autonomous_release/test_release_gates.py

@@ -0,0 +1,95 @@
+from dataclasses import replace
+from datetime import datetime, timezone
+
+import pytest
+
+from aios_autonomous_release_candidate.approval import ApprovalVerificationResult
+from aios_autonomous_release_candidate.errors import ReleaseBlocked
+from aios_autonomous_release_candidate.models import ReleasePlan
+from aios_autonomous_release_candidate.release import ReleaseController, ReleaseState
+from aios_autonomous_release_candidate.release_gate import ReleaseEvidence, ReleaseGate
+from aios_autonomous_release_candidate.tag import TagReference, TagVerifier
+
+
+OID = "a" * 40
+TAG_OID = "b" * 40
+DIGEST = "c" * 64
+
+
+def plan() -> ReleasePlan:
+    return ReleasePlan(
+        task_id="release-task",
+        repository="aios/aios",
+        branch="integration/autonomous-release",
+        version="V2R1C4",
+        target_oid=OID,
+        tree_oid="d" * 40,
+        artifact_sha256=DIGEST,
+        manifest_sha256="e" * 64,
+        package_closure_sha256="3" * 64,
+        evidence_closure_oid="f" * 40,
+        approval_plan_id="owner-release-v2r1c4",
+        rollback_approval_plan_id="owner-rollback-v2r1c4",
+        rollback_artifact_sha256="1" * 64,
+        expected_tag_oid=TAG_OID,
+        expected_target_oid=OID,
+        production_config_sha256="2" * 64,
+    )
+
+
+def approval(plan_id: str) -> ApprovalVerificationResult:
+    return ApprovalVerificationResult(
+        approval_plan_id=plan_id,
+        task_id="release-task",
+        signing_key_id="owner-kms-1",
+        verified_at=datetime(2026, 7, 26, tzinfo=timezone.utc),
+    )
+
+
+def test_deploy_and_rollback_require_distinct_verified_approvals() -> None:
+    controller = ReleaseController()
+    controller.attach_plan(plan())
+    controller.mark_built()
+    controller.verify(artifact_ok=True, evidence_ok=True)
+    controller.request_owner_approval()
+    with pytest.raises(ReleaseBlocked, match="OWNER_APPROVAL_PLAN_MISMATCH"):
+        controller.approve(approval("wrong-plan"))
+    controller.approve(approval("owner-release-v2r1c4"))
+    controller.start_deployment()
+    controller.mark_deployed()
+    controller.request_rollback("candidate acceptance regression")
+    with pytest.raises(ReleaseBlocked, match="ROLLBACK_APPROVAL_PLAN_MISMATCH"):
+        controller.approve_rollback(approval("owner-release-v2r1c4"))
+    controller.approve_rollback(approval("owner-rollback-v2r1c4"))
+    controller.start_rollback()
+    assert controller.mark_rolled_back().state == ReleaseState.ROLLED_BACK
+
+
+def test_annotated_tag_uses_object_and_peeled_oids_separately() -> None:
+    tag = TagReference("refs/tags/V2R1C4", "tag", TAG_OID, OID)
+    TagVerifier().verify(tag, expected_tag_oid=TAG_OID, expected_target_oid=OID)
+    with pytest.raises(ReleaseBlocked, match="TAG_PEELED_TARGET_MISMATCH"):
+        TagVerifier().verify(tag, expected_tag_oid=TAG_OID, expected_target_oid="3" * 40)
+
+
+def test_release_gate_requires_four_source_tag_and_artifact_closure() -> None:
+    evidence = ReleaseEvidence(
+        expected_version="V2R1C4",
+        registry_version="V2R1C4",
+        baseline_version="V2R1C4",
+        local_tag_oid=TAG_OID,
+        remote_tag_oid=TAG_OID,
+        gitea_tag_oid=TAG_OID,
+        expected_tag_oid=TAG_OID,
+        local_peeled_target_oid=OID,
+        remote_peeled_target_oid=OID,
+        gitea_peeled_target_oid=OID,
+        expected_target_oid=OID,
+        artifact_sha256=DIGEST,
+        manifest_sha256="e" * 64,
+        evidence_sha256="4" * 64,
+        rollback_artifact_sha256="1" * 64,
+    )
+    ReleaseGate().verify(evidence)
+    with pytest.raises(ReleaseBlocked, match="TAG_GITEA_MISMATCH"):
+        ReleaseGate().verify(replace(evidence, gitea_tag_oid="5" * 40))

tests/autonomous_release/test_secret_manager_and_execution.py

@@ -0,0 +1,116 @@
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from aios_autonomous_release_candidate.errors import CredentialUnavailable
+from aios_autonomous_release_candidate.evidence import EvidenceChain
+from aios_autonomous_release_candidate.execution_manager import ExecutionManager
+from aios_autonomous_release_candidate.models import CredentialRequest, ResourceBinding
+from aios_autonomous_release_candidate.native_adapters import OpenClawNativeAdapter
+from aios_autonomous_release_candidate.secret_manager import SecretManagerV2
+from aios_autonomous_release_candidate.store import StateStore
+from test_native_adapter_contracts import Provider
+
+
+NOW = datetime(2026, 7, 26, tzinfo=timezone.utc)
+
+
+class Transport:
+    def __init__(self) -> None:
+        self.calls = []
+
+    def exchange(self, operation, payload):
+        self.calls.append((operation, payload))
+        base = {
+            "task_id": payload["task_id"],
+            "credential_id": payload.get("credential_id", "credential-openclaw-1"),
+            "plaintext_secret_returned": False,
+        }
+        if operation == "issue":
+            return {
+                **base,
+                "credential_reference": "systemd-credential:openclaw-gateway-candidate",
+                "state": "ISSUED",
+                "expires_at": payload["expires_at"],
+            }
+        if operation == "audit":
+            return {**base, "events": [{"operation": "issue", "secrets_included": False}]}
+        if operation in {"project", "status"}:
+            return {**base, "state": "PROJECTED"}
+        if operation == "verify_revocation":
+            return {**base, "state": "REVOKED", "post_revocation_auth": "REJECTED"}
+        return {**base, "state": "REVOKED"}
+
+
+def credential_request() -> CredentialRequest:
+    return CredentialRequest(
+        request_id="request-1",
+        task_id="task-execution",
+        session_id="session-1",
+        worker_id="worker-1",
+        audience="openclaw-candidate",
+        consumer_service="aios-autonomous-release-worker-candidate.service",
+        binding=ResourceBinding("aios/aios", "patch/execution", "a" * 40, ("services/test",)),
+        operations=("native_session.spawn",),
+        provider="openclaw",
+        runtime_user="aiman",
+        approval_plan_id="owner-stage-b-native-sessions-20260726",
+        not_before=NOW,
+        expires_at=NOW + timedelta(minutes=5),
+        nonce="nonce-execution-1",
+        ttl_seconds=300,
+        max_uses=1,
+    )
+
+
+def test_secret_manager_uses_reference_only_protocol() -> None:
+    transport = Transport()
+    manager = SecretManagerV2(transport)
+    receipt = manager.issue(credential_request(), NOW)
+    assert receipt.reference == "systemd-credential:openclaw-gateway-candidate"
+    assert receipt.plaintext_secret_returned is False
+    manager.project(receipt)
+    manager.status(receipt)
+    manager.revoke(receipt)
+    manager.verify_revocation(receipt)
+    events = manager.audit(receipt)
+    assert events[0]["secrets_included"] is False
+    assert [operation for operation, _ in transport.calls] == [
+        "issue",
+        "project",
+        "status",
+        "revoke",
+        "verify_revocation",
+        "audit",
+    ]
+
+
+def test_secret_manager_rejects_plaintext_fields() -> None:
+    transport = Transport()
+
+    def unsafe(operation, payload):
+        return {"task_id": payload["task_id"], "credential_id": "credential-1", "api_key": "redacted"}
+
+    transport.exchange = unsafe
+    with pytest.raises(CredentialUnavailable, match="SECRET_MANAGER_PLAINTEXT_REJECTED"):
+        SecretManagerV2(transport).issue(credential_request(), NOW)
+
+
+def test_execution_manager_records_simulation_without_native_claim(tmp_path: Path) -> None:
+    evidence = EvidenceChain(StateStore(tmp_path / "state.sqlite3"), "task-execution")
+    adapter = OpenClawNativeAdapter(
+        Provider("openclaw"), "systemd-credential:openclaw-gateway-candidate"
+    )
+    result = ExecutionManager(evidence).run_native_canary(
+        adapter,
+        task_id="task-execution",
+        parent_session_id="parent-1",
+        recipient_binding={"recipient_id": "weixin-owner"},
+        text="candidate canary",
+    )
+    assert result.broker_session_verified is True
+    assert result.openclaw_native_child_session_verified is False
+    assert result.simulation is True
+    assert result.session_closed is True
+    assert evidence.verify()

tests/autonomous_release/test_state_and_workflow.py

@@ -0,0 +1,89 @@
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+import stat
+
+import pytest
+
+from aios_autonomous_release_candidate.errors import StateConflict
+from aios_autonomous_release_candidate.models import (
+    ResourceBinding,
+    RiskClass,
+    StepState,
+    TaskPlan,
+    WorkflowStep,
+)
+from aios_autonomous_release_candidate.store import StateStore
+from aios_autonomous_release_candidate.workflow import WorkflowEngine
+
+
+OID = "a" * 40
+
+
+def plan() -> TaskPlan:
+    return TaskPlan(
+        task_id="workflow-test",
+        intent="exercise durable workflow",
+        binding=ResourceBinding("aios/aios", "patch/workflow-test", OID, ("services/test",)),
+        risk=RiskClass.LOW,
+        steps=(
+            WorkflowStep("build", "builder", ("file.write",), max_attempts=2, timeout_seconds=30),
+            WorkflowStep("test", "tester", ("test.execute",), depends_on=("build",), max_attempts=2),
+        ),
+        evidence_requirements=("tests",),
+    )
+
+
+def test_store_is_private_and_rejects_schema_drift(tmp_path: Path) -> None:
+    state = tmp_path / "private" / "state.sqlite3"
+    StateStore(state)
+    assert stat.S_IMODE(state.stat().st_mode) == 0o600
+    assert stat.S_IMODE(state.parent.stat().st_mode) & 0o027 == 0
+
+    with StateStore(state).transaction() as connection:
+        connection.execute("UPDATE metadata SET value='999' WHERE key='schema_version'")
+    with pytest.raises(StateConflict, match="SCHEMA_VERSION_UNSUPPORTED"):
+        StateStore(state)
+
+
+def test_dependency_retry_and_terminal_failure_are_durable(tmp_path: Path) -> None:
+    store = StateStore(tmp_path / "state.sqlite3")
+    engine = WorkflowEngine(store)
+    task = plan()
+    engine.plan(task)
+
+    assert engine.next_steps(task).ready_steps == ("build",)
+    assert store.list_steps(task.task_id)["build"].state == StepState.READY
+    engine.claim(task.task_id, "build", "writer", "session-1")
+    engine.start(task.task_id, "build", "writer", "session-1")
+    engine.complete(task.task_id, "build", result_digest="b" * 64)
+
+    assert engine.next_steps(task).ready_steps == ("test",)
+    engine.claim(task.task_id, "test", "tester", "session-2")
+    engine.start(task.task_id, "test", "tester", "session-2")
+    assert engine.fail(task, "test", "transient", retryable=True) == StepState.PENDING
+    assert engine.next_steps(task).ready_steps == ("test",)
+    engine.claim(task.task_id, "test", "tester", "session-3")
+    engine.start(task.task_id, "test", "tester", "session-3")
+    assert engine.fail(task, "test", "still broken", retryable=True) == StepState.FAILED
+    decision = engine.next_steps(task)
+    assert decision.blocked is True
+    assert decision.reason == "terminal_step_failure"
+
+
+def test_claim_is_atomic_under_contention(tmp_path: Path) -> None:
+    store = StateStore(tmp_path / "state.sqlite3")
+    engine = WorkflowEngine(store)
+    task = plan()
+    engine.plan(task)
+    engine.next_steps(task)
+
+    def claim(worker: str) -> str:
+        try:
+            engine.claim(task.task_id, "build", worker, f"session-{worker}")
+            return "claimed"
+        except StateConflict:
+            return "rejected"
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(pool.map(claim, ("one", "two")))
+    assert sorted(results) == ["claimed", "rejected"]

tests/test_wechat_send_backend.py

@@ -1,11 +1,20 @@
 from __future__ import annotations
 
 import importlib.util
+import io
+import os
+import signal
+import subprocess
 import sys
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from types import SimpleNamespace
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import Mock, call, patch
 
 
 BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
@@ -80,6 +89,450 @@ class WechatSendBackendTests(TestCase):
         self.assertEqual(args[1]["content"], "hello")
         self.assertFalse(args[1]["dry_run"])
 
+    def test_ucc_auth_is_authorization_header_not_argv_or_output(self):
+        secret = "ucc-super-secret-token"
+
+        class Response:
+            status = 200
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+            def read(self, _limit=-1):
+                return b'{"ok":true,"status":"sent"}'
+
+        urlopen = Mock(return_value=Response())
+        output = io.StringIO()
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_TOKEN", secret),
+            patch.object(self.backend, "_open_no_redirect", urlopen),
+            redirect_stdout(output),
+            redirect_stderr(output),
+        ):
+            status, payload = self.backend._post_ucc(
+                "/api/internal/wechat/send", {"target": "chat-001", "content": "hello"}
+            )
+
+        request = urlopen.call_args.args[0]
+        self.assertEqual(request.get_header("Authorization"), f"Bearer {secret}")
+        self.assertEqual(status, 200)
+        self.assertEqual(payload, {"ok": True, "status": "sent"})
+        self.assertNotIn(secret, " ".join(self.backend._openclaw_base_args()))
+        self.assertNotIn(secret, output.getvalue())
+        self.assertFalse({"--gateway", "--token", "--password"} & set(self.backend._openclaw_base_args()))
+
+    def test_ucc_paths_are_canonical_first_unique_and_bounded(self):
+        with patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"):
+            paths = self.backend._ucc_send_paths()
+
+        self.assertEqual(paths[0], "/api/internal/wechat/send")
+        self.assertLessEqual(len(paths), 2)
+        self.assertEqual(len(paths), len(set(paths)))
+        self.assertTrue(all(path.startswith("/") for path in paths))
+
+    def test_ucc_accepts_only_explicit_true_and_sent_when_not_dry_run(self):
+        self.assertTrue(self.backend._ucc_response_sent({"ok": True, "status": "sent"}))
+        for response in (
+            {"ok": 1, "status": "sent"},
+            {"ok": True, "status": "accepted"},
+            {"ok": False, "status": "sent"},
+            {"status": "sent"},
+        ):
+            with self.subTest(response=response):
+                self.assertFalse(self.backend._ucc_response_sent(response))
+
+    def test_ambiguous_canonical_failures_never_fallback(self):
+        secret = "raw-upstream-secret"
+        failures = (
+            TimeoutError(secret),
+            ConnectionError(secret),
+            self.backend._BackendProtocolError(secret),
+            RuntimeError(secret),
+        )
+        for failure in failures:
+            with self.subTest(failure=type(failure).__name__):
+                with (
+                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
+                    patch.object(self.backend, "_post_ucc", side_effect=failure) as post,
+                ):
+                    result = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello", message_id="msg-001"
+                    ).to_dict()
+
+                self.assertFalse(result["ok"])
+                self.assertEqual(result["code"], "ucc_adapter_unavailable")
+                self.assertEqual(post.call_count, 1)
+                self.assertNotIn(secret, repr(result))
+
+    def test_only_explicit_http_404_or_405_uses_legacy_fallback(self):
+        for http_status in (404, 405):
+            with self.subTest(http_status=http_status):
+                missing = urllib.error.HTTPError(
+                    "http://127.0.0.1/api/internal/wechat/send",
+                    http_status,
+                    "not found",
+                    {},
+                    None,
+                )
+                with (
+                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
+                    patch.object(
+                        self.backend,
+                        "_post_ucc",
+                        side_effect=[missing, (200, {"ok": True, "status": "sent"})],
+                    ) as post,
+                ):
+                    result = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello", message_id="msg-001"
+                    ).to_dict()
+
+                self.assertTrue(result["ok"])
+                self.assertEqual(result["endpoint"], "/internal/channels/wechat/send")
+                self.assertEqual(post.call_count, 2)
+
+    def test_non_not_found_http_error_never_falls_back(self):
+        denied = urllib.error.HTTPError(
+            "http://127.0.0.1/api/internal/wechat/send", 401, "denied", {}, None
+        )
+        with patch.object(self.backend, "_post_ucc", side_effect=denied) as post:
+            result = self.backend.send_ucc_adapter(target="chat-001", content="hello").to_dict()
+
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "ucc_adapter_http_error")
+        post.assert_called_once()
+
+    def test_ucc_rejection_does_not_expose_raw_error_or_secret(self):
+        secret = "provider-password=secret-value"
+        with patch.object(
+            self.backend,
+            "_post_ucc",
+            return_value=(200, {"ok": False, "status": "failed", "error": secret}),
+        ):
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello", message_id="msg-001"
+            ).to_dict()
+
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "ucc_adapter_rejected")
+        self.assertNotIn(secret, repr(result))
+
+    def test_dry_run_requires_dry_run_status_and_never_reports_sent(self):
+        with patch.object(
+            self.backend,
+            "_post_ucc",
+            return_value=(200, {"ok": True, "status": "dry_run"}),
+        ):
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello", dry_run=True
+            ).to_dict()
+
+        self.assertTrue(result["ok"])
+        self.assertEqual(result["status"], "dry_run")
+        self.assertNotEqual(result["status"], "sent")
+
+        for dry_run, upstream_status in ((True, "sent"), (False, "dry_run")):
+            with self.subTest(dry_run=dry_run, upstream_status=upstream_status):
+                with patch.object(
+                    self.backend,
+                    "_post_ucc",
+                    return_value=(200, {"ok": True, "status": upstream_status}),
+                ):
+                    rejected = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello", dry_run=dry_run
+                    ).to_dict()
+                self.assertFalse(rejected["ok"])
+                self.assertEqual(rejected["status"], "failed")
+
+    def test_invalid_provider_message_id_never_reverses_confirmed_send(self):
+        invalid_values = (
+            "\ud800",
+            "contains\x00control",
+            "x" * (self.backend.MAX_IDENTIFIER_BYTES + 1),
+            {"not": "a string"},
+        )
+        for provider_message_id in invalid_values:
+            with self.subTest(value=repr(provider_message_id)[:40]):
+                with patch.object(
+                    self.backend,
+                    "_post_ucc",
+                    return_value=(
+                        200,
+                        {
+                            "ok": True,
+                            "status": "sent",
+                            "provider_message_id": provider_message_id,
+                        },
+                    ),
+                ):
+                    result = self.backend.send_ucc_adapter(
+                        target="chat-001", content="hello"
+                    ).to_dict()
+
+                self.assertTrue(result["ok"])
+                self.assertEqual(result["status"], "sent")
+                self.assertNotIn("provider_message_id", result)
+
+    def test_openclaw_2026_6_11_commands_have_no_unsupported_global_gateway(self):
+        completed = SimpleNamespace(returncode=0, stdout='{"channels": []}')
+        with (
+            patch.object(self.backend, "OPENCLAW_BIN", "/usr/local/bin/openclaw"),
+            patch.object(self.backend, "_run_openclaw", return_value=completed) as run,
+        ):
+            self.assertIsNone(self.backend.discover_openclaw_wechat_channel())
+
+        command = run.call_args.args[0]
+        self.assertEqual(command, ["/usr/local/bin/openclaw", "channels", "list", "--json"])
+        self.assertFalse({"--gateway", "--token", "--password"} & set(command))
+
+    def test_openclaw_output_overflow_terminates_while_process_is_running(self):
+        writer = (
+            "import os,time;"
+            f"os.write(1,b'x'*{self.backend.MAX_RESPONSE_BYTES + 4096});"
+            "time.sleep(10)"
+        )
+        started = time.monotonic()
+        completed = self.backend._run_openclaw([sys.executable, "-c", writer], timeout=4)
+        elapsed = time.monotonic() - started
+
+        self.assertLess(elapsed, 2.0)
+        self.assertNotEqual(completed.returncode, 0)
+        self.assertEqual(completed.stdout, "")
+        self.assertEqual(completed.stderr, "")
+
+    def test_openclaw_timeout_returns_failure_and_terminates_promptly(self):
+        started = time.monotonic()
+        completed = self.backend._run_openclaw(
+            [sys.executable, "-c", "import time;time.sleep(10)"], timeout=1
+        )
+        elapsed = time.monotonic() - started
+
+        self.assertLess(elapsed, 3.0)
+        self.assertNotEqual(completed.returncode, 0)
+        self.assertEqual(completed.stdout, "")
+        self.assertEqual(completed.stderr, "")
+
+    def test_timeout_kills_inherited_pipe_descendant_after_parent_exits(self):
+        if os.name != "posix" or not Path("/proc").is_dir():
+            self.skipTest("POSIX process-group and procfs contract")
+
+        def process_is_running(pid):
+            try:
+                state = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[2]
+            except (FileNotFoundError, IndexError, OSError):
+                return False
+            return state not in {"X", "Z"}
+
+        with tempfile.TemporaryDirectory() as directory:
+            pid_path = Path(directory) / "descendant.pid"
+            descendant = "import time;time.sleep(30)"
+            launcher = (
+                "import pathlib,subprocess,sys;"
+                f"child=subprocess.Popen([sys.executable,'-c',{descendant!r}]);"
+                f"pathlib.Path({str(pid_path)!r}).write_text(str(child.pid),encoding='utf-8')"
+            )
+            started = time.monotonic()
+            completed = self.backend._run_openclaw(
+                [sys.executable, "-c", launcher], timeout=1
+            )
+            elapsed = time.monotonic() - started
+            self.assertTrue(pid_path.is_file())
+            descendant_pid = int(pid_path.read_text(encoding="utf-8"))
+            try:
+                deadline = time.monotonic() + 2
+                while process_is_running(descendant_pid) and time.monotonic() < deadline:
+                    time.sleep(0.02)
+                self.assertFalse(process_is_running(descendant_pid))
+            finally:
+                if process_is_running(descendant_pid):
+                    os.kill(descendant_pid, signal.SIGKILL)
+
+        self.assertLess(elapsed, 3.0)
+        self.assertNotEqual(completed.returncode, 0)
+
+    def test_process_tree_termination_uses_posix_process_group(self):
+        if os.name != "posix":
+            self.skipTest("POSIX process-group contract")
+
+        class Process:
+            pid = 43210
+
+            def __init__(self):
+                self.wait_count = 0
+
+            def poll(self):
+                return None
+
+            def wait(self, timeout):
+                self.wait_count += 1
+                if self.wait_count == 1:
+                    raise subprocess.TimeoutExpired(["openclaw"], timeout)
+                return -signal.SIGKILL
+
+        process = Process()
+        with patch.object(self.backend.os, "killpg") as killpg:
+            self.backend._terminate_process_tree(process)
+
+        self.assertEqual(
+            killpg.call_args_list,
+            [
+                call(process.pid, signal.SIGTERM),
+                call(process.pid, signal.SIGKILL),
+            ],
+        )
+
+    def test_unsupported_and_generic_chat_backends_fail_closed(self):
+        with (
+            patch.object(self.backend, "SEND_BACKEND", "openclaw-chat"),
+            patch.object(self.backend, "ucc_adapter_alive", return_value=False),
+        ):
+            selected = self.backend.select_backend()
+            result = self.backend.send_message(target="chat-001", content="hello")
+
+        self.assertEqual(selected["backend"], "unavailable")
+        self.assertEqual(selected["reason"], "unsupported_backend")
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "unsupported_backend")
+
+    def test_invalid_or_oversized_input_is_rejected_before_network(self):
+        with patch.object(self.backend, "_post_ucc") as post:
+            empty = self.backend.send_ucc_adapter(target="", content="hello").to_dict()
+            oversized = self.backend.send_ucc_adapter(
+                target="chat-001", content="x" * (self.backend.MAX_CONTENT_BYTES + 1)
+            ).to_dict()
+
+        self.assertEqual(empty["code"], "invalid_input")
+        self.assertEqual(oversized["code"], "input_too_large")
+        post.assert_not_called()
+
+    def test_unpaired_surrogate_inputs_fail_closed_before_network(self):
+        cases = (
+            {"target": "\ud800", "content": "hello"},
+            {"target": "chat-001", "content": "\ud800"},
+            {"target": "chat-001", "content": "hello", "account_id": "\ud800"},
+            {"target": "chat-001", "content": "hello", "message_id": "\ud800"},
+            {"target": "chat-001", "content": "hello", "metadata": {"bad": "\ud800"}},
+        )
+        with patch.object(self.backend, "_post_ucc") as post:
+            for kwargs in cases:
+                with self.subTest(field=next(reversed(kwargs))):
+                    result = self.backend.send_ucc_adapter(**kwargs).to_dict()
+                    self.assertEqual(result["code"], "invalid_input")
+        post.assert_not_called()
+
+    def test_deeply_nested_metadata_is_bounded_before_network(self):
+        metadata = {}
+        for _ in range(1200):
+            metadata = {"nested": metadata}
+
+        with patch.object(self.backend, "_post_ucc") as post:
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello", metadata=metadata
+            ).to_dict()
+
+        self.assertEqual(result["code"], "input_too_large")
+        post.assert_not_called()
+
+    def test_adapter_origin_must_be_allowlisted(self):
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_URL", "https://evil.example:8443"),
+            patch.object(
+                self.backend,
+                "UCC_ADAPTER_ALLOWED_ORIGINS",
+                ("http://127.0.0.1:18800",),
+            ),
+        ):
+            self.assertIsNone(self.backend._adapter_base_url())
+
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_URL", "https://adapter.example:8443"),
+            patch.object(
+                self.backend,
+                "UCC_ADAPTER_ALLOWED_ORIGINS",
+                ("https://adapter.example:8443",),
+            ),
+        ):
+            self.assertEqual(
+                self.backend._adapter_base_url(), "https://adapter.example:8443"
+            )
+
+    def test_adapter_origin_rejects_scheme_userinfo_fragment_and_path(self):
+        invalid_origins = (
+            "ftp://127.0.0.1:18800",
+            "http://user:password@127.0.0.1:18800",
+            "http://127.0.0.1:18800#fragment",
+            "http://127.0.0.1:18800/base-path",
+        )
+        for origin in invalid_origins:
+            with self.subTest(origin=origin):
+                with (
+                    patch.object(self.backend, "UCC_ADAPTER_URL", origin),
+                    patch.object(
+                        self.backend,
+                        "UCC_ADAPTER_ALLOWED_ORIGINS",
+                        ("http://127.0.0.1:18800",),
+                    ),
+                ):
+                    self.assertIsNone(self.backend._adapter_base_url())
+
+    def test_adapter_path_rejects_decoded_traversal(self):
+        for path in (
+            "/api/%2e%2e/send",
+            "/api/%2E%2E/send",
+            "/api/%2525252e%2525252e/send",
+            "/api/%252525252e%252525252e/send",
+            "/api/../send",
+            "/api/%zz/send",
+        ):
+            with self.subTest(path=path):
+                self.assertIsNone(self.backend._safe_path(path))
+
+    def test_adapter_path_keeps_intended_safe_canonical_paths(self):
+        for path in (
+            "/api/internal/wechat/send",
+            "/internal/channels/wechat/send",
+            "/api/%7Ehealth/status",
+        ):
+            with self.subTest(path=path):
+                self.assertEqual(self.backend._safe_path(path), path)
+
+    def test_redirects_are_refused_and_authorization_cannot_cross_origin(self):
+        handler = self.backend._NoRedirectHandler()
+        self.assertIsNone(
+            handler.redirect_request(
+                None,
+                None,
+                302,
+                "redirect",
+                {"Location": "https://evil.example/steal"},
+                "https://evil.example/steal",
+            )
+        )
+
+        redirected = urllib.error.HTTPError(
+            "http://127.0.0.1:18800/api/internal/wechat/send",
+            302,
+            "redirect",
+            {"Location": "https://evil.example/steal"},
+            None,
+        )
+        with (
+            patch.object(self.backend, "UCC_ADAPTER_TOKEN", "header-secret"),
+            patch.object(self.backend, "_open_no_redirect", side_effect=redirected) as opened,
+        ):
+            result = self.backend.send_ucc_adapter(
+                target="chat-001", content="hello"
+            ).to_dict()
+
+        request = opened.call_args.args[0]
+        self.assertEqual(urllib.parse.urlsplit(request.full_url).netloc, "127.0.0.1:18800")
+        self.assertEqual(request.get_header("Authorization"), "Bearer header-secret")
+        opened.assert_called_once()
+        self.assertFalse(result["ok"])
+        self.assertEqual(result["code"], "ucc_adapter_http_error")
+
 
 if __name__ == "__main__":
     import unittest

versions/V2R1C4.json

@@ -0,0 +1,13 @@
+{
+  "version": "V2R1C4",
+  "repository": "aios/aios",
+  "target_oid": "2c48b9fd395f08ca87ad9bb94f9f49d06de8d291",
+  "tag_oid": "2f6c1652f400dfb02e6a80b4b11d433659db494a",
+  "peeled_target_oid": "2c48b9fd395f08ca87ad9bb94f9f49d06de8d291",
+  "artifact_sha256": "c04cf2f1b1b6b77fc632c60642eb434fc577216178bb0bfec5fe89deeda6df4d",
+  "evidence_closure_oid": "053e9c992ad20198cff9bab956fe6ab649b7d21a",
+  "approval_plan_id": "sha256:348408f38c54536e3a2a3765ac70cc6bd3884bcab761d5f5d9f82d1597beec9e",
+  "previous_version": "V2R1C3",
+  "deployment_status": "TAG_PUBLISHED",
+  "release_timestamp": "2026-08-05T03:29:05Z"
+}

versions/V2R1C5.json

@@ -0,0 +1,35 @@
+{
+  "approval_plan_id": "sha256:34c4031c3014498ac2b4bd846026c0ca372a7d9247f3c24816d3700735006be9",
+  "artifact_composition": "PASS",
+  "artifact_sha256": "1a9d5b6d240584eb683c86c55014b05bc8f6f0113b2cae053c88758034aefbba",
+  "baseline_promoted": false,
+  "direct_main_push": false,
+  "evidence_closure_oid": "5e2cf8276f5093f3352b9a3252c701a44c5ab807",
+  "force_push": false,
+  "independent_validation": "PASS",
+  "maintenance_mode": "SINGLE_MAINTAINER",
+  "owner_approval": "OWNER_EXACT_OID_APPROVED",
+  "previous_version": "V2R1C3",
+  "privacy_migration": "PASS",
+  "production_baseline_before": "V2R1C3",
+  "production_deployed": false,
+  "raw_owner_findings": 0,
+  "registered_at": "2026-08-05T06:45:02.796699+00:00",
+  "registered_by": "aios-autonomy-agent",
+  "repository": "aios/aios",
+  "required_approvals": 1,
+  "required_distinct_approvers": 1,
+  "rollback_tag_oid": "a159d0ec4ea558137c1567f558ef41bfd9f5122d",
+  "rollback_target_oid": "7db5e458b225b6eabe1bb59fc119602abde8a629",
+  "rollback_version": "V2R1C3",
+  "schema": "aios.v2r1c5.registry.v1",
+  "second_reviewer_required": false,
+  "secrets_committed": false,
+  "status": "TAG_PUBLISHED_PENDING_PRODUCTION_ACCEPTANCE",
+  "tag": "V2R1C5",
+  "tag_object": "9e65bd59baea392294c0f253046d3ab116f63145",
+  "tag_ref": "refs/tags/V2R1C5",
+  "target_oid": "aa60e891547abd7e215af7cf2bf3f21fd23541da",
+  "tree_oid": "cae5ca9b6d923810187f373afe2d9bfe37215ba4",
+  "version": "V2R1C5"
+}