review(codex): 固化 V2R1 最新发布审查结论

.agent/blockers.json

@@ -0,0 +1,57 @@
+{
+  "schema": "aios.v2.agent-blockers.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "updated_at": "2026-07-21T04:30:00Z",
+  "blockers": [
+    {
+      "blocker_id": "BLK-OPENCLAW-PLAINTEXT-SECRETS",
+      "severity": "high",
+      "status": "resolved",
+      "root_cause": "OpenClaw 三处 managed plaintext。",
+      "fix": "迁移为单一 file Secret Provider 并轮换 Gateway Token。",
+      "verification": "最终 audit 四类计数均为 0，OneAPI 实际调用通过。",
+      "commit": "bf01a13f3aaf42a386232ed2f5409feb0f79fcf6",
+      "resolved_at": "2026-07-20T10:18:36Z"
+    },
+    {
+      "blocker_id": "BLK-AIOS-V2-SIGNING-MATERIAL-MISSING",
+      "severity": "high",
+      "status": "resolved",
+      "root_cause": "Ed25519 Key ID 与 HMAC 实现不一致且材料缺失。",
+      "fix": "真实 Ed25519 合同及受保护材料。",
+      "verification": "0600、sign/verify、篡改拒绝通过。",
+      "commit": "8a36f7832bc105e4b9d32ffbe018abed2ff1b1ab",
+      "resolved_at": "2026-07-20T10:18:36Z"
+    },
+    {
+      "blocker_id": "BLK-CANDIDATE-NOT-MERGED",
+      "severity": "high",
+      "status": "resolved",
+      "root_cause": "OpenClaw 任务未包含 Candidate。",
+      "fix": "--no-ff 合并最新 Candidate。",
+      "verification": "Candidate 为 OpenClaw HEAD 祖先且 OID 同步。",
+      "commit": "f55dfd975e9d716b690c7693742e7cf2beaa55e7",
+      "resolved_at": "2026-07-20T10:18:36Z"
+    },
+    {
+      "blocker_id": "BLK-OPENCLAW-RUN-PROVENANCE-MISSING",
+      "severity": "high",
+      "status": "resolved",
+      "root_cause": "旧插件只有进程内 runId/sender Map，缺少 Owner-signed Schema、持久 Store 和限定 Git Writer。",
+      "fix": "部署 aios-tenant-boundary rc2，增加 Ed25519 issuer/store、Run/Sender/OID/期限绑定、一次性 permit 与七路径专用 Writer。",
+      "verification": "Provenance openclaw-v2r1-20260721T031523Z 完成 heartbeat 写入、Commit、普通 Push 和审计；README.md 越权写入被拒绝；Node 测试 9/9。",
+      "commit": "25c25abd3ed19091fe078d4d7a271c57f23e87fb",
+      "resolved_at": "2026-07-21T03:23:44Z"
+    },
+    {
+      "blocker_id": "BLK-CONTROLLER-LEGACY-GITEA-CHECK",
+      "severity": "high",
+      "status": "resolved",
+      "root_cause": "硬编码旧仓库。",
+      "fix": "配置化 V2R1 OID 健康检查并使用只读 Credential。",
+      "verification": "12 项测试、detached clean、fully_synced。",
+      "commit": "6188da26453505dacde9d6d298191437164c8d27",
+      "resolved_at": "2026-07-20T10:18:36Z"
+    }
+  ]
+}

.agent/claim.json

@@ -0,0 +1,13 @@
+{
+  "schema": "aios.v2.agent-claim.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "executor": "codex",
+  "executor_instance": "windows-chatgpt-owner-supervisor",
+  "task_branch": "task/codex/AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "candidate_branch": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "candidate_commit": "ba7669c3b49f98cf3500af6ea966ba0db472ff3b",
+  "claimed_at": "2026-07-20T08:51:25Z",
+  "owner_observation": "五项 Blocker、Gate A/B、离线验收、OpenClaw 在线合同、双服务重启与回滚均通过；Hermes 完整在线 Worker 生命周期仍缺服务端点证据，Codex 维持 STOP_PREPRODUCTION。",
+  "owner_observation_at": "2026-07-21T04:30:00Z",
+  "reviewed_openclaw_commit_before_evidence": "b158041b8633e658c34ce182f3c8aa97b4bf5f9a"
+}

.agent/credential-status.json

