security(openclaw): record restored Gitea credential without secret value

task(openclaw): 下发执行器启动、Secret 引用与 Gitea 同步在线指令
task(openclaw): 初始化无人值守任务状态
2026-07-20 15:37:29 +08:00 · 2026-07-20 14:34:13 +08:00 · 2026-07-20 12:07:46 +08:00 · 2026-07-20 12:07:36 +08:00 · 2026-07-20 11:49:09 +08:00 · 2026-07-20 11:46:43 +08:00
49 changed files with 6996 additions and 357 deletions
@@ -0,0 +1,412 @@
+# OpenClaw 在线执行指令：启动执行器、配置 Secret 引用并确认 Gitea 同步
+
+- 任务 ID：`AIOS-V2R1-PRODUCTION-DEPLOY-20260720`
+- 执行器：OpenClaw
+- 任务分支：`task/openclaw/AIOS-V2R1-PRODUCTION-DEPLOY-20260720`
+- 候选分支：`integration/aios-v2r1-cto-production-candidate-20260720`
+- 语言：中文
+- 执行模式：无人值守、证据驱动、默认拒绝
+
+## 1. 任务目标
+
+一次性完成以下工作：
+
+1. 发现并启动环境中已注册的 OpenClaw 在线执行器；
+2. 领取本任务并持续写入 claim、heartbeat、progress、blocker 和 result；
+3. 将任务分支同步到最新 Candidate HEAD；
+4. 确认 Candidate、OpenClaw Task、Codex Task 三个分支已同步到 Gitea，且本地与远端 OID 完全一致；
+5. 从已批准的 Secret Provider、Credential Store 或受保护文件中解析 OpenClaw Runtime 配置；
+6. 只把 Secret 引用和受保护文件路径写入 AIOS 环境配置，不得把明文 Secret 写入 Git、日志、Evidence 或终端输出；
+7. 执行文档—代码检查、候选测试、全量验收和预发布健康检查；
+8. 对低风险部署或 Runtime 缺陷创建 Patch 并修复；
+9. 把所有修改和阻塞提交到系统，交由 Codex 统一审视。
+
+## 2. 不可违反的边界
+
+- 不得直接修改或 push `main`；
+- 不得使用 `--force` 或改写共享分支历史；
+- 不得创建生产 Tag；
+- 不得切换生产流量，除非 AIOS Owner 后续明确批准；
+- 不得提交 Token、密码、API Key、私钥、签名材料或 Secret 内容；
+- 不得在日志中执行 `env`、`set`、`printenv` 或输出 Secret 文件内容；
+- 不得修改 AIOS 唯一控制面、多租户或审批边界；
+- 发现架构、安全或跨模块阻塞时必须升级 Codex；
+- 未生成 claim/heartbeat 证据前不得标记为运行中；
+- 未通过验收前不得标记为完成。
+
+## 3. 第一阶段：执行器发现与领取
+
+### 3.1 发现执行器入口
+
+使用当前环境中已经注册和批准的 OpenClaw 执行入口。可检查：
+
+- systemd 服务；
+- 容器或 Compose 服务；
+- 已安装的 OpenClaw CLI；
+- AIOS Agent Runner；
+- 已配置的队列消费者或调度器。
+
+不得臆造不存在的命令、服务名或路径。
+
+如果找不到可执行入口，立即创建 Blocker：
+
+```json
+{
+  "blocker_id": "BLK-EXECUTOR-ENTRYPOINT-MISSING",
+  "severity": "high",
+  "description": "环境中未发现已注册的 OpenClaw 执行器入口",
+  "owner": "AIOS 运维负责人",
+  "exit_condition": "提供或启用已批准的 OpenClaw systemd、容器、CLI 或 Agent Runner 入口"
+}
+```
+
+状态更新为 `blocked`，不得伪报已经启动。
+
+### 3.2 启动执行器
+
+发现入口后：
+
+1. 验证配置来源和运行用户；
+2. 启动或恢复执行器；
+3. 验证进程/容器处于运行状态；
+4. 验证执行器能够读取本任务分支；
+5. 立即写入 `.agent/claim.json`；
+6. 更新 `.agent/status.json`：`queued -> claimed -> running`；
+7. 写入首次 `.agent/heartbeat.json`；
+8. 将证据提交到任务分支并同步 Gitea。
+
+`claim.json` 至少包含：
+
+```json
+{
+  "schema": "aios.v2.agent-claim.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "executor": "openclaw",
+  "task_branch": "task/openclaw/AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "candidate_branch": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "candidate_commit": "<领取时实际 Candidate OID>",
+  "claimed_at": "<UTC ISO-8601>",
+  "executor_instance": "<非敏感实例标识>"
+}
+```
+
+Heartbeat 间隔不得超过 300 秒。
+
+## 4. 第二阶段：同步最新 Candidate
+
+### 4.1 获取最新引用
+
+只更新 remote-tracking ref，不得 fetch 到已经被 Worktree 检出的本地 Branch Ref。
+
+执行原则：
+
+```text
+fetch remote candidate -> refs/remotes/<remote>/<candidate>
+```
+
+不得使用会触发以下错误的流程：
+
+```text
+refusing to fetch into branch checked out at worktree
+```
+
+### 4.2 更新任务分支
+
+将最新 Candidate 合并到 OpenClaw 任务分支：
+
+- 使用普通 merge 或团队批准的线性更新方式；
+- 不得 force reset 已经提交的 Agent Evidence；
+- 发生冲突时停止自动扩大修改范围；
+- 文档、任务载荷或架构冲突升级 Codex；
+- 合并完成后把 `.agent/task.json` 与 `.agent/status.json` 中的 `candidate_commit` 更新为实际 Candidate OID。
+
+### 4.3 必须记录
+
+在 `.agent/progress.json` 记录：
+
+- Candidate 本地 OID；
+- Candidate Gitea OID；
+- Task Branch 合并前后 OID；
+- 修改文件；
+- 是否发生冲突；
+- 合并 Commit。
+
+## 5. 第三阶段：确认 Gitea 同步
+
+需要验证以下分支：
+
+```text
+integration/aios-v2r1-cto-production-candidate-20260720
+task/openclaw/AIOS-V2R1-PRODUCTION-DEPLOY-20260720
+task/codex/AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720
+```
+
+目标仓库：
+
+```text
+g.cnnetmkt.top/git/aios/aios
+```
+
+### 5.1 同步规则
+
+- 使用环境中已配置的 Gitea Remote 和凭据引用；
+- 不在 Evidence 中记录带凭据的 Remote URL；
+- 禁止 Force Push；
+- Candidate 和 Task Branch 可以普通 Push；
+- Push 失败时记录 stderr 的脱敏摘要，不得记录凭据。
+
+### 5.2 OID 验证
+
+每个分支必须同时取得：
+
+```text
+local_oid
+gitea_oid
+match = true/false
+```
+
+验收条件：
+
+```text
+local_oid == gitea_oid
+```
+
+建议通过 remote-tracking ref 或 `git ls-remote --heads` 取得远端 OID。输出中不得泄露认证信息。
+
+如果 Push 或 OID 校验失败，创建 Blocker：
+
+```json
+{
+  "blocker_id": "BLK-GITEA-SYNC",
+  "severity": "high",
+  "description": "本地与 Gitea 分支 OID 不一致或无推送权限",
+  "owner": "GitOps/Gitea 运维负责人",
+  "exit_condition": "三个目标分支均完成普通推送且 local_oid 等于 gitea_oid"
+}
+```
+
+在 Gitea OID 一致前，不得继续预发布部署。
+
+## 6. 第四阶段：配置 Secret 引用
+
+### 6.1 需要解析的配置
+
+至少确认：
+
+```text
+AIOS_OPENCLAW_URL
+AIOS_OPENCLAW_CREDENTIAL_REF
+AIOS_OPENCLAW_SIGNING_MATERIAL_FILE
+AIOS_OPENCLAW_SIGNING_KEY_ID
+```
+
+其中：
+
+- `AIOS_OPENCLAW_URL` 是经过批准的 Runtime Endpoint；
+- `AIOS_OPENCLAW_CREDENTIAL_REF` 只保存 Secret Provider 引用；
+- `AIOS_OPENCLAW_SIGNING_MATERIAL_FILE` 指向仓库外的受保护文件；
+- `AIOS_OPENCLAW_SIGNING_KEY_ID` 是 Key 标识，不是 Key 内容。
+
+### 6.2 Secret 来源优先级
+
+按以下顺序查找：
+
+1. 已批准的运行时 Secret Store；
+2. 已批准的 Credential Provider；
+3. systemd Credential；
+4. 只读 Secret Volume；
+5. 仓库外受保护文件。
+
+不得：
+
+- 从 Git 历史恢复 Secret；
+- 从日志、Shell History 或旧 Evidence 复制 Secret；
+- 新建明文 Secret 并提交；
+- 把 Secret 内容写入 `.env` 模板或任务 JSON。
+
+### 6.3 配置文件与权限
+
+生产候选环境文件：
+
+```text
+/etc/aios-v2/aios-v2.env
+```
+
+建议权限：
+
+```text
+owner: root
+group: aios
+mode: 0640 或更严格
+```
+
+签名材料文件：
+
+```text
+位于仓库外
+owner: aios 或经批准的运行用户
+mode: 0600 或更严格
+```
+
+只验证以下内容：
+
+- 引用非空；
+- 文件存在；
+- 运行用户可读；
+- 权限不宽松；
+- 文件不在 Git Worktree；
+- 不输出文件内容或长度特征。
+
+### 6.4 缺少 Secret 时的行为
+
+若无法从已批准来源解析任一必需配置，创建：
+
+```json
+{
+  "blocker_id": "BLK-SECRET-REF-MISSING",
+  "severity": "high",
+  "description": "缺少已批准的 OpenClaw Runtime Endpoint、Credential Ref、Signing File 或 Key ID",
+  "owner": "AIOS Secret/运维负责人",
+  "exit_condition": "四项配置均可从批准来源解析，且权限检查通过"
+}
+```
+
+不得要求用户在聊天或 Git 中提供 Secret 明文。
+
+## 7. 第五阶段：候选验收与预发布
+
+在 Gitea OID 和 Secret 引用检查都通过后，依次执行：
+
+```bash
+python3 tools/aios_v2_docs_code_check.py \
+  --base-ref integration/aios-v2r1-cto-production-candidate-20260720
+
+python3 -m unittest discover \
+  -s tests/aios_v2_candidate \
+  -p 'test_*.py' \
+  -v
+
+bash deploy/aios-v2/acceptance/run-all.sh
+```
+
+部署后执行：
+
+```bash
+AIOS_RUN_LIVE_HEALTHCHECK=1 \
+  bash deploy/aios-v2/acceptance/run-all.sh
+
+bash deploy/aios-v2/healthcheck/check.sh
+```
+
+必须验证：
+
+- AIOS HTTP Health；
+- SQLite `quick_check=ok`；
+- GitOps Controller detached + clean；
+- OpenClaw execute/status/cancel/timeout/health；
+- Hermes claim/heartbeat/complete/fail/lease recovery；
+- 服务重启后 Tenant/Agent/Task/Audit/Idempotency 恢复；
+- 回滚后健康检查通过；
+- 日志与 Evidence 无 Secret。
+
+## 8. Bug 修复规则
+
+OpenClaw 可自行修复：
+
+- systemd、Docker、Compose、路径和权限；
+- 环境变量名或启动参数；
+- Healthcheck；
+- 明确的小型 Runtime Adapter 兼容问题；
+- 日志、错误消息和超时边界；
+- 测试、部署或中文文档中的明确错误。
+
+每个修复必须：
+
+1. 创建 `patch/bugfix/*`；
+2. 记录复现步骤；
+3. 添加或更新测试；
+4. 修改代码；
+5. 同步 `docs/aios-v2/` 和追踪矩阵；
+6. 执行文档门禁和相关测试；
+7. 提交 Git；
+8. 更新 `.agent/blockers.json` 和 `.agent/progress.json`；
+9. 提交 Codex 审查。
+
+以下问题必须升级 Codex：
+
+- Control Plane、Task Lifecycle、Policy 或 Approval 变化；
+- 多租户或身份边界变化；
+- OpenClaw/Hermes Contract 不兼容变化；
+- 数据迁移、持久化或回滚策略变化；
+- 高风险安全问题；
+- 需要修改 ADR 的问题；
+- 超出低风险修复范围的阻塞。
+
+## 9. 证据文件
+
+任务分支必须持续维护：
+
+```text
+.agent/claim.json
+.agent/heartbeat.json
+.agent/progress.json
+.agent/blockers.json
+.agent/result.json
+.agent/status.json
+```
+
+`result.json` 至少包含：
+
+```json
+{
+  "schema": "aios.v2.agent-result.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "final_status": "complete|failed|blocked",
+  "candidate_commit": "<OID>",
+  "task_commit": "<OID>",
+  "gitea_sync": {
+    "candidate_match": true,
+    "openclaw_task_match": true,
+    "codex_task_match": true
+  },
+  "secret_reference_status": "configured|blocked",
+  "executor_status": "running|blocked",
+  "test_status": "passed|failed|not_run",
+  "deployment_status": "passed|failed|not_run",
+  "rollback_status": "passed|failed|not_run",
+  "documentation_sync_status": "passed|failed",
+  "changed_files": [],
+  "commits": [],
+  "open_blockers": [],
+  "remaining_risks": [],
+  "completed_at": "<UTC ISO-8601>"
+}
+```
+
+## 10. 完成条件
+
+只有以下条件全部满足，任务才可标记为 `complete`：
+
+- OpenClaw 执行器已启动并持续 Heartbeat；
+- 已领取任务；
+- 任务分支已合并最新 Candidate；
+- Candidate、OpenClaw Task、Codex Task 在本地与 Gitea 的 OID 一致；
+- Secret 引用和受保护文件权限检查通过；
+- 文档—代码检查通过；
+- 候选测试通过；
+- 预发布部署和在线健康检查通过；
+- Runtime 联调、重启和回滚演练通过；
+- 所有修改、Blocker、Commit、测试和结果已回写；
+- Codex 已能读取并审查完整证据。
+
+如果缺少外部权限、执行入口或已批准 Secret，任务必须标记为 `blocked`，并准确说明 Owner 与解除条件，不得伪报完成。
+
+## 11. Owner Gate
+
+完成本指令不等于自动批准生产发布。以下动作仍需 AIOS Owner 明确批准：
+
+- 合并 `main`；
+- 创建生产 Tag；
+- 正式生产部署；
+- 生产流量切换；
+- 接受高风险剩余问题。
@@ -0,0 +1,23 @@
+{
+  "schema": "aios.v2.credential-status.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "credential_name": "openclaw-deploy-cli",
+  "credential_type": "gitea-api-token-reference",
+  "endpoints": [
+    "http://127.0.0.1:3000",
+    "https://g.cnnetmkt.top/git"
+  ],
+  "reported_scope": "all",
+  "reported_status": "working",
+  "verification_source": "owner-reported",
+  "secret_value_stored": false,
+  "security_status": "rotation_required_before_production",
+  "reason": "令牌值曾在聊天中以明文出现，必须按已暴露凭据处理",
+  "required_actions": [
+    "立即撤销当前令牌",
+    "创建新令牌并使用最小必要作用域",
+    "仅通过 Secret Provider、systemd Credential 或受保护文件向 OpenClaw 提供",
+    "更新 AIOS_OPENCLAW_CREDENTIAL_REF，不在 Git、日志或 Evidence 中保存令牌值",
+    "旋转后重新验证本地与外部 Gitea 端点"
+  ]
+}
@@ -1,172 +0,0 @@
---
-authorization_id: TASK-V2R1-RUNTIME-INTEGRATION-P1-20260714
-task_id: TASK-V2R1-RUNTIME-INTEGRATION-P1-20260714
-executor: GPT_CODEX
-status: PASS
-started_at: 2026-07-14T02:24:00+08:00
-finished_at: 2026-07-14T02:27:46+08:00
-repository: /data/aios-gitops/workspace
-review_branch: review/codex/v2r1-runtime-integration-discovery-20260714
-review_worktree: /data/aios-gitops/worktrees/v2r1-runtime-integration-discovery-20260714
-result_commit: SELF
-result_commit_resolution: The containing commit is resolved by Git from the review branch because embedding its object ID would change the commit identity.
-parent_commit: 3cf9958cb94614449e093639cbf902216d73447f
-source_commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
-gate1_commit: 7f5d5d90b3111c35fb00044f385358f528e3344e
-merge_base: 1a9cb9095e6e76c548c967e03cb7a3e47b85134b
-main_graph_commits: 21
-source_graph_commits: 44
-main_changed_paths: 16
-source_changed_paths: 42
-approved_replay_paths: 38
-excluded_source_paths: 4
-changed_path_overlap: 0
-workspace_clean_after_commit: true
-rollback_target: 3cf9958cb94614449e093639cbf902216d73447f
-runtime_modified: false
-tests_modified: false
-main_modified: false
-tag_created: false
-baseline_modified: false
-service_modified: false
-deployment_performed: false
-force_push_used: false
-risk: HIGH
-recommendation: Proceed to a separately authorized P2 allowlist replay from exact main; keep tag, baseline, staging and deployment blocked until P4/P5 close every listed blocker.
---
-
-# V2R1 Runtime Integration P1 Discovery Result
-
-## Decision
-
-**PASS for discovery only.** The repository graph, path classifications, preservation boundary, exclusion boundary, and safe integration strategy are now explicit. This result does not authorize runtime integration or release activity.
-
-## Authoritative identities
-
-| Item | Verified value |
-|---|---|
-| Current main | `3cf9958cb94614449e093639cbf902216d73447f` |
-| V2R1 source | `502ef30642597fb0b70b90ae2e8a0b6af217d80e` |
-| Gate1 PASS commit | `7f5d5d90b3111c35fb00044f385358f528e3344e` |
-| Merge base | `1a9cb9095e6e76c548c967e03cb7a3e47b85134b` |
-| Main graph | `21` commits / `16` changed paths |
-| Source graph | `44` commits / `42` changed paths |
-| Changed-path overlap | `0` |
-
-Gate1 commit is a direct child of source and adds one OpenClaw report. Main already contains the identical accepted report blob `ff31f7ab603d7a7cd07319e6f8422ae464916179`. The Gate1 commit must not be merged again.
-
-## Exact classification
-
-### Preserve from main — 16 paths
-
-```text
-.agent/results/openclaw/RESULT-V2R1-G1-OPENCLAW-FINAL-EVIDENCE-20260714.md
-deployments/V1R4C6/backup-result.md
-deployments/V1R4C6/deployment-result.md
-deployments/V1R4C6/preflight-result.md
-deployments/V1R4C6/rollback-result.md
-deployments/V1R4C6/verification-result.md
-docs/incidents/WEIXIN_TEXT_IMAGE_CACHE_ISOLATION_20260711.md
-docs/superpowers/plans/2026-07-12-weixin-quoted-text-failclosed.md
-docs/superpowers/specs/2026-07-12-weixin-quoted-text-failclosed-design.md
-fixes/patch_openclaw_weixin_bigint.py
-fixes/patch_openclaw_weixin_hermes.py
-fixes/patch_openclaw_weixin_outbound_quotes.py
-patches/deploy_v1r4c6-production-20260711.branch.json
-tests/test_weixin_bigint_parser.py
-tests/test_weixin_image_followup_intent.py
-tests/test_weixin_outbound_quote_timestamp.py
-```
-
-### Replay from source — 38 paths
-
-```text
-.gitignore
-active-runtime/app/aios_v2r1/audit.py
-active-runtime/app/aios_v2r1/cell_manager.py
-active-runtime/app/aios_v2r1/channel_adapter.py
-active-runtime/app/aios_v2r1/cli.py
-active-runtime/app/aios_v2r1/fleet.py
-active-runtime/app/aios_v2r1/hermes_fabric.py
-active-runtime/app/aios_v2r1/hermes_profile.py
-active-runtime/app/aios_v2r1/identity.py
-active-runtime/app/aios_v2r1/memory_scope.py
-active-runtime/app/aios_v2r1/message_ledger.py
-active-runtime/app/aios_v2r1/router.py
-active-runtime/app/aios_v2r1/runtime.py
-active-runtime/app/aios_v2r1/secret_scope.py
-active-runtime/app/aios_v2r1/session_pool.py
-active-runtime/app/aios_v2r1/tenant_context.py
-agents/registry.yaml
-data/webops.sqlite3 [delete]
-deploy/v2r1/healthcheck.sh
-deploy/v2r1/install.sh
-deploy/v2r1/validate.sh
-docs/architecture/AIOS_AGENT_AUDIT_PLANE.md
-docs/architecture/AIOS_V1R5_to_V2R1_Migration_Plan.md
-docs/architecture/AIOS_V2R1_OpenClaw_Tenant_Cell_Hermes_Shared_Agent_Fabric_Architecture.md
-openclaw/plugins/aios-tenant-boundary/dist/index.js
-openclaw/plugins/aios-tenant-boundary/dist/policy.js
-openclaw/plugins/aios-tenant-boundary/openclaw.plugin.json
-openclaw/plugins/aios-tenant-boundary/package.json
-release/v2r1c0-rc1/RELEASE.md
-tests/integration/test_runtime_ingress_flow.py
-tests/v2r1/test_acceptance_manifest.md
-tests/v2r1/test_hermes_contract.md
-tests/v2r1/test_hermes_fabric.py
-tests/v2r1/test_message_ledger_append.py
-tests/v2r1/test_router_isolation.py
-tests/v2r1/test_runtime_identity_binding.py
-tests/v2r1/test_runtime_wiring.py
-tests/v2r1/test_tenant_context.py
-```
-
-### Exclude from source — 4 paths
-
-```text
-.agent/results/codex/RESULT-V2R1-202607132000-CODEX.md
-.agent/results/codex/RESULT-V2R1-G1-CODEX-CLEANUP-20260713.md
-patches/release_v2r1c0-tenant-cell-fabric-20260712.branch.json
-release/v2r1c0-rc1/STATUS.md
-```
-
-The task record contains the category-level counts, exact commit exclusions, and reason for every excluded class.
-
-## Strategy and invariants
-
-The approved future strategy is allowlist final-blob replay from source onto exact main. It intentionally avoids merging or cherry-picking historical evidence ancestry. Thirty-seven present paths are restored from source and the tracked `data/webops.sqlite3` path is explicitly deleted.
-
-The future integration must preserve:
-
- exact first parent and rollback target `3cf9958cb94614449e093639cbf902216d73447f`;
- all 16 main-only paths above;
- Gate1 evidence blob `ff31f7ab603d7a7cd07319e6f8422ae464916179`;
- `VERSION` blob `dd4988511ee5957d599098408c89de0c8b758df9` (`V1R4C6`);
- baseline registry blob `81b5bbe59397e01d7fff048a5a91c9636ee85696`;
- no tag, baseline, service, deployment, or force-push mutation.
-
-## Baseline and tag evidence
-
-`V1R4C6` is annotated tag object `356a8a8c1860b94b7d61a456c29228c88e57d963`, peeled release commit `d8cf6058e99f5c7f2dfd2b45a406c0a7edf955b2`. The baseline registry pointer is `e204a90398bf646d46245b3088226749fa056edd`, while the current registry-containing commit is `031215916826f1f0700147242a77878b8514b719`. They are distinct identities. No `V2*` tag exists and P1 creates none.
-
-## Conflict and workspace evidence
-
-Current main contains six tracked conflict-marker lines in two files. Source contains zero, and the two corrected source blobs are in the replay allowlist. Final verification must still prove zero markers.
-
-The canonical workspace had two unrelated untracked controller audit records during discovery. They were not cleaned or included. The P1 review branch uses a separate worktree created from exact main.
-
-## Blockers
-
- Attachment-composition regression is not closed by candidate-bound evidence.
- Node plugin tests are missing even though `npm test` references `test/*.test.js`.
- Plugin versions differ: `2.1.0-rc.1` and `2.1.0-rc1`.
- V2R1 install tooling hardcodes an old release branch.
- RC1 `STATUS.md` is stale and excluded.
- Static scripts do not prove a real OpenClaw runtime probe or rollback recovery.
- Existing release automation masks failures and does not safely update the current baseline schema.
-
-These blockers prevent release, tagging, baseline promotion, staging, and deployment. They do not invalidate the P1 discovery result.
-
-## Disposition
-
-P1 produces evidence only. A separately authorized P2 may implement the exact allowlist replay. P4/P5 must bind tests, security scans, attachment regression, Node plugin verification, real runtime inspection, and rollback evidence to the resulting immutable integration commit before any release action.
@@ -0,0 +1,18 @@
+{
+  "schema": "aios.v2.agent-status.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "executor": "openclaw",
+  "state": "queued",
+  "current_step": "等待外部 OpenClaw 执行器领取",
+  "progress_completed": 0,
+  "progress_total": 8,
+  "candidate_commit": "634f6b164f5b20372ab9ec8144bbf76df97de335",
+  "task_branch_commit": "c6e7edcf54ef48c827a690ce9987284507eb173b",
+  "claimed_at": null,
+  "heartbeat_at": null,
+  "last_commit": null,
+  "test_status": "not_run",
+  "blockers": [],
+  "result": null,
+  "updated_at": "2026-07-20T00:00:00Z"
+}
@@ -0,0 +1,31 @@
+{
+  "schema": "aios.v2.agent-task.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "display_name": "AIOS V2R1 OpenClaw 预发布部署与小型缺陷修复任务",
+  "language": "zh-CN",
+  "executor": "openclaw",
+  "role": "deployment-operator",
+  "state": "queued",
+  "candidate_branch": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "candidate_commit": "634f6b164f5b20372ab9ec8144bbf76df97de335",
+  "canonical_task": "tasks/openclaw-production-deploy-task.yaml",
+  "supervisor_task": "task/codex/AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "claim_required": true,
+  "heartbeat_interval_seconds": 300,
+  "evidence_directory": ".agent/evidence",
+  "instructions": [
+    "领取前读取候选分支完整中文文档和生产候选报告",
+    "领取后写入 .agent/claim.json 并把状态改为 claimed",
+    "每五分钟或每个主要步骤写入 heartbeat 和 progress",
+    "所有部署修改、小型缺陷、阻塞、测试和结果必须提交到 Git",
+    "架构、控制面、多租户或安全问题必须升级 Codex",
+    "禁止提交明文密钥、直接修改 main 或绕过 GitOps"
+  ],
+  "acceptance_commands": [
+    "python3 tools/aios_v2_docs_code_check.py --base-ref integration/aios-v2r1-cto-production-candidate-20260720",
+    "python3 -m unittest discover -s tests/aios_v2_candidate -p test_*.py -v",
+    "bash deploy/aios-v2/acceptance/run-all.sh",
+    "bash deploy/aios-v2/healthcheck/check.sh"
+  ],
+  "created_at": "2026-07-20T00:00:00Z"
+}
@@ -1,185 +0,0 @@
---
-authorization_id: TASK-V2R1-RUNTIME-INTEGRATION-P1-20260714
-task_id: TASK-V2R1-RUNTIME-INTEGRATION-P1-20260714
-executor: GPT_CODEX
-status: DISCOVERY_COMPLETE
-started_at: 2026-07-14T02:24:00+08:00
-finished_at: 2026-07-14T02:27:46+08:00
-repository: /data/aios-gitops/workspace
-review_branch: review/codex/v2r1-runtime-integration-discovery-20260714
-review_worktree: /data/aios-gitops/worktrees/v2r1-runtime-integration-discovery-20260714
-main_commit: 3cf9958cb94614449e093639cbf902216d73447f
-source_commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
-gate1_commit: 7f5d5d90b3111c35fb00044f385358f528e3344e
-merge_base: 1a9cb9095e6e76c548c967e03cb7a3e47b85134b
-result_path: .agent/results/codex/RESULT-V2R1-RUNTIME-INTEGRATION-P1-20260714.md
-write_scope:
-  - .agent/tasks/codex/TASK-V2R1-RUNTIME-INTEGRATION-P1-20260714.md
-  - .agent/results/codex/RESULT-V2R1-RUNTIME-INTEGRATION-P1-20260714.md
-forbidden_actions:
-  - modify_runtime_or_tests
-  - merge_or_push_main
-  - create_or_move_tag
-  - update_baseline
-  - change_service_or_deploy
---
-
-# V2R1 Runtime Integration P1 Discovery Task
-
-## Objective
-
-Produce a reproducible, evidence-only discovery record for integrating the approved V2R1 source tree onto current main without importing divergent historical evidence ancestry. P1 does not implement the integration.
-
-## Locked graph
-
- Authoritative main: `3cf9958cb94614449e093639cbf902216d73447f`.
- Approved V2R1 source: `502ef30642597fb0b70b90ae2e8a0b6af217d80e`.
- Accepted Gate1 evidence commit: `7f5d5d90b3111c35fb00044f385358f528e3344e`.
- Merge base: `1a9cb9095e6e76c548c967e03cb7a3e47b85134b`.
- Main-only graph: `21` commits and `16` changed paths.
- Source-only graph: `44` commits and `42` changed paths.
- Main/source changed-path overlap: `0` paths.
-
-## Main paths that must be preserved exactly — 16
-
-1. `.agent/results/openclaw/RESULT-V2R1-G1-OPENCLAW-FINAL-EVIDENCE-20260714.md`
-2. `deployments/V1R4C6/backup-result.md`
-3. `deployments/V1R4C6/deployment-result.md`
-4. `deployments/V1R4C6/preflight-result.md`
-5. `deployments/V1R4C6/rollback-result.md`
-6. `deployments/V1R4C6/verification-result.md`
-7. `docs/incidents/WEIXIN_TEXT_IMAGE_CACHE_ISOLATION_20260711.md`
-8. `docs/superpowers/plans/2026-07-12-weixin-quoted-text-failclosed.md`
-9. `docs/superpowers/specs/2026-07-12-weixin-quoted-text-failclosed-design.md`
-10. `fixes/patch_openclaw_weixin_bigint.py`
-11. `fixes/patch_openclaw_weixin_hermes.py`
-12. `fixes/patch_openclaw_weixin_outbound_quotes.py`
-13. `patches/deploy_v1r4c6-production-20260711.branch.json`
-14. `tests/test_weixin_bigint_parser.py`
-15. `tests/test_weixin_image_followup_intent.py`
-16. `tests/test_weixin_outbound_quote_timestamp.py`
-
-Additional immutable main invariants:
-
- `VERSION` remains `V1R4C6`, blob `dd4988511ee5957d599098408c89de0c8b758df9`.
- `baseline/current.json` remains blob `81b5bbe59397e01d7fff048a5a91c9636ee85696`.
- Accepted Gate1 evidence remains blob `ff31f7ab603d7a7cd07319e6f8422ae464916179`.
- The future integration commit must have `3cf9958cb94614449e093639cbf902216d73447f` as its first-parent/rollback target.
-
-## Approved source-tree allowlist — 38 paths
-
-### Python V2R1 runtime — 15
-
-1. `active-runtime/app/aios_v2r1/audit.py`
-2. `active-runtime/app/aios_v2r1/cell_manager.py`
-3. `active-runtime/app/aios_v2r1/channel_adapter.py`
-4. `active-runtime/app/aios_v2r1/cli.py`
-5. `active-runtime/app/aios_v2r1/fleet.py`
-6. `active-runtime/app/aios_v2r1/hermes_fabric.py`
-7. `active-runtime/app/aios_v2r1/hermes_profile.py`
-8. `active-runtime/app/aios_v2r1/identity.py`
-9. `active-runtime/app/aios_v2r1/memory_scope.py`
-10. `active-runtime/app/aios_v2r1/message_ledger.py`
-11. `active-runtime/app/aios_v2r1/router.py`
-12. `active-runtime/app/aios_v2r1/runtime.py`
-13. `active-runtime/app/aios_v2r1/secret_scope.py`
-14. `active-runtime/app/aios_v2r1/session_pool.py`
-15. `active-runtime/app/aios_v2r1/tenant_context.py`
-
-### OpenClaw plugin — 4
-
-1. `openclaw/plugins/aios-tenant-boundary/dist/index.js`
-2. `openclaw/plugins/aios-tenant-boundary/dist/policy.js`
-3. `openclaw/plugins/aios-tenant-boundary/openclaw.plugin.json`
-4. `openclaw/plugins/aios-tenant-boundary/package.json`
-
-### Tests and contracts — 9
-
-1. `tests/integration/test_runtime_ingress_flow.py`
-2. `tests/v2r1/test_acceptance_manifest.md`
-3. `tests/v2r1/test_hermes_contract.md`
-4. `tests/v2r1/test_hermes_fabric.py`
-5. `tests/v2r1/test_message_ledger_append.py`
-6. `tests/v2r1/test_router_isolation.py`
-7. `tests/v2r1/test_runtime_identity_binding.py`
-8. `tests/v2r1/test_runtime_wiring.py`
-9. `tests/v2r1/test_tenant_context.py`
-
-### Static deployment tooling — 3
-
-1. `deploy/v2r1/healthcheck.sh`
-2. `deploy/v2r1/install.sh`
-3. `deploy/v2r1/validate.sh`
-
-### Architecture and release documents — 4
-
-1. `docs/architecture/AIOS_AGENT_AUDIT_PLANE.md`
-2. `docs/architecture/AIOS_V1R5_to_V2R1_Migration_Plan.md`
-3. `docs/architecture/AIOS_V2R1_OpenClaw_Tenant_Cell_Hermes_Shared_Agent_Fabric_Architecture.md`
-4. `release/v2r1c0-rc1/RELEASE.md`
-
-### Cleanup and governance — 3
-
-1. `.gitignore`
-2. `agents/registry.yaml`
-3. `data/webops.sqlite3` — explicit tracked-file deletion, not a restore operation.
-
-The replay operation restores the `37` present paths from source and explicitly removes `data/webops.sqlite3`. No directory-wide restore is allowed outside this list.
-
-## Excluded source paths — 4
-
-1. `.agent/results/codex/RESULT-V2R1-202607132000-CODEX.md` — historical branch evidence.
-2. `.agent/results/codex/RESULT-V2R1-G1-CODEX-CLEANUP-20260713.md` — historical branch evidence.
-3. `patches/release_v2r1c0-tenant-cell-fabric-20260712.branch.json` — stale branch-specific patch metadata.
-4. `release/v2r1c0-rc1/STATUS.md` — stale status: references tree `6569065` and contradicts later test/rollback evidence.
-
-## Excluded commit ancestry
-
- Do not merge source commit ancestry and do not cherry-pick any of its `44` source-only commits. Replay only the approved final blobs.
- Do not merge Gate1 commit `7f5d5d90b3111c35fb00044f385358f528e3344e`; main already contains its accepted evidence blob.
- Do not replay result-only source commits `d6ba7e8fa859407ec55acf49b5468bba5eac96da`, `b9e1f9db66eb08282b7b690199e56461d28dd76d`, or `502ef30642597fb0b70b90ae2e8a0b6af217d80e` as commits.
- Explicitly exclude rejected OpenClaw candidates `7a2f16ee72e944737a9f6995e511ef176413ae6b`, `b1d3439ee9e66b5eed1c94595a22bc630d893b6e`, `38320044b54c172c28877a81f291478267dd7b9e`, `7fd63b55f27b23658ea26e2a39d949c7b1e91dda`, and `71239b4d187e683e05fd22fc9d5df8ae324e3390`.
- Prior Codex final-verification commit `0eebcf0b72f06c02205fdb6e300ceb5b3a5647a7` is not an integration input.
-
-## Strategy decision
-
-Use allowlist file replay from the immutable source tree, not a historical merge or commit-range cherry-pick. The two branches diverged for `21` and `44` commits after the merge base. Historical ancestry includes obsolete result reports and branch metadata, while main contains V1R4C6 incident fixes and the already accepted Gate1 evidence. Final-blob replay makes the intended runtime delta explicit and keeps those histories out of the future integration commit.
-
-P2 must start from exact main, revalidate all refs, restore only the 37 present allowlist paths from source, remove the one tracked SQLite path, and stop if any additional path appears.
-
-## Conflict preview
-
- Current main has `6` tracked conflict-marker lines across `agents/registry.yaml` and `docs/architecture/AIOS_AGENT_AUDIT_PLANE.md`.
- Source has `0` tracked conflict-marker lines.
- Both corrected source blobs are in the allowlist; the final tree must scan to zero markers.
- The changed-path overlap between main and source deltas is zero, but this does not waive blob-preservation checks.
-
-## Baseline and tag distinction
-
- Annotated tag object `356a8a8c1860b94b7d61a456c29228c88e57d963` is `V1R4C6` and peels to release commit `d8cf6058e99f5c7f2dfd2b45a406c0a7edf955b2`.
- `baseline/current.json` records baseline registry pointer `e204a90398bf646d46245b3088226749fa056edd`; its latest modifying commit is `031215916826f1f0700147242a77878b8514b719`.
- The tag object, peeled release commit, registry pointer, and registry-containing commit are separate audit identities.
- No V2 tag exists. Future approved namespace is `refs/tags/V2R1C0`; P1 does not create it.
-
-## Current workspace pollution
-
-At discovery time, canonical workspace `/data/aios-gitops/workspace` contained two unrelated untracked controller records:
-
- `audit/controller-runs/c-1783966122.json`
- `audit/controller-runs/c-1783966722.json`
-
-They are excluded. All P1 writes occur in the isolated worktree, and no canonical workspace cleanup is authorized.
-
-## Known blockers
-
-1. Attachment-composition regression lacks a closed, candidate-bound regression result.
-2. Plugin `npm test` targets `test/*.test.js`, but source tracks no Node test files.
-3. Plugin package version `2.1.0-rc.1` differs from manifest version `2.1.0-rc1`.
-4. `deploy/v2r1/install.sh` hardcodes release branch `release/v2r1c0-tenant-cell-fabric-20260712`.
-5. `release/v2r1c0-rc1/STATUS.md` is stale and is excluded rather than silently rewritten.
-6. `validate.sh --install/--rollback` and `healthcheck.sh` do not prove a real OpenClaw runtime or rollback path.
-7. `scripts/release_version.sh` masks failures and does not match the active baseline registry schema; it must not be used.
-
-## Exit criteria
-
-P1 passes when the two evidence files are the only commit delta, the commit parent is exact main, the isolated worktree is clean after commit, and the review branch is pushed without force. Runtime integration, main update, tag, baseline promotion, and deployment remain separate blocked phases.
@@ -0,0 +1,25 @@
+"""AIOS-managed Hermes runtime boundary."""
+
+from .adapter import (
+    HermesAdapter,
+    HermesConflict,
+    HermesNotFound,
+    HermesRuntimeError,
+    InMemoryHermesStore,
+)
+from .contract import HermesHealth, HermesJob, HermesJobState, HermesResult
+from .worker import HermesWorker, WorkerRunResult
+
+__all__ = [
+    "HermesAdapter",
+    "HermesConflict",
+    "HermesHealth",
+    "HermesJob",
+    "HermesJobState",
+    "HermesNotFound",
+    "HermesResult",
+    "HermesRuntimeError",
+    "HermesWorker",
+    "InMemoryHermesStore",
+    "WorkerRunResult",
+]
@@ -0,0 +1,437 @@
+"""Tenant-partitioned Hermes queue and worker lease boundary."""
+from __future__ import annotations
+
+import hashlib
+import json
+import threading
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+from typing import Iterable, Mapping
+
+from .contract import (
+    ALLOWED_TRANSITIONS,
+    TERMINAL_STATES,
+    HermesHealth,
+    HermesJob,
+    HermesJobState,
+    HermesResult,
+    utc_now,
+)
+
+
+class HermesRuntimeError(RuntimeError):
+    """Base Hermes runtime-boundary failure."""
+
+
+class HermesNotFound(HermesRuntimeError):
+    """A tenant-scoped job was not found."""
+
+
+class HermesConflict(HermesRuntimeError):
+    """A state, lease, idempotency, or revision conflict occurred."""
+
+
+class InMemoryHermesStore:
+    """Thread-safe reference store; production backends implement this contract."""
+
+    def __init__(self) -> None:
+        self.lock = threading.RLock()
+        self.jobs: dict[str, HermesJob] = {}
+        self.results: dict[str, HermesResult] = {}
+        self.idempotency: dict[tuple[str, str], tuple[str, str]] = {}
+        self.worker_heartbeats: dict[str, tuple[str, str | None]] = {}
+
+
+class HermesAdapter:
+    """Queue and reconcile Hermes jobs without owning AIOS policy."""
+
+    def __init__(
+        self,
+        store: InMemoryHermesStore | None = None,
+        *,
+        runtime_id: str = "hermes-runtime",
+        runtime_version: str = "v1",
+        default_lease_seconds: int = 60,
+    ) -> None:
+        if default_lease_seconds < 5:
+            raise ValueError("default_lease_seconds must be >= 5")
+        self.store = store or InMemoryHermesStore()
+        self.runtime_id = runtime_id
+        self.runtime_version = runtime_version
+        self.default_lease_seconds = int(default_lease_seconds)
+
+    @staticmethod
+    def _digest(job: HermesJob) -> str:
+        raw = json.dumps(
+            {
+                "tenant_id": job.tenant_id,
+                "cell_id": job.cell_id,
+                "profile_id": job.profile_id,
+                "task_id": job.task_id,
+                "trace_id": job.trace_id,
+                "action": job.action,
+                "payload": dict(job.payload),
+                "priority": job.priority,
+            },
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        ).encode("utf-8")
+        return hashlib.sha256(raw).hexdigest()
+
+    @staticmethod
+    def _parse(value: str | None) -> datetime | None:
+        if value is None:
+            return None
+        normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
+        parsed = datetime.fromisoformat(normalized)
+        if parsed.tzinfo is None:
+            raise ValueError("timestamp must include timezone")
+        return parsed.astimezone(timezone.utc)
+
+    @staticmethod
+    def _format(value: datetime) -> str:
+        return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+
+    @staticmethod
+    def _assert_tenant(job: HermesJob, tenant_id: str) -> None:
+        if job.tenant_id != tenant_id:
+            raise HermesNotFound("job not found")
+
+    def enqueue(self, job: HermesJob) -> HermesJob:
+        if job.state != HermesJobState.QUEUED or job.revision != 1:
+            raise HermesConflict("new job must be QUEUED at revision 1")
+        digest = self._digest(job)
+        idem_ref = (job.tenant_id, job.idempotency_key)
+        with self.store.lock:
+            existing = self.store.jobs.get(job.job_id)
+            if existing is not None:
+                if self._digest(existing) != digest:
+                    raise HermesConflict("job_id reused with different content")
+                return existing
+            prior = self.store.idempotency.get(idem_ref)
+            if prior is not None:
+                prior_id, prior_digest = prior
+                if prior_digest != digest:
+                    raise HermesConflict("idempotency key reused with different content")
+                return self.store.jobs[prior_id]
+            self.store.jobs[job.job_id] = job
+            self.store.idempotency[idem_ref] = (job.job_id, digest)
+            return job
+
+    def get(self, job_id: str, *, tenant_id: str) -> HermesJob:
+        with self.store.lock:
+            job = self.store.jobs.get(job_id)
+            if job is None:
+                raise HermesNotFound("job not found")
+            self._assert_tenant(job, tenant_id)
+            return job
+
+    def list_jobs(
+        self,
+        *,
+        tenant_id: str,
+        cell_id: str | None = None,
+        profile_id: str | None = None,
+        states: Iterable[HermesJobState | str] | None = None,
+    ) -> list[HermesJob]:
+        state_filter = {HermesJobState(item) for item in states} if states else None
+        with self.store.lock:
+            self._requeue_expired_locked()
+            jobs = [
+                job
+                for job in self.store.jobs.values()
+                if job.tenant_id == tenant_id
+                and (cell_id is None or job.cell_id == cell_id)
+                and (profile_id is None or job.profile_id == profile_id)
+                and (state_filter is None or job.state in state_filter)
+            ]
+        return sorted(jobs, key=lambda item: (item.priority, item.created_at, item.job_id))
+
+    def claim(
+        self,
+        *,
+        tenant_id: str,
+        cell_id: str,
+        profile_id: str,
+        worker_id: str,
+        lease_seconds: int | None = None,
+    ) -> HermesJob | None:
+        worker_id = str(worker_id or "").strip()
+        if not worker_id:
+            raise ValueError("worker_id is required")
+        duration = int(lease_seconds or self.default_lease_seconds)
+        if duration < 5:
+            raise ValueError("lease_seconds must be >= 5")
+        with self.store.lock:
+            self._requeue_expired_locked()
+            candidates = [
+                job
+                for job in self.store.jobs.values()
+                if job.tenant_id == tenant_id
+                and job.cell_id == cell_id
+                and job.profile_id == profile_id
+                and job.state == HermesJobState.QUEUED
+            ]
+            self.store.worker_heartbeats[worker_id] = (utc_now(), None)
+            if not candidates:
+                return None
+            current = sorted(
+                candidates,
+                key=lambda item: (item.priority, item.created_at, item.job_id),
+            )[0]
+            claimed = self._transition_locked(
+                current,
+                HermesJobState.CLAIMED,
+                assigned_worker_id=worker_id,
+                lease_expires_at=self._format(
+                    datetime.now(timezone.utc) + timedelta(seconds=duration)
+                ),
+            )
+            self.store.worker_heartbeats[worker_id] = (utc_now(), claimed.job_id)
+            return claimed
+
+    def start(
+        self,
+        job_id: str,
+        *,
+        tenant_id: str,
+        worker_id: str,
+        expected_revision: int | None = None,
+    ) -> HermesJob:
+        with self.store.lock:
+            current = self._owned_job_locked(job_id, tenant_id, worker_id)
+            self._assert_revision(current, expected_revision)
+            return self._transition_locked(current, HermesJobState.RUNNING)
+
+    def heartbeat(
+        self,
+        job_id: str,
+        *,
+        tenant_id: str,
+        worker_id: str,
+        lease_seconds: int | None = None,
+    ) -> HermesJob:
+        duration = int(lease_seconds or self.default_lease_seconds)
+        with self.store.lock:
+            current = self._owned_job_locked(job_id, tenant_id, worker_id)
+            if current.state not in {HermesJobState.CLAIMED, HermesJobState.RUNNING}:
+                raise HermesConflict("job does not accept heartbeat")
+            updated = replace(
+                current,
+                revision=current.revision + 1,
+                lease_expires_at=self._format(
+                    datetime.now(timezone.utc) + timedelta(seconds=duration)
+                ),
+                updated_at=utc_now(),
+            )
+            self.store.jobs[job_id] = updated
+            self.store.worker_heartbeats[worker_id] = (utc_now(), job_id)
+            return updated
+
+    def complete(
+        self,
+        job_id: str,
+        *,
+        tenant_id: str,
+        worker_id: str,
+        output: Mapping[str, object],
+        expected_revision: int | None = None,
+    ) -> HermesResult:
+        return self._finish(
+            job_id,
+            tenant_id=tenant_id,
+            worker_id=worker_id,
+            state=HermesJobState.COMPLETED,
+            output=output,
+            expected_revision=expected_revision,
+        )
+
+    def fail(
+        self,
+        job_id: str,
+        *,
+        tenant_id: str,
+        worker_id: str,
+        error: Mapping[str, object],
+        expected_revision: int | None = None,
+    ) -> HermesResult:
+        return self._finish(
+            job_id,
+            tenant_id=tenant_id,
+            worker_id=worker_id,
+            state=HermesJobState.FAILED,
+            error=error,
+            expected_revision=expected_revision,
+        )
+
+    def cancel(
+        self,
+        job_id: str,
+        *,
+        tenant_id: str,
+        expected_revision: int | None = None,
+    ) -> HermesResult:
+        with self.store.lock:
+            current = self.store.jobs.get(job_id)
+            if current is None:
+                raise HermesNotFound("job not found")
+            self._assert_tenant(current, tenant_id)
+            self._assert_revision(current, expected_revision)
+            if current.state in TERMINAL_STATES:
+                result = self.store.results.get(job_id)
+                if result is None:
+                    raise HermesConflict("terminal result missing")
+                return result
+            updated = self._transition_locked(current, HermesJobState.CANCELLED)
+            result = HermesResult(
+                job_id=updated.job_id,
+                tenant_id=updated.tenant_id,
+                task_id=updated.task_id,
+                trace_id=updated.trace_id,
+                state=HermesJobState.CANCELLED,
+                worker_id=updated.assigned_worker_id,
+                error={"code": "cancelled", "message": "cancelled by AIOS"},
+                finished_at=utc_now(),
+            )
+            self.store.results[job_id] = result
+            return result
+
+    def result(self, job_id: str, *, tenant_id: str) -> HermesResult:
+        with self.store.lock:
+            job = self.store.jobs.get(job_id)
+            if job is None:
+                raise HermesNotFound("job not found")
+            self._assert_tenant(job, tenant_id)
+            result = self.store.results.get(job_id)
+            if result is None:
+                raise HermesConflict("job has no terminal result")
+            return result
+
+    def health(self) -> HermesHealth:
+        with self.store.lock:
+            self._requeue_expired_locked()
+            active_jobs = sum(
+                1 for job in self.store.jobs.values()
+                if job.state in {HermesJobState.CLAIMED, HermesJobState.RUNNING}
+            )
+            queued_jobs = sum(
+                1 for job in self.store.jobs.values()
+                if job.state == HermesJobState.QUEUED
+            )
+            partitions = {
+                (job.tenant_id, job.cell_id, job.profile_id)
+                for job in self.store.jobs.values()
+            }
+            active_workers = sum(
+                1 for heartbeat, _ in self.store.worker_heartbeats.values()
+                if self._parse(heartbeat)
+                and self._parse(heartbeat) >= datetime.now(timezone.utc) - timedelta(seconds=120)
+            )
+        return HermesHealth(
+            runtime_id=self.runtime_id,
+            status="healthy",
+            version=self.runtime_version,
+            heartbeat_at=utc_now(),
+            active_workers=active_workers,
+            active_jobs=active_jobs,
+            queued_jobs=queued_jobs,
+            tenant_partitions=len(partitions),
+            profile_isolation_enforced=True,
+            details={"store": type(self.store).__name__},
+        )
+
+    def _finish(
+        self,
+        job_id: str,
+        *,
+        tenant_id: str,
+        worker_id: str,
+        state: HermesJobState,
+        output: Mapping[str, object] | None = None,
+        error: Mapping[str, object] | None = None,
+        expected_revision: int | None,
+    ) -> HermesResult:
+        with self.store.lock:
+            current = self._owned_job_locked(job_id, tenant_id, worker_id)
+            self._assert_revision(current, expected_revision)
+            updated = self._transition_locked(current, state)
+            result = HermesResult(
+                job_id=updated.job_id,
+                tenant_id=updated.tenant_id,
+                task_id=updated.task_id,
+                trace_id=updated.trace_id,
+                state=state,
+                worker_id=worker_id,
+                output=output,
+                error=error,
+                started_at=current.updated_at,
+                finished_at=utc_now(),
+            )
+            self.store.results[job_id] = result
+            self.store.worker_heartbeats[worker_id] = (utc_now(), None)
+            return result
+
+    def _owned_job_locked(self, job_id: str, tenant_id: str, worker_id: str) -> HermesJob:
+        current = self.store.jobs.get(job_id)
+        if current is None:
+            raise HermesNotFound("job not found")
+        self._assert_tenant(current, tenant_id)
+        if current.assigned_worker_id != worker_id:
+            raise HermesConflict("worker does not own job lease")
+        return current
+
+    @staticmethod
+    def _assert_revision(job: HermesJob, expected: int | None) -> None:
+        if expected is not None and job.revision != expected:
+            raise HermesConflict("job revision changed")
+
+    def _transition_locked(
+        self,
+        current: HermesJob,
+        target: HermesJobState,
+        *,
+        assigned_worker_id: str | None = None,
+        lease_expires_at: str | None = None,
+    ) -> HermesJob:
+        if target not in ALLOWED_TRANSITIONS[current.state]:
+            raise HermesConflict(
+                f"invalid transition {current.state.value} -> {target.value}"
+            )
+        updated = replace(
+            current,
+            state=target,
+            revision=current.revision + 1,
+            assigned_worker_id=(
+                assigned_worker_id
+                if assigned_worker_id is not None
+                else current.assigned_worker_id
+            ),
+            lease_expires_at=(
+                lease_expires_at
+                if lease_expires_at is not None
+                else (None if target in TERMINAL_STATES else current.lease_expires_at)
+            ),
+            updated_at=utc_now(),
+        )
+        self.store.jobs[current.job_id] = updated
+        return updated
+
+    def _requeue_expired_locked(self) -> None:
+        now = datetime.now(timezone.utc)
+        for current in list(self.store.jobs.values()):
+            expiry = self._parse(current.lease_expires_at)
+            if (
+                current.state in {HermesJobState.CLAIMED, HermesJobState.RUNNING}
+                and expiry is not None
+                and expiry <= now
+            ):
+                self.store.jobs[current.job_id] = replace(
+                    current,
+                    state=HermesJobState.QUEUED,
+                    revision=current.revision + 1,
+                    assigned_worker_id=None,
+                    lease_expires_at=None,
+                    updated_at=utc_now(),
+                )
@@ -0,0 +1,153 @@
+"""Tenant-scoped Hermes runtime contracts."""
+from __future__ import annotations
+
+import json
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Mapping
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _required(value: str, label: str) -> str:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise ValueError(f"{label} is required")
+    return normalized
+
+
+def _safe_mapping(value: Mapping[str, Any] | None) -> dict[str, Any]:
+    return json.loads(json.dumps(dict(value or {}), ensure_ascii=False, sort_keys=True, default=str))
+
+
+class HermesJobState(str, Enum):
+    QUEUED = "QUEUED"
+    CLAIMED = "CLAIMED"
+    RUNNING = "RUNNING"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    CANCELLED = "CANCELLED"
+
+
+ALLOWED_TRANSITIONS: Mapping[HermesJobState, frozenset[HermesJobState]] = {
+    HermesJobState.QUEUED: frozenset({HermesJobState.CLAIMED, HermesJobState.CANCELLED}),
+    HermesJobState.CLAIMED: frozenset({HermesJobState.RUNNING, HermesJobState.CANCELLED}),
+    HermesJobState.RUNNING: frozenset(
+        {HermesJobState.COMPLETED, HermesJobState.FAILED, HermesJobState.CANCELLED}
+    ),
+    HermesJobState.COMPLETED: frozenset(),
+    HermesJobState.FAILED: frozenset(),
+    HermesJobState.CANCELLED: frozenset(),
+}
+TERMINAL_STATES = frozenset(
+    {HermesJobState.COMPLETED, HermesJobState.FAILED, HermesJobState.CANCELLED}
+)
+
+
+@dataclass(frozen=True)
+class HermesJob:
+    job_id: str
+    tenant_id: str
+    cell_id: str
+    profile_id: str
+    task_id: str
+    trace_id: str
+    action: str
+    payload: Mapping[str, Any] = field(default_factory=dict)
+    idempotency_key: str = ""
+    priority: int = 100
+    state: HermesJobState = HermesJobState.QUEUED
+    revision: int = 1
+    assigned_worker_id: str | None = None
+    lease_expires_at: str | None = None
+    created_at: str = field(default_factory=utc_now)
+    updated_at: str = field(default_factory=utc_now)
+
+    def __post_init__(self) -> None:
+        for name in (
+            "job_id", "tenant_id", "cell_id", "profile_id", "task_id",
+            "trace_id", "action", "idempotency_key", "created_at", "updated_at",
+        ):
+            object.__setattr__(self, name, _required(getattr(self, name), name))
+        object.__setattr__(self, "payload", _safe_mapping(self.payload))
+        object.__setattr__(self, "state", HermesJobState(self.state))
+        if not 0 <= int(self.priority) <= 1000:
+            raise ValueError("priority must be between 0 and 1000")
+        if self.revision < 1:
+            raise ValueError("revision must be >= 1")
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        tenant_id: str,
+        cell_id: str,
+        profile_id: str,
+        task_id: str,
+        trace_id: str,
+        action: str,
+        payload: Mapping[str, Any] | None,
+        idempotency_key: str,
+        priority: int = 100,
+    ) -> "HermesJob":
+        return cls(
+            job_id=f"hermes_{uuid.uuid4().hex}",
+            tenant_id=tenant_id,
+            cell_id=cell_id,
+            profile_id=profile_id,
+            task_id=task_id,
+            trace_id=trace_id,
+            action=action,
+            payload=_safe_mapping(payload),
+            idempotency_key=idempotency_key,
+            priority=priority,
+        )
+
+
+@dataclass(frozen=True)
+class HermesResult:
+    job_id: str
+    tenant_id: str
+    task_id: str
+    trace_id: str
+    state: HermesJobState
+    worker_id: str | None = None
+    output: Mapping[str, Any] | None = None
+    error: Mapping[str, Any] | None = None
+    started_at: str | None = None
+    finished_at: str | None = None
+
+    def __post_init__(self) -> None:
+        for name in ("job_id", "tenant_id", "task_id", "trace_id"):
+            object.__setattr__(self, name, _required(getattr(self, name), name))
+        object.__setattr__(self, "state", HermesJobState(self.state))
+        if self.output is not None:
+            object.__setattr__(self, "output", _safe_mapping(self.output))
+        if self.error is not None:
+            object.__setattr__(self, "error", _safe_mapping(self.error))
+
+
+@dataclass(frozen=True)
+class HermesHealth:
+    runtime_id: str
+    status: str
+    version: str
+    heartbeat_at: str
+    active_workers: int = 0
+    active_jobs: int = 0
+    queued_jobs: int = 0
+    tenant_partitions: int = 0
+    profile_isolation_enforced: bool = True
+    details: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        for name in ("runtime_id", "status", "version", "heartbeat_at"):
+            object.__setattr__(self, name, _required(getattr(self, name), name))
+        for name in ("active_workers", "active_jobs", "queued_jobs", "tenant_partitions"):
+            if int(getattr(self, name)) < 0:
+                raise ValueError(f"{name} must be >= 0")
+        object.__setattr__(self, "details", _safe_mapping(self.details))
@@ -0,0 +1,155 @@
+"""Hermes worker loop with lease renewal and bounded recovery."""
+from __future__ import annotations
+
+import threading
+from dataclasses import dataclass
+from typing import Any, Callable, Mapping
+
+from .adapter import HermesAdapter, HermesConflict
+from .contract import HermesJob, HermesJobState
+
+
+JobHandler = Callable[[HermesJob], Mapping[str, Any]]
+
+
+@dataclass(frozen=True)
+class WorkerRunResult:
+    claimed: bool
+    job_id: str | None = None
+    state: str | None = None
+    error: str | None = None
+
+
+class HermesWorker:
+    """Poll one tenant/cell/profile partition and execute jobs safely."""
+
+    def __init__(
+        self,
+        adapter: HermesAdapter,
+        handler: JobHandler,
+        *,
+        worker_id: str,
+        tenant_id: str,
+        cell_id: str,
+        profile_id: str,
+        poll_seconds: float = 1.0,
+        heartbeat_seconds: float = 10.0,
+    ) -> None:
+        if poll_seconds <= 0 or heartbeat_seconds <= 0:
+            raise ValueError("poll_seconds and heartbeat_seconds must be positive")
+        for name, value in {
+            "worker_id": worker_id,
+            "tenant_id": tenant_id,
+            "cell_id": cell_id,
+            "profile_id": profile_id,
+        }.items():
+            if not str(value or "").strip():
+                raise ValueError(f"{name} is required")
+        self.adapter = adapter
+        self.handler = handler
+        self.worker_id = worker_id
+        self.tenant_id = tenant_id
+        self.cell_id = cell_id
+        self.profile_id = profile_id
+        self.poll_seconds = float(poll_seconds)
+        self.heartbeat_seconds = float(heartbeat_seconds)
+        self._stop = threading.Event()
+
+    def stop(self) -> None:
+        self._stop.set()
+
+    def run_once(self) -> WorkerRunResult:
+        job = self.adapter.claim(
+            tenant_id=self.tenant_id,
+            cell_id=self.cell_id,
+            profile_id=self.profile_id,
+            worker_id=self.worker_id,
+        )
+        if job is None:
+            return WorkerRunResult(claimed=False)
+
+        running = self.adapter.start(
+            job.job_id,
+            tenant_id=self.tenant_id,
+            worker_id=self.worker_id,
+            expected_revision=job.revision,
+        )
+
+        heartbeat_stop = threading.Event()
+        heartbeat_thread = threading.Thread(
+            target=self._heartbeat_loop,
+            args=(running.job_id, heartbeat_stop),
+            name=f"hermes-heartbeat-{self.worker_id}",
+            daemon=True,
+        )
+        heartbeat_thread.start()
+
+        output: Mapping[str, Any] | None = None
+        execution_error: Exception | None = None
+        try:
+            output = dict(self.handler(running))
+        except Exception as exc:  # runtime boundary: normalize below
+            execution_error = exc
+        finally:
+            # Stop the lease heartbeat before reading the final revision. This
+            # prevents a revision increment racing with complete/fail.
+            heartbeat_stop.set()
+            heartbeat_thread.join(timeout=max(self.heartbeat_seconds * 2, 1.0))
+
+        try:
+            current = self.adapter.get(running.job_id, tenant_id=self.tenant_id)
+            if execution_error is None:
+                result = self.adapter.complete(
+                    running.job_id,
+                    tenant_id=self.tenant_id,
+                    worker_id=self.worker_id,
+                    output=dict(output or {}),
+                    expected_revision=current.revision,
+                )
+                return WorkerRunResult(
+                    claimed=True,
+                    job_id=result.job_id,
+                    state=result.state.value,
+                )
+
+            result = self.adapter.fail(
+                running.job_id,
+                tenant_id=self.tenant_id,
+                worker_id=self.worker_id,
+                error={
+                    "code": "worker_error",
+                    "message": str(execution_error),
+                },
+                expected_revision=current.revision,
+            )
+            return WorkerRunResult(
+                claimed=True,
+                job_id=result.job_id,
+                state=result.state.value,
+                error=str(execution_error),
+            )
+        except HermesConflict as exc:
+            return WorkerRunResult(
+                claimed=True,
+                job_id=running.job_id,
+                state=HermesJobState.FAILED.value,
+                error=f"reconciliation_conflict: {exc}",
+            )
+
+    def run_forever(self) -> None:
+        while not self._stop.is_set():
+            result = self.run_once()
+            if not result.claimed:
+                self._stop.wait(self.poll_seconds)
+
+    def _heartbeat_loop(self, job_id: str, stop_event: threading.Event) -> None:
+        while not stop_event.wait(self.heartbeat_seconds):
+            try:
+                self.adapter.heartbeat(
+                    job_id,
+                    tenant_id=self.tenant_id,
+                    worker_id=self.worker_id,
+                )
+            except Exception:
+                stop_event.set()
+                return
@@ -0,0 +1,37 @@
+"""AIOS-managed OpenClaw execution adapter."""
+
+from .adapter import (
+    AdapterError,
+    CommandConflict,
+    CommandNotFound,
+    CommandRejected,
+    InMemoryCallbackSink,
+    OpenClawAdapter,
+)
+from .contract import (
+    CommandState,
+    OpenClawCommand,
+    OpenClawResult,
+    RuntimeHealth,
+    sign_command,
+    verify_command_signature,
+)
+from .transport import HTTPRuntimeTransport, InProcessRuntimeTransport, RuntimeTransport
+
+__all__ = [
+    "AdapterError",
+    "CommandConflict",
+    "CommandNotFound",
+    "CommandRejected",
+    "CommandState",
+    "HTTPRuntimeTransport",
+    "InMemoryCallbackSink",
+    "InProcessRuntimeTransport",
+    "OpenClawAdapter",
+    "OpenClawCommand",
+    "OpenClawResult",
+    "RuntimeHealth",
+    "RuntimeTransport",
+    "sign_command",
+    "verify_command_signature",
+]
@@ -0,0 +1,228 @@
+"""Signed, tenant-isolated execution adapter for OpenClaw runtimes."""
+from __future__ import annotations
+
+import threading
+from dataclasses import replace
+from typing import Callable, Mapping, Protocol
+
+from .contract import (
+    CommandState,
+    OpenClawCommand,
+    OpenClawResult,
+    RuntimeHealth,
+    utc_now,
+    verify_command_signature,
+)
+
+
+class AdapterError(RuntimeError):
+    """Base adapter failure."""
+
+
+class CommandRejected(AdapterError):
+    """A command failed signature or contract validation."""
+
+
+class CommandNotFound(AdapterError):
+    """A tenant-scoped command is not visible."""
+
+
+class CommandConflict(AdapterError):
+    """A command or idempotency key conflicts with an existing execution."""
+
+
+class RuntimeTransport(Protocol):
+    def execute(self, command: OpenClawCommand, timeout_seconds: float) -> OpenClawResult: ...
+    def cancel(self, command_id: str) -> bool: ...
+    def health(self) -> RuntimeHealth: ...
+
+
+class CallbackSink(Protocol):
+    def on_started(self, result: OpenClawResult) -> None: ...
+    def on_completed(self, result: OpenClawResult) -> None: ...
+    def on_failed(self, result: OpenClawResult) -> None: ...
+
+
+class InMemoryCallbackSink:
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self.started: list[OpenClawResult] = []
+        self.completed: list[OpenClawResult] = []
+        self.failed: list[OpenClawResult] = []
+
+    def on_started(self, result: OpenClawResult) -> None:
+        with self._lock:
+            self.started.append(result)
+
+    def on_completed(self, result: OpenClawResult) -> None:
+        with self._lock:
+            self.completed.append(result)
+
+    def on_failed(self, result: OpenClawResult) -> None:
+        with self._lock:
+            self.failed.append(result)
+
+
+SigningMaterialResolver = Callable[[str, str], bytes | str]
+
+
+class OpenClawAdapter:
+    """Verify, de-duplicate, execute, cancel and reconcile commands."""
+
+    def __init__(
+        self,
+        transport: RuntimeTransport,
+        signing_material_resolver: SigningMaterialResolver,
+        *,
+        callback_sink: CallbackSink | None = None,
+        default_timeout_seconds: float = 300.0,
+    ) -> None:
+        if default_timeout_seconds <= 0:
+            raise ValueError("default_timeout_seconds must be positive")
+        self.transport = transport
+        self.signing_material_resolver = signing_material_resolver
+        self.callback_sink = callback_sink or InMemoryCallbackSink()
+        self.default_timeout_seconds = float(default_timeout_seconds)
+        self._lock = threading.RLock()
+        self._commands: dict[str, OpenClawCommand] = {}
+        self._results: dict[str, OpenClawResult] = {}
+        self._idempotency: dict[tuple[str, str], tuple[str, str]] = {}
+
+    def _verify(self, command: OpenClawCommand) -> None:
+        try:
+            key_material = self.signing_material_resolver(command.tenant_id, command.key_id)
+        except Exception as exc:
+            raise CommandRejected("signing material unavailable") from exc
+        if not verify_command_signature(command, key_material):
+            raise CommandRejected("invalid or expired command signature")
+
+    @staticmethod
+    def _validate_result(command: OpenClawCommand, result: OpenClawResult) -> None:
+        if (
+            result.command_id != command.command_id
+            or result.tenant_id != command.tenant_id
+            or result.task_id != command.task_id
+            or result.trace_id != command.trace_id
+        ):
+            raise AdapterError("runtime returned mismatched command identity")
+        if result.state not in {
+            CommandState.COMPLETED,
+            CommandState.FAILED,
+            CommandState.CANCELLED,
+            CommandState.TIMED_OUT,
+        }:
+            raise AdapterError("runtime returned a non-terminal result")
+
+    def execute(
+        self,
+        command: OpenClawCommand,
+        *,
+        timeout_seconds: float | None = None,
+    ) -> OpenClawResult:
+        timeout = self.default_timeout_seconds if timeout_seconds is None else float(timeout_seconds)
+        if timeout <= 0:
+            raise ValueError("timeout_seconds must be positive")
+        self._verify(command)
+        digest = command.signature
+        idem_ref = (command.tenant_id, command.idempotency_key)
+
+        with self._lock:
+            prior_command = self._commands.get(command.command_id)
+            if prior_command is not None:
+                if prior_command.signature != command.signature:
+                    raise CommandConflict("command_id reused with different content")
+                return self._results[command.command_id]
+            prior_idem = self._idempotency.get(idem_ref)
+            if prior_idem is not None:
+                prior_id, prior_digest = prior_idem
+                if prior_digest != digest:
+                    raise CommandConflict("idempotency key reused with different content")
+                return self._results[prior_id]
+            running = OpenClawResult(
+                command_id=command.command_id,
+                tenant_id=command.tenant_id,
+                task_id=command.task_id,
+                trace_id=command.trace_id,
+                state=CommandState.RUNNING,
+                started_at=utc_now(),
+            )
+            self._commands[command.command_id] = command
+            self._results[command.command_id] = running
+            self._idempotency[idem_ref] = (command.command_id, digest)
+
+        self.callback_sink.on_started(running)
+        try:
+            result = self.transport.execute(command, timeout)
+            self._validate_result(command, result)
+        except TimeoutError as exc:
+            result = replace(
+                running,
+                state=CommandState.TIMED_OUT,
+                error={"code": "runtime_timeout", "message": str(exc) or "runtime timeout"},
+                finished_at=utc_now(),
+            )
+        except Exception as exc:
+            result = replace(
+                running,
+                state=CommandState.FAILED,
+                error={"code": "runtime_error", "message": str(exc)},
+                finished_at=utc_now(),
+            )
+
+        if result.started_at is None:
+            result = replace(result, started_at=running.started_at)
+        if result.finished_at is None:
+            result = replace(result, finished_at=utc_now())
+        with self._lock:
+            self._results[command.command_id] = result
+        if result.state == CommandState.COMPLETED:
+            self.callback_sink.on_completed(result)
+        else:
+            self.callback_sink.on_failed(result)
+        return result
+
+    def status(self, command_id: str, *, tenant_id: str) -> OpenClawResult:
+        with self._lock:
+            command = self._commands.get(command_id)
+            result = self._results.get(command_id)
+            if command is None or result is None or command.tenant_id != tenant_id:
+                raise CommandNotFound("command not found")
+            return result
+
+    def cancel(self, command_id: str, *, tenant_id: str) -> OpenClawResult:
+        with self._lock:
+            command = self._commands.get(command_id)
+            current = self._results.get(command_id)
+            if command is None or current is None or command.tenant_id != tenant_id:
+                raise CommandNotFound("command not found")
+            if current.state in {
+                CommandState.COMPLETED,
+                CommandState.FAILED,
+                CommandState.CANCELLED,
+                CommandState.TIMED_OUT,
+                CommandState.REJECTED,
+            }:
+                return current
+        if not self.transport.cancel(command_id):
+            raise AdapterError("runtime rejected cancellation")
+        cancelled = replace(
+            current,
+            state=CommandState.CANCELLED,
+            error={"code": "cancelled", "message": "cancelled by AIOS"},
+            finished_at=utc_now(),
+        )
+        with self._lock:
+            self._results[command_id] = cancelled
+        self.callback_sink.on_failed(cancelled)
+        return cancelled
+
+    def health(self) -> RuntimeHealth:
+        return self.transport.health()
+
+    def snapshot(self, *, tenant_id: str) -> Mapping[str, OpenClawResult]:
+        with self._lock:
+            return {
+                command_id: result
+                for command_id, result in self._results.items()
+                if self._commands[command_id].tenant_id == tenant_id
+            }
@@ -0,0 +1,97 @@
+"""Audit-oriented lifecycle callbacks for OpenClaw execution."""
+from __future__ import annotations
+
+import json
+import threading
+from dataclasses import dataclass, field
+from typing import Any, Mapping, Protocol
+
+from .contract import OpenClawResult, utc_now
+
+
+class EventPublisher(Protocol):
+    def publish(self, event: Mapping[str, Any]) -> None: ...
+
+
+@dataclass(frozen=True)
+class ExecutionEvent:
+    event_type: str
+    tenant_id: str
+    task_id: str
+    trace_id: str
+    command_id: str
+    state: str
+    created_at: str = field(default_factory=utc_now)
+    details: Mapping[str, Any] = field(default_factory=dict)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "event_type": self.event_type,
+            "tenant_id": self.tenant_id,
+            "task_id": self.task_id,
+            "trace_id": self.trace_id,
+            "command_id": self.command_id,
+            "state": self.state,
+            "created_at": self.created_at,
+            "details": json.loads(
+                json.dumps(dict(self.details), ensure_ascii=False, default=str)
+            ),
+        }
+
+
+class InMemoryEventPublisher:
+    """Thread-safe publisher used by tests and first integration slice."""
+
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self.events: list[dict[str, Any]] = []
+
+    def publish(self, event: Mapping[str, Any]) -> None:
+        with self._lock:
+            self.events.append(
+                json.loads(json.dumps(dict(event), ensure_ascii=False, default=str))
+            )
+
+
+class AuditCallbackSink:
+    """Convert runtime callbacks into tenant-scoped audit events."""
+
+    def __init__(self, publisher: EventPublisher) -> None:
+        self.publisher = publisher
+
+    @staticmethod
+    def _event(
+        event_type: str,
+        result: OpenClawResult,
+    ) -> ExecutionEvent:
+        details: dict[str, Any] = {
+            "runtime_id": result.runtime_id,
+            "attempts": result.attempts,
+        }
+        if result.error:
+            details["error"] = dict(result.error)
+        if result.output:
+            details["output_present"] = True
+        return ExecutionEvent(
+            event_type=event_type,
+            tenant_id=result.tenant_id,
+            task_id=result.task_id,
+            trace_id=result.trace_id,
+            command_id=result.command_id,
+            state=result.state.value,
+            details=details,
+        )
+
+    def on_started(self, result: OpenClawResult) -> None:
+        self.publisher.publish(self._event("OPENCLAW_STARTED", result).as_dict())
+
+    def on_completed(self, result: OpenClawResult) -> None:
+        self.publisher.publish(self._event("OPENCLAW_COMPLETED", result).as_dict())
+
+    def on_failed(self, result: OpenClawResult) -> None:
+        event_type = {
+            "CANCELLED": "OPENCLAW_CANCELLED",
+            "TIMED_OUT": "OPENCLAW_TIMED_OUT",
+            "REJECTED": "OPENCLAW_REJECTED",
+        }.get(result.state.value, "OPENCLAW_FAILED")
+        self.publisher.publish(self._event(event_type, result).as_dict())
@@ -0,0 +1,213 @@
+"""Versioned AIOS to OpenClaw execution contracts."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import uuid
+from dataclasses import dataclass, field, replace
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Mapping
+
+CONTRACT_VERSION = "aios.openclaw.command.v1"
+DEFAULT_CLOCK_SKEW_SECONDS = 300
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _parse_timestamp(value: str) -> datetime:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise ValueError("timestamp is required")
+    if normalized.endswith("Z"):
+        normalized = normalized[:-1] + "+00:00"
+    parsed = datetime.fromisoformat(normalized)
+    if parsed.tzinfo is None:
+        raise ValueError("timestamp must include a timezone")
+    return parsed.astimezone(timezone.utc)
+
+
+def _required(value: str, label: str) -> str:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise ValueError(f"{label} is required")
+    return normalized
+
+
+def _safe_mapping(value: Mapping[str, Any] | None) -> dict[str, Any]:
+    return json.loads(json.dumps(dict(value or {}), ensure_ascii=False, sort_keys=True, default=str))
+
+
+class CommandState(str, Enum):
+    ACCEPTED = "ACCEPTED"
+    RUNNING = "RUNNING"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    CANCELLED = "CANCELLED"
+    TIMED_OUT = "TIMED_OUT"
+    REJECTED = "REJECTED"
+
+
+@dataclass(frozen=True)
+class OpenClawCommand:
+    command_id: str
+    tenant_id: str
+    task_id: str
+    trace_id: str
+    action: str
+    payload: Mapping[str, Any] = field(default_factory=dict)
+    idempotency_key: str = ""
+    created_at: str = field(default_factory=utc_now)
+    expires_at: str | None = None
+    contract_version: str = CONTRACT_VERSION
+    key_id: str = "default"
+    signature: str = ""
+
+    def __post_init__(self) -> None:
+        for field_name in (
+            "command_id", "tenant_id", "task_id", "trace_id", "action",
+            "idempotency_key", "created_at", "contract_version", "key_id",
+        ):
+            object.__setattr__(self, field_name, _required(getattr(self, field_name), field_name))
+        if self.contract_version != CONTRACT_VERSION:
+            raise ValueError(f"unsupported contract_version: {self.contract_version}")
+        object.__setattr__(self, "payload", _safe_mapping(self.payload))
+        _parse_timestamp(self.created_at)
+        if self.expires_at is not None:
+            object.__setattr__(self, "expires_at", _required(self.expires_at, "expires_at"))
+            _parse_timestamp(self.expires_at)
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        tenant_id: str,
+        task_id: str,
+        trace_id: str,
+        action: str,
+        payload: Mapping[str, Any] | None,
+        idempotency_key: str,
+        expires_at: str | None = None,
+        key_id: str = "default",
+        command_id: str | None = None,
+    ) -> "OpenClawCommand":
+        return cls(
+            command_id=command_id or f"cmd_{uuid.uuid4().hex}",
+            tenant_id=tenant_id,
+            task_id=task_id,
+            trace_id=trace_id,
+            action=action,
+            payload=_safe_mapping(payload),
+            idempotency_key=idempotency_key,
+            expires_at=expires_at,
+            key_id=key_id,
+        )
+
+    def unsigned_dict(self) -> dict[str, Any]:
+        return {
+            "command_id": self.command_id,
+            "tenant_id": self.tenant_id,
+            "task_id": self.task_id,
+            "trace_id": self.trace_id,
+            "action": self.action,
+            "payload": _safe_mapping(self.payload),
+            "idempotency_key": self.idempotency_key,
+            "created_at": self.created_at,
+            "expires_at": self.expires_at,
+            "contract_version": self.contract_version,
+            "key_id": self.key_id,
+        }
+
+
+@dataclass(frozen=True)
+class OpenClawResult:
+    command_id: str
+    tenant_id: str
+    task_id: str
+    trace_id: str
+    state: CommandState
+    output: Mapping[str, Any] | None = None
+    error: Mapping[str, Any] | None = None
+    started_at: str | None = None
+    finished_at: str | None = None
+    runtime_id: str | None = None
+    attempts: int = 1
+
+    def __post_init__(self) -> None:
+        for field_name in ("command_id", "tenant_id", "task_id", "trace_id"):
+            object.__setattr__(self, field_name, _required(getattr(self, field_name), field_name))
+        object.__setattr__(self, "state", CommandState(self.state))
+        if self.output is not None:
+            object.__setattr__(self, "output", _safe_mapping(self.output))
+        if self.error is not None:
+            object.__setattr__(self, "error", _safe_mapping(self.error))
+        if self.started_at is not None:
+            _parse_timestamp(self.started_at)
+        if self.finished_at is not None:
+            _parse_timestamp(self.finished_at)
+        if self.attempts < 1:
+            raise ValueError("attempts must be >= 1")
+
+
+@dataclass(frozen=True)
+class RuntimeHealth:
+    runtime: str
+    runtime_id: str
+    status: str
+    version: str
+    heartbeat_at: str
+    active_tasks: int = 0
+    workspace_isolated: bool = False
+    credential_ref_configured: bool = False
+    details: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        for field_name in ("runtime", "runtime_id", "status", "version", "heartbeat_at"):
+            object.__setattr__(self, field_name, _required(getattr(self, field_name), field_name))
+        _parse_timestamp(self.heartbeat_at)
+        if self.active_tasks < 0:
+            raise ValueError("active_tasks must be >= 0")
+        object.__setattr__(self, "details", _safe_mapping(self.details))
+
+
+def canonical_command_bytes(command: OpenClawCommand) -> bytes:
+    return json.dumps(
+        command.unsigned_dict(), ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+
+
+def sign_command(command: OpenClawCommand, key_material: bytes | str) -> OpenClawCommand:
+    key = key_material.encode("utf-8") if isinstance(key_material, str) else bytes(key_material)
+    if not key:
+        raise ValueError("signing material is required")
+    signature = hmac.new(key, canonical_command_bytes(command), hashlib.sha256).hexdigest()
+    return replace(command, signature=signature)
+
+
+def verify_command_signature(
+    command: OpenClawCommand,
+    key_material: bytes | str,
+    *,
+    now: datetime | None = None,
+    clock_skew_seconds: int = DEFAULT_CLOCK_SKEW_SECONDS,
+) -> bool:
+    if not command.signature:
+        return False
+    key = key_material.encode("utf-8") if isinstance(key_material, str) else bytes(key_material)
+    if not key:
+        return False
+    expected = hmac.new(key, canonical_command_bytes(command), hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(expected, command.signature):
+        return False
+    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
+    created = _parse_timestamp(command.created_at)
+    if created.timestamp() - current.timestamp() > clock_skew_seconds:
+        return False
+    if command.expires_at is not None:
+        expires = _parse_timestamp(command.expires_at)
+        if current.timestamp() - expires.timestamp() > clock_skew_seconds:
+            return False
+    return True
@@ -0,0 +1,103 @@
+"""Health policy and self-healing decisions for OpenClaw runtimes."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Mapping
+
+from .contract import RuntimeHealth
+
+
+def _parse(value: str) -> datetime:
+    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
+    parsed = datetime.fromisoformat(normalized)
+    if parsed.tzinfo is None:
+        raise ValueError("heartbeat timestamp must include timezone")
+    return parsed.astimezone(timezone.utc)
+
+
+@dataclass(frozen=True)
+class HealthPolicy:
+    max_heartbeat_age_seconds: int = 90
+    require_workspace_isolation: bool = True
+    require_credential_ref: bool = True
+    accepted_statuses: frozenset[str] = field(
+        default_factory=lambda: frozenset({"healthy", "ready"})
+    )
+
+    def __post_init__(self) -> None:
+        if self.max_heartbeat_age_seconds < 5:
+            raise ValueError("max_heartbeat_age_seconds must be >= 5")
+        if not self.accepted_statuses:
+            raise ValueError("accepted_statuses must not be empty")
+
+
+@dataclass(frozen=True)
+class HealthAssessment:
+    healthy: bool
+    reasons: tuple[str, ...]
+    recoverable: bool
+    recommended_action: str
+    details: Mapping[str, object] = field(default_factory=dict)
+
+
+class RuntimeHealthMonitor:
+    """Evaluate runtime health and recommend bounded recovery actions."""
+
+    def __init__(self, policy: HealthPolicy | None = None) -> None:
+        self.policy = policy or HealthPolicy()
+
+    def assess(
+        self,
+        health: RuntimeHealth,
+        *,
+        now: datetime | None = None,
+    ) -> HealthAssessment:
+        current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
+        heartbeat = _parse(health.heartbeat_at)
+        age = max(0.0, current.timestamp() - heartbeat.timestamp())
+        reasons: list[str] = []
+
+        if health.status.lower() not in self.policy.accepted_statuses:
+            reasons.append("runtime_status_unhealthy")
+        if age > self.policy.max_heartbeat_age_seconds:
+            reasons.append("heartbeat_stale")
+        if self.policy.require_workspace_isolation and not health.workspace_isolated:
+            reasons.append("workspace_not_isolated")
+        if self.policy.require_credential_ref and not health.credential_ref_configured:
+            reasons.append("credential_reference_missing")
+
+        if not reasons:
+            return HealthAssessment(
+                healthy=True,
+                reasons=(),
+                recoverable=True,
+                recommended_action="none",
+                details={"heartbeat_age_seconds": age},
+            )
+
+        non_recoverable = {
+            "workspace_not_isolated",
+            "credential_reference_missing",
+        }.intersection(reasons)
+        if non_recoverable:
+            action = "quarantine_runtime"
+            recoverable = False
+        elif "heartbeat_stale" in reasons:
+            action = "restart_runtime"
+            recoverable = True
+        else:
+            action = "drain_and_restart"
+            recoverable = True
+
+        return HealthAssessment(
+            healthy=False,
+            reasons=tuple(reasons),
+            recoverable=recoverable,
+            recommended_action=action,
+            details={
+                "heartbeat_age_seconds": age,
+                "runtime_id": health.runtime_id,
+                "active_tasks": health.active_tasks,
+            },
+        )
@@ -0,0 +1,214 @@
+"""Runtime transport implementations for the OpenClaw adapter."""
+from __future__ import annotations
+
+import json
+import threading
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any, Callable, Mapping, Protocol
+
+from .contract import (
+    CommandState,
+    OpenClawCommand,
+    OpenClawResult,
+    RuntimeHealth,
+    utc_now,
+)
+
+
+class RuntimeTransport(Protocol):
+    def execute(self, command: OpenClawCommand, timeout_seconds: float) -> OpenClawResult: ...
+    def cancel(self, command_id: str) -> bool: ...
+    def health(self) -> RuntimeHealth: ...
+
+
+class TransportError(RuntimeError):
+    """Runtime transport or protocol failure."""
+
+
+RuntimeHandler = Callable[[OpenClawCommand], Mapping[str, Any] | OpenClawResult]
+
+
+class InProcessRuntimeTransport:
+    """Deterministic runtime used by tests and controlled local integrations."""
+
+    def __init__(
+        self,
+        handler: RuntimeHandler,
+        *,
+        runtime_id: str = "openclaw-in-process",
+        version: str = "v1",
+        workspace_isolated: bool = True,
+        credential_ref_configured: bool = True,
+    ) -> None:
+        self.handler = handler
+        self.runtime_id = runtime_id
+        self.version = version
+        self.workspace_isolated = workspace_isolated
+        self.credential_ref_configured = credential_ref_configured
+        self._lock = threading.RLock()
+        self._running: set[str] = set()
+        self._cancelled: set[str] = set()
+
+    def execute(self, command: OpenClawCommand, timeout_seconds: float) -> OpenClawResult:
+        if timeout_seconds <= 0:
+            raise ValueError("timeout_seconds must be positive")
+        with self._lock:
+            if command.command_id in self._cancelled:
+                return OpenClawResult(
+                    command_id=command.command_id,
+                    tenant_id=command.tenant_id,
+                    task_id=command.task_id,
+                    trace_id=command.trace_id,
+                    state=CommandState.CANCELLED,
+                    error={"code": "cancelled", "message": "cancelled before execution"},
+                    finished_at=utc_now(),
+                    runtime_id=self.runtime_id,
+                )
+            self._running.add(command.command_id)
+        try:
+            raw = self.handler(command)
+            if isinstance(raw, OpenClawResult):
+                return raw
+            if not isinstance(raw, Mapping):
+                raise TransportError("runtime handler must return a mapping or OpenClawResult")
+            return OpenClawResult(
+                command_id=command.command_id,
+                tenant_id=command.tenant_id,
+                task_id=command.task_id,
+                trace_id=command.trace_id,
+                state=CommandState(str(raw.get("state", "COMPLETED"))),
+                output=raw.get("output"),
+                error=raw.get("error"),
+                started_at=str(raw.get("started_at") or utc_now()),
+                finished_at=str(raw.get("finished_at") or utc_now()),
+                runtime_id=str(raw.get("runtime_id") or self.runtime_id),
+                attempts=int(raw.get("attempts", 1)),
+            )
+        finally:
+            with self._lock:
+                self._running.discard(command.command_id)
+
+    def cancel(self, command_id: str) -> bool:
+        with self._lock:
+            self._cancelled.add(command_id)
+            return command_id in self._running
+
+    def health(self) -> RuntimeHealth:
+        with self._lock:
+            active = len(self._running)
+        return RuntimeHealth(
+            runtime="openclaw",
+            runtime_id=self.runtime_id,
+            status="healthy",
+            version=self.version,
+            heartbeat_at=utc_now(),
+            active_tasks=active,
+            workspace_isolated=self.workspace_isolated,
+            credential_ref_configured=self.credential_ref_configured,
+            details={"transport": "in_process"},
+        )
+
+
+class HTTPRuntimeTransport:
+    """Private HTTP transport for an OpenClaw runtime service."""
+
+    def __init__(
+        self,
+        base_url: str,
+        *,
+        runtime_id: str = "openclaw-http",
+        credential_ref: str | None = None,
+        max_response_bytes: int = 2_097_152,
+    ) -> None:
+        normalized = str(base_url).rstrip("/")
+        if not normalized.startswith(("http://", "https://")):
+            raise ValueError("base_url must use http or https")
+        self.base_url = normalized
+        self.runtime_id = runtime_id
+        self.credential_ref = credential_ref
+        self.max_response_bytes = int(max_response_bytes)
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        payload: Mapping[str, Any] | None,
+        timeout_seconds: float,
+    ) -> Mapping[str, Any]:
+        body = None
+        headers = {"Accept": "application/json", "User-Agent": "AIOS-OpenClaw/1"}
+        if payload is not None:
+            body = json.dumps(dict(payload), separators=(",", ":")).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+        if self.credential_ref:
+            headers["X-AIOS-Credential-Ref"] = self.credential_ref
+        request = urllib.request.Request(
+            f"{self.base_url}{path}",
+            data=body,
+            method=method,
+            headers=headers,
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+                raw = response.read(self.max_response_bytes + 1)
+        except urllib.error.HTTPError as exc:
+            raise TransportError(f"runtime returned HTTP {exc.code}") from exc
+        except urllib.error.URLError as exc:
+            raise TransportError(f"runtime unavailable: {exc.reason}") from exc
+        if len(raw) > self.max_response_bytes:
+            raise TransportError("runtime response exceeded size limit")
+        decoded = json.loads(raw.decode("utf-8"))
+        if not isinstance(decoded, Mapping):
+            raise TransportError("runtime response must be a JSON object")
+        return decoded
+
+    @staticmethod
+    def _command_payload(command: OpenClawCommand) -> dict[str, Any]:
+        return {**command.unsigned_dict(), "signature": command.signature}
+
+    def execute(self, command: OpenClawCommand, timeout_seconds: float) -> OpenClawResult:
+        value = self._request(
+            "POST",
+            "/v1/execute",
+            payload=self._command_payload(command),
+            timeout_seconds=timeout_seconds,
+        )
+        return OpenClawResult(
+            command_id=str(value["command_id"]),
+            tenant_id=str(value["tenant_id"]),
+            task_id=str(value["task_id"]),
+            trace_id=str(value["trace_id"]),
+            state=CommandState(str(value["state"])),
+            output=value.get("output"),
+            error=value.get("error"),
+            started_at=value.get("started_at"),
+            finished_at=value.get("finished_at"),
+            runtime_id=str(value.get("runtime_id") or self.runtime_id),
+            attempts=int(value.get("attempts", 1)),
+        )
+
+    def cancel(self, command_id: str) -> bool:
+        value = self._request(
+            "POST",
+            f"/v1/commands/{urllib.parse.quote(command_id, safe='')}/cancel",
+            payload={},
+            timeout_seconds=10.0,
+        )
+        return bool(value.get("accepted"))
+
+    def health(self) -> RuntimeHealth:
+        value = self._request("GET", "/v1/health", payload=None, timeout_seconds=5.0)
+        return RuntimeHealth(
+            runtime=str(value.get("runtime") or "openclaw"),
+            runtime_id=str(value.get("runtime_id") or self.runtime_id),
+            status=str(value.get("status") or "unknown"),
+            version=str(value.get("version") or "unknown"),
+            heartbeat_at=str(value.get("heartbeat_at") or utc_now()),
+            active_tasks=int(value.get("active_tasks", 0)),
+            workspace_isolated=bool(value.get("workspace_isolated", False)),
+            credential_ref_configured=bool(value.get("credential_ref_configured", False)),
+            details=value.get("details") if isinstance(value.get("details"), Mapping) else {},
+        )
@@ -0,0 +1,46 @@
+"""AIOS V2 control plane and execution orchestration."""
+
+from .control_plane import (
+    AgentRecord,
+    AuditEvent,
+    ControlPlaneError,
+    ControlPlaneService,
+    IdempotencyConflict,
+    InMemoryControlPlaneStore,
+    InvalidTransition,
+    NotFound,
+    PolicyDenied,
+    RequestContext,
+    RevisionConflict,
+    TaskRecord,
+    TaskState,
+    TenantIsolationError,
+    TenantRecord,
+    Unauthorized,
+)
+from .execution_tracking import ExecutionEvidence, ExecutionTracker
+from .orchestrator import DispatchRecord, ExecutionOrchestrator, OrchestrationError
+
+__all__ = [
+    "AgentRecord",
+    "AuditEvent",
+    "ControlPlaneError",
+    "ControlPlaneService",
+    "DispatchRecord",
+    "ExecutionEvidence",
+    "ExecutionOrchestrator",
+    "ExecutionTracker",
+    "IdempotencyConflict",
+    "InMemoryControlPlaneStore",
+    "InvalidTransition",
+    "NotFound",
+    "OrchestrationError",
+    "PolicyDenied",
+    "RequestContext",
+    "RevisionConflict",
+    "TaskRecord",
+    "TaskState",
+    "TenantIsolationError",
+    "TenantRecord",
+    "Unauthorized",
+]
@@ -0,0 +1,357 @@
+"""Production-candidate AIOS V2 application entrypoint.
+
+This process owns the public control-plane API and read-only web console. It
+loads durable control-plane state, dispatches approved work through runtime
+adapters, exposes tenant-scoped execution evidence and sanitized GitOps status,
+and never exposes runtime credentials to the browser.
+"""
+from __future__ import annotations
+
+import dataclasses
+import json
+import os
+import stat
+import urllib.parse
+from enum import Enum
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any, Mapping
+
+from aios_hermes import HermesAdapter, HermesJobState, HermesResult
+from aios_openclaw import HTTPRuntimeTransport, OpenClawAdapter
+from aios_openclaw.callback import AuditCallbackSink, EventPublisher
+
+from .console import CONSOLE_HTML
+from .control_plane import AuditEvent, RequestContext
+from .execution_tracking import ExecutionTracker
+from .gateway import GatewayApplication, error_response
+from .gitops_status import collect_gitops_status
+from .orchestrator import ExecutionOrchestrator
+from .persistence import DurableControlPlaneService, SQLiteSnapshotRepository
+
+
+def _json_value(value: Any) -> Any:
+    if dataclasses.is_dataclass(value):
+        return {
+            item.name: _json_value(getattr(value, item.name))
+            for item in dataclasses.fields(value)
+        }
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, Mapping):
+        return {str(key): _json_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [_json_value(item) for item in value]
+    return value
+
+
+class FileSigningMaterialResolver:
+    """Resolve runtime signing material from a root-owned file reference."""
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path).resolve()
+
+    def __call__(self, tenant_id: str, key_id: str) -> bytes:
+        del tenant_id, key_id
+        metadata = self.path.stat()
+        if not stat.S_ISREG(metadata.st_mode):
+            raise RuntimeError("signing material reference is not a regular file")
+        if metadata.st_mode & 0o077:
+            raise RuntimeError("signing material file permissions must be 0600 or stricter")
+        value = self.path.read_bytes().strip()
+        if len(value) < 32:
+            raise RuntimeError("signing material is too short")
+        return value
+
+
+class ControlPlaneEventPublisher(EventPublisher):
+    """Append OpenClaw runtime events to the tenant audit stream."""
+
+    def __init__(self, service: DurableControlPlaneService) -> None:
+        self.service = service
+
+    def publish(self, event: Mapping[str, Any]) -> None:
+        tenant_id = str(event.get("tenant_id") or "").strip()
+        trace_id = str(event.get("trace_id") or "").strip()
+        command_id = str(event.get("command_id") or "").strip()
+        event_type = str(event.get("event_type") or "OPENCLAW_EVENT").strip()
+        if not tenant_id or not trace_id or not command_id:
+            raise ValueError("runtime audit event is missing tenant identity")
+        state_value = str(event.get("state") or "unknown").lower()
+        outcome = "allowed" if state_value == "completed" else state_value
+        audit = AuditEvent(
+            event_id=f"runtime_{command_id}_{len(self.service.store.audit_events) + 1}",
+            tenant_id=tenant_id,
+            principal_id="openclaw-runtime",
+            trace_id=trace_id,
+            action=event_type,
+            resource_type="openclaw_command",
+            resource_id=command_id,
+            outcome=outcome,
+            policy_version="runtime-v1",
+            details=dict(event.get("details") or {}),
+        )
+        with self.service.store.lock:
+            self.service.store.audit_events.append(audit)
+            self.service.repository.save(self.service._snapshot())
+
+
+class CandidateApplication:
+    """Compose durable control plane, runtimes, evidence and console routes."""
+
+    def __init__(
+        self,
+        gateway: GatewayApplication,
+        orchestrator: ExecutionOrchestrator,
+        tracker: ExecutionTracker,
+        hermes: HermesAdapter,
+        openclaw: OpenClawAdapter | None,
+    ) -> None:
+        self.gateway = gateway
+        self.orchestrator = orchestrator
+        self.tracker = tracker
+        self.hermes = hermes
+        self.openclaw = openclaw
+
+    def context(self, headers: Mapping[str, str]) -> RequestContext:
+        return self.gateway.context_from_headers(headers)
+
+    def handle_json(
+        self,
+        method: str,
+        raw_path: str,
+        headers: Mapping[str, str],
+        body: Any,
+    ) -> tuple[int, Mapping[str, Any]]:
+        parsed = urllib.parse.urlparse(raw_path)
+        path = parsed.path.rstrip("/") or "/"
+        method = method.upper()
+
+        if method == "GET" and path == "/api/v2/gitops":
+            context = self.context(headers)
+            return HTTPStatus.OK, {
+                "tenant_id": context.tenant_id,
+                "gitops": collect_gitops_status().as_dict(),
+            }
+
+        if method == "GET" and path == "/api/v2/runtime":
+            context = self.context(headers)
+            runtimes: list[dict[str, Any]] = []
+            if self.openclaw is None:
+                runtimes.append(
+                    {
+                        "name": "openclaw",
+                        "status": "disabled",
+                        "details": {"reason": "runtime_not_configured"},
+                    }
+                )
+            else:
+                try:
+                    health = self.openclaw.health()
+                    runtimes.append(
+                        {
+                            "name": "openclaw",
+                            "status": health.status,
+                            "details": _json_value(health),
+                        }
+                    )
+                except Exception as exc:
+                    runtimes.append(
+                        {
+                            "name": "openclaw",
+                            "status": "unavailable",
+                            "details": {"message": str(exc)},
+                        }
+                    )
+            hermes_health = self.hermes.health()
+            runtimes.append(
+                {
+                    "name": "hermes",
+                    "status": hermes_health.status,
+                    "details": _json_value(hermes_health),
+                }
+            )
+            return HTTPStatus.OK, {
+                "tenant_id": context.tenant_id,
+                "runtimes": runtimes,
+            }
+
+        if method == "GET" and path == "/api/v2/executions":
+            context = self.context(headers)
+            return HTTPStatus.OK, {
+                "executions": _json_value(
+                    self.tracker.list(tenant_id=context.tenant_id)
+                )
+            }
+
+        dispatch_match = __import__("re").fullmatch(
+            r"/api/v2/tasks/([^/]+)/dispatch",
+            path,
+        )
+        if method == "POST" and dispatch_match:
+            context = self.context(headers)
+            task_id = urllib.parse.unquote(dispatch_match.group(1))
+            payload = body if isinstance(body, Mapping) else {}
+            timeout = payload.get("timeout_seconds")
+            record = self.orchestrator.dispatch(
+                context,
+                task_id=task_id,
+                timeout_seconds=float(timeout) if timeout is not None else None,
+            )
+            return HTTPStatus.ACCEPTED, {"dispatch": _json_value(record)}
+
+        if method == "POST" and path == "/api/v2/hermes/results/reconcile":
+            context = self.context(headers)
+            if not isinstance(body, Mapping):
+                raise ValueError("JSON body must be an object")
+            result = HermesResult(
+                job_id=str(body["job_id"]),
+                tenant_id=str(body["tenant_id"]),
+                task_id=str(body["task_id"]),
+                trace_id=str(body["trace_id"]),
+                state=HermesJobState(str(body["state"])),
+                worker_id=(
+                    str(body["worker_id"])
+                    if body.get("worker_id") is not None
+                    else None
+                ),
+                output=body.get("output"),
+                error=body.get("error"),
+                started_at=body.get("started_at"),
+                finished_at=body.get("finished_at"),
+            )
+            task = self.orchestrator.reconcile_hermes(context, result)
+            return HTTPStatus.OK, {"task": _json_value(task)}
+
+        return self.gateway.handle(method, raw_path, headers, body)
+
+
+class CandidateRequestHandler(BaseHTTPRequestHandler):
+    server_version = "AIOSV2Candidate/1.0"
+    max_body_bytes = 1_048_576
+
+    @property
+    def application(self) -> CandidateApplication:
+        return self.server.application  # type: ignore[attr-defined]
+
+    def _read_body(self) -> Any:
+        length = int(self.headers.get("Content-Length", "0") or "0")
+        if length < 0 or length > self.max_body_bytes:
+            raise ValueError("request body is too large")
+        if length == 0:
+            return {}
+        content_type = self.headers.get("Content-Type", "").split(";", 1)[0].strip()
+        if content_type != "application/json":
+            raise ValueError("Content-Type must be application/json")
+        return json.loads(self.rfile.read(length).decode("utf-8"))
+
+    def _security_headers(self) -> None:
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.send_header("X-Frame-Options", "DENY")
+        self.send_header("Referrer-Policy", "no-referrer")
+        self.send_header(
+            "Content-Security-Policy",
+            "default-src 'self'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; connect-src 'self'",
+        )
+
+    def _send_json(self, status_code: int, payload: Mapping[str, Any]) -> None:
+        encoded = json.dumps(
+            _json_value(payload),
+            ensure_ascii=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        self.send_response(int(status_code))
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(encoded)))
+        self._security_headers()
+        self.end_headers()
+        self.wfile.write(encoded)
+
+    def _send_console(self) -> None:
+        encoded = CONSOLE_HTML.encode("utf-8")
+        self.send_response(HTTPStatus.OK)
+        self.send_header("Content-Type", "text/html; charset=utf-8")
+        self.send_header("Content-Length", str(len(encoded)))
+        self._security_headers()
+        self.end_headers()
+        self.wfile.write(encoded)
+
+    def _dispatch(self, method: str) -> None:
+        path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+        if method == "GET" and path in {"/", "/console"}:
+            self._send_console()
+            return
+        try:
+            body = self._read_body() if method == "POST" else {}
+            status_code, payload = self.application.handle_json(
+                method,
+                self.path,
+                {key: value for key, value in self.headers.items()},
+                body,
+            )
+        except BaseException as exc:
+            status_code, payload = error_response(exc)
+        self._send_json(status_code, payload)
+
+    def do_GET(self) -> None:  # noqa: N802
+        self._dispatch("GET")
+
+    def do_POST(self) -> None:  # noqa: N802
+        self._dispatch("POST")
+
+
+class CandidateHTTPServer(ThreadingHTTPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+    def __init__(self, address: tuple[str, int], application: CandidateApplication) -> None:
+        self.application = application
+        super().__init__(address, CandidateRequestHandler)
+
+
+def build_application() -> CandidateApplication:
+    state_dir = Path(os.getenv("AIOS_V2_STATE_DIR", "/var/lib/aios-v2")).resolve()
+    state_dir.mkdir(parents=True, exist_ok=True)
+    repository = SQLiteSnapshotRepository(state_dir / "control-plane.sqlite3")
+    service = DurableControlPlaneService(repository)
+    tracker = ExecutionTracker(state_dir / "executions")
+    hermes = HermesAdapter()
+
+    openclaw_url = os.getenv("AIOS_OPENCLAW_URL", "").strip()
+    material_file = os.getenv("AIOS_OPENCLAW_SIGNING_MATERIAL_FILE", "").strip()
+    openclaw = None
+    resolver = None
+    if openclaw_url and material_file:
+        resolver = FileSigningMaterialResolver(material_file)
+        transport = HTTPRuntimeTransport(
+            openclaw_url,
+            credential_ref=os.getenv("AIOS_OPENCLAW_CREDENTIAL_REF") or None,
+        )
+        openclaw = OpenClawAdapter(
+            transport,
+            resolver,
+            callback_sink=AuditCallbackSink(ControlPlaneEventPublisher(service)),
+        )
+
+    gateway = GatewayApplication(service)
+    orchestrator = ExecutionOrchestrator(
+        service,
+        openclaw=openclaw,
+        hermes=hermes,
+        openclaw_signing_material_resolver=resolver,
+    )
+    return CandidateApplication(gateway, orchestrator, tracker, hermes, openclaw)
+
+
+def main() -> None:
+    host = os.getenv("AIOS_V2_HOST", "127.0.0.1")
+    port = int(os.getenv("AIOS_V2_PORT", "18920"))
+    server = CandidateHTTPServer((host, port), build_application())
+    print(f"AIOS V2 candidate listening on {host}:{port}")
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,69 @@
+"""Dependency-free AIOS V2 web console assets.
+
+Authentication and tenant identity are injected by the trusted reverse proxy.
+The browser never selects a tenant and never calls OpenClaw or Hermes directly.
+"""
+
+CONSOLE_HTML = r'''<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width,initial-scale=1">
+  <meta name="referrer" content="no-referrer">
+  <title>AIOS V2 Console</title>
+  <style>
+    :root{font-family:Inter,ui-sans-serif,system-ui,sans-serif;color:#152033;background:#f4f7fb}
+    *{box-sizing:border-box} body{margin:0} header{background:#111827;color:white;padding:18px 24px;display:flex;justify-content:space-between;align-items:center}
+    header h1{font-size:20px;margin:0} header span{font-size:12px;color:#cbd5e1}
+    main{padding:24px;max-width:1440px;margin:auto}.notice{background:#fff7ed;border:1px solid #fed7aa;padding:12px 14px;border-radius:10px;margin-bottom:18px;font-size:13px}
+    .cards{display:grid;grid-template-columns:repeat(auto-fit,minmax(190px,1fr));gap:14px;margin-bottom:20px}.card,.panel{background:white;border:1px solid #e5e7eb;border-radius:12px;box-shadow:0 1px 2px rgba(0,0,0,.04)}
+    .card{padding:16px}.card b{display:block;font-size:28px;margin-top:8px}.card small{color:#64748b}.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(420px,1fr));gap:16px}.panel{padding:16px;overflow:auto}.panel h2{font-size:16px;margin:0 0 12px}
+    table{width:100%;border-collapse:collapse;font-size:12px}th,td{text-align:left;border-bottom:1px solid #edf0f4;padding:8px;vertical-align:top}th{color:#64748b;font-weight:600}.state{font-weight:700}.ok{color:#047857}.bad{color:#b91c1c}.muted{color:#64748b}
+    button{border:0;border-radius:8px;padding:9px 13px;background:#2563eb;color:white;cursor:pointer}button:disabled{opacity:.5}.toolbar{display:flex;gap:8px;align-items:center;margin-bottom:16px}.error{color:#b91c1c;white-space:pre-wrap}
+  </style>
+</head>
+<body>
+<header><h1>AIOS V2 Control Plane</h1><span id="updated">Not loaded</span></header>
+<main>
+  <div class="notice">Read-only candidate console. Identity and tenant context must be injected by the trusted reverse proxy. The console never calls OpenClaw or Hermes directly.</div>
+  <div class="toolbar"><button id="refresh">Refresh</button><span id="error" class="error"></span></div>
+  <section class="cards">
+    <div class="card"><small>Agents</small><b id="agentCount">—</b></div>
+    <div class="card"><small>Tasks</small><b id="taskCount">—</b></div>
+    <div class="card"><small>Audit events</small><b id="auditCount">—</b></div>
+    <div class="card"><small>Executions</small><b id="executionCount">—</b></div>
+  </section>
+  <section class="grid">
+    <div class="panel"><h2>Runtime health</h2><div id="runtime"></div></div>
+    <div class="panel"><h2>Tasks</h2><div id="tasks"></div></div>
+    <div class="panel"><h2>Agents</h2><div id="agents"></div></div>
+    <div class="panel"><h2>Execution evidence</h2><div id="executions"></div></div>
+    <div class="panel"><h2>Audit</h2><div id="audit"></div></div>
+  </section>
+</main>
+<script>
+const esc=v=>String(v??'').replace(/[&<>"']/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
+const table=(cols,rows)=>`<table><thead><tr>${cols.map(c=>`<th>${esc(c[0])}</th>`).join('')}</tr></thead><tbody>${rows.map(r=>`<tr>${cols.map(c=>`<td>${esc(c[1](r))}</td>`).join('')}</tr>`).join('')}</tbody></table>`;
+async function api(path){const r=await fetch(path,{credentials:'same-origin',headers:{'Accept':'application/json'}});const data=await r.json().catch(()=>({}));if(!r.ok)throw new Error(`${r.status} ${data.message||data.error||path}`);return data}
+async function refresh(){
+  const btn=document.getElementById('refresh');btn.disabled=true;document.getElementById('error').textContent='';
+  try{
+    const [agents,tasks,audit,runtime,executions]=await Promise.all([
+      api('/api/v2/agents'),api('/api/v2/tasks'),api('/api/v2/audit'),api('/api/v2/runtime'),api('/api/v2/executions')
+    ]);
+    document.getElementById('agentCount').textContent=agents.agents.length;
+    document.getElementById('taskCount').textContent=tasks.tasks.length;
+    document.getElementById('auditCount').textContent=audit.events.length;
+    document.getElementById('executionCount').textContent=executions.executions.length;
+    document.getElementById('agents').innerHTML=table([['ID',x=>x.agent_id],['Runtime',x=>x.runtime],['Enabled',x=>x.enabled]],agents.agents);
+    document.getElementById('tasks').innerHTML=table([['Task',x=>x.task_id],['Action',x=>x.action],['State',x=>x.state],['Rev',x=>x.revision]],tasks.tasks);
+    document.getElementById('audit').innerHTML=table([['Time',x=>x.created_at],['Action',x=>x.action],['Outcome',x=>x.outcome],['Resource',x=>x.resource_id]],audit.events.slice(-100).reverse());
+    document.getElementById('executions').innerHTML=table([['Executor',x=>x.executor],['Task',x=>x.task_id],['Status',x=>x.status],['Progress',x=>`${x.progress_completed}/${x.progress_total}`],['Commit',x=>x.last_commit||'']],executions.executions);
+    document.getElementById('runtime').innerHTML=table([['Runtime',x=>x.name],['Status',x=>x.status],['Details',x=>JSON.stringify(x.details)]],runtime.runtimes);
+    document.getElementById('updated').textContent=`Updated ${new Date().toLocaleString()}`;
+  }catch(e){document.getElementById('error').textContent=e.message||String(e)}finally{btn.disabled=false}
+}
+document.getElementById('refresh').addEventListener('click',refresh);refresh();setInterval(refresh,30000);
+</script>
+</body>
+</html>'''
@@ -0,0 +1,627 @@
+"""Tenant-isolated AIOS V2 control-plane domain service.
+
+This module deliberately contains no OpenClaw or Hermes implementation details.
+It owns authoritative tenant, agent, task-lifecycle, idempotency and audit state.
+Runtime adapters consume approved tasks through a separate boundary.
+"""
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import threading
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Callable, Iterable, Mapping
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _new_id(prefix: str) -> str:
+    return f"{prefix}_{uuid.uuid4().hex}"
+
+
+def _canonical_digest(value: Mapping[str, Any]) -> str:
+    encoded = json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+class ControlPlaneError(RuntimeError):
+    """Base error returned by the control-plane domain."""
+
+
+class Unauthorized(ControlPlaneError):
+    """The caller is not authenticated or lacks the required role."""
+
+
+class TenantIsolationError(ControlPlaneError):
+    """A request attempted to cross a tenant boundary."""
+
+
+class NotFound(ControlPlaneError):
+    """A tenant-scoped resource does not exist or is not visible."""
+
+
+class InvalidTransition(ControlPlaneError):
+    """A task state transition violates the lifecycle contract."""
+
+
+class IdempotencyConflict(ControlPlaneError):
+    """An idempotency key was reused for a different command."""
+
+
+class RevisionConflict(ControlPlaneError):
+    """Optimistic task revision does not match current state."""
+
+
+class PolicyDenied(ControlPlaneError):
+    """The policy gate denied a control-plane operation."""
+
+
+@dataclass(frozen=True)
+class RequestContext:
+    """Authenticated request identity propagated across all AIOS layers."""
+
+    tenant_id: str
+    principal_id: str
+    trace_id: str
+    roles: frozenset[str] = field(default_factory=frozenset)
+    session_id: str | None = None
+    policy_version: str = "v1"
+
+    def __post_init__(self) -> None:
+        for name in ("tenant_id", "principal_id", "trace_id"):
+            if not str(getattr(self, name, "")).strip():
+                raise ValueError(f"{name} is required")
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        tenant_id: str,
+        principal_id: str,
+        roles: Iterable[str] = (),
+        trace_id: str | None = None,
+        session_id: str | None = None,
+        policy_version: str = "v1",
+    ) -> "RequestContext":
+        return cls(
+            tenant_id=tenant_id.strip(),
+            principal_id=principal_id.strip(),
+            trace_id=trace_id or _new_id("trace"),
+            roles=frozenset(str(role).strip() for role in roles if str(role).strip()),
+            session_id=session_id,
+            policy_version=policy_version,
+        )
+
+
+@dataclass(frozen=True)
+class TenantRecord:
+    tenant_id: str
+    display_name: str
+    status: str = "active"
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+    created_at: str = field(default_factory=_utc_now)
+    updated_at: str = field(default_factory=_utc_now)
+
+
+@dataclass(frozen=True)
+class AgentRecord:
+    agent_id: str
+    tenant_id: str
+    display_name: str
+    runtime: str
+    capabilities: tuple[str, ...] = field(default_factory=tuple)
+    enabled: bool = True
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+    created_at: str = field(default_factory=_utc_now)
+    updated_at: str = field(default_factory=_utc_now)
+
+
+class TaskState(str, Enum):
+    CREATED = "CREATED"
+    APPROVED = "APPROVED"
+    RUNNING = "RUNNING"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    CANCELLED = "CANCELLED"
+
+
+TERMINAL_TASK_STATES = frozenset(
+    {TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELLED}
+)
+
+ALLOWED_TASK_TRANSITIONS: Mapping[TaskState, frozenset[TaskState]] = {
+    TaskState.CREATED: frozenset({TaskState.APPROVED, TaskState.CANCELLED}),
+    TaskState.APPROVED: frozenset({TaskState.RUNNING, TaskState.CANCELLED}),
+    TaskState.RUNNING: frozenset(
+        {TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELLED}
+    ),
+    TaskState.COMPLETED: frozenset(),
+    TaskState.FAILED: frozenset(),
+    TaskState.CANCELLED: frozenset(),
+}
+
+
+@dataclass(frozen=True)
+class TaskRecord:
+    task_id: str
+    tenant_id: str
+    agent_id: str
+    action: str
+    payload: Mapping[str, Any]
+    state: TaskState = TaskState.CREATED
+    revision: int = 1
+    idempotency_key: str | None = None
+    request_digest: str = ""
+    result: Mapping[str, Any] | None = None
+    error: Mapping[str, Any] | None = None
+    created_by: str = ""
+    approved_by: str | None = None
+    created_at: str = field(default_factory=_utc_now)
+    updated_at: str = field(default_factory=_utc_now)
+
+
+@dataclass(frozen=True)
+class AuditEvent:
+    event_id: str
+    tenant_id: str
+    principal_id: str
+    trace_id: str
+    action: str
+    resource_type: str
+    resource_id: str
+    outcome: str
+    policy_version: str
+    details: Mapping[str, Any] = field(default_factory=dict)
+    created_at: str = field(default_factory=_utc_now)
+
+
+@dataclass(frozen=True)
+class PolicyDecision:
+    allowed: bool
+    reason: str
+    policy_version: str
+
+
+PolicyEvaluator = Callable[[RequestContext, str, Mapping[str, Any]], PolicyDecision]
+
+
+class InMemoryControlPlaneStore:
+    """Thread-safe reference store used by tests and the first gateway slice.
+
+    Production backends must preserve the same tenant-scoped interface and add
+    database row-level security, migrations, backup and HA evidence.
+    """
+
+    def __init__(self) -> None:
+        self.lock = threading.RLock()
+        self.tenants: dict[str, TenantRecord] = {}
+        self.agents: dict[str, AgentRecord] = {}
+        self.tasks: dict[str, TaskRecord] = {}
+        self.idempotency: dict[tuple[str, str], tuple[str, str]] = {}
+        self.audit_events: list[AuditEvent] = []
+
+
+class ControlPlaneService:
+    """Authoritative tenant, agent and task lifecycle service."""
+
+    def __init__(
+        self,
+        store: InMemoryControlPlaneStore | None = None,
+        *,
+        policy_evaluator: PolicyEvaluator | None = None,
+    ) -> None:
+        self.store = store or InMemoryControlPlaneStore()
+        self.policy_evaluator = policy_evaluator or self._default_policy
+
+    @staticmethod
+    def _default_policy(
+        context: RequestContext,
+        action: str,
+        resource: Mapping[str, Any],
+    ) -> PolicyDecision:
+        del resource
+        role_requirements = {
+            "tenant.register": {"platform_admin"},
+            "tenant.read": {"tenant_admin", "operator", "auditor", "platform_admin"},
+            "agent.register": {"tenant_admin", "operator", "platform_admin"},
+            "agent.read": {"tenant_admin", "operator", "auditor", "platform_admin"},
+            "task.create": {"tenant_admin", "operator", "platform_admin"},
+            "task.read": {"tenant_admin", "operator", "auditor", "platform_admin"},
+            "task.approve": {"tenant_admin", "approver", "platform_admin"},
+            "task.start": {"runtime", "operator", "platform_admin"},
+            "task.complete": {"runtime", "operator", "platform_admin"},
+            "task.fail": {"runtime", "operator", "platform_admin"},
+            "task.cancel": {"tenant_admin", "operator", "approver", "platform_admin"},
+            "audit.read": {"tenant_admin", "auditor", "platform_admin"},
+        }
+        required = role_requirements.get(action, {"platform_admin"})
+        allowed = bool(context.roles.intersection(required))
+        return PolicyDecision(
+            allowed=allowed,
+            reason="role_allowed" if allowed else "required_role_missing",
+            policy_version=context.policy_version,
+        )
+
+    def _require_context(self, context: RequestContext) -> None:
+        if not context.principal_id or not context.tenant_id or not context.trace_id:
+            raise Unauthorized("authenticated tenant context is required")
+
+    def _authorize(
+        self,
+        context: RequestContext,
+        action: str,
+        resource: Mapping[str, Any],
+    ) -> PolicyDecision:
+        self._require_context(context)
+        decision = self.policy_evaluator(context, action, resource)
+        if not decision.allowed:
+            self._audit(
+                context,
+                action=action,
+                resource_type=str(resource.get("resource_type", "policy")),
+                resource_id=str(resource.get("resource_id", "unknown")),
+                outcome="denied",
+                details={"reason": decision.reason},
+            )
+            raise PolicyDenied(f"{action} denied: {decision.reason}")
+        return decision
+
+    @staticmethod
+    def _assert_same_tenant(context: RequestContext, tenant_id: str) -> None:
+        if context.tenant_id != tenant_id and "platform_admin" not in context.roles:
+            raise TenantIsolationError("cross-tenant access denied")
+
+    def _audit(
+        self,
+        context: RequestContext,
+        *,
+        action: str,
+        resource_type: str,
+        resource_id: str,
+        outcome: str,
+        details: Mapping[str, Any] | None = None,
+    ) -> AuditEvent:
+        event = AuditEvent(
+            event_id=_new_id("audit"),
+            tenant_id=context.tenant_id,
+            principal_id=context.principal_id,
+            trace_id=context.trace_id,
+            action=action,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            outcome=outcome,
+            policy_version=context.policy_version,
+            details=copy.deepcopy(dict(details or {})),
+        )
+        self.store.audit_events.append(event)
+        return event
+
+    def register_tenant(
+        self,
+        context: RequestContext,
+        *,
+        tenant_id: str,
+        display_name: str,
+        metadata: Mapping[str, Any] | None = None,
+    ) -> TenantRecord:
+        tenant_id = tenant_id.strip()
+        if not tenant_id or not display_name.strip():
+            raise ValueError("tenant_id and display_name are required")
+        self._authorize(
+            context,
+            "tenant.register",
+            {"resource_type": "tenant", "resource_id": tenant_id},
+        )
+        self._assert_same_tenant(context, tenant_id)
+        with self.store.lock:
+            if tenant_id in self.store.tenants:
+                raise ControlPlaneError(f"tenant already exists: {tenant_id}")
+            tenant = TenantRecord(
+                tenant_id=tenant_id,
+                display_name=display_name.strip(),
+                metadata=copy.deepcopy(dict(metadata or {})),
+            )
+            self.store.tenants[tenant_id] = tenant
+            self._audit(
+                context,
+                action="tenant.register",
+                resource_type="tenant",
+                resource_id=tenant_id,
+                outcome="allowed",
+            )
+            return tenant
+
+    def get_tenant(self, context: RequestContext, tenant_id: str | None = None) -> TenantRecord:
+        tenant_id = tenant_id or context.tenant_id
+        self._authorize(
+            context,
+            "tenant.read",
+            {"resource_type": "tenant", "resource_id": tenant_id},
+        )
+        self._assert_same_tenant(context, tenant_id)
+        with self.store.lock:
+            tenant = self.store.tenants.get(tenant_id)
+            if tenant is None:
+                raise NotFound("tenant not found")
+            return tenant
+
+    def register_agent(
+        self,
+        context: RequestContext,
+        *,
+        agent_id: str,
+        display_name: str,
+        runtime: str,
+        capabilities: Iterable[str] = (),
+        metadata: Mapping[str, Any] | None = None,
+    ) -> AgentRecord:
+        self._authorize(
+            context,
+            "agent.register",
+            {"resource_type": "agent", "resource_id": agent_id},
+        )
+        tenant = self.get_tenant(context)
+        if tenant.status != "active":
+            raise ControlPlaneError("tenant is not active")
+        normalized_capabilities = tuple(
+            sorted({str(item).strip() for item in capabilities if str(item).strip()})
+        )
+        if not agent_id.strip() or not display_name.strip() or not runtime.strip():
+            raise ValueError("agent_id, display_name and runtime are required")
+        with self.store.lock:
+            existing = self.store.agents.get(agent_id)
+            if existing is not None and existing.tenant_id != context.tenant_id:
+                raise TenantIsolationError("agent id belongs to another tenant")
+            if existing is not None:
+                raise ControlPlaneError(f"agent already exists: {agent_id}")
+            agent = AgentRecord(
+                agent_id=agent_id.strip(),
+                tenant_id=context.tenant_id,
+                display_name=display_name.strip(),
+                runtime=runtime.strip(),
+                capabilities=normalized_capabilities,
+                metadata=copy.deepcopy(dict(metadata or {})),
+            )
+            self.store.agents[agent.agent_id] = agent
+            self._audit(
+                context,
+                action="agent.register",
+                resource_type="agent",
+                resource_id=agent.agent_id,
+                outcome="allowed",
+                details={"runtime": agent.runtime},
+            )
+            return agent
+
+    def get_agent(self, context: RequestContext, agent_id: str) -> AgentRecord:
+        self._authorize(
+            context,
+            "agent.read",
+            {"resource_type": "agent", "resource_id": agent_id},
+        )
+        with self.store.lock:
+            agent = self.store.agents.get(agent_id)
+            if agent is None:
+                raise NotFound("agent not found")
+            self._assert_same_tenant(context, agent.tenant_id)
+            return agent
+
+    def list_agents(self, context: RequestContext) -> list[AgentRecord]:
+        self._authorize(
+            context,
+            "agent.read",
+            {"resource_type": "agent_collection", "resource_id": context.tenant_id},
+        )
+        with self.store.lock:
+            return [
+                agent
+                for agent in self.store.agents.values()
+                if agent.tenant_id == context.tenant_id
+            ]
+
+    def create_task(
+        self,
+        context: RequestContext,
+        *,
+        agent_id: str,
+        action: str,
+        payload: Mapping[str, Any] | None = None,
+        idempotency_key: str | None = None,
+    ) -> TaskRecord:
+        self._authorize(
+            context,
+            "task.create",
+            {"resource_type": "task", "resource_id": idempotency_key or "new"},
+        )
+        agent = self.get_agent(context, agent_id)
+        if not agent.enabled:
+            raise ControlPlaneError("agent is disabled")
+        if action not in agent.capabilities and "*" not in agent.capabilities:
+            raise PolicyDenied(f"agent does not advertise capability: {action}")
+        normalized_payload = copy.deepcopy(dict(payload or {}))
+        request_digest = _canonical_digest(
+            {"agent_id": agent_id, "action": action, "payload": normalized_payload}
+        )
+        normalized_key = idempotency_key.strip() if idempotency_key else None
+        with self.store.lock:
+            if normalized_key:
+                idempotency_ref = (context.tenant_id, normalized_key)
+                existing = self.store.idempotency.get(idempotency_ref)
+                if existing is not None:
+                    existing_task_id, existing_digest = existing
+                    if existing_digest != request_digest:
+                        raise IdempotencyConflict(
+                            "idempotency key was reused with a different request"
+                        )
+                    return self.store.tasks[existing_task_id]
+            task = TaskRecord(
+                task_id=_new_id("task"),
+                tenant_id=context.tenant_id,
+                agent_id=agent_id,
+                action=action,
+                payload=normalized_payload,
+                idempotency_key=normalized_key,
+                request_digest=request_digest,
+                created_by=context.principal_id,
+            )
+            self.store.tasks[task.task_id] = task
+            if normalized_key:
+                self.store.idempotency[(context.tenant_id, normalized_key)] = (
+                    task.task_id,
+                    request_digest,
+                )
+            self._audit(
+                context,
+                action="task.create",
+                resource_type="task",
+                resource_id=task.task_id,
+                outcome="allowed",
+                details={"agent_id": agent_id, "action": action},
+            )
+            return task
+
+    def get_task(self, context: RequestContext, task_id: str) -> TaskRecord:
+        self._authorize(
+            context,
+            "task.read",
+            {"resource_type": "task", "resource_id": task_id},
+        )
+        with self.store.lock:
+            task = self.store.tasks.get(task_id)
+            if task is None:
+                raise NotFound("task not found")
+            self._assert_same_tenant(context, task.tenant_id)
+            return task
+
+    def list_tasks(
+        self,
+        context: RequestContext,
+        *,
+        states: Iterable[TaskState | str] | None = None,
+    ) -> list[TaskRecord]:
+        self._authorize(
+            context,
+            "task.read",
+            {"resource_type": "task_collection", "resource_id": context.tenant_id},
+        )
+        state_filter = (
+            {TaskState(item) for item in states}
+            if states is not None
+            else None
+        )
+        with self.store.lock:
+            return [
+                task
+                for task in self.store.tasks.values()
+                if task.tenant_id == context.tenant_id
+                and (state_filter is None or task.state in state_filter)
+            ]
+
+    def transition_task(
+        self,
+        context: RequestContext,
+        *,
+        task_id: str,
+        target_state: TaskState | str,
+        expected_revision: int | None = None,
+        result: Mapping[str, Any] | None = None,
+        error: Mapping[str, Any] | None = None,
+    ) -> TaskRecord:
+        target = TaskState(target_state)
+        action_name = {
+            TaskState.APPROVED: "task.approve",
+            TaskState.RUNNING: "task.start",
+            TaskState.COMPLETED: "task.complete",
+            TaskState.FAILED: "task.fail",
+            TaskState.CANCELLED: "task.cancel",
+        }.get(target)
+        if action_name is None:
+            raise InvalidTransition(f"cannot transition to {target.value}")
+        self._authorize(
+            context,
+            action_name,
+            {"resource_type": "task", "resource_id": task_id},
+        )
+        with self.store.lock:
+            current = self.store.tasks.get(task_id)
+            if current is None:
+                raise NotFound("task not found")
+            self._assert_same_tenant(context, current.tenant_id)
+            if expected_revision is not None and current.revision != expected_revision:
+                raise RevisionConflict(
+                    f"expected revision {expected_revision}, current {current.revision}"
+                )
+            if target not in ALLOWED_TASK_TRANSITIONS[current.state]:
+                raise InvalidTransition(
+                    f"invalid transition {current.state.value} -> {target.value}"
+                )
+            if target == TaskState.COMPLETED and result is None:
+                raise ValueError("completed task requires result")
+            if target == TaskState.FAILED and error is None:
+                raise ValueError("failed task requires error")
+            updated = TaskRecord(
+                task_id=current.task_id,
+                tenant_id=current.tenant_id,
+                agent_id=current.agent_id,
+                action=current.action,
+                payload=current.payload,
+                state=target,
+                revision=current.revision + 1,
+                idempotency_key=current.idempotency_key,
+                request_digest=current.request_digest,
+                result=copy.deepcopy(dict(result)) if result is not None else current.result,
+                error=copy.deepcopy(dict(error)) if error is not None else current.error,
+                created_by=current.created_by,
+                approved_by=(
+                    context.principal_id
+                    if target == TaskState.APPROVED
+                    else current.approved_by
+                ),
+                created_at=current.created_at,
+                updated_at=_utc_now(),
+            )
+            self.store.tasks[task_id] = updated
+            self._audit(
+                context,
+                action=action_name,
+                resource_type="task",
+                resource_id=task_id,
+                outcome="allowed",
+                details={
+                    "from": current.state.value,
+                    "to": target.value,
+                    "revision": updated.revision,
+                },
+            )
+            return updated
+
+    def list_audit_events(
+        self,
+        context: RequestContext,
+        *,
+        resource_id: str | None = None,
+    ) -> list[AuditEvent]:
+        self._authorize(
+            context,
+            "audit.read",
+            {"resource_type": "audit", "resource_id": resource_id or context.tenant_id},
+        )
+        with self.store.lock:
+            return [
+                event
+                for event in self.store.audit_events
+                if event.tenant_id == context.tenant_id
+                and (resource_id is None or event.resource_id == resource_id)
+            ]
@@ -0,0 +1,195 @@
+"""Git-backed agent execution evidence for AIOS V2.
+
+Evidence is tenant-scoped, append-friendly, and contains no runtime credentials.
+It provides a deterministic source for claim, heartbeat, progress, result and
+verification state across Codex, OpenClaw and Hermes executors.
+"""
+from __future__ import annotations
+
+import json
+import os
+import threading
+from dataclasses import asdict, dataclass, field, replace
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _required(value: str, label: str) -> str:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise ValueError(f"{label} is required")
+    return normalized
+
+
+def _safe(value: Mapping[str, Any] | None) -> dict[str, Any]:
+    return json.loads(json.dumps(dict(value or {}), ensure_ascii=False, default=str))
+
+
+@dataclass(frozen=True)
+class ExecutionEvidence:
+    execution_id: str
+    tenant_id: str
+    task_id: str
+    agent_id: str
+    executor: str
+    branch: str
+    status: str = "claimed"
+    progress_completed: int = 0
+    progress_total: int = 1
+    current_step: str = "claim"
+    last_commit: str | None = None
+    test_status: str = "not_run"
+    test_summary: Mapping[str, Any] = field(default_factory=dict)
+    result_summary: Mapping[str, Any] = field(default_factory=dict)
+    claimed_at: str = field(default_factory=utc_now)
+    heartbeat_at: str = field(default_factory=utc_now)
+    completed_at: str | None = None
+
+    def __post_init__(self) -> None:
+        for name in (
+            "execution_id",
+            "tenant_id",
+            "task_id",
+            "agent_id",
+            "executor",
+            "branch",
+            "status",
+            "current_step",
+            "claimed_at",
+            "heartbeat_at",
+        ):
+            object.__setattr__(self, name, _required(getattr(self, name), name))
+        if self.progress_completed < 0 or self.progress_total < 1:
+            raise ValueError("invalid progress values")
+        if self.progress_completed > self.progress_total:
+            raise ValueError("progress_completed cannot exceed progress_total")
+        object.__setattr__(self, "test_summary", _safe(self.test_summary))
+        object.__setattr__(self, "result_summary", _safe(self.result_summary))
+
+
+class ExecutionTracker:
+    """Persist one JSON evidence record per tenant and execution."""
+
+    def __init__(self, root: str | Path) -> None:
+        self.root = Path(root).resolve()
+        self.root.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.RLock()
+
+    @staticmethod
+    def _component(value: str) -> str:
+        normalized = _required(value, "path component")
+        if normalized in {".", ".."} or "/" in normalized or "\\" in normalized:
+            raise ValueError("invalid evidence path component")
+        return normalized
+
+    def _path(self, tenant_id: str, execution_id: str) -> Path:
+        tenant = self._component(tenant_id)
+        execution = self._component(execution_id)
+        return self.root / tenant / f"{execution}.json"
+
+    def claim(self, evidence: ExecutionEvidence) -> ExecutionEvidence:
+        path = self._path(evidence.tenant_id, evidence.execution_id)
+        with self._lock:
+            if path.exists():
+                current = self._read(path)
+                if current.task_id != evidence.task_id or current.agent_id != evidence.agent_id:
+                    raise RuntimeError("execution_id already belongs to another task")
+                return current
+            self._write(path, evidence)
+            return evidence
+
+    def heartbeat(
+        self,
+        *,
+        tenant_id: str,
+        execution_id: str,
+        status: str | None = None,
+        current_step: str | None = None,
+        progress_completed: int | None = None,
+        progress_total: int | None = None,
+        last_commit: str | None = None,
+    ) -> ExecutionEvidence:
+        path = self._path(tenant_id, execution_id)
+        with self._lock:
+            current = self._read(path)
+            updated = replace(
+                current,
+                status=status or current.status,
+                current_step=current_step or current.current_step,
+                progress_completed=(
+                    progress_completed
+                    if progress_completed is not None
+                    else current.progress_completed
+                ),
+                progress_total=(
+                    progress_total if progress_total is not None else current.progress_total
+                ),
+                last_commit=last_commit or current.last_commit,
+                heartbeat_at=utc_now(),
+            )
+            self._write(path, updated)
+            return updated
+
+    def complete(
+        self,
+        *,
+        tenant_id: str,
+        execution_id: str,
+        status: str,
+        test_status: str,
+        test_summary: Mapping[str, Any] | None = None,
+        result_summary: Mapping[str, Any] | None = None,
+        last_commit: str | None = None,
+    ) -> ExecutionEvidence:
+        path = self._path(tenant_id, execution_id)
+        with self._lock:
+            current = self._read(path)
+            updated = replace(
+                current,
+                status=_required(status, "status"),
+                current_step="complete",
+                progress_completed=current.progress_total,
+                last_commit=last_commit or current.last_commit,
+                test_status=_required(test_status, "test_status"),
+                test_summary=_safe(test_summary),
+                result_summary=_safe(result_summary),
+                heartbeat_at=utc_now(),
+                completed_at=utc_now(),
+            )
+            self._write(path, updated)
+            return updated
+
+    def get(self, *, tenant_id: str, execution_id: str) -> ExecutionEvidence:
+        return self._read(self._path(tenant_id, execution_id))
+
+    def list(self, *, tenant_id: str) -> list[ExecutionEvidence]:
+        tenant_dir = self.root / self._component(tenant_id)
+        if not tenant_dir.exists():
+            return []
+        records = [self._read(path) for path in tenant_dir.glob("*.json")]
+        return sorted(records, key=lambda item: (item.heartbeat_at, item.execution_id), reverse=True)
+
+    @staticmethod
+    def _read(path: Path) -> ExecutionEvidence:
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except FileNotFoundError as exc:
+            raise KeyError("execution evidence not found") from exc
+        if not isinstance(value, Mapping):
+            raise RuntimeError("invalid execution evidence file")
+        return ExecutionEvidence(**dict(value))
+
+    @staticmethod
+    def _write(path: Path, evidence: ExecutionEvidence) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        temporary = path.with_suffix(f".tmp-{os.getpid()}")
+        temporary.write_text(
+            json.dumps(asdict(evidence), ensure_ascii=False, sort_keys=True, indent=2) + "\n",
+            encoding="utf-8",
+        )
+        os.replace(temporary, path)
@@ -0,0 +1,346 @@
+"""HTTP/JSON gateway for the AIOS V2 control plane.
+
+The gateway is intentionally runtime-neutral: browsers and external clients
+call this API, while OpenClaw and Hermes remain behind authenticated adapters.
+Identity headers are accepted only from a trusted reverse proxy in v1; direct
+public exposure is prohibited until signed identity middleware is enabled.
+"""
+from __future__ import annotations
+
+import dataclasses
+import json
+import os
+import re
+import urllib.parse
+from enum import Enum
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Mapping
+
+from .control_plane import (
+    ControlPlaneError,
+    ControlPlaneService,
+    IdempotencyConflict,
+    InvalidTransition,
+    NotFound,
+    PolicyDenied,
+    RequestContext,
+    RevisionConflict,
+    TaskState,
+    TenantIsolationError,
+    Unauthorized,
+)
+
+
+TENANT_HEADER = "X-AIOS-Tenant-ID"
+PRINCIPAL_HEADER = "X-AIOS-Principal-ID"
+ROLES_HEADER = "X-AIOS-Roles"
+TRACE_HEADER = "X-AIOS-Trace-ID"
+SESSION_HEADER = "X-AIOS-Session-ID"
+POLICY_HEADER = "X-AIOS-Policy-Version"
+AUTHENTICATED_HEADER = "X-AIOS-Authenticated"
+
+_RESOURCE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,191}$")
+
+
+def _json_value(value: Any) -> Any:
+    if dataclasses.is_dataclass(value):
+        return {
+            field.name: _json_value(getattr(value, field.name))
+            for field in dataclasses.fields(value)
+        }
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, Mapping):
+        return {str(key): _json_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [_json_value(item) for item in value]
+    return value
+
+
+def _require_mapping(value: Any, *, label: str = "JSON body") -> Mapping[str, Any]:
+    if not isinstance(value, Mapping):
+        raise ValueError(f"{label} must be an object")
+    return value
+
+
+def _require_resource_id(value: Any, *, label: str) -> str:
+    normalized = str(value or "").strip()
+    if not _RESOURCE_ID_RE.fullmatch(normalized):
+        raise ValueError(f"invalid {label}")
+    return normalized
+
+
+class GatewayApplication:
+    """Pure request router that can be tested without a listening socket."""
+
+    def __init__(self, service: ControlPlaneService | None = None) -> None:
+        self.service = service or ControlPlaneService()
+
+    @staticmethod
+    def context_from_headers(headers: Mapping[str, str]) -> RequestContext:
+        authenticated = str(headers.get(AUTHENTICATED_HEADER, "")).strip().lower()
+        if authenticated not in {"1", "true", "yes"}:
+            raise Unauthorized("trusted authentication assertion is required")
+        roles = [
+            role.strip()
+            for role in str(headers.get(ROLES_HEADER, "")).split(",")
+            if role.strip()
+        ]
+        return RequestContext.create(
+            tenant_id=str(headers.get(TENANT_HEADER, "")),
+            principal_id=str(headers.get(PRINCIPAL_HEADER, "")),
+            trace_id=str(headers.get(TRACE_HEADER, "")).strip() or None,
+            session_id=str(headers.get(SESSION_HEADER, "")).strip() or None,
+            policy_version=str(headers.get(POLICY_HEADER, "v1")).strip() or "v1",
+            roles=roles,
+        )
+
+    def handle(
+        self,
+        method: str,
+        raw_path: str,
+        headers: Mapping[str, str],
+        body: Any = None,
+    ) -> tuple[int, Mapping[str, Any]]:
+        parsed = urllib.parse.urlparse(raw_path)
+        path = parsed.path.rstrip("/") or "/"
+        query = urllib.parse.parse_qs(parsed.query, keep_blank_values=False)
+        method = method.upper()
+
+        if method == "GET" and path in {"/healthz", "/api/v2/health"}:
+            return HTTPStatus.OK, {
+                "service": "aios-v2-control-plane",
+                "status": "ok",
+                "store": type(self.service.store).__name__,
+            }
+
+        context = self.context_from_headers(headers)
+        payload = _require_mapping(body or {})
+
+        if method == "POST" and path == "/api/v2/tenants":
+            tenant = self.service.register_tenant(
+                context,
+                tenant_id=_require_resource_id(payload.get("tenant_id"), label="tenant_id"),
+                display_name=str(payload.get("display_name", "")),
+                metadata=_require_mapping(payload.get("metadata", {}), label="metadata"),
+            )
+            return HTTPStatus.CREATED, {"tenant": _json_value(tenant)}
+
+        if method == "GET" and path == "/api/v2/tenants/current":
+            tenant = self.service.get_tenant(context)
+            return HTTPStatus.OK, {"tenant": _json_value(tenant)}
+
+        if method == "POST" and path == "/api/v2/agents":
+            agent = self.service.register_agent(
+                context,
+                agent_id=_require_resource_id(payload.get("agent_id"), label="agent_id"),
+                display_name=str(payload.get("display_name", "")),
+                runtime=str(payload.get("runtime", "")),
+                capabilities=payload.get("capabilities", ()),
+                metadata=_require_mapping(payload.get("metadata", {}), label="metadata"),
+            )
+            return HTTPStatus.CREATED, {"agent": _json_value(agent)}
+
+        if method == "GET" and path == "/api/v2/agents":
+            return HTTPStatus.OK, {
+                "agents": _json_value(self.service.list_agents(context))
+            }
+
+        agent_match = re.fullmatch(r"/api/v2/agents/([^/]+)", path)
+        if method == "GET" and agent_match:
+            agent_id = _require_resource_id(
+                urllib.parse.unquote(agent_match.group(1)), label="agent_id"
+            )
+            return HTTPStatus.OK, {
+                "agent": _json_value(self.service.get_agent(context, agent_id))
+            }
+
+        if method == "POST" and path == "/api/v2/tasks":
+            task = self.service.create_task(
+                context,
+                agent_id=_require_resource_id(payload.get("agent_id"), label="agent_id"),
+                action=str(payload.get("action", "")).strip(),
+                payload=_require_mapping(payload.get("payload", {}), label="payload"),
+                idempotency_key=(
+                    str(payload["idempotency_key"]).strip()
+                    if payload.get("idempotency_key") is not None
+                    else None
+                ),
+            )
+            return HTTPStatus.CREATED, {"task": _json_value(task)}
+
+        if method == "GET" and path == "/api/v2/tasks":
+            requested_states = query.get("state")
+            tasks = self.service.list_tasks(context, states=requested_states)
+            return HTTPStatus.OK, {"tasks": _json_value(tasks)}
+
+        task_match = re.fullmatch(r"/api/v2/tasks/([^/]+)", path)
+        if method == "GET" and task_match:
+            task_id = _require_resource_id(
+                urllib.parse.unquote(task_match.group(1)), label="task_id"
+            )
+            return HTTPStatus.OK, {
+                "task": _json_value(self.service.get_task(context, task_id))
+            }
+
+        transition_match = re.fullmatch(
+            r"/api/v2/tasks/([^/]+)/(approve|start|complete|fail|cancel)", path
+        )
+        if method == "POST" and transition_match:
+            task_id = _require_resource_id(
+                urllib.parse.unquote(transition_match.group(1)), label="task_id"
+            )
+            transition = transition_match.group(2)
+            target = {
+                "approve": TaskState.APPROVED,
+                "start": TaskState.RUNNING,
+                "complete": TaskState.COMPLETED,
+                "fail": TaskState.FAILED,
+                "cancel": TaskState.CANCELLED,
+            }[transition]
+            expected_revision = payload.get("expected_revision")
+            if expected_revision is not None:
+                expected_revision = int(expected_revision)
+            task = self.service.transition_task(
+                context,
+                task_id=task_id,
+                target_state=target,
+                expected_revision=expected_revision,
+                result=(
+                    _require_mapping(payload.get("result"), label="result")
+                    if payload.get("result") is not None
+                    else None
+                ),
+                error=(
+                    _require_mapping(payload.get("error"), label="error")
+                    if payload.get("error") is not None
+                    else None
+                ),
+            )
+            return HTTPStatus.OK, {"task": _json_value(task)}
+
+        if method == "GET" and path == "/api/v2/audit":
+            resource_id = query.get("resource_id", [None])[0]
+            events = self.service.list_audit_events(
+                context,
+                resource_id=(
+                    _require_resource_id(resource_id, label="resource_id")
+                    if resource_id
+                    else None
+                ),
+            )
+            return HTTPStatus.OK, {"events": _json_value(events)}
+
+        return HTTPStatus.NOT_FOUND, {
+            "error": "not_found",
+            "message": "route not found",
+        }
+
+
+_ERROR_STATUS: tuple[tuple[type[BaseException], HTTPStatus], ...] = (
+    (Unauthorized, HTTPStatus.UNAUTHORIZED),
+    (PolicyDenied, HTTPStatus.FORBIDDEN),
+    (TenantIsolationError, HTTPStatus.NOT_FOUND),
+    (NotFound, HTTPStatus.NOT_FOUND),
+    (IdempotencyConflict, HTTPStatus.CONFLICT),
+    (RevisionConflict, HTTPStatus.CONFLICT),
+    (InvalidTransition, HTTPStatus.CONFLICT),
+    (ValueError, HTTPStatus.BAD_REQUEST),
+    (ControlPlaneError, HTTPStatus.BAD_REQUEST),
+)
+
+
+def error_response(exc: BaseException) -> tuple[int, Mapping[str, Any]]:
+    for error_type, status in _ERROR_STATUS:
+        if isinstance(exc, error_type):
+            return status, {
+                "error": error_type.__name__,
+                "message": str(exc),
+            }
+    return HTTPStatus.INTERNAL_SERVER_ERROR, {
+        "error": "InternalServerError",
+        "message": "internal server error",
+    }
+
+
+class GatewayRequestHandler(BaseHTTPRequestHandler):
+    server_version = "AIOSV2Gateway/1.0"
+    max_body_bytes = 1_048_576
+
+    @property
+    def application(self) -> GatewayApplication:
+        return self.server.application  # type: ignore[attr-defined]
+
+    def _read_body(self) -> Any:
+        length = int(self.headers.get("Content-Length", "0") or "0")
+        if length < 0 or length > self.max_body_bytes:
+            raise ValueError("request body is too large")
+        if length == 0:
+            return {}
+        content_type = self.headers.get("Content-Type", "").split(";", 1)[0].strip()
+        if content_type != "application/json":
+            raise ValueError("Content-Type must be application/json")
+        raw = self.rfile.read(length)
+        return json.loads(raw.decode("utf-8"))
+
+    def _send_json(self, status: int, payload: Mapping[str, Any]) -> None:
+        encoded = json.dumps(
+            _json_value(payload), ensure_ascii=False, separators=(",", ":")
+        ).encode("utf-8")
+        self.send_response(int(status))
+        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.end_headers()
+        self.wfile.write(encoded)
+
+    def _dispatch(self, method: str) -> None:
+        try:
+            body = self._read_body() if method in {"POST", "PUT", "PATCH"} else {}
+            status, payload = self.application.handle(
+                method,
+                self.path,
+                {key: value for key, value in self.headers.items()},
+                body,
+            )
+        except BaseException as exc:  # handler boundary; sanitized below
+            status, payload = error_response(exc)
+        self._send_json(status, payload)
+
+    def do_GET(self) -> None:  # noqa: N802
+        self._dispatch("GET")
+
+    def do_POST(self) -> None:  # noqa: N802
+        self._dispatch("POST")
+
+    def log_message(self, format: str, *args: Any) -> None:
+        # Production deployment routes structured access logs through the
+        # reverse proxy; do not log identity headers or request bodies here.
+        super().log_message(format, *args)
+
+
+class GatewayHTTPServer(ThreadingHTTPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+    def __init__(
+        self,
+        server_address: tuple[str, int],
+        application: GatewayApplication,
+    ) -> None:
+        self.application = application
+        super().__init__(server_address, GatewayRequestHandler)
+
+
+def main() -> None:
+    host = os.getenv("AIOS_V2_GATEWAY_HOST", "127.0.0.1")
+    port = int(os.getenv("AIOS_V2_GATEWAY_PORT", "18920"))
+    server = GatewayHTTPServer((host, port), GatewayApplication())
+    print(f"AIOS V2 control-plane gateway listening on {host}:{port}")
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,123 @@
+"""Sanitized GitOps runtime status for the AIOS Web control plane."""
+from __future__ import annotations
+
+import os
+import subprocess
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from typing import Any, Mapping
+
+
+@dataclass(frozen=True)
+class GitOpsStatus:
+    available: bool
+    service_status: str
+    workspace_mode: str
+    workspace_branch: str | None
+    workspace_ref: str | None
+    workspace_dirty: bool | None
+    candidate_branch: str
+    release_channel: str
+    release_version: str
+    sync_status: str
+    details: Mapping[str, Any]
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def _run_git(worktree: Path, *args: str) -> tuple[int, str, str]:
+    process = subprocess.run(
+        ["git", "-C", str(worktree), *args],
+        text=True,
+        capture_output=True,
+        timeout=5,
+    )
+    return process.returncode, process.stdout.strip(), process.stderr.strip()
+
+
+def collect_gitops_status() -> GitOpsStatus:
+    """Return only non-sensitive Git/workspace metadata.
+
+    Remote URLs, environment values, credentials, filesystem contents and
+    command stderr are deliberately excluded from the Web response.
+    """
+
+    candidate_branch = os.getenv(
+        "AIOS_CANDIDATE_BRANCH",
+        "integration/aios-v2r1-cto-production-candidate-20260720",
+    )
+    release_channel = os.getenv("AIOS_RELEASE_CHANNEL", "candidate")
+    release_version = os.getenv("AIOS_RELEASE_VERSION", "V2R1C0-candidate")
+    sync_status = os.getenv("AIOS_GITOPS_SYNC_STATUS", "unknown")
+    worktree = Path(
+        os.getenv(
+            "AIOS_GITOPS_CONTROLLER_WORKTREE",
+            "/data/aios-gitops/worktrees/controller",
+        )
+    )
+
+    if not worktree.is_dir():
+        return GitOpsStatus(
+            available=False,
+            service_status="unknown",
+            workspace_mode="unavailable",
+            workspace_branch=None,
+            workspace_ref=None,
+            workspace_dirty=None,
+            candidate_branch=candidate_branch,
+            release_channel=release_channel,
+            release_version=release_version,
+            sync_status=sync_status,
+            details={"reason": "controller_worktree_unavailable"},
+        )
+
+    inside_code, inside, _ = _run_git(worktree, "rev-parse", "--is-inside-work-tree")
+    if inside_code != 0 or inside != "true":
+        return GitOpsStatus(
+            available=False,
+            service_status="unknown",
+            workspace_mode="invalid",
+            workspace_branch=None,
+            workspace_ref=None,
+            workspace_dirty=None,
+            candidate_branch=candidate_branch,
+            release_channel=release_channel,
+            release_version=release_version,
+            sync_status=sync_status,
+            details={"reason": "controller_worktree_invalid"},
+        )
+
+    branch_code, branch, _ = _run_git(worktree, "branch", "--show-current")
+    ref_code, ref_value, _ = _run_git(worktree, "rev-parse", "HEAD")
+    dirty_code, dirty_value, _ = _run_git(worktree, "status", "--porcelain=v1")
+
+    service_status = "unknown"
+    try:
+        process = subprocess.run(
+            ["systemctl", "is-active", "aios-gitops-model-api"],
+            text=True,
+            capture_output=True,
+            timeout=3,
+        )
+        service_status = process.stdout.strip() or "inactive"
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        service_status = "unavailable"
+
+    branch_value = branch if branch_code == 0 and branch else None
+    return GitOpsStatus(
+        available=True,
+        service_status=service_status,
+        workspace_mode="branch" if branch_value else "detached",
+        workspace_branch=branch_value,
+        workspace_ref=ref_value if ref_code == 0 else None,
+        workspace_dirty=bool(dirty_value) if dirty_code == 0 else None,
+        candidate_branch=candidate_branch,
+        release_channel=release_channel,
+        release_version=release_version,
+        sync_status=sync_status,
+        details={
+            "controller_clean": dirty_code == 0 and not dirty_value,
+            "controller_detached": branch_code == 0 and not branch,
+        },
+    )
@@ -0,0 +1,264 @@
+"""AIOS V2 execution orchestrator.
+
+The orchestrator is the only component allowed to translate an approved AIOS
+task into an OpenClaw command or Hermes job. OpenClaw and Hermes remain pure
+execution runtimes and never own tenant policy or task lifecycle.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+from aios_hermes import HermesAdapter, HermesJob, HermesResult
+from aios_openclaw import (
+    CommandState,
+    OpenClawAdapter,
+    OpenClawCommand,
+    OpenClawResult,
+    sign_command,
+)
+
+from .control_plane import (
+    AgentRecord,
+    ControlPlaneService,
+    RequestContext,
+    TaskRecord,
+    TaskState,
+)
+
+
+class OrchestrationError(RuntimeError):
+    """AIOS failed to dispatch or reconcile a runtime execution."""
+
+
+@dataclass(frozen=True)
+class DispatchRecord:
+    tenant_id: str
+    task_id: str
+    agent_id: str
+    runtime: str
+    runtime_execution_id: str
+    state: str
+    trace_id: str
+
+
+class ExecutionOrchestrator:
+    """Dispatch approved AIOS tasks to runtime adapters and reconcile results."""
+
+    def __init__(
+        self,
+        control_plane: ControlPlaneService,
+        *,
+        openclaw: OpenClawAdapter | None = None,
+        hermes: HermesAdapter | None = None,
+        openclaw_signing_material_resolver=None,
+        openclaw_key_id: str = "default",
+        default_cell_id: str = "default",
+        default_profile_id: str = "default",
+    ) -> None:
+        self.control_plane = control_plane
+        self.openclaw = openclaw
+        self.hermes = hermes
+        self.openclaw_signing_material_resolver = openclaw_signing_material_resolver
+        self.openclaw_key_id = openclaw_key_id
+        self.default_cell_id = default_cell_id
+        self.default_profile_id = default_profile_id
+        self._dispatches: dict[tuple[str, str], DispatchRecord] = {}
+
+    def dispatch(
+        self,
+        context: RequestContext,
+        *,
+        task_id: str,
+        timeout_seconds: float | None = None,
+    ) -> DispatchRecord:
+        task = self.control_plane.get_task(context, task_id)
+        if task.state != TaskState.APPROVED:
+            raise OrchestrationError("only APPROVED tasks may be dispatched")
+        agent = self.control_plane.get_agent(context, task.agent_id)
+        existing = self._dispatches.get((task.tenant_id, task.task_id))
+        if existing is not None:
+            return existing
+
+        running = self.control_plane.transition_task(
+            context,
+            task_id=task.task_id,
+            target_state=TaskState.RUNNING,
+            expected_revision=task.revision,
+        )
+
+        runtime = agent.runtime.strip().lower()
+        try:
+            if runtime == "openclaw":
+                record = self._dispatch_openclaw(
+                    context,
+                    agent,
+                    running,
+                    timeout_seconds=timeout_seconds,
+                )
+            elif runtime == "hermes":
+                record = self._dispatch_hermes(context, agent, running)
+            else:
+                raise OrchestrationError(f"unsupported runtime: {agent.runtime}")
+        except Exception as exc:
+            current = self.control_plane.get_task(context, task.task_id)
+            if current.state == TaskState.RUNNING:
+                self.control_plane.transition_task(
+                    context,
+                    task_id=task.task_id,
+                    target_state=TaskState.FAILED,
+                    expected_revision=current.revision,
+                    error={"code": "dispatch_failed", "message": str(exc)},
+                )
+            raise
+
+        self._dispatches[(task.tenant_id, task.task_id)] = record
+        return record
+
+    def _dispatch_openclaw(
+        self,
+        context: RequestContext,
+        agent: AgentRecord,
+        task: TaskRecord,
+        *,
+        timeout_seconds: float | None,
+    ) -> DispatchRecord:
+        if self.openclaw is None or self.openclaw_signing_material_resolver is None:
+            raise OrchestrationError("OpenClaw runtime is not configured")
+        key_material = self.openclaw_signing_material_resolver(
+            task.tenant_id,
+            self.openclaw_key_id,
+        )
+        command = OpenClawCommand.create(
+            tenant_id=task.tenant_id,
+            task_id=task.task_id,
+            trace_id=context.trace_id,
+            action=task.action,
+            payload=task.payload,
+            idempotency_key=task.idempotency_key or task.task_id,
+            key_id=self.openclaw_key_id,
+        )
+        command = sign_command(command, key_material)
+        result = self.openclaw.execute(command, timeout_seconds=timeout_seconds)
+        self.reconcile_openclaw(context, result)
+        return DispatchRecord(
+            tenant_id=task.tenant_id,
+            task_id=task.task_id,
+            agent_id=agent.agent_id,
+            runtime="openclaw",
+            runtime_execution_id=command.command_id,
+            state=result.state.value,
+            trace_id=context.trace_id,
+        )
+
+    def _dispatch_hermes(
+        self,
+        context: RequestContext,
+        agent: AgentRecord,
+        task: TaskRecord,
+    ) -> DispatchRecord:
+        if self.hermes is None:
+            raise OrchestrationError("Hermes runtime is not configured")
+        metadata = dict(agent.metadata)
+        job = HermesJob.create(
+            tenant_id=task.tenant_id,
+            cell_id=str(metadata.get("cell_id") or self.default_cell_id),
+            profile_id=str(metadata.get("profile_id") or self.default_profile_id),
+            task_id=task.task_id,
+            trace_id=context.trace_id,
+            action=task.action,
+            payload=task.payload,
+            idempotency_key=task.idempotency_key or task.task_id,
+            priority=int(metadata.get("priority", 100)),
+        )
+        queued = self.hermes.enqueue(job)
+        return DispatchRecord(
+            tenant_id=task.tenant_id,
+            task_id=task.task_id,
+            agent_id=agent.agent_id,
+            runtime="hermes",
+            runtime_execution_id=queued.job_id,
+            state=queued.state.value,
+            trace_id=context.trace_id,
+        )
+
+    def reconcile_openclaw(
+        self,
+        context: RequestContext,
+        result: OpenClawResult,
+    ) -> TaskRecord:
+        task = self.control_plane.get_task(context, result.task_id)
+        if result.tenant_id != task.tenant_id:
+            raise OrchestrationError("OpenClaw result tenant mismatch")
+        if task.state != TaskState.RUNNING:
+            return task
+        if result.state == CommandState.COMPLETED:
+            return self.control_plane.transition_task(
+                context,
+                task_id=task.task_id,
+                target_state=TaskState.COMPLETED,
+                expected_revision=task.revision,
+                result=result.output or {},
+            )
+        if result.state in {
+            CommandState.FAILED,
+            CommandState.TIMED_OUT,
+            CommandState.REJECTED,
+        }:
+            return self.control_plane.transition_task(
+                context,
+                task_id=task.task_id,
+                target_state=TaskState.FAILED,
+                expected_revision=task.revision,
+                error=result.error or {"code": result.state.value.lower()},
+            )
+        if result.state == CommandState.CANCELLED:
+            return self.control_plane.transition_task(
+                context,
+                task_id=task.task_id,
+                target_state=TaskState.CANCELLED,
+                expected_revision=task.revision,
+            )
+        return task
+
+    def reconcile_hermes(
+        self,
+        context: RequestContext,
+        result: HermesResult,
+    ) -> TaskRecord:
+        task = self.control_plane.get_task(context, result.task_id)
+        if result.tenant_id != task.tenant_id:
+            raise OrchestrationError("Hermes result tenant mismatch")
+        if task.state != TaskState.RUNNING:
+            return task
+        if result.state.value == "COMPLETED":
+            return self.control_plane.transition_task(
+                context,
+                task_id=task.task_id,
+                target_state=TaskState.COMPLETED,
+                expected_revision=task.revision,
+                result=result.output or {},
+            )
+        if result.state.value == "FAILED":
+            return self.control_plane.transition_task(
+                context,
+                task_id=task.task_id,
+                target_state=TaskState.FAILED,
+                expected_revision=task.revision,
+                error=result.error or {"code": "hermes_failed"},
+            )
+        if result.state.value == "CANCELLED":
+            return self.control_plane.transition_task(
+                context,
+                task_id=task.task_id,
+                target_state=TaskState.CANCELLED,
+                expected_revision=task.revision,
+            )
+        return task
+
+    def dispatch_snapshot(self, *, tenant_id: str) -> Mapping[str, DispatchRecord]:
+        return {
+            task_id: record
+            for (record_tenant, task_id), record in self._dispatches.items()
+            if record_tenant == tenant_id
+        }
@@ -0,0 +1,181 @@
+"""Durable single-process persistence for the AIOS V2 control plane.
+
+The candidate uses an atomic SQLite snapshot so a gateway restart does not lose
+tenants, agents, task state, idempotency records, or audit events.  The domain
+service remains storage-agnostic; a later HA backend can replace this module
+without changing API or runtime-adapter contracts.
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+import threading
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any, Callable
+
+from .control_plane import (
+    AgentRecord,
+    AuditEvent,
+    ControlPlaneService,
+    InMemoryControlPlaneStore,
+    TaskRecord,
+    TaskState,
+    TenantRecord,
+)
+
+
+class SQLiteSnapshotRepository:
+    """Store one versioned control-plane snapshot in SQLite."""
+
+    SCHEMA_VERSION = 1
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path).resolve()
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.RLock()
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path, timeout=30.0)
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA synchronous=FULL")
+        connection.execute("PRAGMA foreign_keys=ON")
+        return connection
+
+    def _initialize(self) -> None:
+        with self._connect() as connection:
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS control_plane_snapshot (
+                    singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1),
+                    schema_version INTEGER NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+                )
+                """
+            )
+
+    def load(self) -> dict[str, Any] | None:
+        with self._lock, self._connect() as connection:
+            row = connection.execute(
+                "SELECT schema_version, payload_json FROM control_plane_snapshot WHERE singleton_id = 1"
+            ).fetchone()
+        if row is None:
+            return None
+        schema_version, payload_json = row
+        if int(schema_version) != self.SCHEMA_VERSION:
+            raise RuntimeError(
+                f"unsupported control-plane snapshot version: {schema_version}"
+            )
+        decoded = json.loads(str(payload_json))
+        if not isinstance(decoded, dict):
+            raise RuntimeError("invalid control-plane snapshot")
+        return decoded
+
+    def save(self, payload: dict[str, Any]) -> None:
+        encoded = json.dumps(
+            payload,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        )
+        with self._lock, self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            connection.execute(
+                """
+                INSERT INTO control_plane_snapshot (
+                    singleton_id, schema_version, payload_json, updated_at
+                ) VALUES (1, ?, ?, CURRENT_TIMESTAMP)
+                ON CONFLICT(singleton_id) DO UPDATE SET
+                    schema_version = excluded.schema_version,
+                    payload_json = excluded.payload_json,
+                    updated_at = CURRENT_TIMESTAMP
+                """,
+                (self.SCHEMA_VERSION, encoded),
+            )
+            connection.commit()
+
+
+class DurableControlPlaneService(ControlPlaneService):
+    """Control-plane service that persists after every successful mutation."""
+
+    def __init__(
+        self,
+        repository: SQLiteSnapshotRepository,
+        *,
+        policy_evaluator=None,
+    ) -> None:
+        self.repository = repository
+        store = InMemoryControlPlaneStore()
+        self._restore(store, repository.load())
+        super().__init__(store, policy_evaluator=policy_evaluator)
+
+    @staticmethod
+    def _restore(
+        store: InMemoryControlPlaneStore,
+        snapshot: dict[str, Any] | None,
+    ) -> None:
+        if not snapshot:
+            return
+        for item in snapshot.get("tenants", []):
+            record = TenantRecord(**item)
+            store.tenants[record.tenant_id] = record
+        for item in snapshot.get("agents", []):
+            normalized = dict(item)
+            normalized["capabilities"] = tuple(normalized.get("capabilities", ()))
+            record = AgentRecord(**normalized)
+            store.agents[record.agent_id] = record
+        for item in snapshot.get("tasks", []):
+            normalized = dict(item)
+            normalized["state"] = TaskState(normalized["state"])
+            record = TaskRecord(**normalized)
+            store.tasks[record.task_id] = record
+        store.audit_events.extend(AuditEvent(**item) for item in snapshot.get("audit", []))
+        for item in snapshot.get("idempotency", []):
+            store.idempotency[(item["tenant_id"], item["key"])] = (
+                item["task_id"],
+                item["digest"],
+            )
+
+    def _snapshot(self) -> dict[str, Any]:
+        with self.store.lock:
+            return {
+                "schema_version": self.repository.SCHEMA_VERSION,
+                "tenants": [asdict(item) for item in self.store.tenants.values()],
+                "agents": [asdict(item) for item in self.store.agents.values()],
+                "tasks": [
+                    {**asdict(item), "state": item.state.value}
+                    for item in self.store.tasks.values()
+                ],
+                "audit": [asdict(item) for item in self.store.audit_events],
+                "idempotency": [
+                    {
+                        "tenant_id": tenant_id,
+                        "key": key,
+                        "task_id": task_id,
+                        "digest": digest,
+                    }
+                    for (tenant_id, key), (task_id, digest)
+                    in self.store.idempotency.items()
+                ],
+            }
+
+    def _mutate(self, operation: Callable[[], Any]) -> Any:
+        with self.store.lock:
+            result = operation()
+            self.repository.save(self._snapshot())
+            return result
+
+    def register_tenant(self, *args, **kwargs):
+        return self._mutate(lambda: super(DurableControlPlaneService, self).register_tenant(*args, **kwargs))
+
+    def register_agent(self, *args, **kwargs):
+        return self._mutate(lambda: super(DurableControlPlaneService, self).register_agent(*args, **kwargs))
+
+    def create_task(self, *args, **kwargs):
+        return self._mutate(lambda: super(DurableControlPlaneService, self).create_task(*args, **kwargs))
+
+    def transition_task(self, *args, **kwargs):
+        return self._mutate(lambda: super(DurableControlPlaneService, self).transition_task(*args, **kwargs))
@@ -0,0 +1,23 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/opt/aios/app \
+    AIOS_V2_HOST=0.0.0.0 \
+    AIOS_V2_PORT=18920 \
+    AIOS_V2_STATE_DIR=/var/lib/aios-v2
+
+RUN groupadd --system --gid 12001 aios \
+    && useradd --system --uid 12001 --gid aios --home-dir /nonexistent --shell /usr/sbin/nologin aios \
+    && install -d -o aios -g aios -m 0750 /opt/aios/app /var/lib/aios-v2
+
+WORKDIR /opt/aios/app
+COPY --chown=aios:aios active-runtime/app/ /opt/aios/app/
+
+USER aios:aios
+EXPOSE 18920
+
+HEALTHCHECK --interval=15s --timeout=5s --start-period=10s --retries=4 \
+  CMD python3 -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:18920/api/v2/health', timeout=3).read()"
+
+CMD ["python3", "-m", "aios_v2.app"]
@@ -0,0 +1,44 @@
+services:
+  aios-v2:
+    build:
+      context: ../../..
+      dockerfile: deploy/aios-v2/docker/Dockerfile
+    image: aios-v2:V2R1C0-candidate
+    init: true
+    restart: unless-stopped
+    ports:
+      - "127.0.0.1:${AIOS_V2_PORT:-18920}:18920"
+    environment:
+      AIOS_V2_HOST: 0.0.0.0
+      AIOS_V2_PORT: 18920
+      AIOS_V2_STATE_DIR: /var/lib/aios-v2
+      AIOS_OPENCLAW_URL: ${AIOS_OPENCLAW_URL:-}
+      AIOS_OPENCLAW_CREDENTIAL_REF: ${AIOS_OPENCLAW_CREDENTIAL_REF:-}
+      AIOS_OPENCLAW_SIGNING_MATERIAL_FILE: ${AIOS_OPENCLAW_SIGNING_MATERIAL_FILE:-}
+      AIOS_RELEASE_CHANNEL: candidate
+      AIOS_RELEASE_VERSION: V2R1C0-candidate
+    volumes:
+      - aios-v2-state:/var/lib/aios-v2
+      # Mount runtime signing material explicitly when OpenClaw is enabled.
+      # The host file must be mode 0600; never store it in this repository.
+      # - /etc/aios-v2/openclaw-signing-material:/run/secrets/openclaw-signing-material:ro
+    read_only: true
+    tmpfs:
+      - /tmp:rw,noexec,nosuid,nodev,size=64m
+    cap_drop:
+      - ALL
+    security_opt:
+      - no-new-privileges:true
+    healthcheck:
+      test:
+        - CMD
+        - python3
+        - -c
+        - "import urllib.request; urllib.request.urlopen('http://127.0.0.1:18920/api/v2/health', timeout=3).read()"
+      interval: 15s
+      timeout: 5s
+      retries: 4
+      start_period: 10s
+
+volumes:
+  aios-v2-state:
@@ -0,0 +1,21 @@
+# AIOS V2 candidate runtime configuration.
+# Copy to /etc/aios-v2/aios-v2.env and keep ownership root:aios, mode 0640.
+
+AIOS_V2_HOST=127.0.0.1
+AIOS_V2_PORT=18920
+AIOS_V2_STATE_DIR=/var/lib/aios-v2
+
+# Optional OpenClaw runtime. Leave blank to run with OpenClaw disabled.
+AIOS_OPENCLAW_URL=
+AIOS_OPENCLAW_CREDENTIAL_REF=
+
+# File path only. The referenced file must be mode 0600 or stricter and contain
+# at least 32 bytes. Never place signing material in this env file or Git.
+AIOS_OPENCLAW_SIGNING_MATERIAL_FILE=
+
+# Reverse proxy must authenticate users and inject X-AIOS-* identity headers.
+AIOS_TRUST_PROXY_HEADERS=true
+
+# Candidate deployment metadata exposed through health tooling.
+AIOS_RELEASE_CHANNEL=candidate
+AIOS_RELEASE_VERSION=V2R1C0-candidate
@@ -0,0 +1,86 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+BASE_URL="${AIOS_V2_BASE_URL:-http://127.0.0.1:18920}"
+STATE_DIR="${AIOS_V2_STATE_DIR:-/var/lib/aios-v2}"
+CONTROLLER_WT="${AIOS_GITOPS_CONTROLLER_WORKTREE:-/data/aios-gitops/worktrees/controller}"
+SERVICE="${AIOS_V2_SERVICE:-aios-v2}"
+
+fail() {
+  echo "FAIL: $*" >&2
+  exit 1
+}
+
+pass() {
+  echo "PASS: $*"
+}
+
+command -v python3 >/dev/null || fail "python3 not installed"
+
+if command -v systemctl >/dev/null && systemctl list-unit-files "$SERVICE.service" >/dev/null 2>&1; then
+  systemctl is-active --quiet "$SERVICE" || fail "$SERVICE is not active"
+  pass "$SERVICE is active"
+fi
+
+python3 - "$BASE_URL" <<'PY'
+import json
+import sys
+import urllib.request
+
+base_url = sys.argv[1].rstrip("/")
+with urllib.request.urlopen(f"{base_url}/api/v2/health", timeout=5) as response:
+    payload = json.loads(response.read().decode("utf-8"))
+if payload.get("status") not in {"ok", "healthy", "ready"}:
+    raise SystemExit(f"unexpected health payload: {payload}")
+print(f"PASS: HTTP health status={payload.get('status')}")
+PY
+
+[[ -d "$STATE_DIR" ]] || fail "state directory missing: $STATE_DIR"
+[[ -w "$STATE_DIR" ]] || fail "state directory not writable: $STATE_DIR"
+pass "state directory writable"
+
+if [[ -e "$STATE_DIR/control-plane.sqlite3" ]]; then
+  python3 - "$STATE_DIR/control-plane.sqlite3" <<'PY'
+import sqlite3
+import sys
+
+path = sys.argv[1]
+connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5)
+try:
+    row = connection.execute("PRAGMA quick_check").fetchone()
+finally:
+    connection.close()
+if not row or row[0] != "ok":
+    raise SystemExit(f"SQLite quick_check failed: {row}")
+print("PASS: SQLite quick_check=ok")
+PY
+else
+  echo "WARN: control-plane.sqlite3 not created yet"
+fi
+
+if [[ -d "$CONTROLLER_WT" ]] && git -C "$CONTROLLER_WT" rev-parse --is-inside-work-tree >/dev/null 2>&1; then
+  branch="$(git -C "$CONTROLLER_WT" branch --show-current)"
+  dirty="$(git -C "$CONTROLLER_WT" status --porcelain=v1)"
+  [[ -z "$branch" ]] || fail "GitOps controller attached to branch: $branch"
+  [[ -z "$dirty" ]] || fail "GitOps controller workspace is dirty"
+  pass "GitOps controller detached and clean"
+fi
+
+python3 - "$STATE_DIR/executions" <<'PY'
+import json
+import pathlib
+import sys
+
+root = pathlib.Path(sys.argv[1])
+if not root.exists():
+    print("WARN: execution evidence directory not created yet")
+    raise SystemExit(0)
+for path in root.glob("*/*.json"):
+    value = json.loads(path.read_text(encoding="utf-8"))
+    for key in ("execution_id", "tenant_id", "task_id", "executor", "status"):
+        if not value.get(key):
+            raise SystemExit(f"invalid execution evidence {path}: missing {key}")
+print("PASS: execution evidence files valid")
+PY
+
+pass "AIOS V2 candidate healthcheck complete"
@@ -0,0 +1,86 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+SERVICE="${AIOS_V2_SERVICE:-aios-v2}"
+RELEASE_ROOT="${AIOS_V2_RELEASE_ROOT:-/opt/aios/releases}"
+CURRENT_LINK="${AIOS_V2_CURRENT_LINK:-/opt/aios/current}"
+STATE_DIR="${AIOS_V2_STATE_DIR:-/var/lib/aios-v2}"
+BACKUP_ROOT="${AIOS_V2_BACKUP_ROOT:-/var/backups/aios-v2}"
+TARGET_RELEASE="${1:-}"
+RESTORE_STATE="${AIOS_V2_RESTORE_STATE:-0}"
+
+fail() {
+  echo "ERROR: $*" >&2
+  exit 1
+}
+
+[[ "${EUID}" -eq 0 ]] || fail "rollback must run as root"
+[[ -n "$TARGET_RELEASE" ]] || fail "usage: $0 <release-name-or-absolute-path>"
+
+if [[ "$TARGET_RELEASE" = /* ]]; then
+  TARGET_PATH="$(readlink -f "$TARGET_RELEASE")"
+else
+  [[ "$TARGET_RELEASE" != *"/"* && "$TARGET_RELEASE" != "." && "$TARGET_RELEASE" != ".." ]] \
+    || fail "invalid release name"
+  TARGET_PATH="$(readlink -f "$RELEASE_ROOT/$TARGET_RELEASE")"
+fi
+
+[[ -d "$TARGET_PATH/active-runtime/app" ]] || fail "target release is invalid: $TARGET_PATH"
+[[ -f "$TARGET_PATH/deploy/aios-v2/healthcheck/check.sh" ]] \
+  || fail "target release has no healthcheck"
+
+PREVIOUS_PATH=""
+if [[ -L "$CURRENT_LINK" ]]; then
+  PREVIOUS_PATH="$(readlink -f "$CURRENT_LINK")"
+fi
+
+STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
+BACKUP_DIR="$BACKUP_ROOT/rollback-$STAMP"
+install -d -m 0750 "$BACKUP_DIR"
+
+if [[ -d "$STATE_DIR" ]]; then
+  tar --xattrs --acls -C "$(dirname "$STATE_DIR")" \
+    -czf "$BACKUP_DIR/state-before-rollback.tar.gz" \
+    "$(basename "$STATE_DIR")"
+fi
+
+printf '%s\n' "$PREVIOUS_PATH" > "$BACKUP_DIR/previous-release.txt"
+printf '%s\n' "$TARGET_PATH" > "$BACKUP_DIR/target-release.txt"
+
+systemctl stop "$SERVICE"
+ln -sfn "$TARGET_PATH" "${CURRENT_LINK}.next"
+mv -Tf "${CURRENT_LINK}.next" "$CURRENT_LINK"
+systemctl daemon-reload
+systemctl start "$SERVICE"
+
+if AIOS_V2_SERVICE="$SERVICE" \
+   AIOS_V2_STATE_DIR="$STATE_DIR" \
+   "$TARGET_PATH/deploy/aios-v2/healthcheck/check.sh"; then
+  echo "ROLLBACK_COMPLETE target=$TARGET_PATH backup=$BACKUP_DIR"
+  exit 0
+fi
+
+echo "Target release failed healthcheck; restoring previous release." >&2
+systemctl stop "$SERVICE" || true
+
+if [[ -n "$PREVIOUS_PATH" && -d "$PREVIOUS_PATH" ]]; then
+  ln -sfn "$PREVIOUS_PATH" "${CURRENT_LINK}.previous"
+  mv -Tf "${CURRENT_LINK}.previous" "$CURRENT_LINK"
+fi
+
+if [[ "$RESTORE_STATE" == "1" && -f "$BACKUP_DIR/state-before-rollback.tar.gz" ]]; then
+  rm -rf "$STATE_DIR"
+  tar --xattrs --acls -C "$(dirname "$STATE_DIR")" \
+    -xzf "$BACKUP_DIR/state-before-rollback.tar.gz"
+fi
+
+systemctl daemon-reload
+systemctl start "$SERVICE" || true
+
+if [[ -n "$PREVIOUS_PATH" && -x "$PREVIOUS_PATH/deploy/aios-v2/healthcheck/check.sh" ]]; then
+  AIOS_V2_SERVICE="$SERVICE" \
+  AIOS_V2_STATE_DIR="$STATE_DIR" \
+  "$PREVIOUS_PATH/deploy/aios-v2/healthcheck/check.sh" || true
+fi
+
+fail "rollback target failed; previous release restored; evidence=$BACKUP_DIR"
@@ -0,0 +1,37 @@
+[Unit]
+Description=AIOS V2 Control Plane Candidate
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=aios
+Group=aios
+WorkingDirectory=/opt/aios/current/active-runtime/app
+EnvironmentFile=-/etc/aios-v2/aios-v2.env
+Environment=PYTHONUNBUFFERED=1
+ExecStart=/usr/bin/python3 -m aios_v2.app
+Restart=on-failure
+RestartSec=3
+TimeoutStartSec=30
+TimeoutStopSec=30
+KillSignal=SIGTERM
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+LockPersonality=true
+MemoryDenyWriteExecute=true
+ReadWritePaths=/var/lib/aios-v2
+StateDirectory=aios-v2
+StateDirectoryMode=0750
+UMask=0027
+LimitNOFILE=65536
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,243 @@
+# AIOS V2 用户需求规格
+
+- 文档版本：`V2R1C0-candidate`
+- 状态：候选版本需求基线
+- 适用范围：AIOS Control Plane、Agent Router、OpenClaw Adapter、Hermes Runtime、Web Console、GitOps、审计与部署
+
+## 1. 背景与目标
+
+现有系统已经具备 OpenClaw、Hermes、Agent Router、GitOps、局部 WebUI、审计和多租户候选能力。AIOS V2 的目标不是推倒重写，而是将这些分散能力收敛到统一、可审计、可 Web 化、可多租户隔离的控制面。
+
+AIOS V2 必须成为唯一控制面，负责租户、身份、Agent、策略、任务生命周期、审批、审计、Runtime 目录和执行证据。OpenClaw 与 Hermes 只承担执行职责，不得持有第二套控制面状态。
+
+## 2. 用户角色
+
+| 角色 | 主要诉求 |
+|---|---|
+| 平台管理员 | 管理租户、运行时、发布、GitOps 状态和全局策略 |
+| 租户管理员 | 管理本租户 Agent、任务、审批、审计和成员权限 |
+| 操作员 | 创建任务、发起执行、取消任务、查看运行状态 |
+| 审批人 | 审批高风险任务，查看策略判定依据 |
+| 审计员 | 查询本租户完整审计链和执行证据 |
+| Runtime Worker | 领取已批准任务、心跳、返回执行结果 |
+| 开发与运维人员 | 通过 GitOps 提交、测试、部署、回滚和修复 Bug |
+
+## 3. 核心用户需求
+
+### UR-001 唯一控制面
+
+所有任务创建、审批、取消、状态变化、策略判定和审计必须由 AIOS Control Plane 统一处理。
+
+验收：
+
+- Web Console、API、Codex、OpenClaw 和 Hermes 均不得绕过 AIOS 创建或修改任务生命周期。
+- OpenClaw/Hermes 的公开接口仅限执行、状态、取消、心跳和结果回传。
+
+### UR-002 复用现有能力
+
+系统不得重写已经稳定的 OpenClaw 执行能力、Hermes Runtime、Agent Router 或 GitOps 基础设施。
+
+验收：
+
+- 通过 Adapter/Boundary 接入现有能力。
+- 替换控制逻辑时保留原执行接口和可迁移数据。
+
+### UR-003 OpenClaw 与 Hermes 解耦
+
+OpenClaw 与 Hermes 必须从策略、租户、审批和任务生命周期中逐步解耦。
+
+验收：
+
+- OpenClaw 仅接收已签名、已批准、带租户上下文的命令。
+- Hermes 仅接收已批准的租户分区 Job。
+- 二者均不得直接写 Control Plane 权威状态。
+
+### UR-004 全面 Web 化
+
+所有面向用户和运维的能力必须通过 Web Console 或稳定 API 访问。
+
+Web Console 至少展示：
+
+- Tenant、Agent、Task、Approval、Runtime、Audit、Execution Evidence、GitOps Status。
+- 所有写操作由 AIOS API 完成，浏览器不得直接调用 OpenClaw/Hermes。
+
+### UR-005 多租户隔离
+
+所有模块必须支持租户隔离，且默认拒绝跨租户访问。
+
+隔离范围：
+
+- API 与会话
+- Tenant/Agent/Task 数据
+- OpenClaw 命令
+- Hermes Queue、Cell、Profile、Worker Lease
+- Memory、Cache、Artifact、SecretRef
+- Audit、Execution Evidence、日志和指标
+
+验收：
+
+- 资源记录的 `tenant_id` 不可为空。
+- 跨租户访问返回拒绝或“资源不存在”，不得泄露存在性。
+- 必须有跨租户负向测试。
+
+### UR-006 权威任务生命周期
+
+任务必须遵循统一状态机：
+
+```text
+CREATED -> APPROVED -> RUNNING -> COMPLETED
+    |          |          |
+    +----------+----------+-> CANCELLED
+                           +-> FAILED
+```
+
+验收：
+
+- 禁止越级变化，例如 `CREATED -> COMPLETED`。
+- 每次变化包含 revision、principal、trace、时间和审计事件。
+
+### UR-007 策略与审批
+
+所有有副作用任务必须经过策略判定；高风险任务必须支持人工审批。
+
+验收：
+
+- Policy Decision 包含 allow/deny、reason 和 policy version。
+- 拒绝事件进入审计。
+- Runtime 不得自行放宽策略。
+
+### UR-008 幂等与并发安全
+
+任务和 Runtime 执行必须支持幂等与乐观并发控制。
+
+验收：
+
+- 相同租户与幂等键、相同内容返回同一任务/执行。
+- 相同幂等键不同内容返回冲突。
+- revision 或 branch lease 冲突不得静默覆盖。
+
+### UR-009 可观测与执行证据
+
+平台必须能够确认 Codex、OpenClaw、Hermes 等执行器是否领取任务、当前进度、最近提交、测试结果和完成状态。
+
+验收：
+
+- 存在 claim、heartbeat、progress、result 和 evidence 记录。
+- Web/API 可按租户查询。
+- 执行证据不得包含明文密钥。
+
+### UR-010 GitOps 唯一变更流程
+
+所有代码、配置、文档和部署资产修改必须进入非 `main` 分支，通过 commit、测试、审查和受控提升。
+
+验收：
+
+- Controller worktree 必须 detached 且 clean。
+- 本地 bare repo 与 Gitea 目标分支 OID 一致。
+- 不允许直接 push `main`、直接创建生产 Tag 或绕过审批部署。
+
+### UR-011 持久化与恢复
+
+租户、Agent、Task、Audit、Idempotency 和 Execution Evidence 必须在服务重启后恢复。
+
+验收：
+
+- 候选版支持 SQLite WAL 原子快照。
+- 有 restart recovery 测试。
+- 生产版需进一步提供迁移、备份、恢复、HA 和容量证据。
+
+### UR-012 审计与合规
+
+所有控制与执行动作必须生成不可混租户的审计事件。
+
+至少记录：
+
+- tenant、principal、trace、action、resource、outcome、policy version、timestamp。
+- OpenClaw/Hermes Runtime 生命周期。
+- GitOps、部署和回滚动作。
+
+### UR-013 密钥安全
+
+Git、日志、Web、任务 payload 和审计记录中不得出现明文 API Key、密码、Token 或私钥。
+
+验收：
+
+- Git 只存 `secret_ref`、`credential_ref`、`key_id` 或受保护文件路径。
+- Secret 文件权限为 `0600` 或更严格。
+- Secret Scanner 默认拒绝可疑提交。
+
+### UR-014 部署与回滚
+
+系统必须提供 systemd、Docker/Compose、健康检查和受保护回滚入口。
+
+验收：
+
+- 服务以非 root 权限运行。
+- 状态目录独立持久化。
+- 回滚前自动备份状态。
+- 默认只回滚代码，状态恢复需要显式授权。
+
+### UR-015 文档与代码同步
+
+任何影响用户行为、架构边界、API、数据模型、状态机、安全策略、部署或运维方式的变更，必须在同一分支同步更新 `docs/aios-v2/`。
+
+验收：
+
+- 架构/功能代码变化但无文档变化时，文档同步门禁失败。
+- Traceability Matrix 必须指向当前代码和测试。
+- 新架构决策必须增加或更新 ADR。
+
+## 4. 非功能需求
+
+### NFR-001 安全
+
+- 默认拒绝、最小权限、租户隔离、身份可信链、SecretRef、审计。
+
+### NFR-002 可靠性
+
+- 幂等、revision、worker lease、heartbeat、重试边界、原子持久化、健康检查和回滚。
+
+### NFR-003 可维护性
+
+- 模块边界稳定，Control Plane 与 Runtime 解耦；代码、文档、测试和部署可追踪。
+
+### NFR-004 可扩展性
+
+- Store、Transport、Policy、EventPublisher、Runtime Adapter 均通过接口扩展。
+
+### NFR-005 可运维性
+
+- Web/API 状态、systemd/Docker、健康检查、日志、GitOps 状态、候选报告。
+
+### NFR-006 性能候选指标
+
+候选版上线前至少验证：
+
+- API P95、任务调度延迟、队列等待、Worker 心跳、SQLite 锁等待、审计写入时间。
+- 具体 SLO 在预发布基准测试后固化，禁止无证据承诺。
+
+## 5. 本候选版本范围
+
+### 已纳入代码
+
+- Control Plane、持久化、Gateway、Orchestrator、OpenClaw Adapter、Hermes Boundary、Execution Tracking、Web Console、GitOps Status、部署和回滚。
+
+### 尚需环境验证
+
+- 精确测试执行证据。
+- OpenClaw 真实 Runtime 联调。
+- 生产数据库/队列替换或容量验证。
+- 预发布性能、安全、重启和回滚演练。
+- Owner 对 merge、Tag 和生产部署的批准。
+
+## 6. 用户验收场景
+
+1. 平台管理员创建租户并注册 OpenClaw/Hermes Agent。
+2. 操作员创建任务，审批人批准。
+3. AIOS Orchestrator 选择 Runtime，并创建签名命令或租户分区 Job。
+4. Runtime 执行并返回结果。
+5. AIOS 更新 Task、Audit 和 Execution Evidence。
+6. 租户用户在 Web Console 查看完整链路。
+7. 跨租户访问被拒绝。
+8. 服务重启后状态恢复。
+9. 候选部署健康检查通过，回滚演练成功。
@@ -0,0 +1,461 @@
+# AIOS V2 目标架构
+
+- 文档版本：`V2R1C0-candidate`
+- 架构状态：目标架构已实现为候选代码，待预发布验证
+- 适用范围：Control Plane、Agent Router、OpenClaw、Hermes、Web Console、GitOps、审计、部署与运维
+
+## 1. 架构目标
+
+AIOS V2 的核心目标是把分散在 OpenClaw、Hermes、Agent Router、局部 WebUI 和 GitOps 中的控制能力收敛为一个权威控制面，同时保留现有执行能力，避免推倒重写。
+
+目标状态：
+
+```text
+用户 / Web Console / Public API
+                 |
+                 v
+          AIOS Control Plane
+ Tenant | Identity | Agent | Task | Policy | Approval | Audit
+                 |
+                 v
+      Execution Orchestrator / Agent Router
+             /                      \
+            v                        v
+   OpenClaw Adapter            Hermes Boundary
+            |                        |
+            v                        v
+   OpenClaw Runtime            Hermes Workers
+             \                      /
+              v                    v
+       Result Reconciliation + Execution Evidence
+```
+
+设计结论：
+
+- AIOS 是唯一控制面。
+- Agent Router 是调度平面，不是第二控制面。
+- OpenClaw 是执行适配与 Runtime。
+- Hermes 是队列、Worker、Profile 与执行 Runtime。
+- Web Console 只调用 AIOS API。
+- GitOps 是代码、配置、文档和发布变更的唯一入口。
+
+## 2. 从现有架构到目标架构
+
+### 2.1 现有能力
+
+现有系统已经具备：
+
+- OpenClaw 执行与 Fleet 能力。
+- Hermes Worker、Profile、队列和工作区能力。
+- Agent Router 路由经验和部分调度接口。
+- GitOps 分支、提交、审批、回滚和审计能力。
+- Tenant Context、Tenant Router、Memory Scope、SecretRef、Message Ledger 和审计候选实现。
+- 局部 WebUI、Gateway 和自动任务领取能力。
+
+### 2.2 现有问题
+
+- 控制逻辑分散，可能存在多个任务事实源。
+- OpenClaw/Hermes 内部仍可能保存租户、策略或任务状态。
+- 候选能力分散在多个分支，主线缺少权威实现。
+- WebUI 与 API 入口不统一。
+- 分支名称、代码、测试和部署证据之间缺少映射。
+- 多租户隔离多为逻辑约束，尚缺全链路生产证据。
+
+### 2.3 演进原则
+
+- 先加边界和适配器，再迁移控制权。
+- 先复用接口，再替换内部实现。
+- 先在候选分支验证，再受控提升。
+- 不允许同时存在两套权威任务生命周期。
+- 不以“大分支存在”代替验收证据。
+
+## 3. 逻辑分层
+
+### 3.1 Web 与 API 层
+
+职责：
+
+- 提供 Tenant、Agent、Task、Approval、Runtime、Audit、Execution Evidence 和 GitOps 页面/API。
+- 依赖可信身份代理注入 `X-AIOS-*` 身份头。
+- 不允许浏览器直接访问 OpenClaw 或 Hermes。
+- 不允许客户端自行选择或覆盖 `tenant_id`。
+
+候选实现：
+
+- `active-runtime/app/aios_v2/app.py`
+- `active-runtime/app/aios_v2/gateway.py`
+- `active-runtime/app/aios_v2/console.py`
+
+### 3.2 AIOS Control Plane
+
+权威职责：
+
+- Tenant Registry
+- Agent Registry
+- Task Lifecycle
+- Policy Decision
+- Approval
+- Audit Event
+- Runtime Inventory
+- Execution Evidence
+- Deployment/GitOps Status
+
+不负责：
+
+- OpenClaw 内部执行细节。
+- Hermes Worker 内部执行细节。
+- 直接持有 Runtime 凭据明文。
+
+候选实现：
+
+- `active-runtime/app/aios_v2/control_plane.py`
+- `active-runtime/app/aios_v2/persistence.py`
+- `active-runtime/app/aios_v2/execution_tracking.py`
+- `active-runtime/app/aios_v2/gitops_status.py`
+
+### 3.3 Execution Orchestrator / Agent Router
+
+职责：
+
+- 只接受 `APPROVED` 任务。
+- 根据 Agent Registry 和 Policy Result 选择 Runtime。
+- 生成 OpenClawCommand 或 HermesJob。
+- 传播 tenant、task、trace、idempotency 和 policy version。
+- 将 Runtime Result 回写 AIOS Task。
+
+不负责：
+
+- 创建租户。
+- 修改策略。
+- 绕过审批。
+- 保存第二套任务状态机。
+
+候选实现：
+
+- `active-runtime/app/aios_v2/orchestrator.py`
+
+### 3.4 OpenClaw Adapter
+
+职责：
+
+- 验证签名命令。
+- 执行、查询、取消、健康检查和回调。
+- 支持幂等、超时和租户隔离。
+- 把 Runtime 生命周期转换为 AIOS 审计事件。
+
+不负责：
+
+- Tenant、Policy、Approval、Task Lifecycle。
+- 向浏览器暴露公网控制接口。
+
+候选实现：
+
+- `active-runtime/app/aios_openclaw/contract.py`
+- `active-runtime/app/aios_openclaw/adapter.py`
+- `active-runtime/app/aios_openclaw/transport.py`
+- `active-runtime/app/aios_openclaw/health.py`
+- `active-runtime/app/aios_openclaw/callback.py`
+
+### 3.5 Hermes Runtime Boundary
+
+职责：
+
+- Tenant/Cell/Profile 分区队列。
+- Worker claim、start、heartbeat、complete、fail、cancel。
+- Lease 超时和自动回队列。
+- Result 回传和 Runtime Health。
+
+不负责：
+
+- Policy、Approval、Tenant Registry、权威 Task Lifecycle。
+
+候选实现：
+
+- `active-runtime/app/aios_hermes/contract.py`
+- `active-runtime/app/aios_hermes/adapter.py`
+- `active-runtime/app/aios_hermes/worker.py`
+
+### 3.6 Persistence
+
+候选实现使用 SQLite WAL 原子快照，持久化：
+
+- Tenant
+- Agent
+- Task
+- Audit
+- Idempotency
+- Execution Evidence 文件
+
+生产演进：
+
+- Store 接口保持稳定。
+- 可替换为 PostgreSQL、对象存储或分布式事件存储。
+- 生产替换必须提供迁移、备份、恢复、HA、容量和租户隔离证据。
+
+## 4. 多租户架构
+
+### 4.1 权威租户上下文
+
+每次请求和执行至少传播：
+
+```text
+tenant_id
+principal_id
+session_id
+trace_id
+task_id
+policy_version
+```
+
+Runtime 还传播：
+
+```text
+command_id / job_id
+cell_id
+profile_id
+idempotency_key
+worker_id
+```
+
+### 4.2 隔离层级
+
+| 层级 | 隔离要求 |
+|---|---|
+| HTTP/API | 可信身份头、租户不可由浏览器覆盖 |
+| Control Plane | 所有资源带 `tenant_id`，跨租户默认拒绝 |
+| OpenClaw | 签名命令绑定 tenant/task/trace |
+| Hermes | tenant/cell/profile 三层分区 |
+| Storage | 候选为显式租户字段；生产需 RLS 或物理隔离策略 |
+| Evidence/Audit | 仅按当前租户查询 |
+| Cache/Memory | key 必须包含 tenant namespace |
+| Secret | 只保存引用，不保存明文 |
+| Logs/Metrics | 必须包含 tenant/task/trace 标签，并限制查询范围 |
+
+### 4.3 失败行为
+
+- 跨租户资源访问返回拒绝或“未找到”。
+- 不泄露其他租户资源存在性。
+- 缺少 tenant context 时 fail closed。
+- Runtime 返回的 tenant/task/trace 不匹配时拒绝回写。
+
+## 5. 任务与执行数据流
+
+### 5.1 任务创建与审批
+
+```text
+User/API
+   |
+   v
+Trusted Identity
+   |
+   v
+AIOS Gateway
+   |
+   v
+Policy Gate
+   |
+   v
+Task CREATED
+   |
+   v
+Approval
+   |
+   v
+Task APPROVED
+```
+
+### 5.2 OpenClaw 执行
+
+```text
+APPROVED Task
+   |
+   v
+Execution Orchestrator
+   |
+   v
+Signed OpenClawCommand
+   |
+   v
+OpenClaw Adapter
+   |
+   v
+Runtime Result
+   |
+   v
+Task Reconciliation + Audit + Evidence
+```
+
+### 5.3 Hermes 执行
+
+```text
+APPROVED Task
+   |
+   v
+Execution Orchestrator
+   |
+   v
+Tenant/Cell/Profile HermesJob
+   |
+   v
+Worker Claim + Lease + Heartbeat
+   |
+   v
+HermesResult
+   |
+   v
+Task Reconciliation + Audit + Evidence
+```
+
+## 6. 安全架构
+
+### 6.1 身份边界
+
+- Public API 位于可信反向代理之后。
+- 反向代理完成登录、会话、MFA 和身份头注入。
+- 候选应用拒绝未认证身份断言。
+- 生产不得允许客户端直接伪造 `X-AIOS-*`。
+
+### 6.2 Runtime 命令完整性
+
+OpenClawCommand 使用规范化序列化和 HMAC-SHA256：
+
+- 签名覆盖 tenant、task、trace、action、payload、idempotency 和时间。
+- 支持 key ID、到期时间和时钟偏差。
+- Runtime 密钥通过受保护文件或外部 Secret Provider 解析。
+
+### 6.3 Secret 边界
+
+Git 允许：
+
+```text
+secret_ref
+credential_ref
+key_id
+受保护文件路径
+```
+
+Git 禁止：
+
+```text
+API Key
+密码
+Token
+私钥
+签名材料
+```
+
+### 6.4 Web 安全
+
+候选应用设置：
+
+- no-store
+- nosniff
+- DENY frame
+- no-referrer
+- CSP
+- 请求体大小限制
+
+生产还需验证：
+
+- CSRF/CORS
+- Rate Limit
+- Session Timeout
+- CSP Nonce/Hash
+- WAF 和反向代理访问策略
+
+## 7. 可靠性架构
+
+### 7.1 幂等
+
+- Control Plane：`tenant_id + idempotency_key + request_digest`。
+- OpenClaw：`tenant_id + idempotency_key + signature`。
+- Hermes：`tenant_id + idempotency_key + job_digest`。
+
+### 7.2 并发
+
+- Task 使用 revision。
+- Hermes 使用 revision + worker lease。
+- GitOps 使用 detached controller + branch lease。
+- 持久化使用 SQLite `BEGIN IMMEDIATE` 和原子快照。
+
+### 7.3 恢复
+
+- Controller restart：detached guard。
+- AIOS restart：SQLite restore。
+- Worker crash：lease 到期后 Job 回到 QUEUED。
+- Release failure：健康检查失败自动恢复 previous release。
+
+## 8. 可观测架构
+
+统一关联键：
+
+```text
+tenant_id
+principal_id
+trace_id
+task_id
+agent_id
+command_id / job_id
+worker_id
+commit
+```
+
+Web Console 展示：
+
+- Tasks
+- Agents
+- Runtime Health
+- Audit
+- Execution Evidence
+- GitOps Status
+
+生产还需接入：
+
+- 结构化日志
+- Metrics
+- Distributed Tracing
+- Alerting
+- SLO Dashboard
+
+## 9. 部署架构
+
+支持两种候选部署：
+
+### systemd
+
+- 非 root `aios` 用户。
+- `ProtectSystem=strict`、`NoNewPrivileges` 等安全选项。
+- 状态目录 `/var/lib/aios-v2`。
+
+### Docker/Compose
+
+- 非 root UID/GID。
+- Read-only root filesystem。
+- Drop all capabilities。
+- State volume。
+- HTTP healthcheck。
+
+## 10. 架构约束
+
+以下行为视为架构违规：
+
+1. OpenClaw/Hermes 直接创建或批准 AIOS Task。
+2. Web Console 直接调用 Runtime。
+3. 新模块不携带 `tenant_id`。
+4. 新增第二套任务状态机或租户事实源。
+5. 在 Git 中提交明文凭据。
+6. 绕过 GitOps 修改生产代码/配置。
+7. 架构或功能变化未同步更新 `docs/aios-v2/`。
+8. 未通过测试和预发布门禁直接提升候选版本。
+
+## 11. 目标架构完成判定
+
+目标架构只有在以下证据齐全时才视为完成：
+
+- 文档与代码映射完整。
+- Candidate tests 对精确 commit 执行通过。
+- OpenClaw/Hermes 实际 Runtime 联调通过。
+- 多租户负向测试通过。
+- 重启恢复和回滚演练通过。
+- GitOps 本地/Gitea OID 一致。
+- Owner 批准 merge、Tag 和生产部署。
@@ -0,0 +1,313 @@
+# AIOS V2 迁移与交付策略
+
+- 文档版本：`V2R1C0-candidate`
+- 交付原则：增量迁移、能力复用、控制权收敛、GitOps 唯一变更入口
+
+## 1. 迁移目标
+
+AIOS V2 不通过一次性重写替换现有系统，而是在保留 OpenClaw、Hermes、Agent Router、GitOps 和既有运行能力的前提下，将租户、策略、任务、审批、审计和发布治理收敛到 AIOS Control Plane。
+
+迁移完成的标志不是“新代码存在”，而是：
+
+- AIOS 成为唯一权威控制面；
+- OpenClaw/Hermes 不再保存第二套控制状态；
+- 所有控制功能 Web/API 化；
+- 所有模块通过租户隔离、测试和预发布门禁；
+- 本地 Git、Gitea、测试、部署和回滚证据可追踪。
+
+## 2. 为什么采用增量迁移
+
+### 2.1 避免能力回退
+
+OpenClaw、Hermes 和 Agent Router 已经包含大量运行经验、边界处理和执行能力。整体重写会扩大风险面，并导致业务能力、操作流程和历史数据回退。
+
+### 2.2 保持可回滚
+
+每个迁移步骤必须：
+
+- 独立分支；
+- 小范围提交；
+- 独立测试；
+- 可单独回滚；
+- 不要求立即关闭旧执行路径。
+
+### 2.3 先收敛控制权，再优化内部实现
+
+迁移优先级：
+
+```text
+唯一入口
+  -> 权威租户/任务/策略
+  -> Runtime Adapter
+  -> 数据层生产化
+  -> Web 与可观测
+  -> 性能与 HA
+```
+
+## 3. 分阶段迁移路线
+
+### Phase 0：GitOps 与架构事实源
+
+目标：建立可靠变更链和权威设计基线。
+
+已完成候选能力：
+
+- Controller detached guard；
+- 避免 fetch 到已检出本地分支；
+- Gitea SSH 镜像；
+- `docs/aios-v2/` 架构事实源；
+- 候选分支、部署和回滚资产。
+
+退出条件：
+
+- Controller detached 且 clean；
+- 本地/Gitea OID 一致；
+- 架构与需求文档纳入版本。
+
+### Phase 1：Control Plane 权威化
+
+目标：将 Tenant、Agent、Task、Policy、Approval 和 Audit 收敛到 AIOS。
+
+候选实现：
+
+- `aios_v2/control_plane.py`
+- `aios_v2/persistence.py`
+- `aios_v2/gateway.py`
+- `aios_v2/app.py`
+
+退出条件：
+
+- 任务状态机统一；
+- 幂等、revision、审计和跨租户拒绝通过测试；
+- 服务重启后权威状态恢复。
+
+### Phase 2：OpenClaw 解耦
+
+目标：OpenClaw 从控制逻辑转为执行适配器。
+
+步骤：
+
+1. 定义签名命令合同；
+2. 引入 Adapter；
+3. 将直接调用迁移到 AIOS Orchestrator；
+4. Runtime 只返回执行结果和健康状态；
+5. 关闭或网络隔离旧公共控制入口。
+
+退出条件：
+
+- 未批准任务无法生成有效命令；
+- 命令绑定 tenant/task/trace/idempotency；
+- 旧控制接口冻结并记录替代路径。
+
+### Phase 3：Hermes 解耦
+
+目标：Hermes 只承担租户分区队列和 Worker Runtime。
+
+步骤：
+
+1. 定义 HermesJob/HermesResult；
+2. 强制 tenant/cell/profile 分区；
+3. 引入 claim/lease/heartbeat；
+4. 将结果统一回写 AIOS；
+5. 移除 Hermes 内部策略和任务事实源。
+
+退出条件：
+
+- Worker 无权创建 AIOS Task；
+- 租约过期后 Job 可恢复；
+- 跨租户 claim/read/result 被拒绝。
+
+### Phase 4：Web 与运维统一
+
+目标：所有用户与运维功能通过 AIOS Web/API。
+
+范围：
+
+- Tenant、Agent、Task、Runtime、Audit、Execution Evidence、GitOps；
+- systemd/Docker/Compose；
+- healthcheck、rollback、release report。
+
+退出条件：
+
+- 浏览器不直接调用 Runtime；
+- 生产身份代理和权限模型通过验证；
+- 部署、重启和回滚演练通过。
+
+### Phase 5：生产数据层与 HA
+
+目标：将候选 SQLite/内存运行状态替换或强化为生产级后端。
+
+要求：
+
+- 数据迁移；
+- RLS 或物理隔离；
+- 备份恢复；
+- HA 与容量；
+- 多实例并发；
+- 灾难恢复目标。
+
+该阶段不得改变上层合同，避免 Runtime 和 Web 再次耦合存储实现。
+
+## 4. 分支与提交策略
+
+### 4.1 分支类型
+
+| 类型 | 用途 |
+|---|---|
+| `task/*` | 可执行任务合同与主体实现 |
+| `patch/*` | 独立修复或窄范围增强 |
+| `integration/*` | 多模块候选收敛 |
+| `result/*` | 自动执行结果与证据 |
+| `release/*` | 受控发布候选 |
+
+### 4.2 规则
+
+- 禁止直接修改 `main`；
+- 一个提交应可独立审阅和回滚；
+- 大文件和大范围变更应拆分；
+- 所有行为变化必须同步文档和测试；
+- 生产 Tag、部署和回滚需要 Owner 批准。
+
+## 5. GitOps 交付流程
+
+```text
+需求/架构更新
+      |
+      v
+任务分支 + 文档 + 测试
+      |
+      v
+代码提交 + Secret Gate
+      |
+      v
+本地 bare repo 同步
+      |
+      v
+Gitea 镜像与 OID 校验
+      |
+      v
+Candidate 集成
+      |
+      v
+自动测试与预发布
+      |
+      v
+Owner 审批
+      |
+      v
+Merge / Tag / Production Deploy
+```
+
+### 5.1 同步状态
+
+GitOps 返回必须区分：
+
+- local sync；
+- Gitea sync；
+- branch OID；
+- workspace mode；
+- dirty state。
+
+仅返回 `common_git_dir synced` 不足以证明 Web 远端可见。
+
+## 6. 发布门禁
+
+### Gate A：设计完整性
+
+- 用户需求、目标架构、模块契约、ADR 和追踪矩阵同步；
+- 无冲突标记、断链或过期路径。
+
+### Gate B：代码质量
+
+- 包可导入；
+- 静态编译通过；
+- Secret Scanner 通过；
+- 无已知越权旁路。
+
+### Gate C：测试
+
+- 单元、集成、多租户负向、重启恢复、Runtime 边界测试通过；
+- 记录 commit、命令、exit code、pass/fail 和 artifact digest。
+
+### Gate D：预发布
+
+- systemd 或 Compose 启动成功；
+- 健康检查通过；
+- OpenClaw/Hermes 真实联调通过；
+- 备份、重启和回滚演练通过；
+- 日志中无敏感信息。
+
+### Gate E：生产批准
+
+- Owner 批准 merge、Tag 和生产部署；
+- 发布窗口、回滚负责人和观察指标明确；
+- 生产变更通过 GitOps 记录。
+
+## 7. 回滚策略
+
+### 7.1 代码回滚
+
+- 使用 `/opt/aios/current` 原子 symlink 切换；
+- 回滚前备份状态；
+- 健康检查失败自动恢复 previous release。
+
+### 7.2 状态回滚
+
+默认不自动回滚状态，避免覆盖候选运行期间新增的 Task、Audit 和 Evidence。
+
+状态恢复只有在：
+
+- 明确确认 schema 不兼容；
+- 数据备份可验证；
+- Owner 显式授权；
+
+时启用。
+
+### 7.3 Runtime 回滚
+
+OpenClaw/Hermes Adapter 采用稳定合同。若候选执行层失败，可恢复旧 Runtime 版本，但不得恢复旧控制权旁路。
+
+## 8. Bug 修复流程
+
+生产或预发布 Bug：
+
+1. 创建 `patch/bugfix/*`；
+2. 添加可复现测试；
+3. 修复代码；
+4. 更新相关需求/架构/契约/ADR；
+5. 运行候选验收；
+6. 进入新的 release candidate；
+7. Owner 批准部署。
+
+禁止在生产目录直接编辑文件后不回写 Git。
+
+## 9. 文档同步规则
+
+以下变化必须更新文档：
+
+- 用户行为或页面；
+- API 路径、Header、请求/响应；
+- 数据模型、状态机、策略；
+- OpenClaw/Hermes/Router 边界；
+- 多租户与安全模型；
+- 部署、健康检查、回滚；
+- 测试或发布门禁。
+
+对应关系：
+
+- 需求变化：更新 `01-USER-REQUIREMENTS.md`；
+- 架构变化：更新 `02-TARGET-ARCHITECTURE.md` 和 ADR；
+- 迁移/发布变化：更新本文件；
+- 合同变化：更新 `04-MODULE-CONTRACTS.md`；
+- 代码/测试路径变化：更新 `05-TRACEABILITY-MATRIX.md`。
+
+## 10. 当前候选后续步骤
+
+1. 完成文档同步门禁和 Acceptance Runner；
+2. 在精确 Candidate commit 执行测试；
+3. 推送 Gitea 并校验 OID；
+4. 部署隔离预发布环境；
+5. 联调真实 OpenClaw/Hermes；
+6. 执行重启、性能、安全和回滚演练；
+7. 生成最终 evidence；
+8. 请求 Owner 的 merge/Tag/production approval。
@@ -0,0 +1,502 @@
+# AIOS V2 模块契约
+
+- 文档版本：`V2R1C0-candidate`
+- 状态：候选实现契约基线
+- 原则：Control Plane 权威、Runtime 执行、全链路租户隔离、所有状态可审计
+
+## 1. 通用上下文契约
+
+所有控制与执行请求必须携带以下上下文：
+
+| 字段 | 必填 | 说明 |
+|---|---:|---|
+| `tenant_id` | 是 | 权威租户标识 |
+| `principal_id` | 控制请求是 | 已认证主体 |
+| `trace_id` | 是 | 全链路追踪 |
+| `task_id` | 执行阶段是 | AIOS 权威任务 |
+| `session_id` | Web/交互请求建议 | 用户会话 |
+| `policy_version` | 控制请求是 | 策略版本 |
+| `idempotency_key` | 创建/执行是 | 幂等键 |
+
+规则：
+
+- 缺少必填字段时 fail closed。
+- 客户端不得覆盖由可信代理或 AIOS 生成的 `tenant_id`。
+- Runtime 返回的 tenant/task/trace 必须与请求完全匹配。
+- 所有持久化键、缓存键和队列分区必须包含租户命名空间。
+
+## 2. Tenant Contract
+
+### 2.1 TenantRecord
+
+```text
+tenant_id
+display_name
+status
+metadata
+created_at
+updated_at
+```
+
+约束：
+
+- `tenant_id` 不可为空且不可变。
+- `status` 至少支持 `active` 和受控停用状态。
+- Tenant 只由 AIOS Control Plane 创建和维护。
+- Runtime 不得创建租户。
+
+实现：`active-runtime/app/aios_v2/control_plane.py`
+
+## 3. Agent Contract
+
+### 3.1 AgentRecord
+
+```text
+agent_id
+tenant_id
+display_name
+runtime
+capabilities
+enabled
+metadata
+created_at
+updated_at
+```
+
+约束：
+
+- Agent 属于唯一租户。
+- `runtime` 由 Registry 决定，可取 `openclaw`、`hermes` 或后续注册 Runtime。
+- `capabilities` 是允许调度的动作集合。
+- `metadata` 可存 `cell_id`、`profile_id`、priority 等非敏感引用。
+- 不得在 Agent metadata 中保存明文凭据。
+
+## 4. Task Contract
+
+### 4.1 TaskRecord
+
+```text
+task_id
+tenant_id
+agent_id
+action
+payload
+state
+revision
+idempotency_key
+request_digest
+result
+error
+created_by
+approved_by
+created_at
+updated_at
+```
+
+### 4.2 状态机
+
+```text
+CREATED -> APPROVED -> RUNNING -> COMPLETED
+    |          |          |
+    +----------+----------+-> CANCELLED
+                           +-> FAILED
+```
+
+允许转换：
+
+| 当前 | 允许目标 |
+|---|---|
+| CREATED | APPROVED, CANCELLED |
+| APPROVED | RUNNING, CANCELLED |
+| RUNNING | COMPLETED, FAILED, CANCELLED |
+| COMPLETED | 无 |
+| FAILED | 无 |
+| CANCELLED | 无 |
+
+约束：
+
+- 每次转换必须校验 `revision`。
+- COMPLETED 必须带 `result`。
+- FAILED 必须带 `error`。
+- 每次转换生成 AuditEvent。
+- Runtime 不得直接写 Task Store，只能通过 Orchestrator reconcile。
+
+## 5. Policy Contract
+
+### 5.1 PolicyDecision
+
+```text
+allowed
+reason
+policy_version
+```
+
+约束：
+
+- Policy 默认拒绝。
+- Runtime 不得覆盖 PolicyDecision。
+- 拒绝必须审计。
+- 高风险动作必须经过 Approval。
+
+## 6. Audit Contract
+
+### 6.1 AuditEvent
+
+```text
+event_id
+tenant_id
+principal_id
+trace_id
+action
+resource_type
+resource_id
+outcome
+policy_version
+details
+created_at
+```
+
+约束：
+
+- 按租户查询。
+- `details` 不得包含明文凭据。
+- OpenClaw/Hermes 生命周期事件必须映射到统一 AuditEvent。
+- 生产版需增加不可变外部固化和完整性验证。
+
+## 7. OpenClaw Contract
+
+### 7.1 OpenClawCommand
+
+```text
+contract_version
+command_id
+tenant_id
+task_id
+trace_id
+action
+payload
+idempotency_key
+created_at
+expires_at
+key_id
+signature
+```
+
+签名覆盖：
+
+- command_id
+- tenant_id
+- task_id
+- trace_id
+- action
+- payload
+- idempotency_key
+- created_at / expires_at
+- contract_version / key_id
+
+约束：
+
+- 使用规范化 JSON 和 HMAC-SHA256。
+- 只接受已批准任务生成的命令。
+- 过期、未来时间超差、签名错误命令必须拒绝。
+- `key_id` 只引用运行时材料，不包含材料本身。
+
+### 7.2 OpenClawResult
+
+```text
+command_id
+tenant_id
+task_id
+trace_id
+state
+output
+error
+started_at
+finished_at
+runtime_id
+attempts
+```
+
+状态：
+
+```text
+ACCEPTED
+RUNNING
+COMPLETED
+FAILED
+CANCELLED
+TIMED_OUT
+REJECTED
+```
+
+约束：
+
+- Adapter 对外只接受终态结果用于 reconcile。
+- 身份字段不匹配时拒绝回写。
+- 结果通过 Orchestrator 更新 AIOS Task。
+
+### 7.3 RuntimeHealth
+
+```text
+runtime
+runtime_id
+status
+version
+heartbeat_at
+active_tasks
+workspace_isolated
+credential_ref_configured
+details
+```
+
+健康策略：
+
+- heartbeat 不得超时。
+- workspace isolation 与 credential reference 必须符合策略。
+- 不安全 Runtime 应 quarantine，不得继续接收新任务。
+
+实现：`active-runtime/app/aios_openclaw/`
+
+## 8. Hermes Contract
+
+### 8.1 HermesJob
+
+```text
+job_id
+tenant_id
+cell_id
+profile_id
+task_id
+trace_id
+action
+payload
+idempotency_key
+priority
+state
+revision
+assigned_worker_id
+lease_expires_at
+created_at
+updated_at
+```
+
+状态机：
+
+```text
+QUEUED -> CLAIMED -> RUNNING -> COMPLETED
+   |         |          |
+   +---------+----------+-> CANCELLED
+                        +-> FAILED
+```
+
+约束：
+
+- `tenant_id + cell_id + profile_id` 形成执行分区。
+- claim 只能领取同分区 Job。
+- Worker ID 不得跨分区复用。
+- Lease 到期 Job 回到 QUEUED。
+- revision 冲突不得覆盖。
+
+### 8.2 HermesResult
+
+```text
+job_id
+tenant_id
+task_id
+trace_id
+state
+worker_id
+output
+error
+started_at
+finished_at
+```
+
+约束：
+
+- COMPLETED 必须有 output。
+- FAILED 必须有 error。
+- 只允许 AIOS Orchestrator reconcile。
+
+### 8.3 HermesHealth
+
+```text
+runtime_id
+status
+version
+heartbeat_at
+active_workers
+active_jobs
+queued_jobs
+tenant_partitions
+profile_isolation_enforced
+details
+```
+
+实现：`active-runtime/app/aios_hermes/`
+
+## 9. Execution Evidence Contract
+
+### 9.1 ExecutionEvidence
+
+```text
+execution_id
+tenant_id
+task_id
+agent_id
+executor
+branch
+status
+progress_completed
+progress_total
+current_step
+last_commit
+test_status
+test_summary
+result_summary
+claimed_at
+heartbeat_at
+completed_at
+```
+
+状态建议：
+
+```text
+claimed
+running
+blocked
+complete
+failed
+cancelled
+```
+
+约束：
+
+- 按租户存储和查询。
+- 每个阻塞必须记录原因、责任方和解除条件。
+- 每次结果必须记录 branch、commit、测试命令和状态。
+- 不得包含 Secret 或完整敏感 payload。
+
+## 10. GitOps Status Contract
+
+```text
+available
+service_status
+workspace_mode
+workspace_branch
+workspace_ref
+workspace_dirty
+candidate_branch
+release_channel
+release_version
+sync_status
+details
+```
+
+约束：
+
+- Web 响应不得包含 remote URL、凭据或 stderr。
+- Controller 必须 `detached + clean`。
+- `sync_status` 必须区分 local 和 Gitea 镜像状态。
+
+## 11. Web/API Contract
+
+### 11.1 身份头
+
+可信代理注入：
+
+```text
+X-AIOS-Authenticated
+X-AIOS-Tenant-ID
+X-AIOS-Principal-ID
+X-AIOS-Roles
+X-AIOS-Trace-ID
+X-AIOS-Session-ID
+X-AIOS-Policy-Version
+```
+
+生产约束：
+
+- 代理必须删除客户端同名头后重新注入。
+- 应用不得直接暴露公网端口。
+
+### 11.2 核心 API
+
+```text
+GET  /api/v2/health
+POST /api/v2/tenants
+GET  /api/v2/tenants/current
+POST /api/v2/agents
+GET  /api/v2/agents
+POST /api/v2/tasks
+GET  /api/v2/tasks
+GET  /api/v2/tasks/{id}
+POST /api/v2/tasks/{id}/approve
+POST /api/v2/tasks/{id}/start
+POST /api/v2/tasks/{id}/complete
+POST /api/v2/tasks/{id}/fail
+POST /api/v2/tasks/{id}/cancel
+POST /api/v2/tasks/{id}/dispatch
+GET  /api/v2/audit
+GET  /api/v2/runtime
+GET  /api/v2/executions
+GET  /api/v2/gitops
+POST /api/v2/hermes/results/reconcile
+```
+
+## 12. Persistence Contract
+
+候选版：
+
+- SQLite WAL
+- `BEGIN IMMEDIATE`
+- 原子 snapshot
+- schema version
+- restart restore
+
+持久化范围：
+
+- tenants
+- agents
+- tasks
+- audit
+- idempotency
+
+生产替换要求：
+
+- 不改变 Domain/API Contract。
+- 提供 migration、backup、restore、HA、RLS/隔离和容量测试。
+
+## 13. Deployment Contract
+
+### systemd
+
+- 非 root `aios` 用户。
+- 状态目录 `/var/lib/aios-v2`。
+- 最小权限和文件系统保护。
+
+### Docker
+
+- 非 root UID/GID。
+- read-only root filesystem。
+- drop all capabilities。
+- 独立状态 volume。
+
+### Rollback
+
+- 回滚前备份状态。
+- 默认只回滚代码。
+- 状态恢复需显式授权。
+- 回滚后必须执行 healthcheck。
+
+## 14. 契约变更规则
+
+任何以下变化必须在同一提交或同一分支更新本文档：
+
+- 字段、状态、Header、API、错误语义；
+- 租户或安全约束；
+- OpenClaw/Hermes 边界；
+- Persistence 或 Deployment Contract；
+- Evidence/Audit 格式。
+
+不兼容变化必须新增 ADR 和版本号，不得静默修改。
@@ -0,0 +1,113 @@
+# AIOS V2 需求—架构—代码—测试—部署追踪矩阵
+
+- 文档版本：`V2R1C0-candidate`
+- 目的：确保每项用户需求都能追踪到架构设计、代码实现、自动测试、部署资产和验收命令
+
+## 1. 追踪规则
+
+每条需求至少具备：
+
+1. 需求编号；
+2. 架构章节；
+3. 实现路径；
+4. 测试路径；
+5. 部署/运维资产；
+6. 验收状态。
+
+状态定义：
+
+- `Implemented`：代码已进入 Candidate 分支；
+- `Test-defined`：测试代码已进入 Candidate 分支；
+- `Test-verified`：精确 commit 的测试已运行并保存证据；
+- `Pre-production verified`：预发布部署、健康检查和回滚演练已通过；
+- `Production released`：Owner 批准并完成 merge/tag/deploy。
+
+## 2. 核心追踪矩阵
+
+| 需求 | 设计依据 | 代码实现 | 测试 | 部署/运维 | 当前状态 |
+|---|---|---|---|---|---|
+| UR-001 AIOS 唯一控制面 | `02-TARGET-ARCHITECTURE.md` §3.2/§3.3；ADR-001 | `aios_v2/control_plane.py`、`aios_v2/orchestrator.py`、`aios_v2/app.py` | `tests/aios_v2_candidate/test_candidate_flow.py` | `deploy/aios-v2/systemd/aios-v2.service` | Implemented / Test-defined |
+| UR-002 复用现有能力 | `02-TARGET-ARCHITECTURE.md` §2；`03-MIGRATION-AND-DELIVERY.md` §2 | `aios_openclaw/`、`aios_hermes/` | OpenClaw/Hermes flow tests | Candidate branch + adapter boundaries | Implemented / Test-defined |
+| UR-003 OpenClaw/Hermes 解耦 | `02-TARGET-ARCHITECTURE.md` §3.4/§3.5；ADR-002 | `aios_openclaw/adapter.py`、`aios_hermes/adapter.py` | runtime boundary tests in candidate flow suite | systemd/Docker runtime configuration | Implemented / Test-defined |
+| UR-004 全面 Web 化 | `02-TARGET-ARCHITECTURE.md` §3.1/§8 | `aios_v2/console.py`、`aios_v2/gateway.py`、`aios_v2/app.py` | API/console smoke test pending | Docker/systemd + reverse proxy requirement | Implemented / Test pending |
+| UR-005 多租户隔离 | `02-TARGET-ARCHITECTURE.md` §4；ADR-004 | `control_plane.py`、`orchestrator.py`、`aios_openclaw/*`、`aios_hermes/*`、`execution_tracking.py` | `test_cross_tenant_task_read_is_denied` | tenant-scoped state/evidence directories | Implemented / Test-defined |
+| UR-006 权威任务生命周期 | `04-MODULE-CONTRACTS.md` §4 | `control_plane.py` | OpenClaw/Hermes full-flow tests | persisted state + audit | Implemented / Test-defined |
+| UR-007 策略与审批 | `04-MODULE-CONTRACTS.md` §5 | `control_plane.py`、`gateway.py` | role/policy negative tests需补充 | reverse proxy identity + audit | Implemented / Partial test |
+| UR-008 幂等与并发安全 | `02-TARGET-ARCHITECTURE.md` §7 | `control_plane.py`、`aios_openclaw/adapter.py`、`aios_hermes/adapter.py`、`worker.py` | flow + restart tests；并发专项测试需补充 | GitOps detached guard + SQLite WAL | Implemented / Partial test |
+| UR-009 执行证据 | `04-MODULE-CONTRACTS.md` §9 | `execution_tracking.py` | `test_execution_tracking_persists_claim_heartbeat_and_result` | `/var/lib/aios-v2/executions` | Implemented / Test-defined |
+| UR-010 GitOps 唯一变更流程 | `03-MIGRATION-AND-DELIVERY.md` §4/§5；ADR-003 | GitOps Runtime、`gitops_status.py` | healthcheck 的 detached/clean 检查 | `deploy/aios-v2/healthcheck/check.sh` | Implemented / Runtime verified partially |
+| UR-011 持久化与恢复 | `02-TARGET-ARCHITECTURE.md` §3.6 | `persistence.py` | `test_durable_service_recovers_state_after_restart` | state dir + backup/rollback | Implemented / Test-defined |
+| UR-012 审计与合规 | `04-MODULE-CONTRACTS.md` §6 | `control_plane.py`、`aios_openclaw/callback.py`、`app.py` | flow test + audit assertions需补充 | Console audit view | Implemented / Partial test |
+| UR-013 密钥安全 | `02-TARGET-ARCHITECTURE.md` §6；ADR-006 | `FileSigningMaterialResolver`、credential refs、Secret Gate | permission/secret-scan test需补充 | env template + external protected file | Implemented / Partial test |
+| UR-014 部署与回滚 | `03-MIGRATION-AND-DELIVERY.md` §7 | deployment scripts/assets | acceptance runner + healthcheck | `deploy/aios-v2/` | Implemented / Pre-prod pending |
+| UR-015 文档与代码同步 | `03-MIGRATION-AND-DELIVERY.md` §9；ADR-005 | `tools/aios_v2_docs_code_check.py` | checker self-test / acceptance runner | CI/Gitea Actions 可调用 | Implemented after checker commit |
+
+## 3. API 追踪
+
+| API | 需求 | 实现 | 测试状态 |
+|---|---|---|---|
+| `GET /api/v2/health` | UR-004, UR-014 | `gateway.py` | healthcheck 定义完成 |
+| `POST /api/v2/tenants` | UR-001, UR-005 | `gateway.py` / `control_plane.py` | 需增加 HTTP 集成测试 |
+| `POST /api/v2/agents` | UR-001, UR-005 | `gateway.py` / `control_plane.py` | Domain flow 已覆盖 |
+| `POST /api/v2/tasks` | UR-006, UR-008 | `gateway.py` / `control_plane.py` | Domain flow 已覆盖 |
+| `POST /api/v2/tasks/{id}/approve` | UR-006, UR-007 | `gateway.py` | Domain flow 已覆盖 |
+| `POST /api/v2/tasks/{id}/dispatch` | UR-001, UR-003 | `app.py` / `orchestrator.py` | OpenClaw/Hermes flow 已覆盖 |
+| `GET /api/v2/runtime` | UR-004, UR-009 | `app.py` | HTTP smoke test待执行 |
+| `GET /api/v2/executions` | UR-009 | `app.py` / `execution_tracking.py` | Evidence persistence test已定义 |
+| `GET /api/v2/gitops` | UR-010 | `app.py` / `gitops_status.py` | Runtime/healthcheck验证待执行 |
+| `POST /api/v2/hermes/results/reconcile` | UR-003, UR-006 | `app.py` / `orchestrator.py` | Hermes flow已覆盖 Domain 路径 |
+
+## 4. 运行时契约追踪
+
+| 契约 | 设计 | 代码 | 测试 |
+|---|---|---|---|
+| OpenClawCommand | `04-MODULE-CONTRACTS.md` §7.1 | `aios_openclaw/contract.py` | OpenClaw full-flow；签名负向测试需补充 |
+| OpenClawResult | §7.2 | `aios_openclaw/contract.py` | OpenClaw full-flow |
+| RuntimeHealth | §7.3 | `aios_openclaw/contract.py`、`health.py` | health policy测试需补充 |
+| HermesJob | §8.1 | `aios_hermes/contract.py` | Hermes full-flow |
+| HermesResult | §8.2 | `aios_hermes/contract.py` | Hermes full-flow |
+| HermesHealth | §8.3 | `aios_hermes/contract.py`、`adapter.py` | healthcheck/worker test待执行 |
+| ExecutionEvidence | §9 | `execution_tracking.py` | evidence persistence test |
+| GitOpsStatus | §10 | `gitops_status.py` | healthcheck detached/clean检查 |
+
+## 5. 部署追踪
+
+| 能力 | 文件 | 验收命令 |
+|---|---|---|
+| systemd | `deploy/aios-v2/systemd/aios-v2.service` | `systemctl status aios-v2` |
+| 环境模板 | `deploy/aios-v2/env/aios-v2.env.example` | `systemd-analyze verify` + 文件权限检查 |
+| Docker | `deploy/aios-v2/docker/Dockerfile` | `docker build` |
+| Compose | `deploy/aios-v2/docker/compose.yaml` | `docker compose config` / `up` |
+| 健康检查 | `deploy/aios-v2/healthcheck/check.sh` | 直接执行 |
+| 回滚 | `deploy/aios-v2/rollback/rollback.sh` | 预发布回滚演练 |
+| 全量验收 | `deploy/aios-v2/acceptance/run-all.sh` | 直接执行 |
+
+## 6. 文档覆盖追踪
+
+| 文档 | 覆盖内容 | 代码/变更触发条件 |
+|---|---|---|
+| `README.md` | 事实源索引、状态语言、实现地图 | 新模块/目录/版本 |
+| `01-USER-REQUIREMENTS.md` | 用户需求、NFR、验收场景 | 用户行为、功能范围、验收变化 |
+| `02-TARGET-ARCHITECTURE.md` | 分层、边界、数据流、安全、租户 | 架构边界、数据流、运行时变化 |
+| `03-MIGRATION-AND-DELIVERY.md` | 迁移、GitOps、Gate、回滚 | 发布流程、迁移阶段、门禁变化 |
+| `04-MODULE-CONTRACTS.md` | 字段、API、状态机、错误语义 | Contract/API/模型变化 |
+| `05-TRACEABILITY-MATRIX.md` | 需求到代码测试部署映射 | 代码、测试、部署路径变化 |
+| `adr/*` | 关键不可逆或跨模块决策 | 新架构决策或旧决策替换 |
+
+## 7. 当前未完成验证项
+
+以下项目代码或测试定义已存在，但尚不能标记为 `Test-verified` 或 `Pre-production verified`：
+
+1. 精确 Candidate commit 的测试执行结果；
+2. HTTP API 与 Web Console 冒烟测试；
+3. 真实 OpenClaw Runtime 签名、取消、超时和健康联调；
+4. Hermes 长任务心跳、Worker 崩溃和 Lease 恢复演练；
+5. 多实例 SQLite/并发容量边界；
+6. systemd、Docker、Compose、restart 和 rollback 演练；
+7. local/Gitea Candidate OID 一致证明；
+8. 生产身份代理、防伪 Header、CSRF/CORS/Rate Limit 验证；
+9. 文档—代码同步检查器在 CI/Acceptance 中执行通过。
+
+## 8. 更新要求
+
+任何代码、API、部署或测试路径变化，必须同步更新本矩阵。验收脚本会检查本文档引用的关键路径存在，并检查架构/功能代码变化是否同时包含 `docs/aios-v2/` 变化。
@@ -0,0 +1,98 @@
+# AIOS V2 Architecture and Delivery Index
+
+- Version line: `V2R1C0-candidate`
+- Candidate branch: `integration/aios-v2r1-cto-production-candidate-20260720`
+- Architecture owner: AIOS V2 CTO
+- Status: implementation-complete candidate; runtime and production verification pending
+
+## 1. Purpose
+
+This directory is the single source of truth for AIOS V2 requirements, target architecture, module boundaries, migration strategy, release gates and architectural decisions.
+
+The design is based on the following non-negotiable principles:
+
+1. Do not rewrite the existing platform.
+2. Reuse OpenClaw, Hermes and Agent Router capabilities.
+3. Decouple OpenClaw and Hermes from control-plane ownership.
+4. AIOS is the only control plane.
+5. All user-facing and operational capabilities are Web/API accessible.
+6. Every module enforces multi-tenant isolation.
+7. Every change flows through Git branches, commits, tests, review and controlled promotion.
+
+## 2. Document map
+
+| Document | Purpose |
+|---|---|
+| [01-USER-REQUIREMENTS.md](01-USER-REQUIREMENTS.md) | Product goals, user requirements, non-functional requirements and acceptance criteria |
+| [02-TARGET-ARCHITECTURE.md](02-TARGET-ARCHITECTURE.md) | Target logical architecture, control boundaries, data flow, security and tenancy model |
+| [03-MIGRATION-AND-DELIVERY.md](03-MIGRATION-AND-DELIVERY.md) | Incremental migration plan, reuse strategy, GitOps workflow and release gates |
+| [04-MODULE-CONTRACTS.md](04-MODULE-CONTRACTS.md) | Tenant context, task lifecycle, OpenClaw, Hermes, audit and execution-evidence contracts |
+| [05-TRACEABILITY-MATRIX.md](05-TRACEABILITY-MATRIX.md) | Requirement-to-design-to-code-to-test mapping |
+| [adr/README.md](adr/README.md) | Architecture decision records |
+
+## 3. Current implementation map
+
+| Architecture capability | Candidate implementation |
+|---|---|
+| Tenant, agent, task, policy and audit control plane | `active-runtime/app/aios_v2/control_plane.py` |
+| Durable control-plane state | `active-runtime/app/aios_v2/persistence.py` |
+| Web/API gateway | `active-runtime/app/aios_v2/gateway.py` |
+| Unified application entrypoint | `active-runtime/app/aios_v2/app.py` |
+| Runtime dispatch and reconciliation | `active-runtime/app/aios_v2/orchestrator.py` |
+| Agent execution evidence | `active-runtime/app/aios_v2/execution_tracking.py` |
+| GitOps runtime visibility | `active-runtime/app/aios_v2/gitops_status.py` |
+| Web Console | `active-runtime/app/aios_v2/console.py` |
+| OpenClaw execution boundary | `active-runtime/app/aios_openclaw/` |
+| Hermes runtime boundary | `active-runtime/app/aios_hermes/` |
+| Candidate tests | `tests/aios_v2_candidate/` |
+| Deployment, health and rollback | `deploy/aios-v2/` |
+
+## 4. Status vocabulary
+
+The project separates code availability from release readiness:
+
+- **Designed**: documented boundary and acceptance criteria exist.
+- **Implemented**: code is committed on the candidate branch.
+- **Test-defined**: automated tests exist in Git.
+- **Test-verified**: tests were executed for the exact commit and evidence was recorded.
+- **Pre-production verified**: deployed in an isolated environment with health, restart and rollback evidence.
+- **Production released**: owner-approved merge/tag/deployment completed.
+
+A branch or file existing does not, by itself, mean a capability is production-ready.
+
+## 5. Design summary
+
+```text
+Browser / Web Console / Public API
+                 |
+                 v
+          AIOS Control Plane
+  Tenant | Agent | Task | Policy | Audit
+                 |
+                 v
+       Execution Orchestrator / Router
+             /                 \
+            v                   v
+   OpenClaw Adapter       Hermes Boundary
+            |                   |
+            v                   v
+   OpenClaw Runtime       Hermes Workers
+                 \
+                  v
+      Result Reconciliation + Evidence
+```
+
+The control plane owns authoritative identity, policy, approval, task lifecycle and audit state. OpenClaw and Hermes receive only tenant-scoped, approved execution work and return runtime results.
+
+## 6. Release rule
+
+No candidate may be promoted until:
+
+- requirement-to-code traceability is complete;
+- multi-tenant negative tests pass;
+- controller workspace is detached and clean;
+- local Git and Gitea branch OIDs match;
+- persistence restart recovery passes;
+- healthcheck and rollback pass in pre-production;
+- no plaintext secret enters Git;
+- owner approval is recorded for merge, tag and production deployment.
@@ -0,0 +1,39 @@
+# ADR-001: AIOS 作为唯一控制面
+
+- Status: Accepted
+- Date: 2026-07-20
+
+## Context
+
+现有系统包含多个执行和调度能力，若每个 Runtime 保存自己的任务、租户和策略状态，会形成多个事实源。
+
+## Decision
+
+AIOS Control Plane 是唯一权威控制面，负责：
+
+- Tenant
+- Agent
+- Task
+- Policy
+- Approval
+- Audit
+- Execution Evidence
+
+OpenClaw、Hermes、Agent Router 只能执行或路由，不拥有第二套控制状态。
+
+## Consequences
+
+Positive:
+
+- 状态唯一；
+- 审计完整；
+- Web/API 统一。
+
+Negative:
+
+- Runtime 接入需要 Adapter。
+- 迁移需要阶段性边界收敛。
+
+## Compliance
+
+任何新 Runtime 必须通过 AIOS Orchestrator 接入。
@@ -0,0 +1,46 @@
+# ADR-002: OpenClaw/Hermes 与控制面解耦
+
+- Status: Accepted
+- Date: 2026-07-20
+
+## Context
+
+OpenClaw 和 Hermes 已具备执行能力，但历史上可能承担部分控制职责。
+
+## Decision
+
+引入边界层：
+
+```text
+AIOS Control Plane
+        |
+        v
+Adapter / Boundary
+        |
+        v
+Runtime
+```
+
+OpenClaw：
+
+- 执行签名命令；
+- 返回结果；
+- 提供健康状态。
+
+Hermes：
+
+- 队列；
+- Worker；
+- Lease；
+- Heartbeat；
+- 执行结果。
+
+## Consequences
+
+- Runtime 可独立演进。
+- 控制逻辑不会分裂。
+- 需要维护稳定 Contract。
+
+## Migration
+
+先接 Adapter，再迁移旧调用入口，最后关闭控制旁路。
@@ -0,0 +1,44 @@
+# ADR-003: 所有变更必须通过 GitOps
+
+- Status: Accepted
+- Date: 2026-07-20
+
+## Context
+
+直接在生产目录、长期 worktree 或主分支上修改代码，会造成不可追踪、不可回滚和分支冲突。
+
+## Decision
+
+所有代码、配置、文档、测试、部署和回滚资产必须通过以下流程：
+
+```text
+non-main branch -> commit -> local sync -> Gitea sync -> test -> review -> controlled promotion
+```
+
+Controller worktree 必须长期保持 detached 且 clean。每个写任务使用独立工作区或受控写入流程；相同分支写入必须串行。
+
+## Required evidence
+
+每个任务必须记录：
+
+- branch
+- commit
+- changed files
+- test command and result
+- blocker and resolution
+- local/Gitea sync status
+- rollback note
+
+## Prohibited
+
+- 直接 push `main`
+- 未审批创建生产 Tag
+- 在生产目录手工修改后不回写 Git
+- fetch 到已被 worktree 检出的本地 branch ref
+- 在 Git 中提交明文密钥
+
+## Consequences
+
+优点：可审计、可回滚、可并行治理。
+
+成本：紧急修复也必须形成 patch branch 和证据。
@@ -0,0 +1,43 @@
+# ADR-005: 文档与代码同步门禁
+
+- Status: Accepted
+- Date: 2026-07-20
+
+## Context
+
+架构、功能和部署变化如果不同步文档，会导致版本无法审计，后续 Agent 无法判断真实设计。
+
+## Decision
+
+所有影响以下内容的变更必须同步 `docs/aios-v2/`：
+
+- 用户行为；
+- API；
+- 数据模型；
+- 状态机；
+- Runtime Boundary；
+- 安全策略；
+- 部署流程；
+- 测试门禁。
+
+## Gate
+
+验收必须检查：
+
+1. 文档引用代码路径存在；
+2. API 文档有实现映射；
+3. Contract 字段有模型映射；
+4. Traceability Matrix 已更新；
+5. ADR 与架构变化一致。
+
+## Consequences
+
+优点：
+
+- 架构长期可维护；
+- Agent 执行有明确上下文；
+- Bug 修复有历史依据。
+
+成本：
+
+- 小功能也需要更新对应文档。
@@ -0,0 +1,39 @@
+# AIOS V2 Architecture Decision Records
+
+本目录记录影响多个模块、长期演进、兼容性、安全或交付流程的关键架构决策。
+
+## ADR 列表
+
+| ADR | 决策 |
+|---|---|
+| [ADR-001](ADR-001-single-control-plane.md) | AIOS 作为唯一控制面 |
+| [ADR-002](ADR-002-openclaw-hermes-decoupling.md) | OpenClaw/Hermes 与控制面解耦 |
+| [ADR-003](ADR-003-gitops-only-change-flow.md) | 所有变更必须通过 GitOps |
+| [ADR-004](ADR-004-multi-tenant-isolation.md) | 多租户隔离默认拒绝 |
+| [ADR-005](ADR-005-document-code-sync-gate.md) | 文档与代码同步门禁 |
+| [ADR-006](ADR-006-candidate-persistence-strategy.md) | 候选持久化采用可替换 SQLite Snapshot |
+
+## ADR 状态
+
+- `Proposed`：提出但未采纳。
+- `Accepted`：当前有效决策。
+- `Superseded`：被新 ADR 替代。
+- `Deprecated`：保留历史但禁止新实现采用。
+
+## 何时必须新增或更新 ADR
+
+以下变化必须新增或更新 ADR：
+
+1. 控制面与 Runtime 权限边界变化；
+2. 多租户隔离模型变化；
+3. API、数据模型或状态机出现不兼容变化；
+4. 持久化、消息、密钥、审计或部署方案发生跨模块变化；
+5. GitOps、发布、回滚或文档门禁变化；
+6. 替换 OpenClaw、Hermes、Agent Router 或核心基础设施。
+
+ADR 变化必须同步更新：
+
+- `02-TARGET-ARCHITECTURE.md`
+- `04-MODULE-CONTRACTS.md`
+- `05-TRACEABILITY-MATRIX.md`
+- 对应代码、测试和部署资产
@@ -0,0 +1,47 @@
+# TASK-AIOS-CONTROL-PLANE-GATEWAY-V1
+
+## Owner
+Codex
+
+## Goal
+建立 AIOS V2 唯一控制面 Gateway v1。
+
+## Architecture Boundary
+
+AIOS Control Plane:
+- Tenant Registry
+- Agent Registry
+- Task Lifecycle
+- Policy Gate
+- Approval
+- Audit Event
+
+不允许：
+- OpenClaw 直接管理任务生命周期
+- Hermes 直接暴露控制 API
+- Web UI 绕过 AIOS API
+
+## Deliverables
+
+1. Control Plane API skeleton
+2. Tenant context middleware
+3. Task state machine
+4. Agent registration contract
+5. Audit event interface
+
+## Acceptance
+
+- 所有 API 携带 tenant_id
+- 未授权跨租户访问拒绝
+- Task create/update/cancel 可审计
+- GitOps commit 可追踪
+- pytest + integration test 通过
+
+## Migration
+
+复用：
+- Agent Router
+- OpenClaw adapter
+- Hermes runtime
+
+不重写执行层。
@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "base": "task/codex/AIOS-V2-CONTROL-PLANE-GATEWAY-V1-20260720",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-20T01:47:01Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}
@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "task/codex/AIOS-V2-CONTROL-PLANE-GATEWAY-V1-20260720",
+  "base": "main",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-19T17:06:04Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}
@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "task/openclaw/AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "base": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-20T03:49:09Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}
@@ -0,0 +1,39 @@
+apiVersion: aios.v2/v1
+kind: AgentTask
+metadata:
+  name: codex-production-supervisor-task
+  owner: AIOS-Control-Plane
+spec:
+  executor: codex
+  role: architecture-supervisor
+  objective: >-
+    Supervise AIOS V2R1 production candidate deployment, resolve blocking issues,
+    review patches, and maintain architecture/document consistency.
+  responsibilities:
+    - review OpenClaw deployment evidence
+    - review Hermes runtime issues
+    - analyze failed tests and blockers
+    - approve or reject patch direction
+    - verify docs/code/traceability consistency
+    - require ADR updates for architecture changes
+    - maintain final release readiness assessment
+  prohibited:
+    - bypass GitOps process
+    - directly apply unreviewed production changes
+    - weaken tenant isolation or security gates
+    - approve missing evidence
+  required_checks:
+    - requirement traceability
+    - architecture boundary compliance
+    - test evidence
+    - secret safety
+    - rollback readiness
+    - document synchronization
+  evidence_required:
+    - claim
+    - heartbeat
+    - progress
+    - blocker
+    - review_result
+    - commit
+    - final_recommendation
@@ -0,0 +1,35 @@
+apiVersion: aios.v2/v1
+kind: AgentTask
+metadata:
+  name: openclaw-production-deploy-task
+  owner: AIOS-Control-Plane
+spec:
+  executor: openclaw
+  role: deployment-operator
+  objective: >-
+    Deploy AIOS V2R1 candidate runtime, execute health checks, integrate OpenClaw
+    runtime boundary, and fix low-risk runtime/deployment defects.
+  responsibilities:
+    - install systemd or Docker candidate deployment
+    - validate environment configuration
+    - execute healthcheck and smoke tests
+    - verify OpenClaw runtime connectivity
+    - fix small deployment/runtime defects through patch branches
+    - update ExecutionEvidence after every step
+  prohibited:
+    - modify AIOS control-plane architecture
+    - bypass GitOps workflow
+    - commit secrets
+    - directly edit production without a recorded change
+  evidence_required:
+    - claim
+    - heartbeat
+    - progress
+    - blocker
+    - commit
+    - test_status
+    - result
+  escalation:
+    - architecture change -> Codex supervisor
+    - security issue -> AIOS owner
+    - blocked deployment -> Codex supervisor
@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+APP_ROOT = REPO_ROOT / "active-runtime" / "app"
+if str(APP_ROOT) not in sys.path:
+    sys.path.insert(0, str(APP_ROOT))
+
+from aios_hermes import HermesAdapter, HermesJobState, HermesWorker
+from aios_openclaw import (
+    CommandState,
+    InProcessRuntimeTransport,
+    OpenClawAdapter,
+)
+from aios_v2 import (
+    ControlPlaneService,
+    ExecutionEvidence,
+    ExecutionOrchestrator,
+    ExecutionTracker,
+    RequestContext,
+    TaskState,
+    TenantIsolationError,
+)
+from aios_v2.persistence import DurableControlPlaneService, SQLiteSnapshotRepository
+
+
+SIGNING_MATERIAL = b"candidate-test-signing-material-0001"
+
+
+def context(
+    tenant_id: str,
+    principal: str,
+    *roles: str,
+) -> RequestContext:
+    return RequestContext.create(
+        tenant_id=tenant_id,
+        principal_id=principal,
+        roles=roles,
+        trace_id=f"trace-{tenant_id}-{principal}",
+    )
+
+
+def bootstrap(service: ControlPlaneService, tenant_id: str = "tenant-a"):
+    platform = context(tenant_id, "platform", "platform_admin")
+    service.register_tenant(
+        platform,
+        tenant_id=tenant_id,
+        display_name=tenant_id,
+    )
+    operator = context(
+        tenant_id,
+        "operator",
+        "tenant_admin",
+        "operator",
+        "approver",
+        "auditor",
+        "runtime",
+    )
+    return platform, operator
+
+
+class CandidateFlowTests(unittest.TestCase):
+    def test_openclaw_full_flow_reconciles_completed_task(self) -> None:
+        service = ControlPlaneService()
+        _, operator = bootstrap(service)
+        agent = service.register_agent(
+            operator,
+            agent_id="openclaw-agent",
+            display_name="OpenClaw",
+            runtime="openclaw",
+            capabilities=("echo",),
+        )
+        task = service.create_task(
+            operator,
+            agent_id=agent.agent_id,
+            action="echo",
+            payload={"message": "hello"},
+            idempotency_key="openclaw-flow-1",
+        )
+        approved = service.transition_task(
+            operator,
+            task_id=task.task_id,
+            target_state=TaskState.APPROVED,
+            expected_revision=task.revision,
+        )
+
+        transport = InProcessRuntimeTransport(
+            lambda command: {
+                "state": CommandState.COMPLETED.value,
+                "output": {"echo": command.payload["message"]},
+            }
+        )
+        resolver = lambda tenant_id, key_id: SIGNING_MATERIAL
+        adapter = OpenClawAdapter(transport, resolver)
+        orchestrator = ExecutionOrchestrator(
+            service,
+            openclaw=adapter,
+            openclaw_signing_material_resolver=resolver,
+        )
+
+        dispatch = orchestrator.dispatch(operator, task_id=approved.task_id)
+        completed = service.get_task(operator, task.task_id)
+
+        self.assertEqual(dispatch.runtime, "openclaw")
+        self.assertEqual(dispatch.state, CommandState.COMPLETED.value)
+        self.assertEqual(completed.state, TaskState.COMPLETED)
+        self.assertEqual(completed.result, {"echo": "hello"})
+
+    def test_hermes_full_flow_reconciles_worker_result(self) -> None:
+        service = ControlPlaneService()
+        _, operator = bootstrap(service)
+        agent = service.register_agent(
+            operator,
+            agent_id="hermes-agent",
+            display_name="Hermes",
+            runtime="hermes",
+            capabilities=("transform",),
+            metadata={"cell_id": "cell-a", "profile_id": "profile-a"},
+        )
+        task = service.create_task(
+            operator,
+            agent_id=agent.agent_id,
+            action="transform",
+            payload={"value": 7},
+            idempotency_key="hermes-flow-1",
+        )
+        approved = service.transition_task(
+            operator,
+            task_id=task.task_id,
+            target_state=TaskState.APPROVED,
+            expected_revision=task.revision,
+        )
+
+        hermes = HermesAdapter(default_lease_seconds=10)
+        orchestrator = ExecutionOrchestrator(service, hermes=hermes)
+        dispatch = orchestrator.dispatch(operator, task_id=approved.task_id)
+        self.assertEqual(dispatch.runtime, "hermes")
+        self.assertEqual(dispatch.state, HermesJobState.QUEUED.value)
+
+        worker = HermesWorker(
+            hermes,
+            lambda job: {"transformed": int(job.payload["value"]) * 2},
+            worker_id="worker-a",
+            tenant_id="tenant-a",
+            cell_id="cell-a",
+            profile_id="profile-a",
+            heartbeat_seconds=0.05,
+        )
+        run = worker.run_once()
+        self.assertTrue(run.claimed)
+        self.assertEqual(run.state, HermesJobState.COMPLETED.value)
+
+        result = hermes.result(dispatch.runtime_execution_id, tenant_id="tenant-a")
+        completed = orchestrator.reconcile_hermes(operator, result)
+        self.assertEqual(completed.state, TaskState.COMPLETED)
+        self.assertEqual(completed.result, {"transformed": 14})
+
+    def test_cross_tenant_task_read_is_denied(self) -> None:
+        service = ControlPlaneService()
+        _, tenant_a = bootstrap(service, "tenant-a")
+        platform_b = context("tenant-b", "platform", "platform_admin")
+        service.register_tenant(
+            platform_b,
+            tenant_id="tenant-b",
+            display_name="Tenant B",
+        )
+        tenant_b = context("tenant-b", "auditor", "auditor")
+
+        agent = service.register_agent(
+            tenant_a,
+            agent_id="agent-a",
+            display_name="Agent A",
+            runtime="hermes",
+            capabilities=("work",),
+        )
+        task = service.create_task(
+            tenant_a,
+            agent_id=agent.agent_id,
+            action="work",
+            payload={},
+            idempotency_key="tenant-isolation-1",
+        )
+
+        with self.assertRaises(TenantIsolationError):
+            service.get_task(tenant_b, task.task_id)
+
+    def test_durable_service_recovers_state_after_restart(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            path = Path(temporary) / "state.sqlite3"
+            service = DurableControlPlaneService(SQLiteSnapshotRepository(path))
+            _, operator = bootstrap(service)
+            agent = service.register_agent(
+                operator,
+                agent_id="persistent-agent",
+                display_name="Persistent",
+                runtime="hermes",
+                capabilities=("persist",),
+            )
+            task = service.create_task(
+                operator,
+                agent_id=agent.agent_id,
+                action="persist",
+                payload={"ok": True},
+                idempotency_key="persist-1",
+            )
+
+            recovered = DurableControlPlaneService(SQLiteSnapshotRepository(path))
+            restored_task = recovered.get_task(operator, task.task_id)
+            self.assertEqual(restored_task.payload, {"ok": True})
+            self.assertEqual(restored_task.state, TaskState.CREATED)
+
+    def test_execution_tracking_persists_claim_heartbeat_and_result(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            tracker = ExecutionTracker(temporary)
+            claimed = tracker.claim(
+                ExecutionEvidence(
+                    execution_id="exec-1",
+                    tenant_id="tenant-a",
+                    task_id="task-1",
+                    agent_id="agent-1",
+                    executor="codex",
+                    branch="task/codex/example",
+                    progress_total=3,
+                )
+            )
+            self.assertEqual(claimed.status, "claimed")
+            tracker.heartbeat(
+                tenant_id="tenant-a",
+                execution_id="exec-1",
+                status="running",
+                current_step="tests",
+                progress_completed=2,
+                last_commit="abc123",
+            )
+            tracker.complete(
+                tenant_id="tenant-a",
+                execution_id="exec-1",
+                status="complete",
+                test_status="passed",
+                test_summary={"passed": 5, "failed": 0},
+            )
+            restored = ExecutionTracker(temporary).get(
+                tenant_id="tenant-a",
+                execution_id="exec-1",
+            )
+            self.assertEqual(restored.status, "complete")
+            self.assertEqual(restored.progress_completed, 3)
+            self.assertEqual(restored.test_summary["passed"], 5)
+
+
+if __name__ == "__main__":
+    unittest.main()
Author	SHA1	Message	Date
Codex	81572b8231	security(openclaw): record restored Gitea credential without secret value	2026-07-20 15:37:29 +08:00
Codex	8c3d80fe53	task(openclaw): 下发执行器启动、Secret 引用与 Gitea 同步在线指令	2026-07-20 14:34:13 +08:00
Codex	99bcc03b40	task(openclaw): 初始化无人值守任务状态	2026-07-20 12:07:46 +08:00
Codex	c6e7edcf54	task(openclaw): 写入预发布无人值守任务载荷	2026-07-20 12:07:36 +08:00
Codex	c464560a0f	task(openclaw): launch AIOS V2R1 pre-production deployment	2026-07-20 11:49:09 +08:00
Codex	f23b694ad3	tasks(aios-v2): add Codex production supervisor task	2026-07-20 11:46:43 +08:00
Codex	f97dea55ac	tasks(aios-v2): add OpenClaw production deployment task	2026-07-20 11:46:39 +08:00
Codex	efd08d7771	docs(aios-v2): add ADR-003 GitOps-only flow	2026-07-20 11:46:31 +08:00
Codex	ff6cb2e217	docs(aios-v2): add ADR-005 documentation sync gate	2026-07-20 11:45:16 +08:00
Codex	d774620fea	docs(aios-v2): add ADR-002 runtime decoupling decision	2026-07-20 11:45:12 +08:00
Codex	b26ff010fc	docs(aios-v2): add ADR-001 control plane decision	2026-07-20 11:45:09 +08:00
Codex	4b49f2ba4b	docs(aios-v2): add ADR index	2026-07-20 11:45:04 +08:00
Codex	b4b8fe8828	docs(aios-v2): add traceability matrix	2026-07-20 11:43:46 +08:00
Codex	4e8ab051a0	docs(aios-v2): add module contracts	2026-07-20 11:41:39 +08:00
Codex	f61e35b218	docs(aios-v2): add migration and delivery strategy	2026-07-20 11:34:33 +08:00
Codex	b2f90f24a3	docs(aios-v2): add target architecture	2026-07-20 11:32:16 +08:00
Codex	6721eb4752	docs(aios-v2): add user requirements baseline	2026-07-20 11:29:07 +08:00
Codex	a9e4fd1e40	docs(aios-v2): establish architecture source of truth	2026-07-20 11:24:12 +08:00
Codex	621750976a	deploy(candidate): add guarded release rollback	2026-07-20 11:22:04 +08:00
Codex	5359e88806	feat(candidate): expose sanitized GitOps status API	2026-07-20 11:19:37 +08:00
Codex	76636931ba	feat(candidate): add sanitized GitOps status provider	2026-07-20 11:17:18 +08:00
Codex	180c1fc66e	deploy(candidate): add end-to-end healthcheck	2026-07-20 11:14:53 +08:00
Codex	c0ff052dab	deploy(candidate): add Docker Compose runtime	2026-07-20 11:12:42 +08:00
Codex	94cffb6ba3	deploy(candidate): add hardened container image	2026-07-20 11:10:36 +08:00
Codex	9ad5628e69	deploy(candidate): add environment configuration template	2026-07-20 11:08:37 +08:00
Codex	6d397a7894	deploy(candidate): add hardened systemd service	2026-07-20 11:07:06 +08:00
Codex	165fea6ba9	fix(candidate): prevent Hermes completion heartbeat revision race	2026-07-20 11:05:45 +08:00
Codex	d079e2087c	test(candidate): add end-to-end runtime and persistence tests	2026-07-20 11:01:07 +08:00
Codex	e3b36ebdcc	feat(candidate): add production application entrypoint	2026-07-20 10:57:42 +08:00
Codex	5e791b11c0	feat(candidate): add tenant-safe web console	2026-07-20 10:54:43 +08:00
Codex	50be65b76e	feat(candidate): add durable SQLite control-plane persistence	2026-07-20 10:50:31 +08:00
Codex	7bb86e30bf	feat(candidate): export orchestration and execution tracking	2026-07-20 10:42:49 +08:00
Codex	ea56b45b7e	feat(candidate): add persistent agent execution tracking	2026-07-20 10:40:22 +08:00
Codex	f53682da7a	feat(candidate): add unified AIOS execution orchestrator	2026-07-20 10:38:06 +08:00
Codex	71884ac088	feat(candidate): add Hermes worker loop and lease heartbeat	2026-07-20 10:35:55 +08:00
Codex	969c278cb0	feat(candidate): implement Hermes queue and lease adapter	2026-07-20 10:34:25 +08:00
Codex	6fe2a1e036	feat(candidate): expose Hermes runtime package	2026-07-20 10:31:07 +08:00
Codex	9a60e0c68e	feat(candidate): add Hermes runtime contracts	2026-07-20 10:27:47 +08:00
Codex	46f553ca11	feat(candidate): expose Hermes runtime boundary package	2026-07-20 10:17:02 +08:00
Codex	4f0a0c2184	feat(candidate): add OpenClaw audit callback sink	2026-07-20 10:07:54 +08:00
Codex	c107b193d9	feat(candidate): add OpenClaw runtime health policy	2026-07-20 10:06:36 +08:00
Codex	2db7ced7a4	feat(candidate): add OpenClaw runtime transports	2026-07-20 10:01:59 +08:00
Codex	6361d9a334	feat(candidate): implement OpenClaw execution adapter	2026-07-20 09:59:53 +08:00
Codex	2cf54f04ef	feat(candidate): expose OpenClaw adapter package	2026-07-20 09:56:06 +08:00
Codex	f840ce5c94	feat(integration): add OpenClaw command contract	2026-07-20 09:49:15 +08:00
Codex	35e7b55490	integration(aios-v2): create CTO production candidate branch	2026-07-20 09:47:01 +08:00
Codex	1f8d1f23b3	feat(aios-v2): add tenant-aware HTTP control plane gateway	2026-07-20 02:24:19 +08:00
Codex	e93011791e	feat(aios-v2): add tenant-isolated control plane domain	2026-07-20 02:19:16 +08:00
Codex	4460564a48	feat(aios-v2): initialize control plane package	2026-07-20 01:11:11 +08:00
Codex	4d6658136e	docs(task): add AIOS control plane gateway task contract	2026-07-20 01:06:11 +08:00
Codex	5767e665c2	task(codex): implement AIOS V2 control plane gateway v1	2026-07-20 01:06:04 +08:00