feat(gitops): add isolated detached worktree manager

docs(aios-v2): add CTO implementation status and gap analysis
create branch for AIOS V2 implementation status documentation
2026-07-20 00:42:56 +08:00 · 2026-07-20 00:25:57 +08:00 · 2026-07-20 00:21:00 +08:00
40 changed files with 777 additions and 5121 deletions
@@ -1,32 +0,0 @@
-{
-  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "created_at": "2026-07-17T14:13:38+00:00",
-  "schema_version": 1,
-  "state_writer": "openclaw",
-  "work_units": [
-    {
-      "fallback": "codex",
-      "gate": "R0_RUNNER_LIVE_RECOVERY",
-      "id": "R0-DISCOVER-SECRET-PROVIDER",
-      "primary": "openclaw",
-      "status": "READY",
-      "track": "R_RUNNER"
-    },
-    {
-      "fallback": "codex",
-      "gate": "P2_REAL_B3_EVIDENCE",
-      "id": "P2-INVENTORY-REAL-EVIDENCE",
-      "primary": "openclaw",
-      "status": "READY",
-      "track": "P_PERSONAL_WEIXIN"
-    },
-    {
-      "fallback": "openclaw",
-      "gate": "P1_REPAIR3_CODE",
-      "id": "P1-POST-CONTRACT-REACK",
-      "primary": "codex",
-      "status": "READY",
-      "track": "P_PERSONAL_WEIXIN"
-    }
-  ]
-}
@@ -1,157 +0,0 @@
-{
-  "adaptive_dispatch": {
-    "codex_first": [
-      "complex_code",
-      "security_boundary",
-      "architecture",
-      "high_risk_review",
-      "strong_model_materially_better"
-    ],
-    "default_orchestrator": "openclaw",
-    "dual_required": [
-      "release_gate",
-      "crypto_or_identity_change",
-      "production_migration_or_rollback"
-    ],
-    "max_attempts_per_work_unit": 2,
-    "on_timeout": [
-      "mark_TIMEOUT",
-      "preserve_logs",
-      "switch_agent"
-    ],
-    "openclaw_first": [
-      "quality_close",
-      "large_or_repetitive",
-      "long_running",
-      "infra_or_git_inventory",
-      "cost_sensitive"
-    ],
-    "single_state_writer_lease_required": true,
-    "strict_role_partition": false,
-    "timeout_can_pass": false
-  },
-  "authoritative_state": {
-    "b3": "BLOCKED_EVIDENCE_REQUIRED",
-    "b4": "NOT_STARTED",
-    "baseline": "V1R4C6",
-    "production_release": "NOT_STARTED",
-    "release": "paused",
-    "repair3": {
-      "base": "8d9c6da90fc8caa69b691d243c6fefb40b6b0575",
-      "candidate": "PASS_CANDIDATE_PREPARED",
-      "commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4",
-      "exact4": "PASS",
-      "result": "49a9c2e0cc898ecf2bbd45f454e007644d7f3045"
-    },
-    "runner": {
-      "artifact_sha256": "85c472c05a1e55a5899b215d3cd41df53912787de529ac9cb3d5308da4158130",
-      "commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
-      "live": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
-      "offline": "PASS",
-      "result": "e789632e359e9232d8ae081e7c4f23ed7b615a9d"
-    }
-  },
-  "contract_id": "V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT",
-  "facts_policy": {
-    "accept": [
-      "committed_git_tree",
-      "full_oid",
-      "worktree_bare_gitea_oid_match",
-      "test_log_with_rc",
-      "immutable_digest",
-      "result_only_commit",
-      "live_evidence"
-    ],
-    "reject": [
-      "verbal_claim",
-      "readme_only",
-      "local_only_commit",
-      "environment_variable_as_secret_ref",
-      "compile_only",
-      "timeout_as_pass",
-      "unrelated_repo",
-      "fabricated_evidence"
-    ]
-  },
-  "forbidden_git": [
-    "reset",
-    "amend",
-    "rebase",
-    "force_push",
-    "direct_main_push",
-    "direct_bare_ref_mutation",
-    "unapproved_gitea_db_mutation",
-    "reuse_accident_worktree",
-    "delete_live_worktree_to_clear_collision"
-  ],
-  "generated_at": "2026-07-17T14:13:38+00:00",
-  "handoff": {
-    "ack": "CONTRACT_ACK=<full_contract_commit_oid>",
-    "advance": "independent_verification_required",
-    "all_agents_match": true
-  },
-  "join_gate": {
-    "requires": [
-      "RUNNER_RECOVERED",
-      "B3_PASS",
-      "B4_PASS",
-      "INTEGRATION_PASS"
-    ]
-  },
-  "model_policy": {
-    "codex": "strongest_model_actually_available_in_runtime",
-    "gpt_5_6": "use_only_if_exposed",
-    "no_unverified_model_claims": true,
-    "openclaw_default": "DeepSeek V4 Flash"
-  },
-  "owner_authorization": {
-    "adaptive_agent_assignment_authorized": true,
-    "agent_substitution_authorized": true,
-    "autonomous_execution": true,
-    "may_bypass_release_controls": false,
-    "may_bypass_secret_controls": false,
-    "may_fabricate_evidence": false,
-    "user_step_by_step_confirmation_required": false
-  },
-  "release_gates": [
-    "MAIN_MERGE",
-    "STAGING_AND_CANARY",
-    "PRODUCTION"
-  ],
-  "schema_version": 3,
-  "state_rules": [
-    "tracks_run_parallel",
-    "runner_block_does_not_block_offline_work_or_evidence_inventory",
-    "join_requires_both_tracks",
-    "failure_stops_dependents",
-    "production_remains_V1R4C6_paused_until_production_pass"
-  ],
-  "status": "ACTIVE",
-  "tracks": {
-    "P_PERSONAL_WEIXIN": {
-      "b3": "BLOCKED_EVIDENCE_REQUIRED",
-      "b4": "NOT_STARTED",
-      "parallel": true,
-      "repair3": "PASS_CANDIDATE_PREPARED",
-      "repair3_commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4"
-    },
-    "R_RUNNER": {
-      "commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
-      "current": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
-      "fallback": "codex",
-      "parallel": true,
-      "primary": "openclaw",
-      "required": [
-        "artifact_verified",
-        "approved_secret_provider_contract",
-        "owner_probe_verified",
-        "service_healthy",
-        "heartbeat_healthy",
-        "gateway_auth",
-        "durable_task_create_read_cancel",
-        "execution_channel",
-        "live_rollback"
-      ]
-    }
-  }
-}
@@ -1,15 +0,0 @@
-{
-  "b3": "BLOCKED_EVIDENCE_REQUIRED",
-  "b4": "NOT_STARTED",
-  "baseline": "V1R4C6",
-  "contract_branch": "task/governance/v2r1-personal-weixin-production-contract-20260717",
-  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "dispatch_commit": null,
-  "production_release": "NOT_STARTED",
-  "release": "paused",
-  "repair3": "PASS_CANDIDATE_PREPARED",
-  "runner": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
-  "schema_version": 1,
-  "single_state_writer": "openclaw",
-  "updated_at": "2026-07-17T14:13:38+00:00"
-}
@@ -1,69 +0,0 @@
-# Codex task: apply and verify AIOS Orchestrator Gateway v1
-
-Contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
-
-Use the strongest actually available Codex model for the initial security review.
-If unavailable or timed out, record TIMEOUT/NOT_EXECUTED and let OpenClaw run
-the deterministic test path. Do not count a timeout as PASS.
-
-## Apply
-
-1. Create a new clean append-only branch and isolated worktree.
-2. Validate the delivery tarball and `SHA256SUMS`.
-3. Run the external `scripts/apply_gateway_patch.sh` with `SOURCE_DIR` and
-   `WORKTREE`; do not alter unrelated paths.
-4. Review all added code, especially:
-   - exact contract and optimistic state-version checks;
-   - event/decision idempotency;
-   - SQLite leases and retry behavior;
-   - Workspace Agent `202` asynchronous callback flow;
-   - Responses background creation and polling;
-   - file-inbox and Git-ref polling;
-   - traversal/symlink protections;
-   - OpenClaw lifecycle synchronization;
-   - systemd sandbox;
-   - install and rollback transactions.
-
-## Tests
-
-Run from the final source tree:
-
- Python compile for all gateway and helper files;
- unit tests;
- JSON schema/config parsing;
- Bash syntax;
- `systemd-analyze verify`;
- event duplicate and payload-conflict tests;
- stale state-version rejection;
- lease/double-claim test;
- Workspace trigger failure and Responses fallback;
- response queued/in-progress/completed polling;
- file-inbox recovery after HTTP failure;
- Git-ref initial baseline and subsequent-change event;
- task start/end and project summary synchronization;
- dispatch traversal and symlink rejection;
- retry/backoff/dead-letter;
- install dry-run;
- rollback test in an isolated fixture;
- `git diff --check`;
- full repository tests;
- final Gitleaks.
-
-## Commit and review
-
-1. Create an append-only implementation commit.
-2. Push without force.
-3. Confirm `WORKTREE_OID == BARE_OID == GITEA_OID`.
-4. Create an independent RESULT-only branch and exact-one result commit.
-5. Record the source bundle SHA-256, implementation OID, RESULT OID, tests,
-   Gitleaks, systemd validation and remaining external configuration blockers.
-
-## Forbidden
-
-No reset, amend, rebase, force push, direct main push, controller mutation,
-secret disclosure, fabricated test result, production deployment or B3 evidence
-fabrication.
-
-Success:
-
-`GATEWAY_CODE_REVIEW=PASS IMPLEMENTATION=<oid> RESULT=<oid> OID_MATCH=YES`
@@ -1,13 +0,0 @@
-# Codex V2R1 adaptive execution
-CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
-Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
-Execute P1-POST-CONTRACT-REACK now:
- read-only verify Repair3 e110cda9968cadc5c7b7179ec5003b05aad11ce4
- read-only verify RESULT 49a9c2e0cc898ecf2bbd45f454e007644d7f3045
- verify lineage from 8d9c6da90fc8caa69b691d243c6fefb40b6b0575, exact4, tests, Gitleaks and RESULT-only scope
- do not modify Repair3 code in this unit
-Act as fallback for OpenClaw only on timeout/unavailability. Use GPT-5.6 only if actually exposed; otherwise strongest visible model. Prefer OpenClaw when quality is comparable or volume is high; prefer Codex for complex/security-critical work.
-Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
-Report:
-CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=codex STATUS=<status> IMPLEMENTATION=<oid-or-NONE> RESULT=<oid-or-NONE> OID_MATCH=<YES-or-NO> BLOCKER=<reason-or-NONE>
-Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.
@@ -1,68 +0,0 @@
-# OpenClaw task: deploy and supervise AIOS Orchestrator Gateway v1
-
-Contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
-
-You are the default orchestrator and deployment supervisor.
-
-## Preconditions
-
- Codex has applied and independently reviewed this source bundle on a clean,
-  append-only branch.
- Implementation and RESULT OIDs are identical across worktree, bare repo and
-  Gitea.
- Final Gitleaks and repository tests are PASS.
- No force push, reset, amend or rebase occurred.
-
-## Responsibilities
-
-1. Validate the source bundle SHA-256 and `SHA256SUMS`.
-2. Review the generated systemd service, poll timer, tmpfiles and sysusers files.
-3. Install through `scripts/install_aios_orchestrator_gateway.sh`.
-4. Configure one or both central-model channels:
-   - published Workspace Agent API channel with `agtch_...` trigger ID and a
-     Workspace Agent access token stored outside Git; or
-   - Responses API fallback with an approved model and API key stored outside Git.
-5. Publish the callback endpoint securely and configure the Workspace Agent
-   action from `workspace-agent/orchestrator-decision-action.openapi.yaml`.
-6. Confirm that the Workspace Agent instructions match
-   `workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md`.
-7. Enable both:
-   - the continuously running gateway service;
-   - the independent `aios-orchestrator-gateway-poll.timer`.
-8. Verify HTTP event ingestion, file-inbox polling, Git-ref polling,
-   Workspace Agent trigger acceptance, callback decision, Responses background
-   polling fallback, dispatch queue generation, retry leases and loop guards.
-9. Verify OpenClaw receives and synchronizes:
-   - TASK_DISPATCHED
-   - TASK_STARTED
-   - TASK_ENDED
-   - PROJECT_SUMMARY
-   - PROJECT_COMPLETED
-10. Run a harmless bootstrap work unit through OpenClaw and Codex and confirm
-    duplicate event submission is idempotent.
-11. Run rollback, reinstall and heartbeat/health verification.
-12. Create one RESULT-only review recording exact commits, artifact digest,
-    tests, systemd security, polling recovery, lifecycle synchronization and
-    rollback.
-
-## Fail-closed boundaries
-
- Do not put Workspace Agent tokens, API keys, callback tokens or gateway
-  credentials in Git, logs, command arguments or RESULT files.
- Do not claim the gateway resolves the existing B3 credential/raw-wire blocker.
- Do not advance B4 or production merely because this gateway is healthy.
- Do not bypass owner/admin production release approvals.
- Do not use reset, amend, rebase, force push, direct main push or bare-ref edits.
-
-## Required final status
-
-Success:
-
-`ORCHESTRATOR_GATEWAY_LIVE=PASS POLLING=PASS OPENCLAW_SYNC=PASS ROLLBACK=PASS`
-
-Blocker:
-
-`ORCHESTRATOR_GATEWAY_LIVE=BLOCKED reason=<stable_reason>`
-
-Keep `BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED` until the
-existing production contract advances all release gates.
@@ -1,13 +0,0 @@
-# OpenClaw V2R1 adaptive orchestration
-CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
-Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
-You are default orchestrator and sole authoritative state writer.
-Execute immediately:
-1. R0-DISCOVER-SECRET-PROVIDER: discover approved provider metadata and owner-probe source without reading or outputting secrets. Environment variables alone are not approved secret refs.
-2. P2-INVENTORY-REAL-EVIDENCE: inventory approved restricted B3 evidence locations; validate existing facts; never synthesize evidence.
-3. Independently verify Codex results before advancing.
-Prefer OpenClaw for comparable-quality, large, repetitive, long-running, infrastructure and cost-sensitive work. Prefer Codex strongest actually available model for complex code, security boundaries, architecture and high-risk review. On timeout mark TIMEOUT and switch agent; never infer PASS.
-Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
-Report:
-CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=openclaw STATUS=<status> COMMIT=<oid-or-NONE> BLOCKER=<reason-or-NONE>
-Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.
@@ -0,0 +1,218 @@
+# AIOS V2 Implementation Status
+
+- 扫描日期：2026-07-20
+- 当前发布基线：`V1R4C6`
+- 基线提交：`e204a90398bf646d46245b3088226749fa056edd`
+- 实施分支：`patch/aios-v2-implementation-status-next-phase-20260720`
+- 决策角色：AIOS V2 CTO
+- 总体判定：**进入主线收敛阶段，不重写现有系统。**
+
+## 1. 执行摘要
+
+AIOS V2 已经具备一批可复用的候选实现，包括签名租户上下文、权威 Tenant Router、Tenant Cell 生命周期、OpenClaw Fleet 适配、Hermes 租户队列与 Profile 隔离、消息账本、内存/密钥作用域以及追加式审计链。GitOps Runtime 当前可用，分支、提交和审计元数据机制可以继续作为所有变更的唯一入口。
+
+当前尚不能宣称 AIOS V2 已交付。`V1R4C6` 的提升理由仍注明未部署生产；V2R1 实现主要存在于候选发布、集成、任务和结果分支，未进入 `main` 或当前基线。请求的 `docs/aios-v2/` 权威设计目录在已核对的 `main`、当前治理分支和抽查的 V2R1 集成/设计分支中未发现，因此架构目标尚未形成可被自动验证的单一事实源。
+
+CTO 决策：
+
+1. 不推倒 OpenClaw、Hermes、Agent Router 和现有 GitOps 能力。
+2. AIOS 收敛为唯一控制面；OpenClaw 变为执行适配器；Hermes 变为租户隔离的 Runtime；Agent Router 变为统一调度与路由平面。
+3. 候选能力先做证据映射、边界硬化和小步集成，再进入基线。
+4. 所有 API、控制能力、运行状态与审计查询必须 Web 化。
+5. 所有模块必须在入口、存储、队列、缓存、密钥、日志、审计和前端会话层显式携带并校验 `tenant_id`。
+6. 所有修改必须通过非 `main` 分支、可审查提交、自动测试和受控提升完成。
+
+## 2. 扫描范围与证据边界
+
+### 已读取
+
+- GitOps Runtime 状态与当前基线。
+- `main`、V2R1 候选发布分支、集成/任务/结果分支的定向差异。
+- `release/v2r1c0-tenant-cell-fabric-20260712` 中的 `active-runtime/app/aios_v2r1` 候选实现。
+- 同一候选分支中的 `tests/v2r1` 合同测试与发布门禁说明。
+- Gateway、自动编排、租户隔离、WebUI、事件总线等相关分支和结果分支的提交证据。
+
+### 限制
+
+- `docs/aios-v2/` 未出现在已核对的权威分支，无法完成“对该目录全部文件逐份比对”。本报告将此项列为 P0，而不是假设文档已经存在。
+- 从 `V1R4C6` 到 `main`、以及部分大型任务分支的累计差异超过 GitOps diff 接口上限。因此本次采用定向目录、最近提交和结果证据扫描；下一阶段必须生成机器可读的能力—提交—测试索引。
+- “分支存在”不等于“能力已合入”；“测试报告存在”不等于“生产门禁已通过”。状态必须区分基线、候选、验证和部署。
+
+## 3. 当前架构扫描
+
+### 3.1 权威基线
+
+| 项目 | 当前状态 | 结论 |
+|---|---|---|
+| 发布基线 | `V1R4C6` | 当前唯一被提升的版本 |
+| 生产部署 | 未确认；基线理由注明未部署生产 | 不得视为生产完成 |
+| GitOps Runtime | Active | 可作为唯一变更通道 |
+| V2R1 代码进入 `main` | 未发现 | 候选能力尚未成为权威实现 |
+| `docs/aios-v2/` | 未发现 | P0 文档与治理缺口 |
+
+### 3.2 候选运行时能力
+
+`release/v2r1c0-tenant-cell-fabric-20260712` 中已发现以下候选模块：
+
+| 模块 | 已有能力 | 当前成熟度 |
+|---|---|---|
+| `tenant_context.py` | HMAC 签名上下文、规范化消息模型、来源校验、基础工具策略 | 候选；缺少密钥轮换、重放防护和集中策略版本 |
+| `router.py` | SQLite 权威租户、渠道、主体和会话路由；fail-closed 校验 | 候选；需统一到控制面并替换单机存储假设 |
+| `cell_manager.py` / `cli.py` | Tenant Cell 注册和状态操作 | 原型；JSON 注册表无并发控制，状态机约束不足 |
+| `fleet.py` | OpenClaw Fleet 命令适配 | 可复用适配层雏形；缺少超时、认证、幂等、审计和能力协商 |
+| `hermes_fabric.py` | 强制租户分区的任务队列、来源限制、领取接口 | 候选；共享 SQLite 仍是逻辑隔离，需生产级队列和配额 |
+| `hermes_profile.py` | HOME、工作区和路径逃逸防护 | 可复用；需容器/进程级隔离证据 |
+| `memory_scope.py` | session/profile/cell/tenant/shared artifact 命名空间 | 可复用合同；需绑定实际存储后端和查询层 |
+| `secret_scope.py` | 仅保存 SecretRef，校验 tenant/cell 所有权 | 可复用合同；尚无真实 broker、轮换和撤销流程 |
+| `message_ledger.py` | 精确消息 ID、TTL、冲突检测和引用解析 | 候选；需迁移、容量、加密、备份和多实例并发验证 |
+| `audit.py` | 追加式哈希链 | 原型；当前链为全局序列，应按租户/审计域隔离并支持外部不可变存储 |
+
+### 3.3 测试与门禁
+
+已发现可执行合同测试，覆盖 Tenant Router 隔离、TenantContext 签名和 Hermes 来源/重复事件处理。候选验收清单明确要求：
+
+- 跨租户内存、Cell、密钥和队列访问必须拒绝。
+- 非当前已认证人工命令不得触发副作用。
+- OpenClaw Runtime Hook 与副作用审批属于必需项。
+- 在可执行测试与运行时证据通过之前，不得生产部署。
+
+因此当前标记为“候选能力存在”，而不是“发布完成”。
+
+### 3.4 控制面与自动化
+
+结果分支显示 Gateway 和自动编排已出现较强测试证据，包括单元/集成测试与 Codex 任务领取、生命周期回写能力。但是这些成果仍分散在结果/集成分支，尚未形成 AIOS 唯一入口，也未完成从任务结果到权威基线的提升闭环。
+
+### 3.5 Web 化现状
+
+仓库存在 WebUI、观测与控制面相关分支，但分支名称和最后提交内容有不一致现象。当前不能仅依据分支名认定功能完成。需要建立 API 路由、页面、权限、租户隔离、E2E 测试和部署证据五项映射。
+
+## 4. 与 AIOS V2 目标架构对比
+
+目标逻辑架构：
+
+```text
+Web Console / Public API
+          |
+AIOS Control Plane
+  - Tenant Registry
+  - Identity & Policy
+  - Agent Registry
+  - Workflow / Task Lifecycle
+  - Approval & Audit
+  - Runtime Inventory
+          |
+Agent Router / Scheduling Plane
+       /              \
+OpenClaw Adapter    Hermes Runtime Adapter
+       |              |
+OpenClaw Cells      Hermes Workers / Profiles
+```
+
+边界规则：
+
+- 只有 AIOS Control Plane 可以创建、审批、取消和审计任务。
+- Router 只负责基于权威上下文的选择、调度和转发，不拥有租户事实源。
+- OpenClaw 不拥有租户、策略或任务生命周期；只执行有签名、可审计、可幂等的命令。
+- Hermes 不拥有控制面策略；只运行隔离 Profile、消费租户队列并返回结果/遥测。
+- Web Console 不直接调用 OpenClaw/Hermes；只调用 AIOS API。
+- 每次调用均传播 `tenant_id`、`principal_id`、`session_id`、`task_id`、`trace_id` 和策略版本。
+
+## 5. Gap Analysis
+
+| ID | 领域 | 当前 | 目标 | Gap | 优先级 |
+|---|---|---|---|---|---|
+| G-001 | 架构事实源 | `docs/aios-v2/` 不在权威分支 | 全部 V2 设计、ADR、接口和门禁可追踪 | 无法自动核对目标架构 | P0 |
+| G-002 | 主线收敛 | V2 能力分散于候选/结果分支 | `main` 上单一集成历史 | 分支存在但未纳管、未提升 | P0 |
+| G-003 | 唯一控制面 | Gateway/Router/执行器均可能承载控制逻辑 | AIOS 是唯一任务与策略入口 | 控制权边界尚未强制 | P0 |
+| G-004 | 多租户强制 | 候选模块多为显式字段和 SQLite 逻辑隔离 | 全链路强制隔离、负向测试、生产存储策略 | 缺少统一 middleware、RLS/物理隔离策略和 E2E 证据 | P0 |
+| G-005 | 审计与重放 | 全局哈希链；上下文有 nonce 但无 replay store | 按租户/审计域不可篡改、可查询、可验证；命令防重放 | 审计域、保留、导出、重放防护不足 | P0 |
+| G-006 | OpenClaw 边界 | Fleet CLI 适配已有 | 版本化执行协议、幂等、超时、取消、回调、审计 | 仍可能被控制面之外直接调用 | P1 |
+| G-007 | Hermes 边界 | 租户队列/Profile 已有 | Runtime-only、配额、公平调度、取消、恢复 | 共享存储和控制逻辑仍需拆分 | P1 |
+| G-008 | Web 化 | 多个局部 Web/UI 分支 | 所有控制能力均有 Web/API，权限一致 | 无统一 Console、API contract 和 E2E 门禁 | P1 |
+| G-009 | 生产数据层 | 多个候选模块使用 SQLite/JSON | HA、迁移、备份、加密、容量和恢复目标 | 原型数据层不能直接生产 | P1 |
+| G-010 | 证据治理 | 分支名与内容可能不一致，大 diff 超限 | 能力—提交—测试—部署机器可读映射 | 无法可靠计算完成度 | P0 |
+| G-011 | 发布闭环 | 基线提升与生产部署脱节 | 构建、测试、安全、审批、提升、部署、回滚闭环 | 当前基线仍注明未部署 | P0 |
+| G-012 | 可观测性 | 局部分支存在观测/WebUI 工作 | tenant/task/agent/trace 统一指标日志追踪 | 统一 telemetry contract 尚未成为门禁 | P1 |
+
+## 6. 能力标记
+
+### 6.1 保留并复用
+
+- GitOps Runtime、非主线分支、提交元数据、审批式 Tag/回滚流程。
+- OpenClaw Fleet 和现有执行能力。
+- Hermes Runtime、Profile、队列和工作区能力。
+- Agent Router 的路由/调度经验和已有接口。
+- V2R1 TenantContext、CanonicalMessage、TenantRouter、MemoryScope、SecretRef、MessageLedger 合同。
+- 现有 Gateway、自动领取、任务生命周期回写和测试资产。
+- 已有 WebUI、审计和观测代码中可以由证据证明的部分。
+
+### 6.2 必须重构而非重写
+
+- 将 OpenClaw 内部的租户、策略、审批和任务状态决策移到 AIOS；保留执行器和 Fleet 能力。
+- 将 Hermes 的控制逻辑移到 AIOS/Router；保留 Runtime、Profile 与 Worker。
+- 合并重复 Router/Gateway 入口，形成一个权威 Tenant/Policy middleware。
+- 将 SQLite/JSON 原型抽象为 repository/queue/audit 接口，允许先兼容、后替换生产后端。
+- 将审计链改为租户或明确审计域分区，并加入签名、外部固化、查询授权和验证作业。
+- 将所有 UI 改为 AIOS API 客户端，禁止浏览器直接访问执行层。
+- 将测试报告升级为可复现命令、commit、exit code、pass count 和 artifact digest。
+
+### 6.3 暂停继续扩展
+
+在 P0 收敛完成前，不新增：
+
+- 绕过 AIOS Gateway 的 OpenClaw/Hermes 公共入口。
+- 未携带 TenantContext 的任务、消息、缓存或后台作业。
+- 只存在于个人/临时分支且没有验收合同的大型功能。
+- 单次超大提交；每个任务必须可独立审查、测试和回滚。
+
+## 7. 实施路线
+
+### Phase 0：权威化与门禁（立即）
+
+1. 创建 `docs/aios-v2/` 单一事实源和 ADR 索引。
+2. 生成能力—提交—测试—部署 Evidence Index。
+3. 提取 V2R1 候选能力到窄范围集成分支，不直接合并整个历史分支。
+4. 建立统一 TenantContext middleware、Policy Decision 和审计接口。
+5. 建立 AIOS Control Plane Gateway 唯一入口与禁止旁路测试。
+6. 建立 CI Gate：多租户负向测试、重放、幂等、权限、Web E2E、审计验证。
+
+### Phase 1：执行层解耦
+
+1. OpenClaw Adapter v1：版本化协议、能力发现、start/stop/execute/cancel/status、幂等键、超时、结果回调。
+2. Hermes Runtime Adapter v1：enqueue/claim/heartbeat/cancel/result、配额、公平调度和 Tenant Profile 强制。
+3. Router 只读取 AIOS 权威注册表与策略结果，不保留第二套租户事实源。
+
+### Phase 2：Web 与生产化
+
+1. AIOS Console：Tenant、Agent、Workflow、Task、Approval、Runtime、Audit、GitOps 页面。
+2. 统一 API、RBAC/ABAC、租户会话隔离、CSRF/CORS、速率限制和操作审计。
+3. 替换/封装原型数据层，完成迁移、备份恢复、HA、容量和灾备演练。
+4. 完成受控基线提升、部署、冒烟和回滚证据。
+
+## 8. 下一阶段出口标准
+
+只有同时满足以下条件，才允许进入 V2R1 集成候选：
+
+- `docs/aios-v2/` 在目标集成分支存在，且架构、ADR、API、租户模型和门禁互相链接。
+- AIOS Gateway 是任务创建、审批、状态变更和取消的唯一公共入口。
+- OpenClaw/Hermes 旁路调用被代码、网络策略或测试明确拒绝。
+- 所有持久化记录和消息均含不可为空的 `tenant_id`，跨租户负向测试通过。
+- TenantContext 具备密钥版本、重放缓存、时钟偏差、撤销和审计策略。
+- 审计按租户/审计域隔离，能够从 Web/API 授权查询并验证完整性。
+- Codex 和 OpenClaw 任务均返回 branch、commit、测试命令、exit code、pass count、风险与回滚说明。
+- 变更拆分为可审查提交，不出现无法由 diff 接口检查的单任务超大提交。
+- 未执行直接 `main` push、Tag、生产部署或密钥变更。
+
+## 9. 当前状态结论
+
+| 结论 | 状态 |
+|---|---|
+| 不重写现有代码 | 已决策 |
+| 优先复用 OpenClaw/Hermes/Agent Router | 已决策 |
+| OpenClaw 与 Hermes 逐步解耦 | 已列入 Phase 1 |
+| AIOS 成为唯一控制面 | P0 实施中 |
+| 所有功能 Web 化 | P1 规划完成，尚未验收 |
+| 所有模块多租户隔离 | 候选合同存在，生产级全链路尚未完成 |
+| 所有修改通过 Git 任务流程 | 当前已遵守 |
+
+**总体完成度判定：架构与候选能力约处于“可收敛、不可发布”阶段。下一步不是继续增加功能，而是建立权威事实源、统一入口、租户门禁和可验证的集成闭环。**
@@ -1,107 +0,0 @@
-# AIOS Orchestrator Gateway v0.2
-
-This component turns the current passive GPT orchestration pattern into a
-durable, unattended loop:
-
-```text
-Gitea/OpenClaw/Codex event
-  -> HTTP event endpoint or polled file inbox
-  -> durable SQLite state machine
-  -> Workspace Agent API trigger
-  -> strict callback decision
-  -> Responses API background fallback and polling
-  -> OpenClaw/Codex dispatch queues
-  -> start/end/result events
-  -> next orchestration cycle
-```
-
-## Reliability mechanisms
-
-The gateway does not depend on one timer, webhook or model call:
-
- HTTP event ingestion.
- Periodic file-inbox polling for missed webhooks and offline producers.
- Local bare-Git ref polling for `main`, governance, patch and result refs.
- Workspace Agent retries with a stable event `Idempotency-Key`.
- Workspace Agent callback timeout and Responses API background fallback.
- Responses status polling until completion or fail-closed terminal status.
- SQLite optimistic state versions, idempotency records and processing leases.
- Exponential retry with a maximum retry count and dead-letter state.
- Same-blocker and maximum-hop guards to stop non-progress loops.
- A systemd timer runs `--poll-once` as an independent reliability watchdog.
-
-Workspace Agent trigger runs are asynchronous and do not expose a retrievable
-final response. The central Workspace Agent must therefore call the
-`submit_orchestration_decision` action.
-
-## OpenClaw synchronization
-
-Every lifecycle stage is durably written to the OpenClaw synchronization queue:
-
- `TASK_DISPATCHED`
- `TASK_STARTED`
- `TASK_ENDED`
- `DECISION_APPLIED`
- `PROJECT_SUMMARY`
- `PROJECT_COMPLETED`
-
-An optional authenticated OpenClaw HTTP endpoint can receive the same records.
-A failed HTTP delivery remains in the lifecycle retry queue. OpenClaw remains
-the default project-status synchronizer and authoritative progress presenter.
-
-## Agent execution contract
-
-Each dispatched JSON envelope contains:
-
- exact contract commit;
- source event;
- clean base commit;
- primary and fallback agent;
- allowed paths;
- required checks;
- forbidden operations;
- completion event callback format.
-
-OpenClaw and Codex must send a `work_unit_started` event before execution and a
-`work_unit_completed`, `work_unit_blocked`, or `work_unit_failed` event when
-they stop. This guarantees that OpenClaw can synchronize every task start,
-finish, and overall project summary.
-
-## Security
-
- Model and callback credentials are read only from root-owned secret files.
- Credentials never enter Git, dispatch payloads or logs.
- The gateway runs as a dedicated unprivileged user.
- Dispatch and sync paths are protected against traversal and symlink targets.
- All state changes require the exact contract commit and current state version.
- Production evidence, credentials and approvals stay fail-closed.
- Git history operations remain controlled by the committed governance contract.
-
-## Deployment sequence
-
-1. Apply this source bundle to a clean append-only branch.
-2. Run unit, repository, shell, JSON, systemd and Gitleaks checks.
-3. Configure one or both model channels:
-   - published Workspace Agent API channel and Workspace Agent access token;
-   - Responses API key and approved fallback model.
-4. Publish the decision callback as a Workspace Agent action or MCP tool.
-5. Install the service through `scripts/install_aios_orchestrator_gateway.sh`.
-6. Run `scripts/verify_aios_orchestrator_gateway.sh`.
-7. Submit a harmless event and verify:
-   - trigger acceptance;
-   - strict decision callback;
-   - dispatch file;
-   - OpenClaw task-start/task-end/project-summary synchronization;
-   - duplicate idempotency;
-   - polling recovery after suppressing the HTTP webhook.
-
-The installer copies an explicit gateway-only allowlist into a release directory
-named by its SHA-256 digest, atomically updates `/opt/aios-orchestrator-gateway/current`,
-and persists `/var/lib/aios-orchestrator-gateway/install-receipt.json` with the
-current and previous release. `--dry-run` performs source validation and emits
-`AIOS_ORCHESTRATOR_INSTALL_DRY_RUN=PASS` without changing installed state.
-
-Rollback reads that root-owned `0600` receipt, verifies both release directory
-names and `SOURCE_DIGEST` files, coordinates the gateway and poll units, and
-atomically switches `current` to the previous release. Repeating the same
-rollback is a no-op and emits `NOOP_ALREADY_ROLLED_BACK`.
@@ -0,0 +1,550 @@
+#!/usr/bin/env python3
+"""Isolated Git worktree lifecycle for AIOS GitOps write requests.
+
+The long-running controller workspace is read-only and remains detached.  Each
+write request gets a short-lived detached worktree.  A branch-scoped lease and
+compare-and-swap ref update prevent concurrent writers from silently replacing
+one another.
+"""
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import threading
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable
+
+
+ZERO_OID = "0" * 40
+BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,179}$")
+
+
+class GitWorkspaceError(RuntimeError):
+    """Base error for isolated worktree operations."""
+
+
+class BranchLeaseBusy(GitWorkspaceError):
+    """Another request currently owns the target branch lease."""
+
+
+class BranchCheckedOut(GitWorkspaceError):
+    """The target branch is attached to a non-ephemeral worktree."""
+
+
+class ConcurrentBranchUpdate(GitWorkspaceError):
+    """The branch changed after the request acquired its lease."""
+
+
+@dataclass
+class WorkspaceLease:
+    branch: str
+    base: str
+    path: Path
+    start_oid: str
+    local_oid_before: str | None
+    remote_oid_before: str | None
+    lock_file: Any
+
+
+class GitOpsWorktreeManager:
+    """Manage one detached worktree per GitOps write request.
+
+    A manager instance is safe for a ``ThreadingHTTPServer``.  Request state is
+    stored in thread-local storage, while a filesystem lease serializes writers
+    targeting the same branch across processes.
+    """
+
+    def __init__(
+        self,
+        controller_workspace: str | Path,
+        worktree_root: str | Path,
+        *,
+        remote: str = "origin",
+        lock_root: str | Path | None = None,
+        author_name: str = "chatgpt-gitops-bot",
+        author_email: str = "chatgpt-gitops-bot@cnnetmkt.top",
+    ) -> None:
+        self.controller_workspace = Path(controller_workspace).resolve()
+        self.worktree_root = Path(worktree_root).resolve()
+        self.remote = remote
+        self.author_name = author_name
+        self.author_email = author_email
+        self._state = threading.local()
+
+        if not self.controller_workspace.exists():
+            raise GitWorkspaceError(
+                f"controller workspace does not exist: {self.controller_workspace}"
+            )
+
+        common = self._git(
+            ["rev-parse", "--path-format=absolute", "--git-common-dir"],
+            cwd=self.controller_workspace,
+        ).stdout.strip()
+        self.common_git_dir = Path(common).resolve()
+        self.worktree_root.mkdir(parents=True, exist_ok=True)
+
+        preferred_lock_root = Path(
+            lock_root
+            or os.getenv("AIOS_GITOPS_BRANCH_LOCK_ROOT", "/run/lock/aios-gitops")
+        )
+        try:
+            preferred_lock_root.mkdir(parents=True, exist_ok=True)
+            self.lock_root = preferred_lock_root.resolve()
+        except PermissionError:
+            fallback = self.worktree_root / ".locks"
+            fallback.mkdir(parents=True, exist_ok=True)
+            self.lock_root = fallback.resolve()
+
+    @staticmethod
+    def validate_branch(branch: str) -> str:
+        value = str(branch or "").strip()
+        invalid = (
+            not BRANCH_RE.fullmatch(value)
+            or value in {"main", "master"}
+            or value.startswith(("main/", "master/"))
+            or value.endswith(("/", ".", ".lock"))
+            or value.startswith(("/", "."))
+            or ".." in value
+            or "//" in value
+            or "@{" in value
+            or "\\" in value
+        )
+        if invalid:
+            raise ValueError(f"invalid or protected branch: {value!r}")
+        return value
+
+    @staticmethod
+    def validate_path(path: str) -> str:
+        raw = str(path or "").replace("\\", "/").strip()
+        candidate = Path(raw)
+        if (
+            not raw
+            or candidate.is_absolute()
+            or ".." in candidate.parts
+            or candidate.parts[0] == ".git"
+        ):
+            raise ValueError(f"invalid repository path: {path!r}")
+        return str(candidate)
+
+    def _git(
+        self,
+        args: Iterable[str],
+        *,
+        cwd: Path | None = None,
+        check: bool = True,
+    ) -> subprocess.CompletedProcess[str]:
+        command = ["git", *list(args)]
+        proc = subprocess.run(
+            command,
+            cwd=str(cwd or self.controller_workspace),
+            text=True,
+            capture_output=True,
+        )
+        if check and proc.returncode != 0:
+            message = proc.stderr.strip() or proc.stdout.strip() or "git command failed"
+            raise GitWorkspaceError(f"{message}; command={command!r}")
+        return proc
+
+    def _git_dir(
+        self, args: Iterable[str], *, check: bool = True
+    ) -> subprocess.CompletedProcess[str]:
+        return self._git(
+            [f"--git-dir={self.common_git_dir}", *list(args)],
+            cwd=self.controller_workspace,
+            check=check,
+        )
+
+    def _rev_parse(self, ref: str) -> str | None:
+        proc = self._git_dir(["rev-parse", "--verify", ref], check=False)
+        return proc.stdout.strip() if proc.returncode == 0 else None
+
+    def _remote_oid(self, branch: str) -> str | None:
+        proc = self._git(
+            ["ls-remote", "--heads", self.remote, f"refs/heads/{branch}"],
+            cwd=self.controller_workspace,
+            check=False,
+        )
+        if proc.returncode != 0 or not proc.stdout.strip():
+            return None
+        return proc.stdout.split()[0]
+
+    def _fetch_remote_tracking(self, branch: str) -> str | None:
+        remote_oid = self._remote_oid(branch)
+        if not remote_oid:
+            return None
+        destination = f"refs/remotes/{self.remote}/{branch}"
+        self._git(
+            [
+                "fetch",
+                "--no-tags",
+                self.remote,
+                f"+refs/heads/{branch}:{destination}",
+            ],
+            cwd=self.controller_workspace,
+        )
+        return remote_oid
+
+    def _checked_out_paths(self, branch: str) -> list[str]:
+        target_ref = f"refs/heads/{branch}"
+        proc = self._git_dir(["worktree", "list", "--porcelain"])
+        paths: list[str] = []
+        current_path: str | None = None
+        for line in proc.stdout.splitlines():
+            if line.startswith("worktree "):
+                current_path = line.removeprefix("worktree ").strip()
+            elif line.startswith("branch ") and line.removeprefix("branch ").strip() == target_ref:
+                if current_path:
+                    paths.append(current_path)
+        return paths
+
+    def _assert_branch_not_checked_out(self, branch: str) -> None:
+        paths = self._checked_out_paths(branch)
+        if paths:
+            raise BranchCheckedOut(
+                f"branch {branch!r} is checked out at: {', '.join(paths)}"
+            )
+
+    def _acquire_lock(self, branch: str):
+        digest = hashlib.sha256(branch.encode("utf-8")).hexdigest()
+        path = self.lock_root / f"{digest}.lock"
+        handle = path.open("a+", encoding="utf-8")
+        try:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            handle.close()
+            raise BranchLeaseBusy(f"branch lease is busy: {branch}") from exc
+        handle.seek(0)
+        handle.truncate()
+        handle.write(json.dumps({"branch": branch, "pid": os.getpid()}))
+        handle.flush()
+        return handle
+
+    def _release_lock(self, handle: Any) -> None:
+        try:
+            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+        finally:
+            handle.close()
+
+    def _current(self) -> WorkspaceLease | None:
+        return getattr(self._state, "lease", None)
+
+    def current_workspace(self) -> Path:
+        lease = self._current()
+        return lease.path if lease else self.controller_workspace
+
+    def begin(self, branch: str, base: str = "main", create: bool = False) -> str:
+        del create  # Existing branches are resumed; they are never destructively reset.
+        if self._current() is not None:
+            raise GitWorkspaceError("request already owns a worktree lease")
+
+        branch = self.validate_branch(branch)
+        base = str(base or "main").strip()
+        lock_file = self._acquire_lock(branch)
+        try:
+            self._assert_branch_not_checked_out(branch)
+            remote_oid = self._fetch_remote_tracking(branch)
+            local_oid = self._rev_parse(f"refs/heads/{branch}")
+
+            if remote_oid and local_oid and remote_oid != local_oid:
+                raise ConcurrentBranchUpdate(
+                    f"local and remote branch diverged before write: {branch}"
+                )
+
+            start_oid = remote_oid or local_oid
+            if not start_oid:
+                base_remote = self._fetch_remote_tracking(base)
+                base_local = self._rev_parse(f"refs/heads/{base}")
+                start_oid = base_remote or base_local or self._rev_parse(base)
+            if not start_oid:
+                raise GitWorkspaceError(f"cannot resolve base ref: {base}")
+
+            suffix = uuid.uuid4().hex[:12]
+            safe_branch = re.sub(r"[^A-Za-z0-9._-]+", "-", branch)[:96]
+            path = self.worktree_root / f"{safe_branch}-{suffix}"
+            self._git_dir(["worktree", "add", "--detach", str(path), start_oid])
+            self._state.lease = WorkspaceLease(
+                branch=branch,
+                base=base,
+                path=path,
+                start_oid=start_oid,
+                local_oid_before=local_oid,
+                remote_oid_before=remote_oid,
+                lock_file=lock_file,
+            )
+            return branch
+        except Exception:
+            self._release_lock(lock_file)
+            raise
+
+    def sh(self, args: list[str], check: bool = False) -> dict[str, Any]:
+        proc = self._git(args, cwd=self.current_workspace(), check=False)
+        if check and proc.returncode != 0:
+            message = proc.stderr.strip() or proc.stdout.strip() or "git command failed"
+            raise GitWorkspaceError(message)
+        return {
+            "code": proc.returncode,
+            "stdout": proc.stdout.strip(),
+            "stderr": proc.stderr.strip(),
+            "cmd": ["git", *args],
+        }
+
+    def write_file(self, path: str, content: str) -> str:
+        lease = self._current()
+        if lease is None:
+            raise GitWorkspaceError("write attempted without an active branch lease")
+        relative = self.validate_path(path)
+        target = lease.path / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(content, encoding="utf-8")
+        return relative
+
+    def write_json(self, path: str, data: Any) -> str:
+        return self.write_file(
+            path, json.dumps(data, ensure_ascii=False, indent=2) + "\n"
+        )
+
+    def _update_local_branch(self, lease: WorkspaceLease, new_oid: str) -> None:
+        self._assert_branch_not_checked_out(lease.branch)
+        ref = f"refs/heads/{lease.branch}"
+        expected = lease.local_oid_before or ZERO_OID
+        proc = self._git_dir(["update-ref", ref, new_oid, expected], check=False)
+        if proc.returncode != 0:
+            raise ConcurrentBranchUpdate(
+                proc.stderr.strip()
+                or f"local branch changed concurrently: {lease.branch}"
+            )
+
+    def _rollback_local_branch(self, lease: WorkspaceLease, new_oid: str) -> None:
+        ref = f"refs/heads/{lease.branch}"
+        if lease.local_oid_before:
+            self._git_dir(
+                ["update-ref", ref, lease.local_oid_before, new_oid], check=False
+            )
+        else:
+            self._git_dir(["update-ref", "-d", ref, new_oid], check=False)
+
+    def _push_remote(self, lease: WorkspaceLease, new_oid: str) -> dict[str, Any]:
+        remote_url = self._git(
+            ["remote", "get-url", self.remote],
+            cwd=lease.path,
+            check=False,
+        )
+        if remote_url.returncode != 0:
+            return {"status": "synced", "target": "common_git_dir"}
+
+        command = ["push", "--porcelain"]
+        if lease.remote_oid_before:
+            command.append(
+                f"--force-with-lease=refs/heads/{lease.branch}:{lease.remote_oid_before}"
+            )
+        command.extend(
+            [self.remote, f"HEAD:refs/heads/{lease.branch}"]
+        )
+        proc = self._git(command, cwd=lease.path, check=False)
+        if proc.returncode != 0:
+            self._rollback_local_branch(lease, new_oid)
+            raise ConcurrentBranchUpdate(
+                proc.stderr.strip()
+                or proc.stdout.strip()
+                or f"remote branch changed concurrently: {lease.branch}"
+            )
+
+        tracking_ref = f"refs/remotes/{self.remote}/{lease.branch}"
+        tracking_expected = lease.remote_oid_before or ZERO_OID
+        self._git_dir(
+            ["update-ref", tracking_ref, new_oid, tracking_expected], check=False
+        )
+        return {
+            "status": "synced",
+            "target": self.remote,
+            "stdout": proc.stdout.strip(),
+            "stderr": proc.stderr.strip(),
+        }
+
+    def commit(self, paths: list[str] | str | None, message: str) -> dict[str, Any]:
+        lease = self._current()
+        if lease is None:
+            raise GitWorkspaceError("commit attempted without an active branch lease")
+
+        raw_paths = paths or ["."]
+        if isinstance(raw_paths, str):
+            raw_paths = [raw_paths]
+        safe_paths = [self.validate_path(p) if p != "." else "." for p in raw_paths]
+        self._git(["add", "--", *safe_paths], cwd=lease.path)
+        status = self._git(["status", "--porcelain"], cwd=lease.path).stdout.strip()
+        if not status:
+            result = {
+                "status": "no_changes",
+                "branch": lease.branch,
+                "rev": lease.start_oid[:12],
+                "sync": {"status": "unchanged"},
+                "workspace_mode": "detached",
+            }
+            self.finish(success=True)
+            return result
+
+        commit_proc = self._git(
+            [
+                "-c",
+                f"user.name={self.author_name}",
+                "-c",
+                f"user.email={self.author_email}",
+                "commit",
+                "-m",
+                message,
+            ],
+            cwd=lease.path,
+        )
+        new_oid = self._git(["rev-parse", "HEAD"], cwd=lease.path).stdout.strip()
+        self._update_local_branch(lease, new_oid)
+        sync = self._push_remote(lease, new_oid)
+        result = {
+            "status": "committed",
+            "branch": lease.branch,
+            "rev": new_oid[:12],
+            "commit": new_oid,
+            "git": {
+                "code": commit_proc.returncode,
+                "stdout": commit_proc.stdout.strip(),
+                "stderr": commit_proc.stderr.strip(),
+            },
+            "sync": sync,
+            "workspace_mode": "detached",
+        }
+        self.finish(success=True)
+        return result
+
+    def finish(self, *, success: bool) -> dict[str, Any] | None:
+        lease = self._current()
+        if lease is None:
+            return None
+        self._state.lease = None
+        retained = False
+        cleanup_error: str | None = None
+        try:
+            status = self._git(
+                ["status", "--porcelain"], cwd=lease.path, check=False
+            ).stdout.strip()
+            if success and not status:
+                proc = self._git_dir(
+                    ["worktree", "remove", str(lease.path)], check=False
+                )
+                if proc.returncode != 0:
+                    cleanup_error = proc.stderr.strip() or proc.stdout.strip()
+                    retained = True
+            else:
+                retained = lease.path.exists()
+        finally:
+            self._release_lock(lease.lock_file)
+            self._git_dir(["worktree", "prune", "--expire", "now"], check=False)
+        return {
+            "branch": lease.branch,
+            "worktree": str(lease.path),
+            "retained": retained,
+            "cleanup_error": cleanup_error,
+        }
+
+    def abort_current(self) -> dict[str, Any] | None:
+        """Release an unfinished request, retaining dirty evidence for review."""
+        lease = self._current()
+        if lease is None:
+            return None
+        status = self._git(
+            ["status", "--porcelain"], cwd=lease.path, check=False
+        ).stdout.strip()
+        return self.finish(success=not bool(status))
+
+    def sync_branch(self, branch: str) -> dict[str, Any]:
+        """Compatibility hook for the legacy API entrypoint."""
+        return {
+            "status": "managed_by_isolated_worktree",
+            "branch": self.validate_branch(branch),
+        }
+
+    def sync_tag(self, tag: str) -> dict[str, Any]:
+        proc = self._git(
+            ["push", self.remote, f"refs/tags/{tag}:refs/tags/{tag}"],
+            cwd=self.controller_workspace,
+            check=False,
+        )
+        if proc.returncode != 0:
+            raise GitWorkspaceError(proc.stderr.strip() or proc.stdout.strip())
+        return {
+            "status": "synced",
+            "target": self.remote,
+            "tag": tag,
+            "stdout": proc.stdout.strip(),
+            "stderr": proc.stderr.strip(),
+        }
+
+    def runtime_status(self) -> dict[str, Any]:
+        branch = self._git(
+            ["branch", "--show-current"], cwd=self.controller_workspace
+        ).stdout.strip()
+        head = self._git(
+            ["rev-parse", "HEAD"], cwd=self.controller_workspace
+        ).stdout.strip()
+        dirty = bool(
+            self._git(
+                ["status", "--porcelain"], cwd=self.controller_workspace
+            ).stdout.strip()
+        )
+        return {
+            "workspace_mode": "branch" if branch else "detached",
+            "workspace_branch": branch or None,
+            "workspace_ref": head,
+            "workspace_dirty": dirty,
+            "common_git_dir": str(self.common_git_dir),
+            "task_worktree_root": str(self.worktree_root),
+        }
+
+
+def remove_abandoned_clean_worktrees(
+    manager: GitOpsWorktreeManager, *, prefix: str | Path | None = None
+) -> list[str]:
+    """Remove clean detached worktrees below the managed root.
+
+    Dirty worktrees are deliberately retained.  This helper is intended for a
+    startup recovery job and never removes worktrees outside the configured
+    managed root.
+    """
+    root = Path(prefix or manager.worktree_root).resolve()
+    removed: list[str] = []
+    proc = manager._git_dir(["worktree", "list", "--porcelain"])
+    candidates: list[Path] = []
+    current: Path | None = None
+    detached = False
+    for line in [*proc.stdout.splitlines(), ""]:
+        if line.startswith("worktree "):
+            if current and detached:
+                candidates.append(current)
+            current = Path(line.removeprefix("worktree ").strip()).resolve()
+            detached = False
+        elif line == "detached":
+            detached = True
+        elif not line and current and detached:
+            candidates.append(current)
+            current = None
+            detached = False
+
+    for path in candidates:
+        if path == manager.controller_workspace or root not in path.parents:
+            continue
+        status = manager._git(["status", "--porcelain"], cwd=path, check=False)
+        if status.returncode == 0 and not status.stdout.strip():
+            removal = manager._git_dir(
+                ["worktree", "remove", str(path)], check=False
+            )
+            if removal.returncode == 0:
+                removed.append(str(path))
+            elif not path.exists():
+                shutil.rmtree(path, ignore_errors=True)
+    manager._git_dir(["worktree", "prune", "--expire", "now"], check=False)
+    return removed
@@ -1,49 +0,0 @@
-{
-  "host": "127.0.0.1",
-  "port": 18920,
-  "database_path": "/var/lib/aios-orchestrator-gateway/state.sqlite3",
-  "event_inbox_dir": "/var/lib/aios-orchestrator-gateway/inbox/events",
-  "decision_inbox_dir": "/var/lib/aios-orchestrator-gateway/inbox/decisions",
-  "dispatch_outbox_dir": "/var/lib/aios-orchestrator-gateway/outbox",
-  "openclaw_sync_dir": "/var/lib/aios-orchestrator-gateway/openclaw-sync",
-  "command_queue_dir": "/var/lib/aios-orchestrator-gateway/commands",
-  "inbound_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
-  "callback_token_file": "/etc/aios-orchestrator-gateway/callback.token",
-  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "workspace_agent": {
-    "trigger_id": "agtch_REPLACE_ME",
-    "access_token_file": "/etc/aios-orchestrator-gateway/workspace-agent.token"
-  },
-  "responses_fallback": {
-    "api_key_file": "/etc/aios-orchestrator-gateway/openai-api.key",
-    "model": "REPLACE_WITH_APPROVED_MODEL"
-  },
-  "polling": {
-    "interval_seconds": 10,
-    "summary_interval_seconds": 300
-  },
-  "loop_controls": {
-    "callback_timeout_seconds": 900,
-    "lease_seconds": 60,
-    "retry_base_seconds": 10,
-    "max_retries": 5,
-    "max_same_blocker_retries": 3,
-    "max_hops_per_gate": 20
-  },
-  "openclaw_sync": {
-    "endpoint": null,
-    "token_file": null
-  },
-  "project_key": "v2r1-personal-weixin-production",
-  "git_poll": {
-    "enabled": true,
-    "bare_repo": "/data/aios-gitops/repo.git",
-    "ref_prefixes": [
-      "refs/heads/main",
-      "refs/heads/task/governance/",
-      "refs/heads/patch/",
-      "refs/heads/result/"
-    ],
-    "interval_seconds": 30
-  }
-}
@@ -0,0 +1,9 @@
+{
+  "type": "branch",
+  "branch": "patch/aios-v2-implementation-status-next-phase-20260720",
+  "base": "main",
+  "created_by": "chatgpt-gitops-bot",
+  "created_at": "2026-07-19T16:21:00Z",
+  "direct_main_push": false,
+  "owner_approval_required": true
+}
@@ -1,145 +0,0 @@
-{
-  "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "title": "AIOS Orchestrator Decision",
-  "type": "object",
-  "additionalProperties": false,
-  "required": [
-    "schema_version",
-    "project_key",
-    "contract_commit",
-    "event_id",
-    "expected_state_version",
-    "verdict",
-    "state_updates",
-    "next_work_units",
-    "requires_human_input",
-    "stop_reason"
-  ],
-  "properties": {
-    "schema_version": {
-      "type": "integer",
-      "const": 1
-    },
-    "project_key": {
-      "type": "string"
-    },
-    "contract_commit": {
-      "type": "string",
-      "pattern": "^[0-9a-f]{40}$"
-    },
-    "event_id": {
-      "type": "string"
-    },
-    "expected_state_version": {
-      "type": "integer",
-      "minimum": 0
-    },
-    "verdict": {
-      "type": "string",
-      "enum": [
-        "BLOCKED",
-        "COMPLETE",
-        "DISPATCH",
-        "NOOP"
-      ]
-    },
-    "state_updates": {
-      "type": "array",
-      "items": {
-        "type": "object",
-        "additionalProperties": false,
-        "required": [
-          "key",
-          "value"
-        ],
-        "properties": {
-          "key": {
-            "type": "string"
-          },
-          "value": {
-            "type": "string"
-          }
-        }
-      }
-    },
-    "next_work_units": {
-      "type": "array",
-      "maxItems": 32,
-      "items": {
-        "type": "object",
-        "additionalProperties": false,
-        "required": [
-          "id",
-          "track_id",
-          "gate_id",
-          "primary_agent",
-          "fallback_agent",
-          "clean_base_commit",
-          "allowed_paths",
-          "required_checks",
-          "forbidden_operations",
-          "instruction"
-        ],
-        "properties": {
-          "id": {
-            "type": "string"
-          },
-          "track_id": {
-            "type": "string"
-          },
-          "gate_id": {
-            "type": "string"
-          },
-          "primary_agent": {
-            "type": "string",
-            "enum": [
-              "openclaw",
-              "codex"
-            ]
-          },
-          "fallback_agent": {
-            "type": "string",
-            "enum": [
-              "openclaw",
-              "codex"
-            ]
-          },
-          "clean_base_commit": {
-            "type": "string",
-            "pattern": "^[0-9a-f]{40}$"
-          },
-          "allowed_paths": {
-            "type": "array",
-            "items": {
-              "type": "string"
-            }
-          },
-          "required_checks": {
-            "type": "array",
-            "items": {
-              "type": "string"
-            }
-          },
-          "forbidden_operations": {
-            "type": "array",
-            "items": {
-              "type": "string"
-            }
-          },
-          "instruction": {
-            "type": "string"
-          }
-        }
-      }
-    },
-    "requires_human_input": {
-      "type": "boolean"
-    },
-    "stop_reason": {
-      "type": [
-        "string",
-        "null"
-      ]
-    }
-  }
-}
@@ -1,102 +0,0 @@
-{
-  "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "title": "AIOS Orchestrator Event",
-  "type": "object",
-  "additionalProperties": false,
-  "required": [
-    "schema_version",
-    "project_key",
-    "contract_commit",
-    "event_id",
-    "source_agent",
-    "event_type",
-    "work_unit_id",
-    "status",
-    "state_version",
-    "implementation_commit",
-    "result_commit",
-    "blocker",
-    "evidence_refs",
-    "payload"
-  ],
-  "properties": {
-    "schema_version": {
-      "const": 1
-    },
-    "project_key": {
-      "type": "string",
-      "minLength": 1,
-      "maxLength": 128
-    },
-    "contract_commit": {
-      "type": "string",
-      "pattern": "^[0-9a-f]{40}$"
-    },
-    "event_id": {
-      "type": "string",
-      "minLength": 1,
-      "maxLength": 128
-    },
-    "source_agent": {
-      "enum": [
-        "openclaw",
-        "codex",
-        "gitea",
-        "system",
-        "workspace-agent"
-      ]
-    },
-    "event_type": {
-      "type": "string"
-    },
-    "work_unit_id": {
-      "type": "string"
-    },
-    "status": {
-      "enum": [
-        "READY",
-        "RUNNING",
-        "PASS",
-        "BLOCKED",
-        "FAIL",
-        "TIMEOUT",
-        "NOT_EXECUTED"
-      ]
-    },
-    "state_version": {
-      "type": "integer",
-      "minimum": 0
-    },
-    "implementation_commit": {
-      "type": [
-        "string",
-        "null"
-      ],
-      "pattern": "^[0-9a-f]{40}$"
-    },
-    "result_commit": {
-      "type": [
-        "string",
-        "null"
-      ],
-      "pattern": "^[0-9a-f]{40}$"
-    },
-    "blocker": {
-      "type": [
-        "string",
-        "null"
-      ],
-      "maxLength": 512
-    },
-    "evidence_refs": {
-      "type": "array",
-      "items": {
-        "type": "string"
-      },
-      "maxItems": 64
-    },
-    "payload": {
-      "type": "object"
-    }
-  }
-}
@@ -1,63 +0,0 @@
-#!/usr/bin/env python3
-"""Seed exactly one orchestration event when the project has no durable state."""
-from __future__ import annotations
-
-import argparse
-from pathlib import Path
-import sys
-
-REPO_ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(REPO_ROOT))
-
-from services.aios_orchestrator_gateway.config import Config
-from services.aios_orchestrator_gateway.models import Event
-from services.aios_orchestrator_gateway.securefs import atomic_write_bytes_shared
-from services.aios_orchestrator_gateway.store import Store
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--config", type=Path, required=True)
-    args = parser.parse_args()
-
-    config = Config.from_file(args.config)
-    store = Store(config.database_path)
-    if store.project_state(config.project_key) is not None:
-        print("ORCHESTRATION_BOOTSTRAP=NOOP reason=STATE_EXISTS")
-        return 0
-
-    event = Event.from_dict(
-        {
-            "schema_version": 1,
-            "project_key": config.project_key,
-            "contract_commit": config.contract_commit,
-            "event_id": "bootstrap-" + config.contract_commit[:24],
-            "source_agent": "system",
-            "event_type": "orchestration_bootstrap",
-            "work_unit_id": "ORCHESTRATION-BOOTSTRAP",
-            "status": "READY",
-            "state_version": 0,
-            "implementation_commit": None,
-            "result_commit": None,
-            "blocker": None,
-            "evidence_refs": [],
-            "payload": {
-                "reason": "initial_gateway_install",
-                "polling_enabled": True,
-                "openclaw_sync_required": True,
-            },
-        }
-    )
-    path = config.event_inbox_dir / "pending" / f"{event.event_id}.json"
-    atomic_write_bytes_shared(
-        path,
-        (event.canonical_json() + "\n").encode(),
-        mode=0o640,
-        group_writable_directory=True,
-    )
-    print(f"ORCHESTRATION_BOOTSTRAP=PASS event_id={event.event_id}")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())
@@ -1,239 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-umask 0077
-
-SOURCE_ROOT="${SOURCE_ROOT:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}"
-INSTALL_ROOT="${INSTALL_ROOT:-/opt/aios-orchestrator-gateway}"
-CONFIG_ROOT="${CONFIG_ROOT:-/etc/aios-orchestrator-gateway}"
-STATE_ROOT="${STATE_ROOT:-/var/lib/aios-orchestrator-gateway}"
-UNIT_ROOT="${UNIT_ROOT:-/etc/systemd/system}"
-SYSUSERS_ROOT="${SYSUSERS_ROOT:-/etc/sysusers.d}"
-TMPFILES_ROOT="${TMPFILES_ROOT:-/etc/tmpfiles.d}"
-DRY_RUN=0
-
-if [[ "${1:-}" == "--dry-run" ]]; then
-  DRY_RUN=1
-fi
-
-fail() {
-  printf 'AIOS_ORCHESTRATOR_INSTALL=BLOCKED reason=%s\n' "$1" >&2
-  exit 2
-}
-
-[[ "$(id -u)" -eq 0 ]] || fail OWNER_INSTALL_REQUIRED
-
-for cmd in git python3 systemd-analyze systemctl systemd-sysusers systemd-tmpfiles sha256sum find sort xargs; do
-  command -v "$cmd" >/dev/null 2>&1 || fail "MISSING_COMMAND_${cmd}"
-done
-
-if [[ -f "$SOURCE_ROOT/SHA256SUMS" ]]; then
-  (cd "$SOURCE_ROOT" && sha256sum --check SHA256SUMS)
-fi
-
-for path in \
-  "$SOURCE_ROOT/services/aios_orchestrator_gateway" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.service" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.service" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles" \
-  "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json"; do
-  [[ -e "$path" ]] || fail "MISSING_SOURCE_${path}"
-done
-
-PYCACHE="/tmp/aios-orchestrator-install-pycache-$$"
-rm -rf "$PYCACHE"
-PYTHONPYCACHEPREFIX="$PYCACHE" python3 -m compileall -q "$SOURCE_ROOT/services" "$SOURCE_ROOT/scripts"
-PYTHONDONTWRITEBYTECODE=1 PYTHONPATH="$SOURCE_ROOT" \
-  python3 -m unittest -v tests.test_aios_orchestrator_gateway
-rm -rf "$PYCACHE"
-for script in "$SOURCE_ROOT"/scripts/*.sh; do bash -n "$script"; done
-python3 -m json.tool "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json" >/dev/null
-systemd-analyze verify \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.service" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.service" \
-  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer"
-
-SOURCE_FILES=(
-  README-AIOS-ORCHESTRATOR-GATEWAY.md
-  config/aios-orchestrator-gateway.example.json
-  schemas/decision.schema.json
-  schemas/event.schema.json
-  scripts/bootstrap_orchestration_loop.py
-  scripts/install_aios_orchestrator_gateway.sh
-  scripts/report_orchestrator_event.py
-  scripts/rollback_aios_orchestrator_gateway.sh
-  scripts/verify_aios_orchestrator_gateway.sh
-  systemd/aios-orchestrator-gateway-poll.service
-  systemd/aios-orchestrator-gateway-poll.timer
-  systemd/aios-orchestrator-gateway.service
-  systemd/aios-orchestrator-gateway.sysusers
-  systemd/aios-orchestrator-gateway.tmpfiles
-  tests/test_aios_orchestrator_gateway.py
-  tests/test_aios_orchestrator_gateway_rollback.sh
-  workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md
-  workspace-agent/orchestrator-decision-action.openapi.yaml
-)
-while IFS= read -r path; do SOURCE_FILES+=("$path"); done < <(
-  cd "$SOURCE_ROOT"
-  find services/aios_orchestrator_gateway -type f ! -path '*/__pycache__/*' ! -name '*.pyc' -print | sort
-)
-for path in "${SOURCE_FILES[@]}"; do
-  [[ -f "$SOURCE_ROOT/$path" && ! -L "$SOURCE_ROOT/$path" ]] || fail "INVALID_SOURCE_FILE_${path}"
-done
-
-SOURCE_DIGEST="$(
-  cd "$SOURCE_ROOT"
-  printf '%s\0' "${SOURCE_FILES[@]}" | sort -z | xargs -0 sha256sum \
-    | sha256sum \
-    | awk '{print $1}'
-)"
-[[ "$SOURCE_DIGEST" =~ ^[0-9a-f]{64}$ ]] || fail SOURCE_DIGEST_INVALID
-RELEASE_DIR="$INSTALL_ROOT/releases/$SOURCE_DIGEST"
-
-if [[ "$DRY_RUN" -eq 1 ]]; then
-  printf 'AIOS_ORCHESTRATOR_INSTALL_DRY_RUN=PASS digest=%s release=%s\n' \
-    "$SOURCE_DIGEST" "$RELEASE_DIR"
-  exit 0
-fi
-
-install -d -m 0755 "$INSTALL_ROOT/releases"
-install -d -m 0700 "$CONFIG_ROOT"
-install -d -m 0700 "$STATE_ROOT"
-
-systemd-sysusers "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers"
-systemd-tmpfiles --create "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles"
-
-if [[ ! -d "$RELEASE_DIR" ]]; then
-  TMP_RELEASE="$INSTALL_ROOT/releases/.tmp-$SOURCE_DIGEST-$$"
-  install -d -m 0755 "$TMP_RELEASE"
-  for path in "${SOURCE_FILES[@]}"; do
-    install -d -m 0755 "$TMP_RELEASE/$(dirname "$path")"
-    cp -a "$SOURCE_ROOT/$path" "$TMP_RELEASE/$path"
-  done
-  printf '%s\n' "$SOURCE_DIGEST" > "$TMP_RELEASE/SOURCE_DIGEST"
-  chmod -R go-w "$TMP_RELEASE"
-  mv "$TMP_RELEASE" "$RELEASE_DIR"
-fi
-
-PREVIOUS_TARGET=""
-if [[ -L "$INSTALL_ROOT/current" ]]; then
-  PREVIOUS_TARGET="$(readlink -f "$INSTALL_ROOT/current")"
-fi
-
-ln -s "$RELEASE_DIR" "$INSTALL_ROOT/.current-new-$$"
-mv -Tf "$INSTALL_ROOT/.current-new-$$" "$INSTALL_ROOT/current"
-
-install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.service" \
-  "$UNIT_ROOT/aios-orchestrator-gateway.service"
-install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.service" \
-  "$UNIT_ROOT/aios-orchestrator-gateway-poll.service"
-install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer" \
-  "$UNIT_ROOT/aios-orchestrator-gateway-poll.timer"
-install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers" \
-  "$SYSUSERS_ROOT/aios-orchestrator-gateway.conf"
-install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles" \
-  "$TMPFILES_ROOT/aios-orchestrator-gateway.conf"
-
-if [[ ! -f "$CONFIG_ROOT/inbound.token" ]]; then
-  python3 - <<'PY' > "$CONFIG_ROOT/inbound.token"
-import secrets
-print(secrets.token_urlsafe(48))
-PY
-fi
-[[ ! -L "$CONFIG_ROOT/inbound.token" ]] || fail INBOUND_TOKEN_SYMLINK
-chown root:aios-orchestrator "$CONFIG_ROOT/inbound.token"
-chmod 0640 "$CONFIG_ROOT/inbound.token"
-
-if [[ ! -f "$CONFIG_ROOT/callback.token" ]]; then
-  python3 - <<'PY' > "$CONFIG_ROOT/callback.token"
-import secrets
-print(secrets.token_urlsafe(48))
-PY
-fi
-[[ ! -L "$CONFIG_ROOT/callback.token" ]] || fail CALLBACK_TOKEN_SYMLINK
-chown root:aios-orchestrator "$CONFIG_ROOT/callback.token"
-chmod 0640 "$CONFIG_ROOT/callback.token"
-
-if [[ ! -f "$CONFIG_ROOT/config.json" ]]; then
-  install -m 0600 \
-    "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json" \
-    "$CONFIG_ROOT/config.json.example"
-  printf 'AIOS_ORCHESTRATOR_INSTALL=BLOCKED reason=CONFIGURATION_REQUIRED digest=%s\n' \
-    "$SOURCE_DIGEST"
-  exit 2
-fi
-
-chown root:aios-orchestrator "$CONFIG_ROOT/config.json"
-chmod 0640 "$CONFIG_ROOT/config.json"
-
-python3 -m json.tool "$CONFIG_ROOT/config.json" >/dev/null
-AIOS_ORCHESTRATOR_CONFIG_OWNER_UID=0 \
-AIOS_ORCHESTRATOR_SECRET_OWNER_UID=0 \
-PYTHONPATH="$INSTALL_ROOT/current" \
-python3 - "$CONFIG_ROOT/config.json" <<'PY'
-import sys
-from pathlib import Path
-from services.aios_orchestrator_gateway.config import Config
-cfg = Config.from_file(Path(sys.argv[1]))
-cfg.inbound_token()
-cfg.callback_token()
-if cfg.workspace_trigger_id:
-    cfg.workspace_token()
-if cfg.responses_api_key_file:
-    cfg.responses_api_key()
-if cfg.openclaw_sync_endpoint:
-    cfg.openclaw_sync_token()
-print("CONFIG_VALIDATE=PASS")
-PY
-
-python3 - "$STATE_ROOT/install-receipt.json" "$SOURCE_DIGEST" "$RELEASE_DIR" "$PREVIOUS_TARGET" <<'PY'
-import json, os, sys, tempfile
-from pathlib import Path
-path = Path(sys.argv[1])
-data = {
-    "schema_version": 1,
-    "source_digest": sys.argv[2],
-    "current_release": sys.argv[3],
-    "previous_release": sys.argv[4] or None,
-    "last_rollback": None,
-}
-path.parent.mkdir(parents=True, exist_ok=True)
-fd, temp = tempfile.mkstemp(prefix=".install-receipt-", dir=path.parent)
-try:
-    with os.fdopen(fd, "w") as handle:
-        json.dump(data, handle, indent=2, sort_keys=True)
-        handle.write("\n")
-        handle.flush()
-        os.fsync(handle.fileno())
-    os.chmod(temp, 0o600)
-    os.replace(temp, path)
-    directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
-    try:
-        os.fsync(directory)
-    finally:
-        os.close(directory)
-finally:
-    try:
-        os.unlink(temp)
-    except FileNotFoundError:
-        pass
-PY
-chown aios-orchestrator:aios-orchestrator "$STATE_ROOT/install-receipt.json"
-chmod 0600 "$STATE_ROOT/install-receipt.json"
-
-systemctl daemon-reload
-systemctl enable --now aios-orchestrator-gateway.service
-systemctl enable --now aios-orchestrator-gateway-poll.timer
-systemctl is-active --quiet aios-orchestrator-gateway.service
-systemctl is-active --quiet aios-orchestrator-gateway-poll.timer
-
-AIOS_ORCHESTRATOR_CONFIG_OWNER_UID=0 \
-AIOS_ORCHESTRATOR_SECRET_OWNER_UID=0 \
-PYTHONPATH="$INSTALL_ROOT/current" \
-python3 "$INSTALL_ROOT/current/scripts/bootstrap_orchestration_loop.py" \
-  --config "$CONFIG_ROOT/config.json"
-systemctl start aios-orchestrator-gateway-poll.service
-
-printf 'AIOS_ORCHESTRATOR_INSTALL=PASS digest=%s previous=%s\n' \
-  "$SOURCE_DIGEST" "${PREVIOUS_TARGET:-NONE}"
@@ -1,130 +0,0 @@
-#!/usr/bin/env python3
-"""Submit an OpenClaw/Codex task lifecycle event over HTTP or a polled inbox."""
-from __future__ import annotations
-
-import argparse
-import json
-import os
-from pathlib import Path
-import sys
-import urllib.error
-import urllib.request
-
-REPO_ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(REPO_ROOT))
-
-from services.aios_orchestrator_gateway.models import Event
-from services.aios_orchestrator_gateway.securefs import atomic_write_bytes_shared, read_secure_json, read_secure_text
-
-
-def parser() -> argparse.ArgumentParser:
-    p = argparse.ArgumentParser()
-    p.add_argument("--project-key", required=True)
-    p.add_argument("--contract-commit", required=True)
-    p.add_argument("--event-id", required=True)
-    p.add_argument("--source-agent", required=True, choices=["openclaw", "codex", "gitea", "system"])
-    p.add_argument("--event-type", required=True)
-    p.add_argument("--work-unit-id", required=True)
-    p.add_argument(
-        "--status",
-        required=True,
-        choices=["READY", "RUNNING", "PASS", "BLOCKED", "FAIL", "TIMEOUT", "NOT_EXECUTED"],
-    )
-    p.add_argument("--state-version", type=int, required=True)
-    p.add_argument("--implementation-commit")
-    p.add_argument("--result-commit")
-    p.add_argument("--blocker")
-    p.add_argument("--evidence-ref", action="append", default=[])
-    p.add_argument("--payload-file", type=Path)
-    p.add_argument("--endpoint")
-    p.add_argument("--token-file", type=Path)
-    p.add_argument("--inbox-dir", type=Path)
-    p.add_argument("--timeout-seconds", type=int, default=15)
-    return p
-
-
-def main() -> int:
-    args = parser().parse_args()
-    if not args.endpoint and not args.inbox_dir:
-        raise SystemExit("either --endpoint or --inbox-dir is required")
-    if args.endpoint and not args.token_file:
-        raise SystemExit("--token-file is required with --endpoint")
-    payload = {}
-    if args.payload_file:
-        if not args.payload_file.is_absolute():
-            raise SystemExit("--payload-file must be absolute")
-        payload = read_secure_json(
-            args.payload_file, expected_uid=os.getuid(), max_bytes=262_144
-        )
-        if not isinstance(payload, dict):
-            raise SystemExit("payload root must be an object")
-
-    event = Event.from_dict(
-        {
-            "schema_version": 1,
-            "project_key": args.project_key,
-            "contract_commit": args.contract_commit,
-            "event_id": args.event_id,
-            "source_agent": args.source_agent,
-            "event_type": args.event_type,
-            "work_unit_id": args.work_unit_id,
-            "status": args.status,
-            "state_version": args.state_version,
-            "implementation_commit": args.implementation_commit,
-            "result_commit": args.result_commit,
-            "blocker": args.blocker,
-            "evidence_refs": args.evidence_ref,
-            "payload": payload,
-        }
-    )
-    encoded = event.canonical_json().encode()
-    http_error: Exception | None = None
-
-    if args.endpoint:
-        token = read_secure_text(
-            args.token_file,
-            expected_uid=None,
-            max_bytes=64_000,
-        )
-        request = urllib.request.Request(
-            args.endpoint.rstrip("/") + "/v1/orchestrator/events",
-            data=encoded,
-            method="POST",
-            headers={
-                "Authorization": f"Bearer {token}",
-                "Content-Type": "application/json",
-                "Idempotency-Key": event.event_id,
-                "User-Agent": "aios-agent-event-reporter/0.2",
-            },
-        )
-        try:
-            with urllib.request.urlopen(request, timeout=args.timeout_seconds) as response:
-                response.read(65_536)
-                if response.status not in {200, 202}:
-                    raise RuntimeError(f"unexpected HTTP {response.status}")
-                print(f"EVENT_SUBMIT=PASS event_id={event.event_id} channel=http")
-                return 0
-        except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError) as exc:
-            http_error = exc
-
-    if args.inbox_dir:
-        pending = args.inbox_dir / "pending"
-        path = pending / f"{event.event_id}.json"
-        atomic_write_bytes_shared(
-            path,
-            (json.dumps(event.to_dict(), indent=2, sort_keys=True) + "\n").encode(),
-            mode=0o640,
-            group_writable_directory=True,
-        )
-        print(f"EVENT_SUBMIT=PASS event_id={event.event_id} channel=file_inbox")
-        return 0
-
-    print(
-        f"EVENT_SUBMIT=FAIL event_id={event.event_id} reason={type(http_error).__name__}",
-        file=sys.stderr,
-    )
-    return 2
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())
@@ -1,102 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-umask 0077
-
-INSTALL_ROOT="${INSTALL_ROOT:-/opt/aios-orchestrator-gateway}"
-STATE_ROOT="${STATE_ROOT:-/var/lib/aios-orchestrator-gateway}"
-RECEIPT="$STATE_ROOT/install-receipt.json"
-
-fail() {
-  printf 'AIOS_ORCHESTRATOR_ROLLBACK=BLOCKED reason=%s\n' "$1" >&2
-  exit 2
-}
-
-[[ "$(id -u)" -eq 0 ]] || fail OWNER_REQUIRED
-[[ -f "$RECEIPT" ]] || fail RECEIPT_MISSING
-[[ ! -L "$RECEIPT" ]] || fail RECEIPT_SYMLINK
-[[ "$(stat -c '%u:%a' "$RECEIPT")" == "0:600" ]] || fail RECEIPT_PERMISSION
-
-readarray -t VALUES < <(
-python3 - "$RECEIPT" "$INSTALL_ROOT/releases" "$INSTALL_ROOT/current" <<'PY'
-import json, os, stat, sys
-from pathlib import Path
-receipt = Path(sys.argv[1])
-releases = Path(sys.argv[2]).resolve()
-current_link = Path(sys.argv[3])
-data = json.loads(receipt.read_text())
-previous = data.get("previous_release")
-current = data.get("current_release")
-if not previous or not current:
-    raise SystemExit(2)
-previous_path = Path(previous).resolve()
-current_path = Path(current).resolve()
-if previous_path.parent != releases or current_path.parent != releases:
-    raise SystemExit(2)
-if not stat.S_ISDIR(os.lstat(previous_path).st_mode) or not stat.S_ISDIR(os.lstat(current_path).st_mode):
-    raise SystemExit(2)
-for release in (previous_path, current_path):
-    if len(release.name) != 64 or any(ch not in "0123456789abcdef" for ch in release.name):
-        raise SystemExit(2)
-    digest_file = release / "SOURCE_DIGEST"
-    if digest_file.is_symlink() or not digest_file.is_file():
-        raise SystemExit(2)
-    if digest_file.read_text(encoding="utf-8").strip() != release.name:
-        raise SystemExit(2)
-last = data.get("last_rollback")
-if isinstance(last, dict) and last.get("to") == current and current_link.is_symlink():
-    if current_link.resolve() == current_path:
-        print("ALREADY_ROLLED_BACK")
-        raise SystemExit(0)
-print(previous_path)
-print(current_path)
-PY
-) || fail RECEIPT_INVALID
-
-if [[ "${VALUES[0]:-}" == "ALREADY_ROLLED_BACK" ]]; then
-  echo "AIOS_ORCHESTRATOR_ROLLBACK=PASS result=NOOP_ALREADY_ROLLED_BACK"
-  exit 0
-fi
-[[ "${#VALUES[@]}" -eq 2 ]] || fail RECEIPT_INVALID
-PREVIOUS="${VALUES[0]}"
-CURRENT="${VALUES[1]}"
-systemctl stop aios-orchestrator-gateway-poll.timer || true
-systemctl stop aios-orchestrator-gateway-poll.service || true
-systemctl stop aios-orchestrator-gateway.service || true
-ln -s "$PREVIOUS" "$INSTALL_ROOT/.current-rollback-$$"
-mv -Tf "$INSTALL_ROOT/.current-rollback-$$" "$INSTALL_ROOT/current"
-systemctl start aios-orchestrator-gateway.service
-systemctl is-active --quiet aios-orchestrator-gateway.service
-systemctl start aios-orchestrator-gateway-poll.timer
-systemctl is-active --quiet aios-orchestrator-gateway-poll.timer
-
-python3 - "$RECEIPT" "$PREVIOUS" "$CURRENT" <<'PY'
-import json, os, sys, tempfile
-from pathlib import Path
-path = Path(sys.argv[1])
-data = json.loads(path.read_text())
-data["current_release"] = sys.argv[2]
-data["previous_release"] = sys.argv[3]
-data["last_rollback"] = {"from": sys.argv[3], "to": sys.argv[2]}
-fd, temp = tempfile.mkstemp(prefix=".rollback-receipt-", dir=path.parent)
-try:
-    with os.fdopen(fd, "w") as handle:
-        json.dump(data, handle, indent=2, sort_keys=True)
-        handle.write("\n")
-        handle.flush()
-        os.fsync(handle.fileno())
-    os.chmod(temp, 0o600)
-    os.chown(temp, 0, 0)
-    os.replace(temp, path)
-    directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
-    try:
-        os.fsync(directory)
-    finally:
-        os.close(directory)
-finally:
-    try:
-        os.unlink(temp)
-    except FileNotFoundError:
-        pass
-PY
-
-printf 'AIOS_ORCHESTRATOR_ROLLBACK=PASS current=%s previous=%s\n' "$PREVIOUS" "$CURRENT"
@@ -1,32 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-
-CONFIG="${1:-/etc/aios-orchestrator-gateway/config.json}"
-PORT="${PORT:-18920}"
-
-python3 -m json.tool "$CONFIG" >/dev/null
-systemd-analyze verify \
-  /etc/systemd/system/aios-orchestrator-gateway.service \
-  /etc/systemd/system/aios-orchestrator-gateway-poll.service \
-  /etc/systemd/system/aios-orchestrator-gateway-poll.timer
-systemctl is-active --quiet aios-orchestrator-gateway.service
-systemctl is-active --quiet aios-orchestrator-gateway-poll.timer
-
-health="$(curl --fail --silent --show-error "http://127.0.0.1:${PORT}/healthz")"
-python3 -c '
-import json, sys
-v=json.load(sys.stdin)
-assert v["service"]=="aios-orchestrator-gateway"
-assert v["status"]=="active"
-assert v["polling"] is True
-assert len(v["contract_commit"])==40
-' <<<"$health"
-
-systemctl start aios-orchestrator-gateway-poll.service
-if journalctl -u aios-orchestrator-gateway.service -n 200 --no-pager \
-  | grep -Eqi 'Bearer |token=|api[_-]?key|secret='; then
-  echo "AIOS_ORCHESTRATOR_VERIFY=BLOCKED reason=SECRET_PATTERN_IN_LOG" >&2
-  exit 2
-fi
-
-echo "AIOS_ORCHESTRATOR_VERIFY=PASS"
@@ -1 +0,0 @@
-"""AIOS services namespace."""
@@ -1,3 +0,0 @@
-"""Reliable central orchestration gateway for AIOS."""
-
-__version__ = "0.2.0"
@@ -1,34 +0,0 @@
-from __future__ import annotations
-
-import argparse
-import logging
-from pathlib import Path
-
-from .config import Config
-from .service import Runtime, serve
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser(description="AIOS Orchestrator Gateway")
-    parser.add_argument("--config", type=Path, required=True)
-    parser.add_argument("--log-level", default="INFO")
-    parser.add_argument(
-        "--poll-once",
-        action="store_true",
-        help="Run one inbox/model/outbox/OpenClaw synchronization polling cycle.",
-    )
-    args = parser.parse_args()
-    logging.basicConfig(
-        level=getattr(logging, args.log_level.upper(), logging.INFO),
-        format="%(asctime)s %(levelname)s %(name)s %(message)s",
-    )
-    if args.poll_once:
-        runtime = Runtime(Config.from_file(args.config))
-        runtime.poll_once()
-        return 0
-    serve(args.config)
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())
@@ -1,178 +0,0 @@
-from __future__ import annotations
-
-import json
-import urllib.error
-import urllib.request
-from typing import Any
-
-from .errors import TriggerError
-from .models import DECISION_SCHEMA, Decision, Event
-
-
-class WorkspaceAgentClient:
-    """Asynchronous Workspace Agent trigger.
-
-    The official endpoint returns only 202 and no run ID or retrievable output.
-    Completion therefore arrives through the gateway callback endpoint.
-    """
-
-    def __init__(self, trigger_id: str, token: str, timeout_seconds: int = 30):
-        self.trigger_id = trigger_id
-        self.token = token
-        self.timeout_seconds = timeout_seconds
-
-    def trigger(self, event: Event) -> None:
-        url = f"https://api.chatgpt.com/v1/workspace_agents/{self.trigger_id}/trigger"
-        prompt = (
-            "Process this AIOS orchestration event under the committed contract. "
-            "When complete, call submit_orchestration_decision with strict JSON. "
-            "Do not fabricate tests, Git state, secrets, evidence, or deployment facts.\n\n"
-            + event.canonical_json()
-        )
-        body = json.dumps(
-            {
-                "conversation_key": event.project_key,
-                "input": prompt,
-            }
-        ).encode()
-        request = urllib.request.Request(
-            url,
-            data=body,
-            method="POST",
-            headers={
-                "Authorization": f"Bearer {self.token}",
-                "Content-Type": "application/json",
-                "Idempotency-Key": event.event_id,
-                "User-Agent": "aios-orchestrator-gateway/0.2",
-            },
-        )
-        try:
-            with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
-                response.read(1024)
-                if response.status != 202:
-                    raise TriggerError(f"workspace_agent_http_{response.status}")
-        except urllib.error.HTTPError as exc:
-            raise TriggerError(f"workspace_agent_http_{exc.code}") from exc
-        except (urllib.error.URLError, TimeoutError, OSError) as exc:
-            raise TriggerError("workspace_agent_trigger_failed") from exc
-
-
-class ResponsesClient:
-    """Background Responses API client with polling and strict output validation."""
-
-    def __init__(self, api_key: str, model: str, timeout_seconds: int = 60):
-        self.api_key = api_key
-        self.model = model
-        self.timeout_seconds = timeout_seconds
-
-    def start(self, event: Event) -> str:
-        body = {
-            "model": self.model,
-            "background": True,
-            "instructions": (
-                "You are the AIOS central orchestration brain. Apply the exact "
-                "contract_commit in the event. Return only a strict decision. "
-                "Never fabricate evidence, secrets, Git facts, tests, approvals, "
-                "or deployment state. Keep external blockers fail-closed."
-            ),
-            "input": event.canonical_json(),
-            "text": {
-                "format": {
-                    "type": "json_schema",
-                    "name": "aios_orchestrator_decision",
-                    "strict": True,
-                    "schema": DECISION_SCHEMA,
-                }
-            },
-            "metadata": {
-                "event_id": event.event_id,
-                "project_key": event.project_key,
-                "contract_commit": event.contract_commit,
-            },
-        }
-        payload = self._request(
-            "POST",
-            "https://api.openai.com/v1/responses",
-            body,
-            idempotency_key=event.event_id,
-        )
-        response_id = payload.get("id")
-        if not isinstance(response_id, str) or not response_id.startswith("resp_"):
-            raise TriggerError("responses_missing_id")
-        return response_id
-
-    def poll(self, response_id: str) -> tuple[str, Decision | None]:
-        payload = self._request(
-            "GET",
-            f"https://api.openai.com/v1/responses/{response_id}",
-            None,
-        )
-        status = payload.get("status")
-        if status in {"queued", "in_progress"}:
-            return str(status), None
-        if status == "completed":
-            text = _extract_output_text(payload)
-            try:
-                return "completed", Decision.from_dict(json.loads(text))
-            except json.JSONDecodeError as exc:
-                raise TriggerError("responses_invalid_json") from exc
-        if status in {"failed", "cancelled", "incomplete"}:
-            raise TriggerError(f"responses_{status}")
-        raise TriggerError("responses_unknown_status")
-
-    def _request(
-        self,
-        method: str,
-        url: str,
-        body: dict[str, Any] | None,
-        *,
-        idempotency_key: str | None = None,
-    ) -> dict[str, Any]:
-        encoded = json.dumps(body).encode() if body is not None else None
-        headers = {
-            "Authorization": f"Bearer {self.api_key}",
-            "Content-Type": "application/json",
-            "User-Agent": "aios-orchestrator-gateway/0.2",
-        }
-        if idempotency_key:
-            headers["Idempotency-Key"] = idempotency_key
-        request = urllib.request.Request(
-            url,
-            data=encoded,
-            method=method,
-            headers=headers,
-        )
-        try:
-            with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
-                raw = response.read(4_194_304)
-                if response.status not in {200, 201}:
-                    raise TriggerError(f"responses_http_{response.status}")
-        except urllib.error.HTTPError as exc:
-            raise TriggerError(f"responses_http_{exc.code}") from exc
-        except (urllib.error.URLError, TimeoutError, OSError) as exc:
-            raise TriggerError("responses_request_failed") from exc
-        try:
-            payload = json.loads(raw)
-        except json.JSONDecodeError as exc:
-            raise TriggerError("responses_invalid_payload") from exc
-        if not isinstance(payload, dict):
-            raise TriggerError("responses_payload_not_object")
-        return payload
-
-
-def _extract_output_text(payload: dict[str, Any]) -> str:
-    output = payload.get("output")
-    if not isinstance(output, list):
-        raise TriggerError("responses_missing_output")
-    for item in output:
-        if not isinstance(item, dict):
-            continue
-        content = item.get("content")
-        if not isinstance(content, list):
-            continue
-        for part in content:
-            if isinstance(part, dict) and part.get("type") == "output_text":
-                text = part.get("text")
-                if isinstance(text, str):
-                    return text
-    raise TriggerError("responses_missing_output_text")
@@ -1,203 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-import os
-from pathlib import Path
-from typing import Any
-
-from .errors import ConfigurationError
-from .securefs import read_secure_json, read_secure_text
-
-
-@dataclass(frozen=True)
-class Config:
-    host: str
-    port: int
-    project_key: str
-    database_path: Path
-    event_inbox_dir: Path
-    decision_inbox_dir: Path
-    dispatch_outbox_dir: Path
-    openclaw_sync_dir: Path
-    command_queue_dir: Path
-    inbound_token_file: Path
-    callback_token_file: Path
-    contract_commit: str
-    workspace_trigger_id: str | None
-    workspace_token_file: Path | None
-    responses_api_key_file: Path | None
-    responses_model: str | None
-    poll_interval_seconds: int
-    summary_interval_seconds: int
-    callback_timeout_seconds: int
-    lease_seconds: int
-    retry_base_seconds: int
-    max_retries: int
-    max_same_blocker_retries: int
-    max_hops_per_gate: int
-    openclaw_sync_endpoint: str | None
-    openclaw_sync_token_file: Path | None
-    git_poll_enabled: bool
-    git_poll_bare_repo: Path | None
-    git_poll_ref_prefixes: tuple[str, ...]
-    git_poll_interval_seconds: int
-
-    @classmethod
-    def from_file(cls, path: Path) -> "Config":
-        owner_uid = int(os.environ.get("AIOS_ORCHESTRATOR_CONFIG_OWNER_UID", "0"))
-        raw = read_secure_json(path, expected_uid=owner_uid)
-        expected = {
-            "host", "port", "project_key", "database_path", "event_inbox_dir",
-            "decision_inbox_dir", "dispatch_outbox_dir", "openclaw_sync_dir",
-            "command_queue_dir", "inbound_token_file", "callback_token_file",
-            "contract_commit", "workspace_agent", "responses_fallback",
-            "polling", "loop_controls", "openclaw_sync", "git_poll",
-        }
-        if set(raw) != expected:
-            raise ConfigurationError("invalid_config_fields")
-
-        def path_value(key: str) -> Path:
-            value = raw.get(key)
-            if not isinstance(value, str):
-                raise ConfigurationError(f"invalid_{key}")
-            result = Path(value)
-            if not result.is_absolute():
-                raise ConfigurationError(f"path_not_absolute:{key}")
-            return result
-
-        def integer(container: dict[str, Any], key: str, lo: int, hi: int) -> int:
-            value = container.get(key)
-            if not isinstance(value, int) or not lo <= value <= hi:
-                raise ConfigurationError(f"invalid_{key}")
-            return value
-
-        host = raw.get("host")
-        port = raw.get("port")
-        project_key = raw.get("project_key")
-        contract = raw.get("contract_commit")
-        if not isinstance(host, str) or not host:
-            raise ConfigurationError("invalid_host")
-        if not isinstance(project_key, str) or not project_key:
-            raise ConfigurationError("invalid_project_key")
-        if not isinstance(port, int) or not 1 <= port <= 65535:
-            raise ConfigurationError("invalid_port")
-        if not isinstance(contract, str) or len(contract) != 40:
-            raise ConfigurationError("invalid_contract_commit")
-
-        workspace = raw.get("workspace_agent")
-        responses = raw.get("responses_fallback")
-        polling = raw.get("polling")
-        loop = raw.get("loop_controls")
-        sync = raw.get("openclaw_sync")
-        git_poll = raw.get("git_poll")
-        if not all(isinstance(x, dict) for x in (workspace, responses, polling, loop, sync, git_poll)):
-            raise ConfigurationError("invalid_nested_config")
-
-        trigger_id = workspace.get("trigger_id")
-        workspace_token = workspace.get("access_token_file")
-        if trigger_id is not None:
-            if not isinstance(trigger_id, str) or not trigger_id.startswith("agtch_"):
-                raise ConfigurationError("invalid_workspace_trigger_id")
-            if not isinstance(workspace_token, str) or not Path(workspace_token).is_absolute():
-                raise ConfigurationError("invalid_workspace_token_file")
-            workspace_token_file = Path(workspace_token)
-        else:
-            workspace_token_file = None
-
-        responses_key = responses.get("api_key_file")
-        responses_model = responses.get("model")
-        if responses_key is not None:
-            if not isinstance(responses_key, str) or not Path(responses_key).is_absolute():
-                raise ConfigurationError("invalid_responses_key_file")
-            if not isinstance(responses_model, str) or not responses_model:
-                raise ConfigurationError("invalid_responses_model")
-            responses_key_file = Path(responses_key)
-        else:
-            responses_key_file = None
-            responses_model = None
-
-        if trigger_id is None and responses_key_file is None:
-            raise ConfigurationError("no_central_model_channel")
-
-        endpoint = sync.get("endpoint")
-        token_file = sync.get("token_file")
-        if endpoint is not None and (not isinstance(endpoint, str) or not endpoint.startswith(("http://", "https://"))):
-            raise ConfigurationError("invalid_openclaw_sync_endpoint")
-        if endpoint is not None:
-            if not isinstance(token_file, str) or not Path(token_file).is_absolute():
-                raise ConfigurationError("invalid_openclaw_sync_token_file")
-            sync_token_file = Path(token_file)
-        else:
-            sync_token_file = None
-
-        git_enabled = git_poll.get("enabled")
-        if not isinstance(git_enabled, bool):
-            raise ConfigurationError("invalid_git_poll_enabled")
-        git_bare_repo_raw = git_poll.get("bare_repo")
-        git_prefixes_raw = git_poll.get("ref_prefixes")
-        if git_enabled:
-            if not isinstance(git_bare_repo_raw, str) or not Path(git_bare_repo_raw).is_absolute():
-                raise ConfigurationError("invalid_git_poll_bare_repo")
-            if (
-                not isinstance(git_prefixes_raw, list)
-                or not git_prefixes_raw
-                or any(not isinstance(item, str) or not item.startswith("refs/") for item in git_prefixes_raw)
-            ):
-                raise ConfigurationError("invalid_git_poll_ref_prefixes")
-            git_bare_repo = Path(git_bare_repo_raw)
-            git_prefixes = tuple(git_prefixes_raw)
-        else:
-            git_bare_repo = None
-            git_prefixes = ()
-
-        return cls(
-            host=host,
-            port=port,
-            project_key=project_key,
-            database_path=path_value("database_path"),
-            event_inbox_dir=path_value("event_inbox_dir"),
-            decision_inbox_dir=path_value("decision_inbox_dir"),
-            dispatch_outbox_dir=path_value("dispatch_outbox_dir"),
-            openclaw_sync_dir=path_value("openclaw_sync_dir"),
-            command_queue_dir=path_value("command_queue_dir"),
-            inbound_token_file=path_value("inbound_token_file"),
-            callback_token_file=path_value("callback_token_file"),
-            contract_commit=contract,
-            workspace_trigger_id=trigger_id,
-            workspace_token_file=workspace_token_file,
-            responses_api_key_file=responses_key_file,
-            responses_model=responses_model,
-            poll_interval_seconds=integer(polling, "interval_seconds", 1, 300),
-            summary_interval_seconds=integer(polling, "summary_interval_seconds", 10, 86400),
-            callback_timeout_seconds=integer(loop, "callback_timeout_seconds", 30, 86400),
-            lease_seconds=integer(loop, "lease_seconds", 10, 3600),
-            retry_base_seconds=integer(loop, "retry_base_seconds", 1, 3600),
-            max_retries=integer(loop, "max_retries", 0, 20),
-            max_same_blocker_retries=integer(loop, "max_same_blocker_retries", 0, 20),
-            max_hops_per_gate=integer(loop, "max_hops_per_gate", 1, 100),
-            openclaw_sync_endpoint=endpoint,
-            openclaw_sync_token_file=sync_token_file,
-            git_poll_enabled=git_enabled,
-            git_poll_bare_repo=git_bare_repo,
-            git_poll_ref_prefixes=git_prefixes,
-            git_poll_interval_seconds=integer(git_poll, "interval_seconds", 5, 3600),
-        )
-
-    def _secret(self, path: Path) -> str:
-        owner_uid = int(os.environ.get("AIOS_ORCHESTRATOR_SECRET_OWNER_UID", "0"))
-        return read_secure_text(path, expected_uid=owner_uid, max_bytes=64_000)
-
-    def inbound_token(self) -> str:
-        return self._secret(self.inbound_token_file)
-
-    def callback_token(self) -> str:
-        return self._secret(self.callback_token_file)
-
-    def workspace_token(self) -> str | None:
-        return self._secret(self.workspace_token_file) if self.workspace_token_file else None
-
-    def responses_api_key(self) -> str | None:
-        return self._secret(self.responses_api_key_file) if self.responses_api_key_file else None
-
-    def openclaw_sync_token(self) -> str | None:
-        return self._secret(self.openclaw_sync_token_file) if self.openclaw_sync_token_file else None
@@ -1,89 +0,0 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-import re
-import os
-import stat
-from typing import Any
-
-from .errors import DispatchError
-from .securefs import (
-    atomic_write_bytes,
-    atomic_write_bytes_shared,
-    ensure_secure_directory,
-    ensure_shared_directory,
-    read_secure_bytes,
-)
-
-
-SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
-
-
-class DispatchSink:
-    """Reliable file-based dispatch adapter for existing OpenClaw/Codex watchers."""
-
-    def __init__(self, outbox_dir: Path, command_queue_dir: Path, contract_commit: str):
-        self.outbox_dir = Path(outbox_dir)
-        self.command_queue_dir = Path(command_queue_dir)
-        self.contract_commit = contract_commit
-        ensure_secure_directory(self.outbox_dir)
-        ensure_shared_directory(self.command_queue_dir)
-
-    def dispatch(self, record: dict[str, Any]) -> tuple[Path, Path]:
-        work_unit_id = record.get("work_unit_id")
-        event_id = record.get("event_id")
-        agent = record.get("assigned_agent")
-        payload_json = record.get("payload_json")
-        if (
-            not isinstance(work_unit_id, str)
-            or not SAFE_ID.fullmatch(work_unit_id)
-            or not isinstance(event_id, str)
-            or not SAFE_ID.fullmatch(event_id)
-            or agent not in {"openclaw", "codex"}
-            or not isinstance(payload_json, str)
-        ):
-            raise DispatchError("invalid_dispatch_record")
-        try:
-            work_unit = json.loads(payload_json)
-        except json.JSONDecodeError as exc:
-            raise DispatchError("invalid_dispatch_payload") from exc
-
-        envelope = {
-            "schema_version": 1,
-            "contract_commit": self.contract_commit,
-            "event_id": event_id,
-            "assigned_agent": agent,
-            "work_unit": work_unit,
-            "callback": {
-                "event_endpoint": "/v1/orchestrator/events",
-                "start_event_type": "work_unit_started",
-                "end_event_type": "work_unit_completed",
-            },
-        }
-        outbox_path = self.outbox_dir / "dispatched" / f"{work_unit_id}.json"
-        queue_path = self.command_queue_dir / agent / "pending" / f"{work_unit_id}.json"
-        ensure_secure_directory(outbox_path.parent)
-        ensure_shared_directory(queue_path.parent)
-
-        # Idempotent immutable payload: existing identical file is accepted.
-        self._write_idempotent_json(outbox_path, envelope, shared=False)
-        self._write_idempotent_json(queue_path, envelope, shared=True)
-        return outbox_path, queue_path
-
-    @staticmethod
-    def _write_idempotent_json(path: Path, value: dict[str, Any], *, shared: bool) -> None:
-        payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode()
-        try:
-            st = os.lstat(path)
-        except FileNotFoundError:
-            if shared:
-                atomic_write_bytes_shared(path, payload, mode=0o640)
-            else:
-                atomic_write_bytes(path, payload, mode=0o600)
-            return
-        if not stat.S_ISREG(st.st_mode):
-            raise DispatchError(f"dispatch_target_not_regular:{path.name}")
-        existing = read_secure_bytes(path, expected_uid=None, max_bytes=1_048_576)
-        if existing != payload:
-            raise DispatchError(f"dispatch_idempotency_conflict:{path.name}")
@@ -1,26 +0,0 @@
-class OrchestratorError(Exception):
-    """Base orchestration error."""
-
-
-class ConfigurationError(OrchestratorError):
-    """Unsafe or invalid configuration."""
-
-
-class ValidationError(OrchestratorError):
-    """Invalid event or decision."""
-
-
-class AuthenticationError(OrchestratorError):
-    """Authentication failed."""
-
-
-class ConflictError(OrchestratorError):
-    """Idempotency, lease, or optimistic concurrency conflict."""
-
-
-class TriggerError(OrchestratorError):
-    """Central model trigger failed."""
-
-
-class DispatchError(OrchestratorError):
-    """Dispatch or synchronization output failed."""
@@ -1,346 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-import hashlib
-import json
-import re
-from typing import Any
-
-from .errors import ValidationError
-
-
-OID_RE = re.compile(r"^[0-9a-f]{40}$")
-ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
-AGENTS = {"openclaw", "codex", "gitea", "system", "workspace-agent"}
-EVENT_STATUSES = {"READY", "RUNNING", "PASS", "BLOCKED", "FAIL", "TIMEOUT", "NOT_EXECUTED"}
-VERDICTS = {"DISPATCH", "BLOCKED", "COMPLETE", "NOOP"}
-EXECUTORS = {"openclaw", "codex"}
-
-
-def req_str(value: dict[str, Any], key: str, pattern: re.Pattern[str] | None = None) -> str:
-    item = value.get(key)
-    if not isinstance(item, str) or not item:
-        raise ValidationError(f"invalid_{key}")
-    if pattern is not None and not pattern.fullmatch(item):
-        raise ValidationError(f"invalid_{key}")
-    return item
-
-
-def opt_oid(item: Any, key: str) -> str | None:
-    if item is None:
-        return None
-    if not isinstance(item, str) or not OID_RE.fullmatch(item):
-        raise ValidationError(f"invalid_{key}")
-    return item
-
-
-def string_list(value: dict[str, Any], key: str, limit: int) -> tuple[str, ...]:
-    raw = value.get(key)
-    if not isinstance(raw, list) or len(raw) > limit:
-        raise ValidationError(f"invalid_{key}")
-    result: list[str] = []
-    for item in raw:
-        if not isinstance(item, str) or not item or len(item) > 1024:
-            raise ValidationError(f"invalid_{key}")
-        result.append(item)
-    return tuple(result)
-
-
-@dataclass(frozen=True)
-class Event:
-    schema_version: int
-    project_key: str
-    contract_commit: str
-    event_id: str
-    source_agent: str
-    event_type: str
-    work_unit_id: str
-    status: str
-    state_version: int
-    implementation_commit: str | None = None
-    result_commit: str | None = None
-    blocker: str | None = None
-    evidence_refs: tuple[str, ...] = ()
-    payload: dict[str, Any] = field(default_factory=dict)
-
-    @classmethod
-    def from_dict(cls, value: dict[str, Any]) -> "Event":
-        allowed = {
-            "schema_version", "project_key", "contract_commit", "event_id",
-            "source_agent", "event_type", "work_unit_id", "status",
-            "state_version", "implementation_commit", "result_commit",
-            "blocker", "evidence_refs", "payload",
-        }
-        if not isinstance(value, dict) or set(value) - allowed:
-            raise ValidationError("invalid_event_fields")
-        if value.get("schema_version") != 1:
-            raise ValidationError("unsupported_event_schema")
-        source = req_str(value, "source_agent")
-        status = req_str(value, "status")
-        if source not in AGENTS:
-            raise ValidationError("invalid_source_agent")
-        if status not in EVENT_STATUSES:
-            raise ValidationError("invalid_event_status")
-        version = value.get("state_version")
-        if not isinstance(version, int) or version < 0:
-            raise ValidationError("invalid_state_version")
-        blocker = value.get("blocker")
-        if blocker is not None and (not isinstance(blocker, str) or len(blocker) > 512):
-            raise ValidationError("invalid_blocker")
-        payload = value.get("payload", {})
-        if not isinstance(payload, dict):
-            raise ValidationError("invalid_payload")
-        if len(json.dumps(payload, sort_keys=True).encode()) > 262_144:
-            raise ValidationError("payload_too_large")
-        return cls(
-            schema_version=1,
-            project_key=req_str(value, "project_key", ID_RE),
-            contract_commit=req_str(value, "contract_commit", OID_RE),
-            event_id=req_str(value, "event_id", ID_RE),
-            source_agent=source,
-            event_type=req_str(value, "event_type", ID_RE),
-            work_unit_id=req_str(value, "work_unit_id", ID_RE),
-            status=status,
-            state_version=version,
-            implementation_commit=opt_oid(value.get("implementation_commit"), "implementation_commit"),
-            result_commit=opt_oid(value.get("result_commit"), "result_commit"),
-            blocker=blocker,
-            evidence_refs=string_list(value, "evidence_refs", 64),
-            payload=payload,
-        )
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "schema_version": 1,
-            "project_key": self.project_key,
-            "contract_commit": self.contract_commit,
-            "event_id": self.event_id,
-            "source_agent": self.source_agent,
-            "event_type": self.event_type,
-            "work_unit_id": self.work_unit_id,
-            "status": self.status,
-            "state_version": self.state_version,
-            "implementation_commit": self.implementation_commit,
-            "result_commit": self.result_commit,
-            "blocker": self.blocker,
-            "evidence_refs": list(self.evidence_refs),
-            "payload": self.payload,
-        }
-
-    def canonical_json(self) -> str:
-        return json.dumps(self.to_dict(), separators=(",", ":"), sort_keys=True)
-
-    def digest(self) -> str:
-        return hashlib.sha256(self.canonical_json().encode()).hexdigest()
-
-
-@dataclass(frozen=True)
-class WorkUnit:
-    id: str
-    track_id: str
-    gate_id: str
-    primary_agent: str
-    fallback_agent: str
-    clean_base_commit: str
-    allowed_paths: tuple[str, ...]
-    required_checks: tuple[str, ...]
-    forbidden_operations: tuple[str, ...]
-    instruction: str
-
-    @classmethod
-    def from_dict(cls, value: dict[str, Any]) -> "WorkUnit":
-        expected = {
-            "id", "track_id", "gate_id", "primary_agent", "fallback_agent",
-            "clean_base_commit", "allowed_paths", "required_checks",
-            "forbidden_operations", "instruction",
-        }
-        if not isinstance(value, dict) or set(value) != expected:
-            raise ValidationError("invalid_work_unit_fields")
-        primary = req_str(value, "primary_agent")
-        fallback = req_str(value, "fallback_agent")
-        if primary not in EXECUTORS or fallback not in EXECUTORS or primary == fallback:
-            raise ValidationError("invalid_agent_assignment")
-        instruction = req_str(value, "instruction")
-        if len(instruction.encode()) > 65_536:
-            raise ValidationError("instruction_too_large")
-        return cls(
-            id=req_str(value, "id", ID_RE),
-            track_id=req_str(value, "track_id", ID_RE),
-            gate_id=req_str(value, "gate_id", ID_RE),
-            primary_agent=primary,
-            fallback_agent=fallback,
-            clean_base_commit=req_str(value, "clean_base_commit", OID_RE),
-            allowed_paths=string_list(value, "allowed_paths", 128),
-            required_checks=string_list(value, "required_checks", 128),
-            forbidden_operations=string_list(value, "forbidden_operations", 64),
-            instruction=instruction,
-        )
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "id": self.id,
-            "track_id": self.track_id,
-            "gate_id": self.gate_id,
-            "primary_agent": self.primary_agent,
-            "fallback_agent": self.fallback_agent,
-            "clean_base_commit": self.clean_base_commit,
-            "allowed_paths": list(self.allowed_paths),
-            "required_checks": list(self.required_checks),
-            "forbidden_operations": list(self.forbidden_operations),
-            "instruction": self.instruction,
-        }
-
-
-@dataclass(frozen=True)
-class Decision:
-    schema_version: int
-    project_key: str
-    contract_commit: str
-    event_id: str
-    expected_state_version: int
-    verdict: str
-    state_updates: dict[str, str]
-    next_work_units: tuple[WorkUnit, ...]
-    requires_human_input: bool
-    stop_reason: str | None
-
-    @classmethod
-    def from_dict(cls, value: dict[str, Any]) -> "Decision":
-        expected = {
-            "schema_version", "project_key", "contract_commit", "event_id",
-            "expected_state_version", "verdict", "state_updates",
-            "next_work_units", "requires_human_input", "stop_reason",
-        }
-        if not isinstance(value, dict) or set(value) != expected:
-            raise ValidationError("invalid_decision_fields")
-        if value.get("schema_version") != 1:
-            raise ValidationError("unsupported_decision_schema")
-        version = value.get("expected_state_version")
-        if not isinstance(version, int) or version < 0:
-            raise ValidationError("invalid_expected_state_version")
-        verdict = req_str(value, "verdict")
-        if verdict not in VERDICTS:
-            raise ValidationError("invalid_verdict")
-        updates = value.get("state_updates")
-        normalized: dict[str, str] = {}
-        if isinstance(updates, dict):
-            update_items = [{"key": key, "value": item} for key, item in updates.items()]
-        elif isinstance(updates, list):
-            update_items = updates
-        else:
-            raise ValidationError("invalid_state_updates")
-        if len(update_items) > 128:
-            raise ValidationError("invalid_state_updates")
-        for update in update_items:
-            if not isinstance(update, dict) or set(update) != {"key", "value"}:
-                raise ValidationError("invalid_state_update")
-            key = update.get("key")
-            item = update.get("value")
-            if not isinstance(key, str) or not ID_RE.fullmatch(key):
-                raise ValidationError("invalid_state_update_key")
-            if not isinstance(item, str) or not item or len(item) > 256:
-                raise ValidationError("invalid_state_update_value")
-            if key in normalized:
-                raise ValidationError("duplicate_state_update_key")
-            normalized[key] = item
-        raw_units = value.get("next_work_units")
-        if not isinstance(raw_units, list) or len(raw_units) > 32:
-            raise ValidationError("invalid_next_work_units")
-        units = tuple(WorkUnit.from_dict(item) for item in raw_units)
-        human = value.get("requires_human_input")
-        if not isinstance(human, bool):
-            raise ValidationError("invalid_requires_human_input")
-        stop = value.get("stop_reason")
-        if stop is not None and (not isinstance(stop, str) or len(stop) > 512):
-            raise ValidationError("invalid_stop_reason")
-        if verdict == "DISPATCH" and not units:
-            raise ValidationError("dispatch_without_work_units")
-        if verdict != "DISPATCH" and units:
-            raise ValidationError("work_units_without_dispatch")
-        return cls(
-            schema_version=1,
-            project_key=req_str(value, "project_key", ID_RE),
-            contract_commit=req_str(value, "contract_commit", OID_RE),
-            event_id=req_str(value, "event_id", ID_RE),
-            expected_state_version=version,
-            verdict=verdict,
-            state_updates=normalized,
-            next_work_units=units,
-            requires_human_input=human,
-            stop_reason=stop,
-        )
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "schema_version": 1,
-            "project_key": self.project_key,
-            "contract_commit": self.contract_commit,
-            "event_id": self.event_id,
-            "expected_state_version": self.expected_state_version,
-            "verdict": self.verdict,
-            "state_updates": [
-                {"key": key, "value": value}
-                for key, value in sorted(self.state_updates.items())
-            ],
-            "next_work_units": [u.to_dict() for u in self.next_work_units],
-            "requires_human_input": self.requires_human_input,
-            "stop_reason": self.stop_reason,
-        }
-
-
-DECISION_SCHEMA: dict[str, Any] = {
-    "type": "object",
-    "additionalProperties": False,
-    "required": [
-        "schema_version", "project_key", "contract_commit", "event_id",
-        "expected_state_version", "verdict", "state_updates",
-        "next_work_units", "requires_human_input", "stop_reason",
-    ],
-    "properties": {
-        "schema_version": {"type": "integer", "const": 1},
-        "project_key": {"type": "string"},
-        "contract_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
-        "event_id": {"type": "string"},
-        "expected_state_version": {"type": "integer", "minimum": 0},
-        "verdict": {"type": "string", "enum": sorted(VERDICTS)},
-        "state_updates": {
-            "type": "array",
-            "items": {
-                "type": "object",
-                "additionalProperties": False,
-                "required": ["key", "value"],
-                "properties": {
-                    "key": {"type": "string"},
-                    "value": {"type": "string"}
-                }
-            }
-        },
-        "next_work_units": {
-            "type": "array",
-            "maxItems": 32,
-            "items": {
-                "type": "object",
-                "additionalProperties": False,
-                "required": [
-                    "id", "track_id", "gate_id", "primary_agent",
-                    "fallback_agent", "clean_base_commit", "allowed_paths",
-                    "required_checks", "forbidden_operations", "instruction",
-                ],
-                "properties": {
-                    "id": {"type": "string"},
-                    "track_id": {"type": "string"},
-                    "gate_id": {"type": "string"},
-                    "primary_agent": {"type": "string", "enum": ["openclaw", "codex"]},
-                    "fallback_agent": {"type": "string", "enum": ["openclaw", "codex"]},
-                    "clean_base_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
-                    "allowed_paths": {"type": "array", "items": {"type": "string"}},
-                    "required_checks": {"type": "array", "items": {"type": "string"}},
-                    "forbidden_operations": {"type": "array", "items": {"type": "string"}},
-                    "instruction": {"type": "string"},
-                },
-            },
-        },
-        "requires_human_input": {"type": "boolean"},
-        "stop_reason": {"type": ["string", "null"]},
-    },
-}
@@ -1,229 +0,0 @@
-from __future__ import annotations
-
-import json
-import os
-from pathlib import Path
-import stat
-from typing import Any
-
-from .errors import ConfigurationError
-
-
-_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
-
-
-def reject_symlink_chain(path: Path) -> None:
-    """Reject symlinks in every existing component of an absolute path."""
-    path = Path(path)
-    if not path.is_absolute():
-        raise ConfigurationError(f"path_not_absolute:{path}")
-    current = Path(path.anchor)
-    for part in path.parts[1:]:
-        current = current / part
-        try:
-            st = os.lstat(current)
-        except FileNotFoundError:
-            continue
-        if stat.S_ISLNK(st.st_mode):
-            raise ConfigurationError(f"symlink_rejected:{current}")
-
-
-def verify_secure_file(
-    path: Path,
-    *,
-    expected_uid: int | None,
-    allowed_write_bits: int = stat.S_IWUSR,
-    require_executable: bool = False,
-) -> os.stat_result:
-    path = Path(path)
-    reject_symlink_chain(path)
-    before = os.lstat(path)
-    if not stat.S_ISREG(before.st_mode):
-        raise ConfigurationError(f"not_regular_file:{path}")
-    if expected_uid is not None and before.st_uid != expected_uid:
-        raise ConfigurationError(f"unexpected_owner:{path}")
-    write_bits = before.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
-    if write_bits & ~allowed_write_bits:
-        raise ConfigurationError(f"permissive_mode:{path}")
-    if require_executable and not (before.st_mode & stat.S_IXUSR):
-        raise ConfigurationError(f"not_executable:{path}")
-    fd = os.open(path, os.O_RDONLY | _NOFOLLOW)
-    try:
-        after = os.fstat(fd)
-        if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
-            raise ConfigurationError(f"inode_changed:{path}")
-    finally:
-        os.close(fd)
-    return before
-
-
-def read_secure_bytes(
-    path: Path,
-    *,
-    expected_uid: int | None,
-    max_bytes: int = 1_048_576,
-) -> bytes:
-    path = Path(path)
-    before = verify_secure_file(path, expected_uid=expected_uid)
-    if before.st_size > max_bytes:
-        raise ConfigurationError(f"file_too_large:{path}")
-    fd = os.open(path, os.O_RDONLY | _NOFOLLOW)
-    try:
-        data = bytearray()
-        while True:
-            chunk = os.read(fd, min(65536, max_bytes + 1 - len(data)))
-            if not chunk:
-                break
-            data.extend(chunk)
-            if len(data) > max_bytes:
-                raise ConfigurationError(f"file_too_large:{path}")
-        return bytes(data)
-    finally:
-        os.close(fd)
-
-
-def read_secure_text(path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576) -> str:
-    try:
-        return read_secure_bytes(path, expected_uid=expected_uid, max_bytes=max_bytes).decode("utf-8").strip()
-    except UnicodeDecodeError as exc:
-        raise ConfigurationError(f"not_utf8:{path}") from exc
-
-
-def read_secure_json(path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576) -> dict[str, Any]:
-    try:
-        value = json.loads(read_secure_text(path, expected_uid=expected_uid, max_bytes=max_bytes))
-    except json.JSONDecodeError as exc:
-        raise ConfigurationError(f"invalid_json:{path}") from exc
-    if not isinstance(value, dict):
-        raise ConfigurationError(f"json_root_not_object:{path}")
-    return value
-
-
-def ensure_secure_directory(path: Path, *, mode: int = 0o700) -> None:
-    path = Path(path)
-    if not path.is_absolute():
-        raise ConfigurationError(f"path_not_absolute:{path}")
-    reject_symlink_chain(path.parent)
-    path.mkdir(parents=True, exist_ok=True, mode=mode)
-    reject_symlink_chain(path)
-    st = os.lstat(path)
-    if not stat.S_ISDIR(st.st_mode):
-        raise ConfigurationError(f"not_directory:{path}")
-    if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
-        raise ConfigurationError(f"permissive_directory:{path}")
-
-
-def atomic_write_bytes(path: Path, payload: bytes, *, mode: int = 0o600) -> None:
-    """Write through a verified directory fd and atomically replace the target."""
-    path = Path(path)
-    ensure_secure_directory(path.parent)
-    parent_before = os.lstat(path.parent)
-    parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | _NOFOLLOW)
-    temp_name = f".{path.name}.tmp-{os.getpid()}-{os.urandom(6).hex()}"
-    try:
-        parent_after = os.fstat(parent_fd)
-        if (parent_before.st_dev, parent_before.st_ino) != (
-            parent_after.st_dev,
-            parent_after.st_ino,
-        ):
-            raise ConfigurationError(f"directory_changed:{path.parent}")
-        fd = os.open(
-            temp_name,
-            os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW,
-            mode,
-            dir_fd=parent_fd,
-        )
-        try:
-            view = memoryview(payload)
-            while view:
-                written = os.write(fd, view)
-                view = view[written:]
-            os.fsync(fd)
-        finally:
-            os.close(fd)
-        os.replace(
-            temp_name,
-            path.name,
-            src_dir_fd=parent_fd,
-            dst_dir_fd=parent_fd,
-        )
-        os.fsync(parent_fd)
-    finally:
-        try:
-            os.unlink(temp_name, dir_fd=parent_fd)
-        except FileNotFoundError:
-            pass
-        os.close(parent_fd)
-
-
-def atomic_write_json(path: Path, value: Any, *, mode: int = 0o600) -> None:
-    payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
-    atomic_write_bytes(path, payload, mode=mode)
-
-def ensure_shared_directory(path: Path, *, group_writable: bool = False) -> None:
-    """Verify/create a setgid queue directory shared with a consumer group."""
-    path = Path(path)
-    if not path.is_absolute():
-        raise ConfigurationError(f"path_not_absolute:{path}")
-    reject_symlink_chain(path.parent)
-    mode = 0o2770 if group_writable else 0o2750
-    try:
-        path.mkdir(parents=True, mode=mode)
-    except FileExistsError:
-        pass
-    else:
-        os.chmod(path, mode)
-    reject_symlink_chain(path)
-    st = os.lstat(path)
-    if not stat.S_ISDIR(st.st_mode):
-        raise ConfigurationError(f"not_directory:{path}")
-    observed = stat.S_IMODE(st.st_mode)
-    required = stat.S_ISGID | stat.S_IRGRP | stat.S_IXGRP
-    if group_writable:
-        required |= stat.S_IWGRP
-    if observed & 0o007 or observed & required != required:
-        raise ConfigurationError(f"invalid_shared_directory_mode:{path}")
-
-def atomic_write_bytes_shared(
-    path: Path,
-    payload: bytes,
-    *,
-    mode: int = 0o640,
-    group_writable_directory: bool = False,
-) -> None:
-    """Atomic write into a setgid/group-readable shared queue directory."""
-    path = Path(path)
-    ensure_shared_directory(path.parent, group_writable=group_writable_directory)
-    parent_before = os.lstat(path.parent)
-    parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | _NOFOLLOW)
-    temp_name = f".{path.name}.tmp-{os.getpid()}-{os.urandom(6).hex()}"
-    try:
-        parent_after = os.fstat(parent_fd)
-        if (parent_before.st_dev, parent_before.st_ino) != (
-            parent_after.st_dev,
-            parent_after.st_ino,
-        ):
-            raise ConfigurationError(f"directory_changed:{path.parent}")
-        fd = os.open(
-            temp_name,
-            os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW,
-            mode,
-            dir_fd=parent_fd,
-        )
-        os.fchmod(fd, mode)
-        try:
-            view = memoryview(payload)
-            while view:
-                written = os.write(fd, view)
-                view = view[written:]
-            os.fsync(fd)
-        finally:
-            os.close(fd)
-        os.replace(temp_name, path.name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
-        os.fsync(parent_fd)
-    finally:
-        try:
-            os.unlink(temp_name, dir_fd=parent_fd)
-        except FileNotFoundError:
-            pass
-        os.close(parent_fd)
@@ -1,642 +0,0 @@
-from __future__ import annotations
-
-import hashlib
-import hmac
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-import json
-import logging
-import os
-from pathlib import Path
-import socket
-import subprocess
-import threading
-import time
-from typing import Any
-from urllib.parse import urlparse
-
-from .clients import ResponsesClient, WorkspaceAgentClient
-from .config import Config
-from .dispatch import DispatchSink
-from .errors import (
-    AuthenticationError,
-    ConflictError,
-    DispatchError,
-    OrchestratorError,
-    TriggerError,
-    ValidationError,
-)
-from .models import Decision, Event
-from .securefs import ensure_secure_directory, ensure_shared_directory, read_secure_json
-from .store import Store
-from .sync import OpenClawSynchronizer, lifecycle_record
-
-
-LOG = logging.getLogger("aios-orchestrator-gateway")
-
-
-class Runtime:
-    def __init__(self, config: Config):
-        self.config = config
-        self.store = Store(config.database_path)
-        self.dispatch = DispatchSink(
-            config.dispatch_outbox_dir,
-            config.command_queue_dir,
-            config.contract_commit,
-        )
-        self.workspace = None
-        workspace_token = config.workspace_token()
-        if config.workspace_trigger_id and workspace_token:
-            self.workspace = WorkspaceAgentClient(
-                config.workspace_trigger_id,
-                workspace_token,
-            )
-        self.responses = None
-        responses_key = config.responses_api_key()
-        if responses_key and config.responses_model:
-            self.responses = ResponsesClient(
-                responses_key,
-                config.responses_model,
-            )
-        self.synchronizer = OpenClawSynchronizer(
-            config.openclaw_sync_dir,
-            endpoint=config.openclaw_sync_endpoint,
-            bearer_token=config.openclaw_sync_token(),
-        )
-        self.worker_id = f"{socket.gethostname()}-{os.getpid()}-{threading.get_native_id()}"
-        self.stop_event = threading.Event()
-        self.worker = threading.Thread(target=self._loop, daemon=True)
-        self._next_git_poll = 0.0
-        for directory in (
-            config.event_inbox_dir,
-            config.decision_inbox_dir,
-            config.dispatch_outbox_dir,
-            config.openclaw_sync_dir,
-        ):
-            ensure_secure_directory(directory)
-        ensure_shared_directory(config.command_queue_dir)
-
-    def start(self) -> None:
-        self.worker.start()
-
-    def stop(self) -> None:
-        self.stop_event.set()
-        if self.worker.is_alive():
-            self.worker.join(timeout=15)
-        self.store.close()
-
-    def submit_event(self, payload: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
-        event = Event.from_dict(payload)
-        if event.contract_commit != self.config.contract_commit:
-            raise ConflictError("contract_version_mismatch")
-        created, stored = self.store.register_event(event)
-        if created:
-            self._enqueue_event_lifecycle(event)
-        return created, stored
-
-    def submit_decision(self, payload: dict[str, Any]) -> int:
-        decision = Decision.from_dict(payload)
-        if decision.contract_commit != self.config.contract_commit:
-            raise ConflictError("contract_version_mismatch")
-        version = self.store.apply_decision(
-            decision,
-            max_hops_per_gate=self.config.max_hops_per_gate,
-            max_same_blocker_retries=self.config.max_same_blocker_retries,
-        )
-        self._enqueue_decision_lifecycle(decision, version)
-        return version
-
-    def _loop(self) -> None:
-        next_summary = 0.0
-        while not self.stop_event.wait(self.config.poll_interval_seconds):
-            try:
-                self.poll_once()
-                if time.monotonic() >= next_summary:
-                    self.enqueue_periodic_summaries()
-                    next_summary = time.monotonic() + self.config.summary_interval_seconds
-            except Exception:
-                LOG.exception("poll_cycle_failed")
-
-    def poll_once(self) -> None:
-        """One deterministic reliability cycle.
-
-        This is deliberately callable from a systemd timer or health supervisor
-        in addition to the internal loop.
-        """
-        self.ingest_file_inboxes()
-        self.poll_git_refs()
-        self.process_new_events()
-        self.process_workspace_timeouts()
-        self.poll_background_responses()
-        self.dispatch_work_units()
-        self.synchronize_openclaw()
-
-    def ingest_file_inboxes(self) -> None:
-        self._ingest_directory(self.config.event_inbox_dir, kind="event")
-        self._ingest_directory(self.config.decision_inbox_dir, kind="decision")
-
-    def _ingest_directory(self, directory: Path, *, kind: str) -> None:
-        pending = directory / "pending"
-        processed = directory / "processed"
-        rejected = directory / "rejected"
-        ensure_shared_directory(pending, group_writable=True)
-        ensure_secure_directory(processed)
-        ensure_secure_directory(rejected)
-        for path in sorted(pending.glob("*.json"))[:128]:
-            try:
-                payload = read_secure_json(path, expected_uid=None)
-                if kind == "event":
-                    self.submit_event(payload)
-                else:
-                    self.submit_decision(payload)
-                target = processed / path.name
-            except Exception as exc:
-                LOG.warning("file_inbox_rejected path=%s error=%s", path, type(exc).__name__)
-                target = rejected / path.name
-            if target.exists():
-                # Same file name is an idempotency replay. Keep original audit copy.
-                path.unlink()
-            else:
-                path.replace(target)
-
-    def poll_git_refs(self) -> None:
-        """Poll authoritative local bare refs so missed webhooks cannot stall the loop."""
-        if not self.config.git_poll_enabled or self.config.git_poll_bare_repo is None:
-            return
-        now = time.monotonic()
-        if now < self._next_git_poll:
-            return
-        self._next_git_poll = now + self.config.git_poll_interval_seconds
-        command = [
-            "git",
-            f"--git-dir={self.config.git_poll_bare_repo}",
-            "for-each-ref",
-            "--format=%(refname) %(objectname)",
-            *self.config.git_poll_ref_prefixes,
-        ]
-        try:
-            result = subprocess.run(
-                command,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                text=True,
-                check=False,
-                timeout=30,
-                env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
-            )
-        except (OSError, subprocess.SubprocessError):
-            LOG.exception("git_ref_poll_failed")
-            return
-        if result.returncode != 0:
-            LOG.warning("git_ref_poll_rc=%s", result.returncode)
-            return
-        refs: dict[str, str] = {}
-        for line in result.stdout.splitlines():
-            refname, sep, oid = line.partition(" ")
-            if not sep or not refname.startswith("refs/") or len(oid) != 40:
-                continue
-            if any(ch not in "0123456789abcdef" for ch in oid):
-                continue
-            refs[refname] = oid
-        initialized, changes = self.store.observe_refs(refs)
-        if not initialized or not changes:
-            return
-        payload_changes = [
-            {"ref": refname, "old_oid": old_oid, "new_oid": new_oid}
-            for refname, old_oid, new_oid in sorted(changes)
-        ]
-        canonical = json.dumps(payload_changes, separators=(",", ":"), sort_keys=True)
-        event_id = "gitpoll-" + hashlib.sha256(canonical.encode()).hexdigest()[:40]
-        state = self.store.project_state(self.config.project_key)
-        state_version = 0 if state is None else int(state["state_version"])
-        event = Event.from_dict(
-            {
-                "schema_version": 1,
-                "project_key": self.config.project_key,
-                "contract_commit": self.config.contract_commit,
-                "event_id": event_id,
-                "source_agent": "gitea",
-                "event_type": "git_refs_changed",
-                "work_unit_id": "GIT-REF-POLL",
-                "status": "READY",
-                "state_version": state_version,
-                "implementation_commit": None,
-                "result_commit": None,
-                "blocker": None,
-                "evidence_refs": [],
-                "payload": {"changes": payload_changes},
-            }
-        )
-        try:
-            self.submit_event(event.to_dict())
-        except ConflictError:
-            LOG.info("git_ref_event_already_registered event_id=%s", event_id)
-
-    def process_new_events(self) -> None:
-        rows = self.store.claim_events(
-            owner=self.worker_id,
-            statuses=("RECEIVED", "TRIGGER_FAILED"),
-            limit=32,
-            lease_seconds=self.config.lease_seconds,
-        )
-        for row in rows:
-            event = Event.from_dict(json.loads(row["event_json"]))
-            attempts = int(row["attempts"])
-            try:
-                if self.workspace is not None:
-                    self.workspace.trigger(event)
-                    self.store.update_event_trigger(
-                        event.event_id,
-                        status="AWAITING_DECISION",
-                        channel="workspace_agent",
-                        remote_id=None,
-                    )
-                elif self.responses is not None:
-                    response_id = self.responses.start(event)
-                    self.store.update_event_trigger(
-                        event.event_id,
-                        status="POLLING_RESPONSE",
-                        channel="responses_api",
-                        remote_id=response_id,
-                        retry_after=self.config.poll_interval_seconds,
-                    )
-                else:
-                    raise TriggerError("no_model_channel")
-            except TriggerError as exc:
-                self._retry_or_dead_letter(event, attempts, str(exc))
-
-    def process_workspace_timeouts(self) -> None:
-        cutoff = int(time.time()) - self.config.callback_timeout_seconds
-        for row in self.store.claim_workspace_timeouts(
-            owner=self.worker_id,
-            older_than=cutoff,
-            limit=32,
-            lease_seconds=self.config.lease_seconds,
-        ):
-            event = Event.from_dict(json.loads(row["event_json"]))
-            if self.responses is None:
-                self._retry_or_dead_letter(
-                    event,
-                    int(row["attempts"]),
-                    "workspace_callback_timeout",
-                )
-                continue
-            try:
-                response_id = self.responses.start(event)
-                self.store.update_event_trigger(
-                    event.event_id,
-                    status="POLLING_RESPONSE",
-                    channel="responses_api",
-                    remote_id=response_id,
-                    retry_after=self.config.poll_interval_seconds,
-                    increment_attempt=True,
-                )
-            except TriggerError as exc:
-                self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
-
-    def poll_background_responses(self) -> None:
-        if self.responses is None:
-            return
-        for row in self.store.claim_response_polls(
-            owner=self.worker_id,
-            limit=32,
-            lease_seconds=self.config.lease_seconds,
-        ):
-            event = Event.from_dict(json.loads(row["event_json"]))
-            response_id = row.get("remote_id")
-            if not isinstance(response_id, str):
-                self._retry_or_dead_letter(event, int(row["attempts"]), "missing_response_id")
-                continue
-            try:
-                status, decision = self.responses.poll(response_id)
-                if status in {"queued", "in_progress"}:
-                    attempts = int(row["attempts"])
-                    if attempts + 1 > self.config.max_retries:
-                        self._retry_or_dead_letter(
-                            event, attempts, f"response_{status}_retry_limit"
-                        )
-                        continue
-                    self.store.update_event_trigger(
-                        event.event_id,
-                        status="POLLING_RESPONSE",
-                        channel="responses_api",
-                        remote_id=response_id,
-                        retry_after=self.config.poll_interval_seconds,
-                        increment_attempt=True,
-                    )
-                    continue
-                if decision is None:
-                    raise TriggerError("completed_without_decision")
-                self.submit_decision(decision.to_dict())
-            except OrchestratorError as exc:
-                self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
-
-    def dispatch_work_units(self) -> None:
-        rows = self.store.claim_outbox(
-            owner=self.worker_id,
-            limit=64,
-            lease_seconds=self.config.lease_seconds,
-        )
-        for row in rows:
-            try:
-                outbox_path, queue_path = self.dispatch.dispatch(row)
-                self.store.mark_outbox(row["work_unit_id"], status="DISPATCHED")
-                payload = {
-                    "assigned_agent": row["assigned_agent"],
-                    "event_id": row["event_id"],
-                    "dispatch_outbox": str(outbox_path),
-                    "command_queue": str(queue_path),
-                }
-                record = lifecycle_record(
-                    contract_commit=self.config.contract_commit,
-                    project_key=self._project_for_event(row["event_id"]),
-                    kind="TASK_DISPATCHED",
-                    work_unit_id=row["work_unit_id"],
-                    payload=payload,
-                )
-                self.store.enqueue_lifecycle(
-                    sync_id=record["sync_id"],
-                    project_key=record["project_key"],
-                    work_unit_id=row["work_unit_id"],
-                    kind=record["kind"],
-                    payload=record,
-                )
-            except Exception:
-                attempts = int(row["attempts"])
-                status = "DEAD_LETTER" if attempts >= self.config.max_retries else "RETRY"
-                self.store.mark_outbox(
-                    row["work_unit_id"],
-                    status=status,
-                    retry_after=self._backoff(attempts),
-                    increment_attempt=True,
-                )
-                LOG.exception("dispatch_failed work_unit=%s", row["work_unit_id"])
-
-    def synchronize_openclaw(self) -> None:
-        rows = self.store.claim_lifecycle(
-            owner=self.worker_id,
-            limit=64,
-            lease_seconds=self.config.lease_seconds,
-        )
-        for row in rows:
-            try:
-                payload = json.loads(row["payload_json"])
-                self.synchronizer.emit(payload)
-                self.store.mark_lifecycle(row["sync_id"], status="SENT")
-            except Exception:
-                attempts = int(row["attempts"])
-                status = "DEAD_LETTER" if attempts >= self.config.max_retries else "RETRY"
-                self.store.mark_lifecycle(
-                    row["sync_id"],
-                    status=status,
-                    retry_after=self._backoff(attempts),
-                    increment_attempt=True,
-                )
-                LOG.exception("openclaw_sync_failed sync_id=%s", row["sync_id"])
-
-    def enqueue_periodic_summaries(self) -> None:
-        for state in self.store.all_project_states():
-            project_key = state["project_key"]
-            if not self.store.summary_due(project_key, self.config.summary_interval_seconds):
-                continue
-            summary = self.store.project_summary(project_key)
-            if summary is None:
-                continue
-            digest = hashlib.sha256(
-                json.dumps(summary, separators=(",", ":"), sort_keys=True).encode()
-            ).hexdigest()
-            record = lifecycle_record(
-                contract_commit=self.config.contract_commit,
-                project_key=project_key,
-                kind="PROJECT_SUMMARY",
-                work_unit_id=None,
-                payload=summary,
-            )
-            self.store.enqueue_lifecycle(
-                sync_id=record["sync_id"],
-                project_key=project_key,
-                work_unit_id=None,
-                kind=record["kind"],
-                payload=record,
-            )
-            self.store.mark_summary(project_key, digest)
-
-    def _enqueue_event_lifecycle(self, event: Event) -> None:
-        if event.event_type == "work_unit_started":
-            kind = "TASK_STARTED"
-        elif event.event_type in {"work_unit_completed", "work_unit_failed", "work_unit_blocked"}:
-            kind = "TASK_ENDED"
-        else:
-            kind = "EVENT_RECEIVED"
-        record = lifecycle_record(
-            contract_commit=self.config.contract_commit,
-            project_key=event.project_key,
-            kind=kind,
-            work_unit_id=event.work_unit_id,
-            payload=event.to_dict(),
-        )
-        self.store.enqueue_lifecycle(
-            sync_id=record["sync_id"],
-            project_key=event.project_key,
-            work_unit_id=event.work_unit_id,
-            kind=kind,
-            payload=record,
-        )
-
-    def _enqueue_decision_lifecycle(self, decision: Decision, version: int) -> None:
-        kind = "PROJECT_COMPLETED" if decision.verdict == "COMPLETE" else "DECISION_APPLIED"
-        record = lifecycle_record(
-            contract_commit=self.config.contract_commit,
-            project_key=decision.project_key,
-            kind=kind,
-            work_unit_id=None,
-            payload={"decision": decision.to_dict(), "state_version": version},
-        )
-        self.store.enqueue_lifecycle(
-            sync_id=record["sync_id"],
-            project_key=decision.project_key,
-            work_unit_id=None,
-            kind=kind,
-            payload=record,
-        )
-        summary = self.store.project_summary(decision.project_key)
-        if summary:
-            summary_record = lifecycle_record(
-                contract_commit=self.config.contract_commit,
-                project_key=decision.project_key,
-                kind="PROJECT_SUMMARY",
-                work_unit_id=None,
-                payload=summary,
-            )
-            self.store.enqueue_lifecycle(
-                sync_id=summary_record["sync_id"],
-                project_key=decision.project_key,
-                work_unit_id=None,
-                kind=summary_record["kind"],
-                payload=summary_record,
-            )
-
-    def _project_for_event(self, event_id: str) -> str:
-        rows = self.store._conn().execute(
-            "SELECT project_key FROM events WHERE event_id=?",
-            (event_id,),
-        ).fetchone()
-        if rows is None:
-            raise ConflictError("event_not_found")
-        return str(rows["project_key"])
-
-    def _retry_or_dead_letter(self, event: Event, attempts: int, reason: str) -> None:
-        next_attempt = attempts + 1
-        if next_attempt > self.config.max_retries:
-            self.store.update_event_trigger(
-                event.event_id,
-                status="DEAD_LETTER",
-                channel=None,
-                remote_id=None,
-                increment_attempt=True,
-            )
-            record = lifecycle_record(
-                contract_commit=self.config.contract_commit,
-                project_key=event.project_key,
-                kind="TASK_ENDED",
-                work_unit_id=event.work_unit_id,
-                payload={
-                    "status": "FAIL",
-                    "reason": reason,
-                    "event_id": event.event_id,
-                    "attempts": next_attempt,
-                },
-            )
-            self.store.enqueue_lifecycle(
-                sync_id=record["sync_id"],
-                project_key=event.project_key,
-                work_unit_id=event.work_unit_id,
-                kind=record["kind"],
-                payload=record,
-            )
-            return
-        self.store.update_event_trigger(
-            event.event_id,
-            status="TRIGGER_FAILED",
-            channel=None,
-            remote_id=None,
-            retry_after=self._backoff(attempts),
-            increment_attempt=True,
-        )
-
-    def _backoff(self, attempts: int) -> int:
-        return min(3600, self.config.retry_base_seconds * (2 ** min(attempts, 10)))
-
-
-class Handler(BaseHTTPRequestHandler):
-    runtime: Runtime
-
-    def log_message(self, fmt: str, *args: object) -> None:
-        LOG.info("%s %s", self.client_address[0], fmt % args)
-
-    def _reply(self, status: int, payload: dict[str, Any]) -> None:
-        body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(body)))
-        self.send_header("Cache-Control", "no-store")
-        self.end_headers()
-        self.wfile.write(body)
-
-    def _body(self) -> dict[str, Any]:
-        try:
-            length = int(self.headers.get("Content-Length", "0"))
-        except ValueError as exc:
-            raise ValidationError("invalid_content_length") from exc
-        if length <= 0 or length > 1_048_576:
-            raise ValidationError("invalid_body_size")
-        try:
-            value = json.loads(self.rfile.read(length))
-        except json.JSONDecodeError as exc:
-            raise ValidationError("invalid_json") from exc
-        if not isinstance(value, dict):
-            raise ValidationError("json_root_not_object")
-        return value
-
-    def _authenticate(self, expected: str) -> None:
-        header = self.headers.get("Authorization", "")
-        if not header.startswith("Bearer "):
-            raise AuthenticationError("missing_bearer")
-        if not hmac.compare_digest(header[7:], expected):
-            raise AuthenticationError("invalid_bearer")
-
-    def do_GET(self) -> None:
-        path = urlparse(self.path).path.rstrip("/") or "/"
-        try:
-            if path == "/healthz":
-                self._reply(
-                    200,
-                    {
-                        "service": "aios-orchestrator-gateway",
-                        "status": "active",
-                        "contract_commit": self.runtime.config.contract_commit,
-                        "polling": True,
-                    },
-                )
-                return
-            if path.startswith("/v1/orchestrator/state/"):
-                self._authenticate(self.runtime.config.inbound_token())
-                key = path.rsplit("/", 1)[-1]
-                state = self.runtime.store.project_summary(key)
-                self._reply(200 if state else 404, state or {"error": "project_not_found"})
-                return
-            self._reply(404, {"error": "not_found"})
-        except AuthenticationError as exc:
-            self._reply(401, {"error": str(exc)})
-        except OrchestratorError as exc:
-            self._reply(400, {"error": str(exc)})
-
-    def do_POST(self) -> None:
-        path = urlparse(self.path).path.rstrip("/")
-        try:
-            if path == "/v1/orchestrator/events":
-                self._authenticate(self.runtime.config.inbound_token())
-                created, event = self.runtime.submit_event(self._body())
-                self._reply(202 if created else 200, {"created": created, "event": event})
-                return
-            if path == "/v1/orchestrator/decisions":
-                self._authenticate(self.runtime.config.callback_token())
-                version = self.runtime.submit_decision(self._body())
-                self._reply(202, {"accepted": True, "state_version": version})
-                return
-            if path == "/v1/orchestrator/poll":
-                self._authenticate(self.runtime.config.inbound_token())
-                self.runtime.poll_once()
-                self._reply(200, {"polled": True})
-                return
-            self._reply(404, {"error": "not_found"})
-        except AuthenticationError as exc:
-            self._reply(401, {"error": str(exc)})
-        except ConflictError as exc:
-            self._reply(409, {"error": str(exc)})
-        except ValidationError as exc:
-            self._reply(422, {"error": str(exc)})
-        except OrchestratorError as exc:
-            self._reply(400, {"error": str(exc)})
-        except Exception:
-            LOG.exception("request_failed path=%s", path)
-            self._reply(500, {"error": "internal_error"})
-
-
-def serve(config_path: Path) -> None:
-    config = Config.from_file(config_path)
-    runtime = Runtime(config)
-    Handler.runtime = runtime
-    server = ThreadingHTTPServer((config.host, config.port), Handler)
-    runtime.start()
-    LOG.info(
-        "gateway_started host=%s port=%s contract=%s polling=%ss",
-        config.host,
-        config.port,
-        config.contract_commit,
-        config.poll_interval_seconds,
-    )
-    try:
-        server.serve_forever(poll_interval=0.5)
-    finally:
-        server.server_close()
-        runtime.stop()
@@ -1,826 +0,0 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-import sqlite3
-import threading
-import time
-from typing import Any
-
-from .errors import ConflictError
-from .models import Decision, Event
-from .securefs import ensure_secure_directory
-
-
-class Store:
-    """Durable state machine with idempotency, optimistic versioning, and leases."""
-
-    def __init__(self, path: Path):
-        self.path = Path(path)
-        ensure_secure_directory(self.path.parent)
-        self._local = threading.local()
-        self._initialize()
-
-    def _connect(self) -> sqlite3.Connection:
-        conn = sqlite3.connect(self.path, timeout=30, isolation_level=None)
-        conn.row_factory = sqlite3.Row
-        conn.execute("PRAGMA journal_mode=WAL")
-        conn.execute("PRAGMA synchronous=FULL")
-        conn.execute("PRAGMA foreign_keys=ON")
-        conn.execute("PRAGMA busy_timeout=30000")
-        return conn
-
-    def _conn(self) -> sqlite3.Connection:
-        conn = getattr(self._local, "conn", None)
-        if conn is None:
-            conn = self._connect()
-            self._local.conn = conn
-        return conn
-
-    def _initialize(self) -> None:
-        conn = self._connect()
-        try:
-            conn.executescript(
-                """
-                CREATE TABLE IF NOT EXISTS project_state (
-                    project_key TEXT PRIMARY KEY,
-                    contract_commit TEXT NOT NULL,
-                    state_version INTEGER NOT NULL,
-                    state_json TEXT NOT NULL,
-                    hops INTEGER NOT NULL DEFAULT 0,
-                    updated_at INTEGER NOT NULL
-                );
-
-                CREATE TABLE IF NOT EXISTS gate_hops (
-                    project_key TEXT NOT NULL,
-                    gate_id TEXT NOT NULL,
-                    hops INTEGER NOT NULL DEFAULT 0,
-                    updated_at INTEGER NOT NULL,
-                    PRIMARY KEY(project_key, gate_id)
-                );
-
-                CREATE TABLE IF NOT EXISTS events (
-                    event_id TEXT PRIMARY KEY,
-                    project_key TEXT NOT NULL,
-                    digest TEXT NOT NULL,
-                    event_json TEXT NOT NULL,
-                    status TEXT NOT NULL,
-                    channel TEXT,
-                    remote_id TEXT,
-                    attempts INTEGER NOT NULL DEFAULT 0,
-                    next_attempt_at INTEGER NOT NULL DEFAULT 0,
-                    lease_owner TEXT,
-                    lease_until INTEGER,
-                    created_at INTEGER NOT NULL,
-                    updated_at INTEGER NOT NULL
-                );
-
-                CREATE INDEX IF NOT EXISTS idx_events_poll
-                ON events(status, next_attempt_at, lease_until, created_at);
-
-                CREATE TABLE IF NOT EXISTS decisions (
-                    event_id TEXT PRIMARY KEY REFERENCES events(event_id),
-                    project_key TEXT NOT NULL,
-                    decision_json TEXT NOT NULL,
-                    state_version_after INTEGER NOT NULL,
-                    created_at INTEGER NOT NULL
-                );
-
-                CREATE TABLE IF NOT EXISTS outbox (
-                    work_unit_id TEXT PRIMARY KEY,
-                    event_id TEXT NOT NULL REFERENCES events(event_id),
-                    assigned_agent TEXT NOT NULL,
-                    payload_json TEXT NOT NULL,
-                    status TEXT NOT NULL,
-                    attempts INTEGER NOT NULL DEFAULT 0,
-                    next_attempt_at INTEGER NOT NULL DEFAULT 0,
-                    lease_owner TEXT,
-                    lease_until INTEGER,
-                    created_at INTEGER NOT NULL,
-                    updated_at INTEGER NOT NULL
-                );
-
-                CREATE INDEX IF NOT EXISTS idx_outbox_poll
-                ON outbox(status, next_attempt_at, lease_until, created_at);
-
-                CREATE TABLE IF NOT EXISTS lifecycle (
-                    sync_id TEXT PRIMARY KEY,
-                    project_key TEXT NOT NULL,
-                    work_unit_id TEXT,
-                    kind TEXT NOT NULL,
-                    payload_json TEXT NOT NULL,
-                    status TEXT NOT NULL,
-                    attempts INTEGER NOT NULL DEFAULT 0,
-                    next_attempt_at INTEGER NOT NULL DEFAULT 0,
-                    lease_owner TEXT,
-                    lease_until INTEGER,
-                    created_at INTEGER NOT NULL,
-                    updated_at INTEGER NOT NULL
-                );
-
-                CREATE INDEX IF NOT EXISTS idx_lifecycle_poll
-                ON lifecycle(status, next_attempt_at, lease_until, created_at);
-
-                CREATE TABLE IF NOT EXISTS blocker_history (
-                    project_key TEXT NOT NULL,
-                    gate_id TEXT NOT NULL,
-                    blocker TEXT NOT NULL,
-                    repetitions INTEGER NOT NULL,
-                    updated_at INTEGER NOT NULL,
-                    PRIMARY KEY(project_key, gate_id, blocker)
-                );
-
-                CREATE TABLE IF NOT EXISTS summaries (
-                    project_key TEXT PRIMARY KEY,
-                    last_emitted_at INTEGER NOT NULL DEFAULT 0,
-                    last_digest TEXT
-                );
-
-                CREATE TABLE IF NOT EXISTS observed_refs (
-                    refname TEXT PRIMARY KEY,
-                    object_id TEXT NOT NULL,
-                    observed_at INTEGER NOT NULL
-                );
-
-                CREATE TABLE IF NOT EXISTS metadata (
-                    key TEXT PRIMARY KEY,
-                    value TEXT NOT NULL
-                );
-                """
-            )
-        finally:
-            conn.close()
-
-    def register_event(self, event: Event) -> tuple[bool, dict[str, Any]]:
-        now = int(time.time())
-        conn = self._conn()
-        conn.execute("BEGIN IMMEDIATE")
-        try:
-            existing = conn.execute(
-                "SELECT digest, event_json FROM events WHERE event_id=?",
-                (event.event_id,),
-            ).fetchone()
-            if existing:
-                if existing["digest"] != event.digest():
-                    raise ConflictError("event_id_payload_conflict")
-                conn.execute("COMMIT")
-                return False, json.loads(existing["event_json"])
-
-            state = conn.execute(
-                "SELECT contract_commit, state_version FROM project_state WHERE project_key=?",
-                (event.project_key,),
-            ).fetchone()
-            if state is None:
-                conn.execute(
-                    """
-                    INSERT INTO project_state(
-                        project_key, contract_commit, state_version, state_json,
-                        hops, updated_at
-                    ) VALUES (?, ?, ?, '{}', 0, ?)
-                    """,
-                    (event.project_key, event.contract_commit, event.state_version, now),
-                )
-            else:
-                if state["contract_commit"] != event.contract_commit:
-                    raise ConflictError("contract_version_mismatch")
-                if state["state_version"] != event.state_version:
-                    raise ConflictError("stale_state_version")
-
-            conn.execute(
-                """
-                INSERT INTO events(
-                    event_id, project_key, digest, event_json, status,
-                    next_attempt_at, created_at, updated_at
-                ) VALUES (?, ?, ?, ?, 'RECEIVED', ?, ?, ?)
-                """,
-                (
-                    event.event_id,
-                    event.project_key,
-                    event.digest(),
-                    event.canonical_json(),
-                    now,
-                    now,
-                    now,
-                ),
-            )
-            conn.execute("COMMIT")
-            return True, event.to_dict()
-        except Exception:
-            conn.execute("ROLLBACK")
-            raise
-
-    def claim_events(
-        self,
-        *,
-        owner: str,
-        statuses: tuple[str, ...],
-        limit: int,
-        lease_seconds: int,
-    ) -> list[dict[str, Any]]:
-        if not statuses:
-            return []
-        now = int(time.time())
-        placeholders = ",".join("?" for _ in statuses)
-        conn = self._conn()
-        conn.execute("BEGIN IMMEDIATE")
-        try:
-            rows = conn.execute(
-                f"""
-                SELECT event_id FROM events
-                WHERE status IN ({placeholders})
-                  AND next_attempt_at <= ?
-                  AND (lease_until IS NULL OR lease_until < ?)
-                ORDER BY created_at
-                LIMIT ?
-                """,
-                (*statuses, now, now, limit),
-            ).fetchall()
-            ids = [row["event_id"] for row in rows]
-            for event_id in ids:
-                conn.execute(
-                    """
-                    UPDATE events SET lease_owner=?, lease_until=?, updated_at=?
-                    WHERE event_id=?
-                    """,
-                    (owner, now + lease_seconds, now, event_id),
-                )
-            conn.execute("COMMIT")
-        except Exception:
-            conn.execute("ROLLBACK")
-            raise
-        if not ids:
-            return []
-        placeholders = ",".join("?" for _ in ids)
-        return [
-            dict(row)
-            for row in conn.execute(
-                f"SELECT * FROM events WHERE event_id IN ({placeholders})",
-                ids,
-            ).fetchall()
-        ]
-
-    def update_event_trigger(
-        self,
-        event_id: str,
-        *,
-        status: str,
-        channel: str | None,
-        remote_id: str | None,
-        retry_after: int = 0,
-        increment_attempt: bool = False,
-    ) -> None:
-        now = int(time.time())
-        self._conn().execute(
-            """
-            UPDATE events
-            SET status=?, channel=?, remote_id=?,
-                attempts=attempts + ?,
-                next_attempt_at=?,
-                lease_owner=NULL, lease_until=NULL, updated_at=?
-            WHERE event_id=?
-            """,
-            (
-                status,
-                channel,
-                remote_id,
-                1 if increment_attempt else 0,
-                now + max(0, retry_after),
-                now,
-                event_id,
-            ),
-        )
-
-    def claim_workspace_timeouts(
-        self,
-        *,
-        owner: str,
-        older_than: int,
-        limit: int,
-        lease_seconds: int,
-    ) -> list[dict[str, Any]]:
-        return self._claim_event_subset(
-            owner=owner,
-            where_sql="status='AWAITING_DECISION' AND channel='workspace_agent' AND updated_at <= ?",
-            where_params=(older_than,),
-            limit=limit,
-            lease_seconds=lease_seconds,
-        )
-
-    def claim_response_polls(
-        self,
-        *,
-        owner: str,
-        limit: int,
-        lease_seconds: int,
-    ) -> list[dict[str, Any]]:
-        now = int(time.time())
-        return self._claim_event_subset(
-            owner=owner,
-            where_sql="status='POLLING_RESPONSE' AND next_attempt_at <= ?",
-            where_params=(now,),
-            limit=limit,
-            lease_seconds=lease_seconds,
-        )
-
-    def _claim_event_subset(
-        self,
-        *,
-        owner: str,
-        where_sql: str,
-        where_params: tuple[Any, ...],
-        limit: int,
-        lease_seconds: int,
-    ) -> list[dict[str, Any]]:
-        allowed = {
-            "status='AWAITING_DECISION' AND channel='workspace_agent' AND updated_at <= ?",
-            "status='POLLING_RESPONSE' AND next_attempt_at <= ?",
-        }
-        if where_sql not in allowed:
-            raise ValueError("invalid_event_claim_query")
-        now = int(time.time())
-        conn = self._conn()
-        conn.execute("BEGIN IMMEDIATE")
-        try:
-            rows = conn.execute(
-                f"""
-                SELECT event_id FROM events
-                WHERE {where_sql}
-                  AND (lease_until IS NULL OR lease_until < ?)
-                ORDER BY updated_at
-                LIMIT ?
-                """,
-                (*where_params, now, limit),
-            ).fetchall()
-            ids = [str(row["event_id"]) for row in rows]
-            for event_id in ids:
-                conn.execute(
-                    """
-                    UPDATE events SET lease_owner=?, lease_until=?
-                    WHERE event_id=?
-                    """,
-                    (owner, now + lease_seconds, event_id),
-                )
-            conn.execute("COMMIT")
-        except Exception:
-            conn.execute("ROLLBACK")
-            raise
-        if not ids:
-            return []
-        placeholders = ",".join("?" for _ in ids)
-        return [
-            dict(row)
-            for row in conn.execute(
-                f"SELECT * FROM events WHERE event_id IN ({placeholders})",
-                ids,
-            ).fetchall()
-        ]
-
-    def apply_decision(
-        self,
-        decision: Decision,
-        *,
-        max_hops_per_gate: int,
-        max_same_blocker_retries: int,
-    ) -> int:
-        now = int(time.time())
-        conn = self._conn()
-        conn.execute("BEGIN IMMEDIATE")
-        try:
-            event = conn.execute(
-                "SELECT project_key FROM events WHERE event_id=?",
-                (decision.event_id,),
-            ).fetchone()
-            if event is None:
-                raise ConflictError("event_not_found")
-            if event["project_key"] != decision.project_key:
-                raise ConflictError("event_project_mismatch")
-
-            existing = conn.execute(
-                "SELECT decision_json, state_version_after FROM decisions WHERE event_id=?",
-                (decision.event_id,),
-            ).fetchone()
-            if existing:
-                if json.loads(existing["decision_json"]) != decision.to_dict():
-                    raise ConflictError("decision_idempotency_conflict")
-                conn.execute("COMMIT")
-                return int(existing["state_version_after"])
-
-            state = conn.execute(
-                "SELECT * FROM project_state WHERE project_key=?",
-                (decision.project_key,),
-            ).fetchone()
-            if state is None:
-                raise ConflictError("project_state_not_found")
-            if state["contract_commit"] != decision.contract_commit:
-                raise ConflictError("contract_version_mismatch")
-            if state["state_version"] != decision.expected_state_version:
-                raise ConflictError("stale_decision_state_version")
-            state_json = json.loads(state["state_json"])
-            gate = str(
-                decision.state_updates.get(
-                    "current_gate", state_json.get("current_gate", "unknown")
-                )
-            )
-            gate_hops = conn.execute(
-                "SELECT hops FROM gate_hops WHERE project_key=? AND gate_id=?",
-                (decision.project_key, gate),
-            ).fetchone()
-            if gate_hops is not None and int(gate_hops["hops"]) >= max_hops_per_gate:
-                raise ConflictError("max_hops_exceeded")
-
-            state_json.update(decision.state_updates)
-            next_version = int(state["state_version"]) + 1
-
-            if decision.stop_reason:
-                row = conn.execute(
-                    """
-                    SELECT repetitions FROM blocker_history
-                    WHERE project_key=? AND gate_id=? AND blocker=?
-                    """,
-                    (decision.project_key, gate, decision.stop_reason),
-                ).fetchone()
-                repetitions = (int(row["repetitions"]) if row else 0) + 1
-                conn.execute(
-                    """
-                    INSERT INTO blocker_history(
-                        project_key, gate_id, blocker, repetitions, updated_at
-                    ) VALUES (?, ?, ?, ?, ?)
-                    ON CONFLICT(project_key, gate_id, blocker)
-                    DO UPDATE SET repetitions=excluded.repetitions,
-                                  updated_at=excluded.updated_at
-                    """,
-                    (decision.project_key, gate, decision.stop_reason, repetitions, now),
-                )
-                if repetitions > max_same_blocker_retries:
-                    raise ConflictError("same_blocker_retry_limit_exceeded")
-
-            conn.execute(
-                """
-                INSERT INTO gate_hops(project_key, gate_id, hops, updated_at)
-                VALUES (?, ?, 1, ?)
-                ON CONFLICT(project_key, gate_id)
-                DO UPDATE SET hops=gate_hops.hops + 1, updated_at=excluded.updated_at
-                """,
-                (decision.project_key, gate, now),
-            )
-
-            conn.execute(
-                """
-                UPDATE project_state
-                SET state_version=?, state_json=?, hops=hops+1, updated_at=?
-                WHERE project_key=?
-                """,
-                (
-                    next_version,
-                    json.dumps(state_json, separators=(",", ":"), sort_keys=True),
-                    now,
-                    decision.project_key,
-                ),
-            )
-            conn.execute(
-                """
-                INSERT INTO decisions(
-                    event_id, project_key, decision_json,
-                    state_version_after, created_at
-                ) VALUES (?, ?, ?, ?, ?)
-                """,
-                (
-                    decision.event_id,
-                    decision.project_key,
-                    json.dumps(decision.to_dict(), separators=(",", ":"), sort_keys=True),
-                    next_version,
-                    now,
-                ),
-            )
-            for unit in decision.next_work_units:
-                conn.execute(
-                    """
-                    INSERT INTO outbox(
-                        work_unit_id, event_id, assigned_agent, payload_json,
-                        status, next_attempt_at, created_at, updated_at
-                    ) VALUES (?, ?, ?, ?, 'PENDING', ?, ?, ?)
-                    """,
-                    (
-                        unit.id,
-                        decision.event_id,
-                        unit.primary_agent,
-                        json.dumps(unit.to_dict(), separators=(",", ":"), sort_keys=True),
-                        now,
-                        now,
-                        now,
-                    ),
-                )
-            conn.execute(
-                """
-                UPDATE events SET status='DECIDED', lease_owner=NULL,
-                    lease_until=NULL, updated_at=?
-                WHERE event_id=?
-                """,
-                (now, decision.event_id),
-            )
-            conn.execute("COMMIT")
-            return next_version
-        except Exception:
-            conn.execute("ROLLBACK")
-            raise
-
-    def claim_outbox(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
-        return self._claim_table(
-            table="outbox",
-            key="work_unit_id",
-            owner=owner,
-            statuses=("PENDING", "RETRY"),
-            limit=limit,
-            lease_seconds=lease_seconds,
-        )
-
-    def mark_outbox(
-        self,
-        work_unit_id: str,
-        *,
-        status: str,
-        retry_after: int = 0,
-        increment_attempt: bool = False,
-    ) -> None:
-        self._mark_table(
-            table="outbox",
-            key="work_unit_id",
-            value=work_unit_id,
-            status=status,
-            retry_after=retry_after,
-            increment_attempt=increment_attempt,
-        )
-
-    def enqueue_lifecycle(
-        self,
-        *,
-        sync_id: str,
-        project_key: str,
-        work_unit_id: str | None,
-        kind: str,
-        payload: dict[str, Any],
-    ) -> None:
-        now = int(time.time())
-        encoded = json.dumps(payload, separators=(",", ":"), sort_keys=True)
-        self._conn().execute(
-            """
-            INSERT OR IGNORE INTO lifecycle(
-                sync_id, project_key, work_unit_id, kind, payload_json,
-                status, next_attempt_at, created_at, updated_at
-            ) VALUES (?, ?, ?, ?, ?, 'PENDING', ?, ?, ?)
-            """,
-            (sync_id, project_key, work_unit_id, kind, encoded, now, now, now),
-        )
-
-    def claim_lifecycle(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
-        return self._claim_table(
-            table="lifecycle",
-            key="sync_id",
-            owner=owner,
-            statuses=("PENDING", "RETRY"),
-            limit=limit,
-            lease_seconds=lease_seconds,
-        )
-
-    def mark_lifecycle(
-        self,
-        sync_id: str,
-        *,
-        status: str,
-        retry_after: int = 0,
-        increment_attempt: bool = False,
-    ) -> None:
-        self._mark_table(
-            table="lifecycle",
-            key="sync_id",
-            value=sync_id,
-            status=status,
-            retry_after=retry_after,
-            increment_attempt=increment_attempt,
-        )
-
-    def _claim_table(
-        self,
-        *,
-        table: str,
-        key: str,
-        owner: str,
-        statuses: tuple[str, ...],
-        limit: int,
-        lease_seconds: int,
-    ) -> list[dict[str, Any]]:
-        if table not in {"outbox", "lifecycle"} or key not in {"work_unit_id", "sync_id"}:
-            raise ValueError("invalid_claim_table")
-        now = int(time.time())
-        placeholders = ",".join("?" for _ in statuses)
-        conn = self._conn()
-        conn.execute("BEGIN IMMEDIATE")
-        try:
-            rows = conn.execute(
-                f"""
-                SELECT {key} FROM {table}
-                WHERE status IN ({placeholders})
-                  AND next_attempt_at <= ?
-                  AND (lease_until IS NULL OR lease_until < ?)
-                ORDER BY created_at
-                LIMIT ?
-                """,
-                (*statuses, now, now, limit),
-            ).fetchall()
-            values = [row[key] for row in rows]
-            for item in values:
-                conn.execute(
-                    f"""
-                    UPDATE {table} SET lease_owner=?, lease_until=?, updated_at=?
-                    WHERE {key}=?
-                    """,
-                    (owner, now + lease_seconds, now, item),
-                )
-            conn.execute("COMMIT")
-        except Exception:
-            conn.execute("ROLLBACK")
-            raise
-        if not values:
-            return []
-        placeholders = ",".join("?" for _ in values)
-        return [
-            dict(row)
-            for row in conn.execute(
-                f"SELECT * FROM {table} WHERE {key} IN ({placeholders})",
-                values,
-            ).fetchall()
-        ]
-
-    def _mark_table(
-        self,
-        *,
-        table: str,
-        key: str,
-        value: str,
-        status: str,
-        retry_after: int,
-        increment_attempt: bool,
-    ) -> None:
-        if table not in {"outbox", "lifecycle"} or key not in {"work_unit_id", "sync_id"}:
-            raise ValueError("invalid_mark_table")
-        now = int(time.time())
-        self._conn().execute(
-            f"""
-            UPDATE {table}
-            SET status=?, attempts=attempts + ?, next_attempt_at=?,
-                lease_owner=NULL, lease_until=NULL, updated_at=?
-            WHERE {key}=?
-            """,
-            (
-                status,
-                1 if increment_attempt else 0,
-                now + max(0, retry_after),
-                now,
-                value,
-            ),
-        )
-
-    def project_state(self, project_key: str) -> dict[str, Any] | None:
-        row = self._conn().execute(
-            "SELECT * FROM project_state WHERE project_key=?",
-            (project_key,),
-        ).fetchone()
-        if row is None:
-            return None
-        result = dict(row)
-        result["state"] = json.loads(result.pop("state_json"))
-        return result
-
-    def all_project_states(self) -> list[dict[str, Any]]:
-        result = []
-        for row in self._conn().execute("SELECT * FROM project_state ORDER BY project_key").fetchall():
-            item = dict(row)
-            item["state"] = json.loads(item.pop("state_json"))
-            result.append(item)
-        return result
-
-    def summary_due(self, project_key: str, interval: int) -> bool:
-        now = int(time.time())
-        row = self._conn().execute(
-            "SELECT last_emitted_at FROM summaries WHERE project_key=?",
-            (project_key,),
-        ).fetchone()
-        return row is None or now - int(row["last_emitted_at"]) >= interval
-
-    def mark_summary(self, project_key: str, digest: str) -> None:
-        now = int(time.time())
-        self._conn().execute(
-            """
-            INSERT INTO summaries(project_key, last_emitted_at, last_digest)
-            VALUES (?, ?, ?)
-            ON CONFLICT(project_key)
-            DO UPDATE SET last_emitted_at=excluded.last_emitted_at,
-                          last_digest=excluded.last_digest
-            """,
-            (project_key, now, digest),
-        )
-
-
-    def project_summary(self, project_key: str) -> dict[str, Any] | None:
-        state = self.project_state(project_key)
-        if state is None:
-            return None
-        conn = self._conn()
-        event_counts = {
-            row["status"]: int(row["count"])
-            for row in conn.execute(
-                """
-                SELECT status, COUNT(*) AS count FROM events
-                WHERE project_key=? GROUP BY status
-                """,
-                (project_key,),
-            ).fetchall()
-        }
-        outbox_counts = {
-            row["status"]: int(row["count"])
-            for row in conn.execute(
-                """
-                SELECT o.status, COUNT(*) AS count
-                FROM outbox o JOIN events e ON e.event_id=o.event_id
-                WHERE e.project_key=? GROUP BY o.status
-                """,
-                (project_key,),
-            ).fetchall()
-        }
-        return {
-            "project_key": project_key,
-            "contract_commit": state["contract_commit"],
-            "state_version": state["state_version"],
-            "state": state["state"],
-            "event_counts": event_counts,
-            "work_unit_counts": outbox_counts,
-            "updated_at": state["updated_at"],
-        }
-
-
-    def metadata_get(self, key: str) -> str | None:
-        row = self._conn().execute(
-            "SELECT value FROM metadata WHERE key=?",
-            (key,),
-        ).fetchone()
-        return None if row is None else str(row["value"])
-
-    def metadata_set(self, key: str, value: str) -> None:
-        self._conn().execute(
-            """
-            INSERT INTO metadata(key, value) VALUES (?, ?)
-            ON CONFLICT(key) DO UPDATE SET value=excluded.value
-            """,
-            (key, value),
-        )
-
-    def observe_refs(self, refs: dict[str, str]) -> tuple[bool, list[tuple[str, str | None, str]]]:
-        """Persist a Git snapshot and return changes after the initial baseline."""
-        now = int(time.time())
-        conn = self._conn()
-        conn.execute("BEGIN IMMEDIATE")
-        try:
-            initialized = conn.execute(
-                "SELECT value FROM metadata WHERE key='git_poll_initialized'"
-            ).fetchone()
-            previous = {
-                row["refname"]: row["object_id"]
-                for row in conn.execute("SELECT refname, object_id FROM observed_refs").fetchall()
-            }
-            changes: list[tuple[str, str | None, str]] = []
-            if initialized is not None:
-                for refname, oid in refs.items():
-                    old = previous.get(refname)
-                    if old != oid:
-                        changes.append((refname, old, oid))
-                for refname, old in previous.items():
-                    if refname not in refs:
-                        changes.append((refname, old, "0" * 40))
-            conn.execute("DELETE FROM observed_refs")
-            for refname, oid in refs.items():
-                conn.execute(
-                    """
-                    INSERT INTO observed_refs(refname, object_id, observed_at)
-                    VALUES (?, ?, ?)
-                    """,
-                    (refname, oid, now),
-                )
-            conn.execute(
-                """
-                INSERT INTO metadata(key, value)
-                VALUES ('git_poll_initialized', '1')
-                ON CONFLICT(key) DO UPDATE SET value='1'
-                """
-            )
-            conn.execute("COMMIT")
-            return initialized is not None, changes
-        except Exception:
-            conn.execute("ROLLBACK")
-            raise
-
-
-    def close(self) -> None:
-        conn = getattr(self._local, "conn", None)
-        if conn is not None:
-            conn.close()
-            self._local.conn = None
@@ -1,108 +0,0 @@
-from __future__ import annotations
-
-import hashlib
-import json
-import os
-import stat
-from pathlib import Path
-import urllib.error
-import urllib.request
-from typing import Any
-
-from .errors import DispatchError
-from .securefs import atomic_write_bytes_shared, ensure_shared_directory, read_secure_bytes
-
-
-class OpenClawSynchronizer:
-    """Synchronize task starts, ends, dispatches, and project summaries.
-
-    A durable file is always emitted. An optional HTTP endpoint is best-effort;
-    failure leaves the lifecycle record pending for retry.
-    """
-
-    def __init__(
-        self,
-        queue_dir: Path,
-        *,
-        endpoint: str | None = None,
-        bearer_token: str | None = None,
-        timeout_seconds: int = 10,
-    ):
-        self.queue_dir = Path(queue_dir)
-        self.endpoint = endpoint
-        self.bearer_token = bearer_token
-        self.timeout_seconds = timeout_seconds
-        ensure_shared_directory(self.queue_dir)
-
-    def emit(self, record: dict[str, Any]) -> Path:
-        sync_id = record.get("sync_id")
-        if not isinstance(sync_id, str) or not sync_id:
-            raise DispatchError("invalid_sync_id")
-        payload = (json.dumps(record, indent=2, sort_keys=True) + "\n").encode()
-        path = self.queue_dir / "pending" / f"{sync_id}.json"
-        ensure_shared_directory(path.parent)
-        try:
-            st = os.lstat(path)
-        except FileNotFoundError:
-            atomic_write_bytes_shared(path, payload, mode=0o640)
-        else:
-            if not stat.S_ISREG(st.st_mode):
-                raise DispatchError("sync_target_not_regular")
-            if read_secure_bytes(path, expected_uid=None, max_bytes=1_048_576) != payload:
-                raise DispatchError("sync_idempotency_conflict")
-
-        if self.endpoint:
-            headers = {
-                "Content-Type": "application/json",
-                "Idempotency-Key": sync_id,
-                "User-Agent": "aios-orchestrator-gateway/0.2",
-            }
-            if self.bearer_token:
-                headers["Authorization"] = f"Bearer {self.bearer_token}"
-            request = urllib.request.Request(
-                self.endpoint,
-                data=payload,
-                method="POST",
-                headers=headers,
-            )
-            try:
-                with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
-                    response.read(65_536)
-                    if response.status not in {200, 201, 202, 204}:
-                        raise DispatchError(f"openclaw_sync_http_{response.status}")
-            except urllib.error.HTTPError as exc:
-                raise DispatchError(f"openclaw_sync_http_{exc.code}") from exc
-            except (urllib.error.URLError, TimeoutError, OSError) as exc:
-                raise DispatchError("openclaw_sync_failed") from exc
-        return path
-
-
-def lifecycle_record(
-    *,
-    contract_commit: str,
-    project_key: str,
-    kind: str,
-    work_unit_id: str | None,
-    payload: dict[str, Any],
-) -> dict[str, Any]:
-    canonical = json.dumps(
-        {
-            "contract_commit": contract_commit,
-            "project_key": project_key,
-            "kind": kind,
-            "work_unit_id": work_unit_id,
-            "payload": payload,
-        },
-        separators=(",", ":"),
-        sort_keys=True,
-    )
-    sync_id = hashlib.sha256(canonical.encode()).hexdigest()
-    return {
-        "schema_version": 1,
-        "sync_id": sync_id,
-        "contract_commit": contract_commit,
-        "project_key": project_key,
-        "kind": kind,
-        "work_unit_id": work_unit_id,
-        "payload": payload,
-    }
@@ -1,37 +0,0 @@
-[Unit]
-Description=AIOS Orchestrator Gateway reliability poll
-After=aios-orchestrator-gateway.service
-ConditionPathExists=/etc/aios-orchestrator-gateway/config.json
-
-[Service]
-Type=oneshot
-User=aios-orchestrator
-Group=aios-orchestrator
-SupplementaryGroups=gitops
-WorkingDirectory=/opt/aios-orchestrator-gateway/current
-ExecStart=/usr/bin/python3 -m services.aios_orchestrator_gateway --config /etc/aios-orchestrator-gateway/config.json --poll-once
-UMask=0077
-NoNewPrivileges=true
-PrivateTmp=true
-PrivateDevices=true
-ProtectSystem=strict
-ProtectHome=true
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectKernelLogs=true
-ProtectControlGroups=true
-ProtectHostname=true
-ProtectProc=invisible
-ProcSubset=pid
-RestrictNamespaces=true
-MemoryDenyWriteExecute=true
-RestrictSUIDSGID=true
-LockPersonality=true
-CapabilityBoundingSet=
-AmbientCapabilities=
-RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
-StateDirectory=aios-orchestrator-gateway
-ReadOnlyPaths=/etc/aios-orchestrator-gateway /opt/aios-orchestrator-gateway
-ReadWritePaths=/var/lib/aios-orchestrator-gateway
-SystemCallFilter=@system-service
-SystemCallErrorNumber=EPERM
@@ -1,12 +0,0 @@
-[Unit]
-Description=Periodically poll AIOS orchestration inboxes and model responses
-
-[Timer]
-OnBootSec=45s
-OnUnitActiveSec=60s
-RandomizedDelaySec=10s
-Persistent=true
-Unit=aios-orchestrator-gateway-poll.service
-
-[Install]
-WantedBy=timers.target
@@ -1,52 +0,0 @@
-[Unit]
-Description=AIOS Orchestrator Gateway
-After=network-online.target
-Wants=network-online.target
-ConditionPathExists=/etc/aios-orchestrator-gateway/config.json
-
-[Service]
-Type=simple
-User=aios-orchestrator
-Group=aios-orchestrator
-SupplementaryGroups=gitops
-WorkingDirectory=/opt/aios-orchestrator-gateway/current
-ExecStart=/usr/bin/python3 -m services.aios_orchestrator_gateway --config /etc/aios-orchestrator-gateway/config.json
-Restart=on-failure
-RestartSec=5s
-TimeoutStartSec=30s
-TimeoutStopSec=30s
-UMask=0077
-
-NoNewPrivileges=true
-PrivateTmp=true
-PrivateDevices=true
-ProtectSystem=strict
-ProtectHome=true
-ProtectKernelTunables=true
-ProtectKernelModules=true
-ProtectKernelLogs=true
-ProtectControlGroups=true
-ProtectClock=true
-ProtectHostname=true
-ProtectProc=invisible
-ProcSubset=pid
-RestrictNamespaces=true
-MemoryDenyWriteExecute=true
-RestrictSUIDSGID=true
-LockPersonality=true
-RestrictRealtime=true
-CapabilityBoundingSet=
-AmbientCapabilities=
-RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
-SystemCallArchitectures=native
-StateDirectory=aios-orchestrator-gateway
-RuntimeDirectory=aios-orchestrator-gateway
-ReadOnlyPaths=/etc/aios-orchestrator-gateway /opt/aios-orchestrator-gateway
-ReadWritePaths=/var/lib/aios-orchestrator-gateway
-SystemCallFilter=@system-service
-SystemCallErrorNumber=EPERM
-StandardOutput=journal
-StandardError=journal
-
-[Install]
-WantedBy=multi-user.target
@@ -1,2 +0,0 @@
-g aios-orchestrator -
-u aios-orchestrator - "AIOS Orchestrator Gateway" /var/lib/aios-orchestrator-gateway /usr/sbin/nologin
@@ -1,19 +0,0 @@
-d /etc/aios-orchestrator-gateway 0700 root root -
-d /var/lib/aios-orchestrator-gateway 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox/events 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox/events/pending 2770 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/inbox/events/processed 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox/events/rejected 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox/decisions 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox/decisions/pending 2770 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/inbox/decisions/processed 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/inbox/decisions/rejected 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/outbox 0700 aios-orchestrator aios-orchestrator -
-d /var/lib/aios-orchestrator-gateway/openclaw-sync 2750 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/openclaw-sync/pending 2750 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/commands 2750 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/commands/openclaw 2750 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/commands/openclaw/pending 2750 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/commands/codex 2750 aios-orchestrator gitops -
-d /var/lib/aios-orchestrator-gateway/commands/codex/pending 2750 aios-orchestrator gitops -
@@ -1,774 +0,0 @@
-from __future__ import annotations
-
-import json
-import os
-import stat
-from pathlib import Path
-import tempfile
-import time
-import unittest
-import urllib.error
-from unittest import mock
-
-from services.aios_orchestrator_gateway.clients import ResponsesClient, WorkspaceAgentClient
-from services.aios_orchestrator_gateway.config import Config
-from services.aios_orchestrator_gateway.dispatch import DispatchSink
-from services.aios_orchestrator_gateway.errors import ConflictError, DispatchError, ValidationError
-from services.aios_orchestrator_gateway.models import Decision, Event
-from services.aios_orchestrator_gateway.service import Runtime
-from services.aios_orchestrator_gateway.store import Store
-from services.aios_orchestrator_gateway.sync import OpenClawSynchronizer, lifecycle_record
-from scripts import report_orchestrator_event
-
-
-CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
-BASE = "8d9c6da90fc8caa69b691d243c6fefb40b6b0575"
-
-
-def event_payload(**overrides):
-    value = {
-        "schema_version": 1,
-        "project_key": "v2r1-personal-weixin",
-        "contract_commit": CONTRACT,
-        "event_id": "evt-001",
-        "source_agent": "openclaw",
-        "event_type": "work_unit_completed",
-        "work_unit_id": "P1-REACK",
-        "status": "PASS",
-        "state_version": 0,
-        "implementation_commit": None,
-        "result_commit": None,
-        "blocker": None,
-        "evidence_refs": [],
-        "payload": {},
-    }
-    value.update(overrides)
-    return value
-
-
-def decision_payload(**overrides):
-    value = {
-        "schema_version": 1,
-        "project_key": "v2r1-personal-weixin",
-        "contract_commit": CONTRACT,
-        "event_id": "evt-001",
-        "expected_state_version": 0,
-        "verdict": "DISPATCH",
-        "state_updates": [
-            {"key": "P1_REPAIR3_CODE", "value": "PASS_CANDIDATE_PREPARED"},
-            {"key": "current_gate", "value": "R0_RUNNER_LIVE_RECOVERY"},
-        ],
-        "next_work_units": [
-            {
-                "id": "R0-DISCOVER",
-                "track_id": "R_RUNNER",
-                "gate_id": "R0_RUNNER_LIVE_RECOVERY",
-                "primary_agent": "openclaw",
-                "fallback_agent": "codex",
-                "clean_base_commit": BASE,
-                "allowed_paths": [],
-                "required_checks": ["secret_provider_metadata"],
-                "forbidden_operations": ["force_push"],
-                "instruction": "Discover provider metadata without reading secrets.",
-            }
-        ],
-        "requires_human_input": False,
-        "stop_reason": None,
-    }
-    value.update(overrides)
-    return value
-
-
-class ModelTests(unittest.TestCase):
-    def test_event_accepts_valid_payload(self):
-        event = Event.from_dict(event_payload())
-        self.assertEqual(event.event_id, "evt-001")
-        self.assertEqual(len(event.digest()), 64)
-
-    def test_event_rejects_extra_fields(self):
-        with self.assertRaises(ValidationError):
-            Event.from_dict(event_payload(extra=True))
-
-    def test_event_rejects_invalid_contract_oid(self):
-        with self.assertRaises(ValidationError):
-            Event.from_dict(event_payload(contract_commit="abc"))
-
-    def test_decision_requires_work_unit_for_dispatch(self):
-        with self.assertRaises(ValidationError):
-            Decision.from_dict(decision_payload(next_work_units=[]))
-
-    def test_decision_rejects_same_primary_fallback(self):
-        value = decision_payload()
-        value["next_work_units"][0]["fallback_agent"] = "openclaw"
-        with self.assertRaises(ValidationError):
-            Decision.from_dict(value)
-
-    def test_non_dispatch_rejects_work_units(self):
-        with self.assertRaises(ValidationError):
-            Decision.from_dict(decision_payload(verdict="BLOCKED"))
-
-
-
-class ClientTests(unittest.TestCase):
-    def test_workspace_trigger_accepts_202_without_run_id(self):
-        event = Event.from_dict(event_payload())
-        response = mock.MagicMock()
-        response.status = 202
-        response.read.return_value = b""
-        response.__enter__.return_value = response
-        response.__exit__.return_value = False
-        with mock.patch(
-            "services.aios_orchestrator_gateway.clients.urllib.request.urlopen",
-            return_value=response,
-        ) as call:
-            WorkspaceAgentClient("agtch_test", "token").trigger(event)
-        request = call.call_args.args[0]
-        self.assertEqual(request.headers["Idempotency-key"], "evt-001")
-
-    def test_responses_background_start_and_poll(self):
-        event = Event.from_dict(event_payload())
-        decision = decision_payload()
-        client = ResponsesClient("key", "gpt-test")
-        with mock.patch.object(
-            client,
-            "_request",
-            side_effect=[
-                {"id": "resp_123", "status": "queued"},
-                {
-                    "id": "resp_123",
-                    "status": "completed",
-                    "output": [
-                        {
-                            "content": [
-                                {
-                                    "type": "output_text",
-                                    "text": json.dumps(decision),
-                                }
-                            ]
-                        }
-                    ],
-                },
-            ],
-        ):
-            response_id = client.start(event)
-            status, parsed = client.poll(response_id)
-        self.assertEqual(response_id, "resp_123")
-        self.assertEqual(status, "completed")
-        self.assertEqual(parsed.event_id, "evt-001")
-
-    def test_responses_poll_keeps_queued_state(self):
-        client = ResponsesClient("key", "gpt-test")
-        with mock.patch.object(
-            client,
-            "_request",
-            return_value={"id": "resp_123", "status": "in_progress"},
-        ):
-            status, parsed = client.poll("resp_123")
-        self.assertEqual(status, "in_progress")
-        self.assertIsNone(parsed)
-
-    def test_responses_poll_queued_in_progress_completed_sequence(self):
-        client = ResponsesClient("key", "gpt-test")
-        completed = {
-            "status": "completed",
-            "output": [{"content": [{"type": "output_text", "text": json.dumps(decision_payload())}]}],
-        }
-        with mock.patch.object(
-            client,
-            "_request",
-            side_effect=[{"status": "queued"}, {"status": "in_progress"}, completed],
-        ):
-            self.assertEqual(client.poll("resp_123"), ("queued", None))
-            self.assertEqual(client.poll("resp_123"), ("in_progress", None))
-            status, decision = client.poll("resp_123")
-        self.assertEqual(status, "completed")
-        self.assertEqual(decision.event_id, "evt-001")
-
-    def test_responses_invalid_completed_decision_fails_closed(self):
-        client = ResponsesClient("key", "gpt-test")
-        payload = {"status": "completed", "output": [{"content": [{"type": "output_text", "text": "{}"}]}]}
-        with mock.patch.object(client, "_request", return_value=payload):
-            with self.assertRaises(ValidationError):
-                client.poll("resp_123")
-
-
-class StoreTests(unittest.TestCase):
-    def setUp(self):
-        self.temp = tempfile.TemporaryDirectory()
-        self.store = Store(Path(self.temp.name) / "state.sqlite3")
-
-    def tearDown(self):
-        self.store.close()
-        self.temp.cleanup()
-
-    def test_event_idempotency(self):
-        event = Event.from_dict(event_payload())
-        created, _ = self.store.register_event(event)
-        duplicate, _ = self.store.register_event(event)
-        self.assertTrue(created)
-        self.assertFalse(duplicate)
-
-    def test_event_payload_conflict_rejected(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        with self.assertRaises(ConflictError):
-            self.store.register_event(Event.from_dict(event_payload(status="FAIL")))
-
-    def test_stale_state_version_rejected(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        with self.assertRaises(ConflictError):
-            self.store.register_event(
-                Event.from_dict(event_payload(event_id="evt-002", state_version=1))
-            )
-
-    def test_lease_claim_prevents_double_claim(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        first = self.store.claim_events(
-            owner="one", statuses=("RECEIVED",), limit=5, lease_seconds=60
-        )
-        second_store = Store(Path(self.temp.name) / "state.sqlite3")
-        try:
-            second = second_store.claim_events(
-                owner="two", statuses=("RECEIVED",), limit=5, lease_seconds=60
-            )
-        finally:
-            second_store.close()
-        self.assertEqual(len(first), 1)
-        self.assertEqual(second, [])
-
-    def test_workspace_timeout_claim_prevents_double_worker(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        self.store.update_event_trigger("evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None)
-        self.store._conn().execute("UPDATE events SET updated_at=0 WHERE event_id='evt-001'")
-        first = self.store.claim_workspace_timeouts(owner="one", older_than=1, limit=5, lease_seconds=60)
-        second = self.store.claim_workspace_timeouts(owner="two", older_than=1, limit=5, lease_seconds=60)
-        self.assertEqual(len(first), 1)
-        self.assertEqual(second, [])
-
-    def test_response_poll_claim_prevents_double_worker(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        self.store.update_event_trigger("evt-001", status="POLLING_RESPONSE", channel="responses_api", remote_id="resp_123")
-        first = self.store.claim_response_polls(owner="one", limit=5, lease_seconds=60)
-        second = self.store.claim_response_polls(owner="two", limit=5, lease_seconds=60)
-        self.assertEqual(len(first), 1)
-        self.assertEqual(second, [])
-
-    def test_max_hops_limit(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        self.store._conn().execute(
-            """
-            INSERT INTO gate_hops(project_key, gate_id, hops, updated_at)
-            VALUES ('v2r1-personal-weixin', 'R0_RUNNER_LIVE_RECOVERY', 1, 0)
-            """
-        )
-        with self.assertRaisesRegex(ConflictError, "max_hops_exceeded"):
-            self.store.apply_decision(
-                Decision.from_dict(decision_payload()),
-                max_hops_per_gate=1,
-                max_same_blocker_retries=3,
-            )
-
-    def test_hop_limit_is_scoped_per_gate(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        first = Decision.from_dict(decision_payload())
-        self.assertEqual(
-            self.store.apply_decision(
-                first, max_hops_per_gate=1, max_same_blocker_retries=3
-            ),
-            1,
-        )
-
-        self.store.register_event(
-            Event.from_dict(event_payload(event_id="evt-002", state_version=1))
-        )
-        second_value = decision_payload(
-            event_id="evt-002",
-            expected_state_version=1,
-            state_updates=[{"key": "current_gate", "value": "P2_REAL_B3_EVIDENCE"}],
-        )
-        second_value["next_work_units"][0]["id"] = "P2-REVIEW"
-        second_value["next_work_units"][0]["gate_id"] = "P2_REAL_B3_EVIDENCE"
-        self.assertEqual(
-            self.store.apply_decision(
-                Decision.from_dict(second_value),
-                max_hops_per_gate=1,
-                max_same_blocker_retries=3,
-            ),
-            2,
-        )
-
-        self.store.register_event(
-            Event.from_dict(event_payload(event_id="evt-003", state_version=2))
-        )
-        third_value = decision_payload(
-            event_id="evt-003",
-            expected_state_version=2,
-            state_updates=[{"key": "current_gate", "value": "P2_REAL_B3_EVIDENCE"}],
-        )
-        third_value["next_work_units"][0]["id"] = "P2-REVIEW-RETRY"
-        third_value["next_work_units"][0]["gate_id"] = "P2_REAL_B3_EVIDENCE"
-        with self.assertRaisesRegex(ConflictError, "max_hops_exceeded"):
-            self.store.apply_decision(
-                Decision.from_dict(third_value),
-                max_hops_per_gate=1,
-                max_same_blocker_retries=3,
-            )
-
-    def test_decision_advances_state_and_creates_outbox(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        version = self.store.apply_decision(
-            Decision.from_dict(decision_payload()),
-            max_hops_per_gate=20,
-            max_same_blocker_retries=3,
-        )
-        self.assertEqual(version, 1)
-        state = self.store.project_state("v2r1-personal-weixin")
-        self.assertEqual(state["state_version"], 1)
-        rows = self.store.claim_outbox(owner="x", limit=10, lease_seconds=60)
-        self.assertEqual(len(rows), 1)
-
-    def test_decision_idempotency(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        decision = Decision.from_dict(decision_payload())
-        first = self.store.apply_decision(
-            decision, max_hops_per_gate=20, max_same_blocker_retries=3
-        )
-        second = self.store.apply_decision(
-            decision, max_hops_per_gate=20, max_same_blocker_retries=3
-        )
-        self.assertEqual(first, second)
-
-    def test_same_blocker_limit(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        decision = Decision.from_dict(
-            decision_payload(
-                verdict="BLOCKED",
-                next_work_units=[],
-                stop_reason="SAME_BLOCKER",
-            )
-        )
-        self.store.apply_decision(
-            decision, max_hops_per_gate=20, max_same_blocker_retries=1
-        )
-        # New event at next state with the same blocker exceeds the configured limit.
-        self.store.register_event(
-            Event.from_dict(event_payload(event_id="evt-002", state_version=1))
-        )
-        decision2 = Decision.from_dict(
-            decision_payload(
-                event_id="evt-002",
-                expected_state_version=1,
-                verdict="BLOCKED",
-                next_work_units=[],
-                stop_reason="SAME_BLOCKER",
-            )
-        )
-        with self.assertRaises(ConflictError):
-            self.store.apply_decision(
-                decision2, max_hops_per_gate=20, max_same_blocker_retries=1
-            )
-
-    def test_project_summary_contains_counts(self):
-        self.store.register_event(Event.from_dict(event_payload()))
-        summary = self.store.project_summary("v2r1-personal-weixin")
-        self.assertEqual(summary["event_counts"]["RECEIVED"], 1)
-
-
-class DispatchTests(unittest.TestCase):
-    def test_dispatch_is_idempotent(self):
-        with tempfile.TemporaryDirectory() as temp:
-            outbox = Path(temp) / "outbox"
-            queue = Path(temp) / "queue"
-            outbox.mkdir(mode=0o700)
-            queue.mkdir(mode=0o2770)
-            queue.chmod(0o2770)
-            sink = DispatchSink(outbox, queue, CONTRACT)
-            record = {
-                "work_unit_id": "R0-DISCOVER",
-                "event_id": "evt-001",
-                "assigned_agent": "openclaw",
-                "payload_json": json.dumps(decision_payload()["next_work_units"][0]),
-            }
-            first = sink.dispatch(record)
-            second = sink.dispatch(record)
-            self.assertEqual(first, second)
-            self.assertTrue(first[0].exists())
-            self.assertTrue(first[1].exists())
-            self.assertEqual(stat.S_IMODE(first[1].stat().st_mode), 0o640)
-
-    def test_dispatch_rejects_path_escape(self):
-        with tempfile.TemporaryDirectory() as temp:
-            outbox = Path(temp) / "outbox"
-            queue = Path(temp) / "queue"
-            outbox.mkdir(mode=0o700)
-            queue.mkdir(mode=0o2770)
-            queue.chmod(0o2770)
-            sink = DispatchSink(outbox, queue, CONTRACT)
-            with self.assertRaises(DispatchError):
-                sink.dispatch(
-                    {
-                        "work_unit_id": "../escape",
-                        "event_id": "evt-001",
-                        "assigned_agent": "openclaw",
-                        "payload_json": "{}",
-                    }
-                )
-
-
-
-    def test_dispatch_rejects_existing_symlink_target(self):
-        with tempfile.TemporaryDirectory() as temp:
-            outbox = Path(temp) / "outbox"
-            queue = Path(temp) / "queue"
-            outbox.mkdir(mode=0o700)
-            queue.mkdir(mode=0o2770)
-            queue.chmod(0o2770)
-            sink = DispatchSink(outbox, queue, CONTRACT)
-            target = Path(temp) / "target"
-            target.write_text("x")
-            pending = outbox / "dispatched"
-            pending.mkdir(mode=0o700)
-            (pending / "R0-DISCOVER.json").symlink_to(target)
-            with self.assertRaises(DispatchError):
-                sink.dispatch(
-                    {
-                        "work_unit_id": "R0-DISCOVER",
-                        "event_id": "evt-001",
-                        "assigned_agent": "openclaw",
-                        "payload_json": json.dumps(decision_payload()["next_work_units"][0]),
-                    }
-                )
-
-
-class SyncTests(unittest.TestCase):
-    def test_lifecycle_record_is_stable(self):
-        one = lifecycle_record(
-            contract_commit=CONTRACT,
-            project_key="v2r1",
-            kind="TASK_STARTED",
-            work_unit_id="R0",
-            payload={"status": "RUNNING"},
-        )
-        two = lifecycle_record(
-            contract_commit=CONTRACT,
-            project_key="v2r1",
-            kind="TASK_STARTED",
-            work_unit_id="R0",
-            payload={"status": "RUNNING"},
-        )
-        self.assertEqual(one["sync_id"], two["sync_id"])
-
-    def test_file_sync_is_idempotent(self):
-        with tempfile.TemporaryDirectory() as temp:
-            root = Path(temp)
-            root.chmod(0o2750)
-            sync = OpenClawSynchronizer(root)
-            record = lifecycle_record(
-                contract_commit=CONTRACT,
-                project_key="v2r1",
-                kind="PROJECT_SUMMARY",
-                work_unit_id=None,
-                payload={"done": 1},
-            )
-            first = sync.emit(record)
-            second = sync.emit(record)
-            self.assertEqual(first, second)
-
-    def test_http_failure_keeps_file_for_retry(self):
-        with tempfile.TemporaryDirectory() as temp:
-            root = Path(temp)
-            root.chmod(0o2750)
-            sync = OpenClawSynchronizer(root, endpoint="https://openclaw.invalid/events")
-            record = lifecycle_record(
-                contract_commit=CONTRACT,
-                project_key="v2r1",
-                kind="PROJECT_COMPLETED",
-                work_unit_id=None,
-                payload={"done": 1},
-            )
-            with mock.patch(
-                "services.aios_orchestrator_gateway.sync.urllib.request.urlopen",
-                side_effect=OSError("offline"),
-            ):
-                with self.assertRaises(DispatchError):
-                    sync.emit(record)
-            pending = root / "pending" / f"{record['sync_id']}.json"
-            self.assertTrue(pending.exists())
-            response = mock.MagicMock(status=202)
-            response.__enter__.return_value = response
-            response.__exit__.return_value = False
-            with mock.patch(
-                "services.aios_orchestrator_gateway.sync.urllib.request.urlopen",
-                return_value=response,
-            ):
-                self.assertEqual(sync.emit(record), pending)
-
-
-class ReporterTests(unittest.TestCase):
-    def test_http_failure_falls_back_to_file_inbox(self):
-        with tempfile.TemporaryDirectory() as temp:
-            root = Path(temp)
-            inbox = root / "inbox"
-            inbox.mkdir(mode=0o2770)
-            inbox.chmod(0o2770)
-            token = root / "token"
-            token.write_text("not-a-real-secret", encoding="utf-8")
-            token.chmod(0o600)
-            argv = [
-                "report_orchestrator_event.py",
-                "--project-key", "v2r1-personal-weixin",
-                "--contract-commit", CONTRACT,
-                "--event-id", "evt-http-fallback",
-                "--source-agent", "codex",
-                "--event-type", "work_unit_completed",
-                "--work-unit-id", "P1-REACK",
-                "--status", "PASS",
-                "--state-version", "0",
-                "--endpoint", "https://gateway.invalid",
-                "--token-file", str(token),
-                "--inbox-dir", str(inbox),
-            ]
-            with mock.patch("sys.argv", argv), mock.patch(
-                "scripts.report_orchestrator_event.urllib.request.urlopen",
-                side_effect=urllib.error.URLError("offline"),
-            ):
-                self.assertEqual(report_orchestrator_event.main(), 0)
-            pending = inbox / "pending/evt-http-fallback.json"
-            self.assertTrue(pending.exists())
-            self.assertEqual(stat.S_IMODE(pending.stat().st_mode), 0o640)
-
-    def test_payload_symlink_is_rejected(self):
-        with tempfile.TemporaryDirectory() as temp:
-            root = Path(temp)
-            target = root / "payload.json"
-            target.write_text("{}", encoding="utf-8")
-            target.chmod(0o600)
-            link = root / "payload-link.json"
-            link.symlink_to(target)
-            argv = [
-                "report_orchestrator_event.py",
-                "--project-key", "v2r1-personal-weixin",
-                "--contract-commit", CONTRACT,
-                "--event-id", "evt-symlink",
-                "--source-agent", "codex",
-                "--event-type", "work_unit_completed",
-                "--work-unit-id", "P1-REACK",
-                "--status", "PASS",
-                "--state-version", "0",
-                "--payload-file", str(link),
-                "--inbox-dir", str(root / "inbox"),
-            ]
-            with mock.patch("sys.argv", argv), self.assertRaisesRegex(
-                Exception, "symlink_rejected"
-            ):
-                report_orchestrator_event.main()
-
-
-class RuntimePollingTests(unittest.TestCase):
-    def setUp(self):
-        self.temp = tempfile.TemporaryDirectory()
-        self.root = Path(self.temp.name)
-        for name in ("events", "decisions", "outbox", "etc"):
-            (self.root / name).mkdir(mode=0o700)
-        for name in ("sync", "commands"):
-            (self.root / name).mkdir(mode=0o2750)
-            (self.root / name).chmod(0o2750)
-        for name, value in (
-            ("inbound.token", "inbound"),
-            ("callback.token", "callback"),
-            ("workspace.token", "workspace"),
-        ):
-            path = self.root / "etc" / name
-            path.write_text(value)
-            path.chmod(0o600)
-        config = {
-            "host": "127.0.0.1",
-            "port": 18920,
-            "project_key": "v2r1-personal-weixin-production",
-            "database_path": str(self.root / "state.sqlite3"),
-            "event_inbox_dir": str(self.root / "events"),
-            "decision_inbox_dir": str(self.root / "decisions"),
-            "dispatch_outbox_dir": str(self.root / "outbox"),
-            "openclaw_sync_dir": str(self.root / "sync"),
-            "command_queue_dir": str(self.root / "commands"),
-            "inbound_token_file": str(self.root / "etc/inbound.token"),
-            "callback_token_file": str(self.root / "etc/callback.token"),
-            "contract_commit": CONTRACT,
-            "workspace_agent": {
-                "trigger_id": "agtch_test",
-                "access_token_file": str(self.root / "etc/workspace.token"),
-            },
-            "responses_fallback": {
-                "api_key_file": None,
-                "model": None,
-            },
-            "polling": {
-                "interval_seconds": 1,
-                "summary_interval_seconds": 10,
-            },
-            "loop_controls": {
-                "callback_timeout_seconds": 30,
-                "lease_seconds": 10,
-                "retry_base_seconds": 1,
-                "max_retries": 2,
-                "max_same_blocker_retries": 2,
-                "max_hops_per_gate": 20,
-            },
-            "openclaw_sync": {"endpoint": None, "token_file": None},
-            "git_poll": {
-                "enabled": False,
-                "bare_repo": None,
-                "ref_prefixes": [],
-                "interval_seconds": 30,
-            },
-        }
-        config_path = self.root / "etc/config.json"
-        config_path.write_text(json.dumps(config))
-        config_path.chmod(0o600)
-        os.environ["AIOS_ORCHESTRATOR_CONFIG_OWNER_UID"] = str(os.getuid())
-        os.environ["AIOS_ORCHESTRATOR_SECRET_OWNER_UID"] = str(os.getuid())
-        self.runtime = Runtime(Config.from_file(config_path))
-
-    def tearDown(self):
-        self.runtime.store.close()
-        self.temp.cleanup()
-        os.environ.pop("AIOS_ORCHESTRATOR_CONFIG_OWNER_UID", None)
-        os.environ.pop("AIOS_ORCHESTRATOR_SECRET_OWNER_UID", None)
-
-    def test_file_event_polling_and_workspace_trigger(self):
-        pending = self.root / "events/pending"
-        pending.mkdir(mode=0o2770)
-        pending.chmod(0o2770)
-        (pending / "evt-001.json").write_text(json.dumps(event_payload()))
-        (pending / "evt-001.json").chmod(0o640)
-        with mock.patch.object(self.runtime.workspace, "trigger") as trigger:
-            self.runtime.poll_once()
-        trigger.assert_called_once()
-        processed = self.root / "events/processed/evt-001.json"
-        self.assertTrue(processed.exists())
-
-    def test_workspace_timeout_uses_responses_fallback(self):
-        self.runtime.submit_event(event_payload())
-        self.runtime.store.update_event_trigger(
-            "evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None
-        )
-        self.runtime.store._conn().execute("UPDATE events SET updated_at=0 WHERE event_id='evt-001'")
-        fallback = mock.Mock()
-        fallback.start.return_value = "resp_fallback"
-        self.runtime.responses = fallback
-        self.runtime.process_workspace_timeouts()
-        row = self.runtime.store._conn().execute(
-            "SELECT status, channel, remote_id FROM events WHERE event_id='evt-001'"
-        ).fetchone()
-        self.assertEqual(tuple(row), ("POLLING_RESPONSE", "responses_api", "resp_fallback"))
-        fallback.start.assert_called_once()
-
-    def test_workspace_callback_timeout_reaches_dead_letter_without_fallback(self):
-        self.runtime.submit_event(event_payload())
-        self.runtime.store.update_event_trigger(
-            "evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None
-        )
-        self.runtime.store._conn().execute(
-            "UPDATE events SET attempts=2, updated_at=0 WHERE event_id='evt-001'"
-        )
-        self.assertIsNone(self.runtime.responses)
-        self.runtime.process_workspace_timeouts()
-        row = self.runtime.store._conn().execute(
-            "SELECT status, attempts FROM events WHERE event_id='evt-001'"
-        ).fetchone()
-        self.assertEqual(tuple(row), ("DEAD_LETTER", 3))
-
-    def test_response_poll_limit_dead_letters_queued_response(self):
-        self.runtime.submit_event(event_payload())
-        self.runtime.store.update_event_trigger(
-            "evt-001",
-            status="POLLING_RESPONSE",
-            channel="responses_api",
-            remote_id="resp-stuck",
-        )
-        self.runtime.store._conn().execute(
-            "UPDATE events SET attempts=2, next_attempt_at=0 WHERE event_id='evt-001'"
-        )
-        fallback = mock.Mock()
-        fallback.poll.return_value = ("queued", None)
-        self.runtime.responses = fallback
-        self.runtime.poll_background_responses()
-        row = self.runtime.store._conn().execute(
-            "SELECT status, attempts FROM events WHERE event_id='evt-001'"
-        ).fetchone()
-        self.assertEqual(tuple(row), ("DEAD_LETTER", 3))
-
-    def test_callback_rejects_wrong_contract(self):
-        self.runtime.submit_event(event_payload())
-        with self.assertRaisesRegex(ConflictError, "contract_version_mismatch"):
-            self.runtime.submit_decision(
-                decision_payload(contract_commit="f" * 40)
-            )
-
-    def test_task_start_and_end_are_synced(self):
-        start = Event.from_dict(
-            event_payload(
-                event_id="evt-start",
-                event_type="work_unit_started",
-                status="RUNNING",
-            )
-        )
-        self.runtime.submit_event(start.to_dict())
-        self.runtime.synchronize_openclaw()
-        pending = list((self.root / "sync/pending").glob("*.json"))
-        self.assertTrue(pending)
-        kinds = {json.loads(p.read_text())["kind"] for p in pending}
-        self.assertIn("TASK_STARTED", kinds)
-
-    def test_decision_dispatch_and_summary_sync(self):
-        self.runtime.submit_event(event_payload())
-        self.runtime.submit_decision(decision_payload())
-        self.runtime.dispatch_work_units()
-        self.runtime.synchronize_openclaw()
-        self.assertTrue(list((self.root / "commands/openclaw/pending").glob("*.json")))
-        kinds = {
-            json.loads(p.read_text())["kind"]
-            for p in (self.root / "sync/pending").glob("*.json")
-        }
-        self.assertIn("TASK_DISPATCHED", kinds)
-        self.assertIn("PROJECT_SUMMARY", kinds)
-
-    def test_poll_once_ingests_decision_file(self):
-        self.runtime.submit_event(event_payload())
-        pending = self.root / "decisions/pending"
-        pending.mkdir(mode=0o2770)
-        pending.chmod(0o2770)
-        (pending / "decision.json").write_text(json.dumps(decision_payload()))
-        (pending / "decision.json").chmod(0o640)
-        self.runtime.poll_once()
-        self.assertTrue((self.root / "decisions/processed/decision.json").exists())
-
-
-    def test_git_ref_polling_baselines_then_emits_change(self):
-        object.__setattr__(self.runtime.config, "git_poll_enabled", True)
-        object.__setattr__(self.runtime.config, "git_poll_bare_repo", self.root / "repo.git")
-        object.__setattr__(self.runtime.config, "git_poll_ref_prefixes", ("refs/heads/main",))
-        oid1 = "1" * 40
-        oid2 = "2" * 40
-        first = mock.Mock(returncode=0, stdout=f"refs/heads/main {oid1}\n", stderr="")
-        second = mock.Mock(returncode=0, stdout=f"refs/heads/main {oid2}\n", stderr="")
-        with mock.patch(
-            "services.aios_orchestrator_gateway.service.subprocess.run",
-            side_effect=[first, second],
-        ):
-            self.runtime._next_git_poll = 0
-            self.runtime.poll_git_refs()
-            self.runtime._next_git_poll = 0
-            self.runtime.poll_git_refs()
-        summary = self.runtime.store.project_summary(
-            "v2r1-personal-weixin-production"
-        )
-        self.assertIsNotNone(summary)
-        self.assertEqual(summary["event_counts"]["RECEIVED"], 1)
-
-    def test_retry_backoff_is_bounded(self):
-        self.assertEqual(self.runtime._backoff(0), 1)
-        self.assertLessEqual(self.runtime._backoff(100), 3600)
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -1,59 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-umask 0077
-
-SOURCE_ROOT="${1:?source root required}"
-ROOT="$(mktemp -d /tmp/aios-orchestrator-rollback-test.XXXXXX)"
-trap 'rm -rf "$ROOT"' EXIT
-INSTALL_ROOT="$ROOT/opt"
-STATE_ROOT="$ROOT/state"
-mkdir -p "$INSTALL_ROOT/releases" "$STATE_ROOT" "$ROOT/bin"
-chmod 0700 "$STATE_ROOT"
-
-OLD="$(printf 'a%.0s' {1..64})"
-NEW="$(printf 'b%.0s' {1..64})"
-mkdir "$INSTALL_ROOT/releases/$OLD" "$INSTALL_ROOT/releases/$NEW"
-printf '%s\n' "$OLD" > "$INSTALL_ROOT/releases/$OLD/SOURCE_DIGEST"
-printf '%s\n' "$NEW" > "$INSTALL_ROOT/releases/$NEW/SOURCE_DIGEST"
-ln -s "$INSTALL_ROOT/releases/$NEW" "$INSTALL_ROOT/current"
-python3 - "$STATE_ROOT/install-receipt.json" "$INSTALL_ROOT/releases/$NEW" "$INSTALL_ROOT/releases/$OLD" <<'PY'
-import json, sys
-with open(sys.argv[1], "w", encoding="utf-8") as handle:
-    json.dump({
-        "schema_version": 1,
-        "source_digest": "b" * 64,
-        "current_release": sys.argv[2],
-        "previous_release": sys.argv[3],
-        "last_rollback": None,
-    }, handle)
-PY
-chmod 0600 "$STATE_ROOT/install-receipt.json"
-cat > "$ROOT/bin/systemctl" <<'SH'
-#!/bin/sh
-printf '%s\n' "$*" >> "${SYSTEMCTL_LOG:?}"
-exit 0
-SH
-chmod 0755 "$ROOT/bin/systemctl"
-: > "$ROOT/systemctl.log"
-
-SYSTEMCTL_LOG="$ROOT/systemctl.log" PATH="$ROOT/bin:/usr/bin:/bin" INSTALL_ROOT="$INSTALL_ROOT" STATE_ROOT="$STATE_ROOT" \
-  bash "$SOURCE_ROOT/scripts/rollback_aios_orchestrator_gateway.sh"
-FIRST="$(readlink -f "$INSTALL_ROOT/current")"
-SYSTEMCTL_LOG="$ROOT/systemctl.log" PATH="$ROOT/bin:/usr/bin:/bin" INSTALL_ROOT="$INSTALL_ROOT" STATE_ROOT="$STATE_ROOT" \
-  bash "$SOURCE_ROOT/scripts/rollback_aios_orchestrator_gateway.sh"
-SECOND="$(readlink -f "$INSTALL_ROOT/current")"
-[[ "$FIRST" == "$INSTALL_ROOT/releases/$OLD" ]]
-[[ "$SECOND" == "$FIRST" ]]
-grep -Fxq 'stop aios-orchestrator-gateway-poll.timer' "$ROOT/systemctl.log"
-grep -Fxq 'stop aios-orchestrator-gateway-poll.service' "$ROOT/systemctl.log"
-grep -Fxq 'start aios-orchestrator-gateway-poll.timer' "$ROOT/systemctl.log"
-
-printf 'tampered\n' > "$INSTALL_ROOT/releases/$OLD/SOURCE_DIGEST"
-if SYSTEMCTL_LOG="$ROOT/systemctl.log" PATH="$ROOT/bin:/usr/bin:/bin" \
-  INSTALL_ROOT="$INSTALL_ROOT" STATE_ROOT="$STATE_ROOT" \
-  bash "$SOURCE_ROOT/scripts/rollback_aios_orchestrator_gateway.sh" >/dev/null 2>&1; then
-  echo 'ROLLBACK_DIGEST_MISMATCH_TEST=FAIL' >&2
-  exit 1
-fi
-echo "ROLLBACK_DIGEST_MISMATCH_TEST=PASS"
-echo "ROLLBACK_REPEAT_TEST=PASS"
@@ -1,31 +0,0 @@
-# AIOS Central Orchestrator Workspace Agent
-
-Contract commit: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
-
-When an API-triggered input arrives:
-
-1. Parse the AIOS event JSON from the trigger message.
-2. Verify `contract_commit` exactly matches the configured contract commit.
-3. Use connected GitOps tools to read the authoritative Git tree, runtime status,
-   gate state, test evidence and full OIDs.
-4. Never invent evidence, credentials, secrets, tests, approvals, Git results,
-   model availability or deployment status.
-5. Apply the adaptive dispatch contract:
-   - Prefer OpenClaw when quality is comparable, work is large/repetitive,
-     monitoring-heavy, infrastructure-heavy or cost-sensitive.
-   - Prefer the strongest actually available Codex model for complex code,
-     security boundaries, architecture and final high-risk review.
-   - On timeout/unavailability, mark the attempt `TIMEOUT` or `NOT_EXECUTED`
-     and assign the same contract to the fallback agent.
-   - Never infer PASS from timeout or missing output.
-6. Every next work unit must include a clean base commit, allowed paths, required
-   checks, forbidden operations, primary agent and fallback agent.
-7. Keep real credential, raw-wire, reviewer, secret-provider and owner-only
-   release gates fail-closed.
-8. Submit exactly one strict decision through the
-   `submit_orchestration_decision` action.
-9. Do not write a second decision for the same event ID.
-10. Stop after the action succeeds.
-
-The decision must satisfy the JSON schema in the connected action. Use the
-incoming `event_id`, `project_key`, `contract_commit`, and `state_version`.
@@ -1,114 +0,0 @@
-openapi: 3.1.0
-info:
-  title: AIOS Orchestrator Decision Callback
-  version: 1.0.0
-servers:
-  - url: https://REPLACE_WITH_GATEWAY_HOST
-paths:
-  /v1/orchestrator/decisions:
-    post:
-      operationId: submit_orchestration_decision
-      summary: Submit one strict orchestration decision for a triggered event
-      security:
-        - bearerAuth: []
-      requestBody:
-        required: true
-        content:
-          application/json:
-            schema:
-              $ref: '#/components/schemas/Decision'
-      responses:
-        '202':
-          description: Accepted
-        '409':
-          description: Contract, idempotency, or state version conflict
-        '422':
-          description: Schema validation failed
-components:
-  securitySchemes:
-    bearerAuth:
-      type: http
-      scheme: bearer
-  schemas:
-    WorkUnit:
-      type: object
-      additionalProperties: false
-      required:
-        - id
-        - track_id
-        - gate_id
-        - primary_agent
-        - fallback_agent
-        - clean_base_commit
-        - allowed_paths
-        - required_checks
-        - forbidden_operations
-        - instruction
-      properties:
-        id: {type: string}
-        track_id: {type: string}
-        gate_id: {type: string}
-        primary_agent:
-          type: string
-          enum: [openclaw, codex]
-        fallback_agent:
-          type: string
-          enum: [openclaw, codex]
-        clean_base_commit:
-          type: string
-          pattern: '^[0-9a-f]{40}$'
-        allowed_paths:
-          type: array
-          items: {type: string}
-        required_checks:
-          type: array
-          items: {type: string}
-        forbidden_operations:
-          type: array
-          items: {type: string}
-        instruction: {type: string}
-    Decision:
-      type: object
-      additionalProperties: false
-      required:
-        - schema_version
-        - project_key
-        - contract_commit
-        - event_id
-        - expected_state_version
-        - verdict
-        - state_updates
-        - next_work_units
-        - requires_human_input
-        - stop_reason
-      properties:
-        schema_version:
-          type: integer
-          const: 1
-        project_key: {type: string}
-        contract_commit:
-          type: string
-          pattern: '^[0-9a-f]{40}$'
-        event_id: {type: string}
-        expected_state_version:
-          type: integer
-          minimum: 0
-        verdict:
-          type: string
-          enum: [DISPATCH, BLOCKED, COMPLETE, NOOP]
-        state_updates:
-          type: array
-          items:
-            type: object
-            additionalProperties: false
-            required: [key, value]
-            properties:
-              key: {type: string}
-              value: {type: string}
-        next_work_units:
-          type: array
-          items:
-            $ref: '#/components/schemas/WorkUnit'
-        requires_human_input: {type: boolean}
-        stop_reason:
-          type: [string, 'null']
Author	SHA1	Message	Date
Codex	c9c18cb8e3	feat(gitops): add isolated detached worktree manager	2026-07-20 00:42:56 +08:00
Codex	c4e587d9cd	docs(aios-v2): add CTO implementation status and gap analysis	2026-07-20 00:25:57 +08:00
Codex	5f2f948b92	create branch for AIOS V2 implementation status documentation	2026-07-20 00:21:00 +08:00