@@ -0,0 +1,18 @@
+{
+  "schema": "aios.agent.credential-status.v1",
+  "task_id": "task/codex/AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "last_updated": "2026-07-21T04:30:00Z",
+  "plaintext_findings_before": 3,
+  "plaintext_findings_after": 0,
+  "gateway_token_rotated": true,
+  "secret_values_recorded": false,
+  "signing_material_configured": true,
+  "signing_key_id": "openclaw-v2r1-candidate-ed25519",
+  "runtime_sign_verify_test": "passed",
+  "controller_credential_scope": "read:repository",
+  "run_provenance_status": "resolved",
+  "provenance_id": "openclaw-v2r1-20260721T031523Z",
+  "openclaw_runtime_plugin_version": "2.1.0-rc.3",
+  "runtime_public_key_only_hot_path": true,
+  "secret_values_recorded_in_evidence": false
+}

.agent/heartbeat.json

@@ -0,0 +1,10 @@
+{
+  "schema": "aios.v2.agent-heartbeat.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "executor": "codex",
+  "heartbeat_at": "2026-07-21T04:30:00Z",
+  "state": "reviewing",
+  "current_step": "FINAL_REVIEW_HERMES_ONLINE_GAP",
+  "reviewed_commit": "b158041b8633e658c34ce182f3c8aa97b4bf5f9a",
+  "openclaw_evidence_commit": "0be3ba09e42b4f6db56461e3ca3a3fcc0a487f31"
+}

.agent/progress.json

@@ -0,0 +1,73 @@
+{
+  "schema": "aios.v2.agent-progress.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "updated_at": "2026-07-21T04:30:00Z",
+  "state": "reviewing",
+  "candidate_oid": "634f6b164f5b20372ab9ec8144bbf76df97de335",
+  "openclaw_task_oid_at_review": "0be3ba09e42b4f6db56461e3ca3a3fcc0a487f31",
+  "codex_task_oid_before_evidence": "be4726178aa8fa59321aafaabcff8a83653cc1b8",
+  "gitea_sync": {
+    "candidate_match": true,
+    "openclaw_task_match": true,
+    "codex_task_match_before_this_evidence": true
+  },
+  "completed_checks": [
+    "worktree",
+    "candidate_merge",
+    "secrets_audit_zero",
+    "ed25519_sign_verify",
+    "controller_fully_synced",
+    "owner_signed_provenance_writer",
+    "unauthorized_path_denied",
+    "provenance_audit_chain",
+    "gate_a",
+    "gate_b_10_of_10",
+    "offline_acceptance_8_0_1",
+    "openclaw_online_contract",
+    "aios_openclaw_restart_recovery",
+    "controlled_rollback_and_forward"
+  ],
+  "review_conclusion": "五项 Blocker、文档/代码 Gate、候选测试、离线验收、OpenClaw 在线合同、重启与回滚均通过；Hermes 完整在线 Worker 生命周期未形成服务端点证据，预发布仍停止。",
+  "check_time_window": {
+    "started_at": "2026-07-20T08:24:30Z",
+    "ended_at": "2026-07-20T09:03:08Z"
+  },
+  "operator": "windows-chatgpt-owner-supervisor",
+  "reviewer": "codex-supervisor + Spark architect/explorer/implementer/validator/docs_reporter",
+  "approver": "AIOS Owner gate recorded by Windows ChatGPT Owner Supervisor",
+  "evidence_artifacts": [
+    "task/openclaw/AIOS-V2R1-PRODUCTION-DEPLOY-20260720@1aa71f233d2c2a6ed7290d0a3ddcb04dcd84febe",
+    "task/codex/AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720:.agent/",
+    "/data/aios-gitops/runtime/backups/gitea-db-before-openclaw-token-rotation-20260720T084251Z.sqlite3",
+    "/home/aiman/.ssh/known_hosts.aios-v2r1-backup-20260720T083647Z",
+    "/var/log/aios-openclaw-v2r1-evidence-repair-20260720.json"
+  ],
+  "independent_reviews": {
+    "architect": "REJECT；只允许修复+复验，禁止生产放行。",
+    "explorer": "OpenClaw evidence OID 已重新闭环，但 Candidate 未合并、Secret 与预发布仍未确认。",
+    "implementer": "IMPLEMENTATION_SAFE_WITH_BLOCKERS；已实施动作无需立即回滚。",
+    "validator": "BLOCKED；Secret、环境、预发布门禁未通过。",
+    "docs_reporter": "已补齐时间窗、证据路径、operator/reviewer/approver、blocker 与 final_decision。"
+  },
+  "final_decision": "STOP_PREPRODUCTION",
+  "aios_v2_env_status": {
+    "exists": true,
+    "owner": "root",
+    "group": "gitops",
+    "mode": "0640",
+    "four_fields_nonempty": true,
+    "credential_ref_target_exists": true,
+    "signing_material_target_exists": true
+  },
+  "open_checks": [
+    "hermes_online_claim_start_heartbeat_complete_fail_cancel",
+    "hermes_online_lease_expiry_worker_crash_recovery",
+    "hermes_online_cell_profile_and_invalid_result_rejection"
+  ],
+  "gate_a": "passed",
+  "gate_b": "passed_10_of_10",
+  "offline_acceptance": "passed_8_failed_0_warning_1",
+  "online_openclaw": "passed",
+  "restart_recovery": "passed",
+  "rollback_drill": "passed"
+}

.agent/result.json

@@ -0,0 +1,27 @@
+{
+  "schema": "aios.v2.agent-result.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "final_status": "PARTIAL",
+  "candidate_commit": "ba7669c3b49f98cf3500af6ea966ba0db472ff3b",
+  "reviewed_commit": "b158041b8633e658c34ce182f3c8aa97b4bf5f9a",
+  "gitea_sync": {
+    "candidate_match": true,
+    "openclaw_task_match": true,
+    "codex_task_match_before_this_evidence": true
+  },
+  "credential_rotation_status": "secretref_resolved",
+  "test_status": "Gate A 通过；Gate B 10/10；离线 8/0/1；OpenClaw 在线、重启恢复、回滚通过；Hermes 完整在线 Worker 生命周期未覆盖",
+  "deployment_status": "STOP_PREPRODUCTION",
+  "rollback_status": "Previous 2a2b702 与 Current b158041 双向演练通过且业务状态保留",
+  "documentation_sync_status": "updated",
+  "open_blockers": [],
+  "release_recommendation": "STOP_PREPRODUCTION",
+  "completed_at": "2026-07-21T04:30:00Z",
+  "openclaw_evidence_commit": "0be3ba09e42b4f6db56461e3ca3a3fcc0a487f31",
+  "independent_validation": "PASS_FOR_IMPLEMENTED_OPENCLAW_BOUNDARY",
+  "operator": "windows-chatgpt-owner-supervisor",
+  "reviewer": "codex-supervisor + Spark roles",
+  "approver": "AIOS Owner gate recorded",
+  "remaining_risk": "Hermes 完整在线 Worker 生命周期缺少可调用服务端点与实时证据",
+  "formal_online_acceptance": "partial"
+}

.agent/status.json

@@ -0,0 +1,18 @@
+{
+  "schema": "aios.v2.agent-status.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "executor": "codex",
+  "state": "reviewing",
+  "claimed_at": "2026-07-20T08:51:25Z",
+  "running_at": "2026-07-20T08:51:25Z",
+  "updated_at": "2026-07-21T04:30:00Z",
+  "last_heartbeat": "2026-07-21T04:30:00Z",
+  "current_step": "FINAL_REVIEW_HERMES_ONLINE_GAP",
+  "candidate_commit": "ba7669c3b49f98cf3500af6ea966ba0db472ff3b",
+  "reviewed_commit": "b158041b8633e658c34ce182f3c8aa97b4bf5f9a",
+  "release_recommendation": "STOP_PREPRODUCTION",
+  "openclaw_evidence_commit": "0be3ba09e42b4f6db56461e3ca3a3fcc0a487f31",
+  "status": "reviewing"
+  ,"formal_online_acceptance": "partial"
+  ,"remaining_gap": "Hermes full online worker lifecycle"
+}

.agent/task.json

@@ -0,0 +1,31 @@
+{
+  "schema": "aios.v2.agent-task.v1",
+  "task_id": "AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "display_name": "AIOS V2R1 Codex 架构监督、阻塞处理与发布审查任务",
+  "language": "zh-CN",
+  "executor": "codex",
+  "role": "architecture-supervisor",
+  "state": "queued",
+  "candidate_branch": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "candidate_commit": "634f6b164f5b20372ab9ec8144bbf76df97de335",
+  "canonical_task": "tasks/codex-production-supervisor-task.yaml",
+  "supervises": "task/openclaw/AIOS-V2R1-PRODUCTION-DEPLOY-20260720",
+  "claim_required": true,
+  "heartbeat_interval_seconds": 300,
+  "evidence_directory": ".agent/evidence",
+  "instructions": [
+    "领取前读取候选分支完整中文文档、ADR、追踪矩阵和候选报告",
+    "领取后写入 .agent/claim.json 并把状态改为 claimed",
+    "持续审阅 OpenClaw 的 claim、heartbeat、progress、blocker、commit 和 result",
+    "分析失败测试、架构偏移、多租户、安全和文档同步阻塞",
+    "需要修改时创建 Patch 分支、增加复现测试并同步文档",
+    "所有阻塞和审查结论必须提交到系统统一审视",
+    "禁止缺少证据时批准发布或削弱安全门禁"
+  ],
+  "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"
+  ],
+  "created_at": "2026-07-20T00:00:00Z"
+}

active-runtime/app/aios_hermes/__init__.py

@@ -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",
+]

active-runtime/app/aios_hermes/adapter.py

@@ -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(),
+                )

active-runtime/app/aios_hermes/contract.py

@@ -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))

active-runtime/app/aios_hermes/worker.py

@@ -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

active-runtime/app/aios_openclaw/__init__.py

@@ -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",
+]

active-runtime/app/aios_openclaw/adapter.py

@@ -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
+            }

active-runtime/app/aios_openclaw/callback.py

@@ -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())

active-runtime/app/aios_openclaw/contract.py

@@ -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

active-runtime/app/aios_openclaw/health.py

@@ -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,
+            },
+        )

active-runtime/app/aios_openclaw/transport.py

@@ -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 {},
+        )

active-runtime/app/aios_v2/__init__.py

@@ -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",
+]

active-runtime/app/aios_v2/app.py

@@ -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()

active-runtime/app/aios_v2/console.py

@@ -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>'''

active-runtime/app/aios_v2/control_plane.py

@@ -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)
+            ]

active-runtime/app/aios_v2/execution_tracking.py

@@ -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)

active-runtime/app/aios_v2/gateway.py

@@ -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()

active-runtime/app/aios_v2/gitops_status.py

@@ -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,
+        },
+    )

active-runtime/app/aios_v2/orchestrator.py

@@ -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
+        }

active-runtime/app/aios_v2/persistence.py

@@ -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))

deploy/aios-v2/docker/Dockerfile

@@ -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"]

deploy/aios-v2/docker/compose.yaml

@@ -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:

deploy/aios-v2/env/aios-v2.env.example

@@ -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

deploy/aios-v2/healthcheck/check.sh

@@ -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"

deploy/aios-v2/rollback/rollback.sh

@@ -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"

deploy/aios-v2/systemd/aios-v2.service

@@ -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

docs/aios-v2/01-USER-REQUIREMENTS.md

@@ -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. 候选部署健康检查通过，回滚演练成功。

docs/aios-v2/02-TARGET-ARCHITECTURE.md

@@ -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 和生产部署。

docs/aios-v2/03-MIGRATION-AND-DELIVERY.md

@@ -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。

docs/aios-v2/04-MODULE-CONTRACTS.md

@@ -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 和版本号，不得静默修改。

docs/aios-v2/05-TRACEABILITY-MATRIX.md

@@ -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/` 变化。

docs/aios-v2/README.md

@@ -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.

docs/aios-v2/adr/ADR-001-single-control-plane.md

@@ -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 接入。

docs/aios-v2/adr/ADR-002-openclaw-hermes-decoupling.md

@@ -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，再迁移旧调用入口，最后关闭控制旁路。

docs/aios-v2/adr/ADR-003-gitops-only-change-flow.md

@@ -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 和证据。

docs/aios-v2/adr/ADR-005-document-code-sync-gate.md

@@ -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 修复有历史依据。
+
+成本：
+
+- 小功能也需要更新对应文档。

docs/aios-v2/adr/README.md

@@ -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`
+- 对应代码、测试和部署资产

docs/tasks/AIOS-CONTROL-PLANE-GATEWAY-V1.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
+
+不重写执行层。

patches/integration_aios-v2r1-cto-production-candidate-20260720.branch.json

@@ -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
+}

patches/task_codex_AIOS-V2-CONTROL-PLANE-GATEWAY-V1-20260720.branch.json

@@ -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
+}

patches/task_codex_AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720.branch.json

@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "task/codex/AIOS-V2R1-PRODUCTION-SUPERVISOR-20260720",
+  "base": "integration/aios-v2r1-cto-production-candidate-20260720",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-20T03:49:13Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}

tasks/codex-production-supervisor-task.yaml

@@ -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

tasks/openclaw-production-deploy-task.yaml

@@ -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

tests/aios_v2_candidate/test_candidate_flow.py

@@ -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()