docs(codex): record A0 automation closure result

fix(codex): isolate watcher runtime directory ownership
feat(codex): add persistent automatic command watcher
2026-07-18 03:48:54 +08:00 · 2026-07-18 03:39:45 +08:00 · 2026-07-18 03:36:46 +08:00 · 2026-07-18 03:01:30 +08:00 · 2026-07-18 03:00:32 +08:00 · 2026-07-18 02:26:14 +08:00
52 changed files with 7149 additions and 1 deletions
@@ -0,0 +1,32 @@
+{
+  "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"
+    }
+  ]
+}
@@ -0,0 +1,157 @@
+{
+  "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"
+      ]
+    }
+  }
+}
@@ -0,0 +1,55 @@
+---
+contract: cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
+work_unit: A0-CODEX-AUTOMATION-CLOSURE
+status: PASS
+base_oid: 9195470ceb78fd57ffface3dcc38f40b75efa9d8
+baseline_remediation_commit: 378cc4d56ab58dde7da8814846ffd64a979b23f6
+orchestrator_implementation_oid: 95fc4d37f0b9502945e5ba97a1f8e5a146768f6b
+integration_merge_commit: 1b3b74fd626028ae5d5f11ae6933953fecf57df5
+implementation_commit: c3984ca2b4388875d2875b4fdfdf95a2984f6ab3
+full_unittest: PASS_78
+full_pytest: PASS_154_PLUS_34_SUBTESTS
+python_compile: PASS
+bash_syntax: PASS
+json_schema_openapi: PASS
+systemd_verify: PASS
+gitleaks_final_tree: PASS
+gitleaks_head_history: PASS_297_COMMITS
+gitleaks_implementation_range: PASS
+codex_auto_claim: PASS
+worktree_oid: c3984ca2b4388875d2875b4fdfdf95a2984f6ab3
+bare_oid: c3984ca2b4388875d2875b4fdfdf95a2984f6ab3
+gitea_oid: c3984ca2b4388875d2875b4fdfdf95a2984f6ab3
+all_equal: true
+---
+
+# A0 Codex automation closure result
+
+## A0-C1 baseline remediation
+
+Lineage showed that `tests/test_wechat_send_backend.py` entered the authoritative ancestry without its source, while no source-deletion commit existed. The approved hardened source/test blobs were restored through their direct lineage. Relevant unittest and pytest each passed 26 tests; the remediation commit was pushed with equal worktree, bare, and Gitea OIDs.
+
+## A0-C2 Gateway integration
+
+The remediation commit and Gateway implementation were combined by a normal two-parent merge. Gateway tests passed, as did the repository suites and the compile, Bash, JSON/schema, OpenAPI, systemd, diff, rollback, and Gitleaks gates.
+
+## A0-C3 persistent Codex watcher
+
+The watcher is installed as a five-second systemd timer under `aios-orchestrator`. It uses a SQLite unique `work_unit_id`, optimistic atomic leases, stale-lease recovery, phase persistence, bounded retries, dead-letter handling, exact contract rejection, durable idempotent lifecycle events, sanitized executor environment, secure no-follow file reads, and a loopback-only Gateway endpoint. External execution is at-most-once: an ambiguous post-launch crash is not replayed.
+
+The first installer run failed closed because systemd-tmpfiles rejected an ownership transition into the Gateway command tree. No task ran. The append-only repair limited tmpfiles to watcher-owned `/var` paths and validated the existing command-tree owner and modes before creating only the absent dead-letter directory.
+
+The harmless live command was created in the production queue and claimed by the timer without manually starting the service. Watcher SQLite recorded `COMPLETED`, `DONE`, `attempts=1`; Gateway SQLite contained exact matching `work_unit_started` and `work_unit_ended` event JSON. The audit chain was:
+
+```text
+COMMAND_CREATED
+-> CODEX_AUTO_CLAIM
+-> TASK_STARTED
+-> TASK_EXECUTED
+-> TASK_ENDED
+-> EVENT_POSTED
+```
+
+The acceptance result retained only an input SHA-256 and exit code. A second live command with a wrong contract OID was moved to dead-letter, created no watcher command row, and posted no Gateway event. Journal scanning found no bearer token, API key, or secret pattern. The superseded broken watcher was disabled; the new timer is enabled and active.
+
+The full all-ref Gitleaks diagnostic found four pre-existing findings only on unrelated refs (three governance/result documents, outside this ancestry). They were not treated as evidence for this work unit. The final tree, the complete `HEAD` ancestry, and the implementation range contained zero findings.
@@ -0,0 +1,15 @@
+{
+  "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"
+}
@@ -0,0 +1,69 @@
+# 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`
@@ -0,0 +1,13 @@
+# 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.
@@ -0,0 +1,68 @@
+# 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.
@@ -0,0 +1,13 @@
+# 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,11 @@
+# AIOS Codex automatic command watcher
+
+The watcher polls the governed `commands/active` queue without a chat trigger. It uses SQLite transactions for exclusive leases, records the exact command digest, rejects contract drift, retries transient failures, dead-letters exhausted work, and resumes from the last durable phase so an end-event retry does not execute a work unit twice.
+
+Commands are JSON documents validated by `schemas/codex-command.schema.json`. `acceptance_probe` is the only built-in action and has no external side effect. Production `codex` work requires an administrator-provisioned absolute executor path; the instruction is passed on standard input, never argv or environment. The executor receives a sanitized environment and must obtain credentials through its approved file or exec provider. If the executor is absent, the task fails closed and follows the retry/dead-letter policy.
+
+The watcher provides at-most-once execution. It writes the `EXECUTING` phase before launching an external executor; if the host dies after an external effect but before a result is durably recorded, recovery does not replay the work and instead dead-letters the ambiguity for human review. Start and end event JSON is persisted before transmission, so response-loss retries reuse the exact event ID and payload and are deduplicated by the Gateway.
+
+Lifecycle evidence is appended to `/var/log/aios-codex-watcher/audit.jsonl` as `CODEX_AUTO_CLAIM`, `TASK_STARTED`, `TASK_EXECUTED`, `TASK_ENDED`, and `EVENT_POSTED`. Inputs, tokens, stdout, and stderr are not logged; only result digests are retained.
+
+Install from a reviewed commit with `sudo scripts/install_aios_codex_watcher.sh`. The installer creates a content-addressed release, atomically updates `current`, writes a receipt, and enables the five-second systemd timer. The existing Gateway inbound token file is referenced in place and never copied.
@@ -0,0 +1,107 @@
+# 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,17 @@
+{
+  "active_dir": "/data/aios-gitops/workspace/agents/commands/active",
+  "completed_dir": "/data/aios-gitops/workspace/agents/commands/completed",
+  "dead_letter_dir": "/data/aios-gitops/workspace/agents/commands/dead-letter",
+  "database_path": "/var/lib/aios-codex-watcher/state.sqlite3",
+  "audit_log_path": "/var/log/aios-codex-watcher/audit.jsonl",
+  "gateway_endpoint": "http://127.0.0.1:18920/v1/orchestrator/events",
+  "gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
+  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
+  "lease_seconds": 60,
+  "retry_seconds": 10,
+  "max_attempts": 5,
+  "poll_seconds": 5,
+  "executor_path": null,
+  "executor_args": [],
+  "executor_timeout_seconds": 1800
+}
@@ -0,0 +1,49 @@
+{
+  "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,25 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://aios.local/schemas/codex-command.schema.json",
+  "title": "AIOS Codex command",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["schema_version", "command_id", "project_key", "work_unit_id", "contract_commit", "state_version", "execution"],
+  "properties": {
+    "schema_version": {"const": 1},
+    "command_id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
+    "project_key": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
+    "work_unit_id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
+    "contract_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "state_version": {"type": "integer", "minimum": 0},
+    "execution": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["kind", "input"],
+      "properties": {
+        "kind": {"enum": ["acceptance_probe", "codex"]},
+        "input": {"type": "string", "maxLength": 65536}
+      }
+    }
+  }
+}
@@ -0,0 +1,145 @@
+{
+  "$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"
+      ]
+    }
+  }
+}
@@ -0,0 +1,102 @@
+{
+  "$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"
+    }
+  }
+}
@@ -0,0 +1,63 @@
+#!/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())
@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+[[ ${EUID} -eq 0 ]] || { echo "root required" >&2; exit 1; }
+ROOT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd -P)
+CONFIG=${1:-/etc/aios-codex-watcher/config.json}
+[[ "$CONFIG" = /* ]] || { echo "config path must be absolute" >&2; exit 1; }
+[[ -f "$CONFIG" && ! -L "$CONFIG" ]] || { echo "pre-provisioned regular config required" >&2; exit 1; }
+
+for path in \
+  services/aios_codex_watcher/__init__.py \
+  services/aios_codex_watcher/__main__.py \
+  services/aios_codex_watcher/watcher.py \
+  systemd/aios-codex-watcher.service \
+  systemd/aios-codex-watcher.timer \
+  systemd/aios-codex-watcher.sysusers \
+  systemd/aios-codex-watcher.tmpfiles; do
+  [[ -f "$ROOT/$path" && ! -L "$ROOT/$path" ]] || { echo "missing or unsafe: $path" >&2; exit 1; }
+done
+
+digest=$(sha256sum \
+  "$ROOT/services/aios_codex_watcher/__init__.py" \
+  "$ROOT/services/aios_codex_watcher/__main__.py" \
+  "$ROOT/services/aios_codex_watcher/watcher.py" | sha256sum | awk '{print $1}')
+release="/opt/aios-codex-watcher/releases/$digest"
+install -d -o root -g root -m 0755 "$release/services/aios_codex_watcher" "$release/bin"
+install -o root -g root -m 0644 "$ROOT/services/aios_codex_watcher/"*.py "$release/services/aios_codex_watcher/"
+
+launcher="$release/bin/aios-codex-watcher"
+printf '%s\n' '#!/usr/bin/env bash' 'set -euo pipefail' \
+  "export PYTHONPATH='$release'" \
+  "exec /usr/bin/python3 -m services.aios_codex_watcher \"\$@\"" >"$launcher"
+chown root:root "$launcher"
+chmod 0755 "$launcher"
+
+install -d -o root -g root -m 0755 /opt/aios-codex-watcher /etc/aios-codex-watcher
+ln -sfn "releases/$digest" /opt/aios-codex-watcher/current.new
+mv -Tf /opt/aios-codex-watcher/current.new /opt/aios-codex-watcher/current
+install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.service" /etc/systemd/system/
+install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.timer" /etc/systemd/system/
+install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.sysusers" /usr/lib/sysusers.d/
+install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.tmpfiles" /usr/lib/tmpfiles.d/
+systemd-sysusers /usr/lib/sysusers.d/aios-codex-watcher.sysusers
+systemd-tmpfiles --create /usr/lib/tmpfiles.d/aios-codex-watcher.tmpfiles
+commands=/data/aios-gitops/workspace/agents/commands
+for directory in "$commands/active" "$commands/completed"; do
+  [[ -d "$directory" && ! -L "$directory" ]] || { echo "missing or unsafe command directory: $directory" >&2; exit 1; }
+  [[ $(stat -c '%U:%G' "$directory") == aios-orchestrator:aios-orchestrator ]] || { echo "unexpected command directory owner: $directory" >&2; exit 1; }
+  mode=$(stat -c '%a' "$directory")
+  (( (8#$mode & 2) == 0 )) || { echo "world-writable command directory: $directory" >&2; exit 1; }
+done
+if [[ -e "$commands/dead-letter" || -L "$commands/dead-letter" ]]; then
+  [[ -d "$commands/dead-letter" && ! -L "$commands/dead-letter" ]] || { echo "unsafe dead-letter directory" >&2; exit 1; }
+else
+  install -d -o aios-orchestrator -g aios-orchestrator -m 0750 "$commands/dead-letter"
+fi
+[[ $(stat -c '%U:%G' "$commands/dead-letter") == aios-orchestrator:aios-orchestrator ]] || { echo "unexpected dead-letter owner" >&2; exit 1; }
+mode=$(stat -c '%a' "$commands/dead-letter")
+(( (8#$mode & 2) == 0 )) || { echo "world-writable dead-letter directory" >&2; exit 1; }
+printf '{"release_sha256":"%s","installed_at":"%s"}\n' "$digest" "$(date -u +%FT%TZ)" >"$release/receipt.json"
+chmod 0644 "$release/receipt.json"
+systemctl daemon-reload
+systemctl enable --now aios-codex-watcher.timer
+printf 'installed_sha256=%s\n' "$digest"
@@ -0,0 +1,239 @@
+#!/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}"
@@ -0,0 +1,130 @@
+#!/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())
@@ -0,0 +1,102 @@
+#!/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"
@@ -0,0 +1,32 @@
+#!/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"
@@ -0,0 +1,652 @@
+#!/usr/bin/env python3
+"""Hardened WeChat outbound backend for AIOS.
+
+Only the dedicated UCC adapter is a supported send backend.  OpenClaw channel
+discovery is retained for diagnostics, but generic OpenClaw chat or agent
+delivery is never promoted to WeChat delivery.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import selectors
+import signal
+import subprocess
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any, Dict, List, Optional, Tuple
+
+
+OPENCLAW_BIN = os.environ.get("AIOS_OPENCLAW_BIN", "openclaw")
+WECHAT_ACCOUNT_ID = os.environ.get("AIOS_WECHAT_ACCOUNT_ID", "")
+WECHAT_CHANNEL = os.environ.get("AIOS_WECHAT_CHANNEL", "")
+SEND_BACKEND = os.environ.get("AIOS_WECHAT_SEND_BACKEND", "")
+UCC_ADAPTER_URL = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_URL", "http://127.0.0.1:18800")
+UCC_ADAPTER_TOKEN = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_TOKEN") or os.environ.get(
+    "AIOS_WECHAT_QR_ADAPTER_TOKEN", ""
+)
+UCC_SEND_PATH = os.environ.get("AIOS_UCC_WECHAT_SEND_PATH", "/api/internal/wechat/send")
+_allowed_origins = os.environ.get("AIOS_UCC_WECHAT_ALLOWED_ORIGINS", "").strip()
+UCC_ADAPTER_ALLOWED_ORIGINS = tuple(
+    origin.strip() for origin in _allowed_origins.split(",") if origin.strip()
+) or (
+    "http://127.0.0.1:18800",
+    "http://[::1]:18800",
+)
+
+MAX_TARGET_BYTES = 512
+MAX_CONTENT_BYTES = 32 * 1024
+MAX_IDENTIFIER_BYTES = 512
+MAX_METADATA_BYTES = 8 * 1024
+MAX_METADATA_DEPTH = 32
+MAX_METADATA_NODES = 1024
+MAX_RESPONSE_BYTES = 64 * 1024
+MAX_ENDPOINT_BYTES = 256
+_DEFAULT_TIMEOUT_SECONDS = 10
+_MAX_TIMEOUT_SECONDS = 30
+_LEGACY_SEND_PATH = "/internal/channels/wechat/send"
+_SAFE_PATH = re.compile(r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]+$")
+_MAX_PATH_DECODE_ROUNDS = 4
+_OPENCLAW_TIMEOUT_RETURN_CODE = 124
+_OPENCLAW_OUTPUT_LIMIT_RETURN_CODE = 125
+_OPENCLAW_SPAWN_ERROR_RETURN_CODE = 126
+
+
+def _bounded_timeout(value: str) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return _DEFAULT_TIMEOUT_SECONDS
+    return min(max(parsed, 1), _MAX_TIMEOUT_SECONDS)
+
+
+REQUEST_TIMEOUT_SECONDS = _bounded_timeout(os.environ.get("AIOS_UCC_WECHAT_TIMEOUT_SECONDS", "10"))
+
+
+class BackendResult:
+    def __init__(self, ok: bool, backend: str, status: str, data: Dict[str, Any]) -> None:
+        self.ok = ok
+        self.backend = backend
+        self.status = status
+        self.data = data
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"ok": self.ok, "backend": self.backend, "status": self.status, **self.data}
+
+
+class _BackendProtocolError(Exception):
+    """Internal marker whose details must never cross the module boundary."""
+
+
+def _openclaw_base_args() -> List[str]:
+    """Return the OpenClaw 2026.6.11-compatible command prefix.
+
+    Gateway/auth global flags are not supported by the installed CLI and would
+    also expose credentials through the process argument vector.
+    """
+
+    return [OPENCLAW_BIN]
+
+
+def _terminate_process_tree(process: Any) -> None:
+    if os.name == "posix":
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except OSError:
+            pass
+        try:
+            process.wait(timeout=0.5)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        time.sleep(0.05)
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except OSError:
+            pass
+        try:
+            process.wait(timeout=0.5)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+        return
+    if process.poll() is not None:
+        return
+    try:
+        process.terminate()
+        process.wait(timeout=1)
+        return
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+    try:
+        process.kill()
+        process.wait(timeout=1)
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+
+
+def _run_openclaw(
+    args: List[str], timeout: int = 15
+) -> subprocess.CompletedProcess[str]:
+    bounded_timeout = min(max(int(timeout), 1), _MAX_TIMEOUT_SECONDS)
+    popen_options: Dict[str, Any] = {
+        "stdout": subprocess.PIPE,
+        "stderr": subprocess.PIPE,
+    }
+    if os.name == "posix":
+        popen_options["start_new_session"] = True
+    elif hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
+        popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+    try:
+        process = subprocess.Popen(args, **popen_options)
+    except (OSError, ValueError):
+        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
+
+    selector = selectors.DefaultSelector()
+    streams = {"stdout": process.stdout, "stderr": process.stderr}
+    buffers = {"stdout": bytearray(), "stderr": bytearray()}
+    try:
+        for name, stream in streams.items():
+            if stream is not None:
+                selector.register(stream, selectors.EVENT_READ, name)
+    except (OSError, ValueError):
+        selector.close()
+        _terminate_process_tree(process)
+        for stream in streams.values():
+            if stream is not None:
+                stream.close()
+        return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
+
+    deadline = time.monotonic() + bounded_timeout
+    total_bytes = 0
+    return_code: Optional[int] = None
+    try:
+        while selector.get_map():
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return_code = _OPENCLAW_TIMEOUT_RETURN_CODE
+                break
+            events = selector.select(min(remaining, 0.1))
+            for key, _mask in events:
+                allowance = MAX_RESPONSE_BYTES - total_bytes
+                try:
+                    chunk = os.read(key.fileobj.fileno(), min(8192, allowance + 1))
+                except OSError:
+                    return_code = _OPENCLAW_SPAWN_ERROR_RETURN_CODE
+                    break
+                if not chunk:
+                    selector.unregister(key.fileobj)
+                    key.fileobj.close()
+                    continue
+                if len(chunk) > allowance:
+                    return_code = _OPENCLAW_OUTPUT_LIMIT_RETURN_CODE
+                    break
+                buffers[key.data].extend(chunk)
+                total_bytes += len(chunk)
+            if return_code is not None:
+                break
+    finally:
+        selector.close()
+
+    if return_code is not None:
+        _terminate_process_tree(process)
+        for stream in streams.values():
+            if stream is not None and not stream.closed:
+                stream.close()
+        return subprocess.CompletedProcess(args, return_code, "", "")
+    try:
+        completed_return_code = process.wait(timeout=1)
+    except subprocess.TimeoutExpired:
+        _terminate_process_tree(process)
+        return subprocess.CompletedProcess(args, _OPENCLAW_TIMEOUT_RETURN_CODE, "", "")
+    stdout = buffers["stdout"].decode("utf-8", errors="replace")
+    stderr = buffers["stderr"].decode("utf-8", errors="replace")
+    return subprocess.CompletedProcess(args, completed_return_code, stdout, stderr)
+
+
+def _json_dict(text: str) -> Dict[str, Any]:
+    try:
+        value = json.loads(text or "{}")
+    except (RecursionError, TypeError, ValueError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def discover_openclaw_wechat_channel() -> Optional[str]:
+    """Discover an explicitly typed WeChat/Weixin channel for diagnostics."""
+
+    if WECHAT_CHANNEL:
+        return WECHAT_CHANNEL
+    try:
+        proc = _run_openclaw(_openclaw_base_args() + ["channels", "list", "--json"], timeout=15)
+    except (OSError, subprocess.SubprocessError, ValueError):
+        return None
+    if proc.returncode != 0:
+        return None
+    data = _json_dict(proc.stdout[:MAX_RESPONSE_BYTES])
+    candidates: List[str] = []
+
+    def walk(value: Any) -> None:
+        if isinstance(value, dict):
+            name = value.get("name") or value.get("channel") or value.get("id")
+            kind = value.get("type") or value.get("kind")
+            if isinstance(name, str) and isinstance(kind, str):
+                label = f"{name} {kind}".lower()
+                if "wechat" in label or "weixin" in label:
+                    candidates.append(name)
+            for child in value.values():
+                walk(child)
+        elif isinstance(value, list):
+            for child in value:
+                walk(child)
+
+    walk(data)
+    return candidates[0] if candidates else None
+
+
+def _auth_headers() -> Dict[str, str]:
+    headers = {"Content-Type": "application/json", "Accept": "application/json"}
+    if UCC_ADAPTER_TOKEN:
+        headers["Authorization"] = "Bearer " + UCC_ADAPTER_TOKEN
+    return headers
+
+
+def _safe_path(value: Any) -> Optional[str]:
+    if not isinstance(value, str):
+        return None
+    path = value if value.startswith("/") else "/" + value
+    try:
+        encoded_path = path.encode("utf-8")
+    except UnicodeEncodeError:
+        return None
+    if len(encoded_path) > MAX_ENDPOINT_BYTES:
+        return None
+    if not _SAFE_PATH.fullmatch(path):
+        return None
+    decoded = path
+    for _ in range(_MAX_PATH_DECODE_ROUNDS):
+        try:
+            next_decoded = urllib.parse.unquote_to_bytes(decoded).decode("utf-8")
+        except UnicodeDecodeError:
+            return None
+        if next_decoded == decoded:
+            break
+        decoded = next_decoded
+    if "%" in decoded or "\\" in decoded or "\x00" in decoded:
+        return None
+    if not _SAFE_PATH.fullmatch(decoded):
+        return None
+    if any(segment in {".", ".."} for segment in decoded.split("/")):
+        return None
+    return path
+
+
+def _unique_paths(paths: List[str]) -> List[str]:
+    out: List[str] = []
+    for candidate in paths:
+        path = _safe_path(candidate)
+        if path and path not in out:
+            out.append(path)
+        if len(out) == 2:
+            break
+    return out
+
+
+def _ucc_send_paths() -> List[str]:
+    canonical = _safe_path(UCC_SEND_PATH) or "/api/internal/wechat/send"
+    return _unique_paths([canonical, _LEGACY_SEND_PATH])
+
+
+def _canonical_origin(value: Any) -> Optional[str]:
+    if not isinstance(value, str) or not value or len(value) > 2048:
+        return None
+    if "\\" in value or any(ord(character) < 32 for character in value):
+        return None
+    try:
+        parsed = urllib.parse.urlsplit(value)
+        port = parsed.port
+    except (UnicodeError, ValueError):
+        return None
+    scheme = parsed.scheme.lower()
+    if scheme not in {"http", "https"} or not parsed.hostname:
+        return None
+    if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
+        return None
+    try:
+        host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
+    except UnicodeError:
+        return None
+    if not host:
+        return None
+    if port is None:
+        port = 80 if scheme == "http" else 443
+    rendered_host = f"[{host}]" if ":" in host else host
+    return f"{scheme}://{rendered_host}:{port}"
+
+
+def _adapter_base_url() -> Optional[str]:
+    origin = _canonical_origin(UCC_ADAPTER_URL)
+    allowed = {
+        canonical
+        for canonical in (_canonical_origin(value) for value in UCC_ADAPTER_ALLOWED_ORIGINS)
+        if canonical
+    }
+    return origin if origin and origin in allowed else None
+
+
+class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    def redirect_request(
+        self,
+        req: Any,
+        fp: Any,
+        code: int,
+        msg: str,
+        headers: Any,
+        newurl: str,
+    ) -> None:
+        del req, fp, code, msg, headers, newurl
+        return None
+
+
+def _open_no_redirect(request: urllib.request.Request, timeout: int) -> Any:
+    opener = urllib.request.build_opener(_NoRedirectHandler())
+    return opener.open(request, timeout=timeout)
+
+
+def ucc_adapter_alive() -> bool:
+    base_url = _adapter_base_url()
+    if not base_url:
+        return False
+    request = urllib.request.Request(
+        base_url + "/health", headers=_auth_headers(), method="GET"
+    )
+    try:
+        with _open_no_redirect(request, timeout=min(5, REQUEST_TIMEOUT_SECONDS)) as response:
+            return 200 <= int(response.status) < 300
+    except (OSError, ValueError, urllib.error.URLError):
+        return False
+
+
+def select_backend() -> Dict[str, Any]:
+    explicit = SEND_BACKEND.strip().lower()
+    if explicit:
+        if explicit == "ucc-adapter":
+            return {"backend": "ucc-adapter", "reason": "explicit"}
+        return {"backend": "unavailable", "reason": "unsupported_backend"}
+    if ucc_adapter_alive():
+        return {"backend": "ucc-adapter", "reason": "ucc_adapter_alive"}
+    return {"backend": "unavailable", "reason": "ucc_adapter_unavailable"}
+
+
+def _post_ucc(path: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
+    base_url = _adapter_base_url()
+    safe_path = _safe_path(path)
+    if not base_url or not safe_path:
+        raise _BackendProtocolError("invalid adapter endpoint")
+    try:
+        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    except (TypeError, ValueError) as exc:
+        raise _BackendProtocolError("invalid request payload") from exc
+    request = urllib.request.Request(
+        base_url + safe_path,
+        data=body,
+        headers=_auth_headers(),
+        method="POST",
+    )
+    with _open_no_redirect(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+        raw = response.read(MAX_RESPONSE_BYTES + 1)
+        if len(raw) > MAX_RESPONSE_BYTES:
+            raise _BackendProtocolError("adapter response too large")
+        try:
+            text = raw.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise _BackendProtocolError("adapter response encoding invalid") from exc
+        data = _json_dict(text)
+        if not data:
+            raise _BackendProtocolError("adapter response invalid")
+        return int(response.status), data
+
+
+def _ucc_response_sent(data: Dict[str, Any]) -> bool:
+    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == "sent"
+
+
+def _ucc_result_is_ok(data: Dict[str, Any], dry_run: bool = False) -> bool:
+    expected_status = "dry_run" if dry_run else "sent"
+    return isinstance(data, dict) and data.get("ok") is True and data.get("status") == expected_status
+
+
+def _utf8_size(value: str) -> Optional[int]:
+    try:
+        return len(value.encode("utf-8"))
+    except UnicodeEncodeError:
+        return None
+
+
+def _valid_identifier(value: Any, required: bool = False) -> bool:
+    if not isinstance(value, str):
+        return False
+    if required and not value.strip():
+        return False
+    if not value:
+        return not required
+    if len(value) > MAX_IDENTIFIER_BYTES:
+        return False
+    size = _utf8_size(value)
+    return size is not None and size <= MAX_IDENTIFIER_BYTES and not any(
+        ord(character) < 32 or ord(character) == 127 for character in value
+    )
+
+
+def _validate_metadata(metadata: Any) -> Optional[str]:
+    if not isinstance(metadata, dict):
+        return "invalid_input"
+    nodes = 0
+    stack: List[Tuple[Any, int]] = [(metadata, 0)]
+    while stack:
+        value, depth = stack.pop()
+        nodes += 1
+        if depth > MAX_METADATA_DEPTH or nodes > MAX_METADATA_NODES:
+            return "input_too_large"
+        if isinstance(value, dict):
+            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
+                return "input_too_large"
+            for key, child in value.items():
+                if (
+                    not isinstance(key, str)
+                    or len(key) > MAX_METADATA_BYTES
+                    or _utf8_size(key) is None
+                ):
+                    return "invalid_input"
+                stack.append((child, depth + 1))
+        elif isinstance(value, list):
+            if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
+                return "input_too_large"
+            stack.extend((child, depth + 1) for child in value)
+        elif isinstance(value, str):
+            if len(value) > MAX_METADATA_BYTES:
+                return "input_too_large"
+            if _utf8_size(value) is None:
+                return "invalid_input"
+        elif value is not None and not isinstance(value, (bool, int, float)):
+            return "invalid_input"
+    try:
+        encoded = json.dumps(
+            metadata,
+            ensure_ascii=False,
+            allow_nan=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    except (RecursionError, TypeError, UnicodeEncodeError, ValueError):
+        return "invalid_input"
+    return "input_too_large" if len(encoded) > MAX_METADATA_BYTES else None
+
+
+def _validate_payload(
+    target: Any,
+    content: Any,
+    account_id: Any,
+    message_id: Any,
+    metadata: Any,
+    dry_run: Any,
+) -> Optional[str]:
+    if not isinstance(target, str) or not isinstance(content, str):
+        return "invalid_input"
+    if not target.strip() or not content.strip() or not isinstance(dry_run, bool):
+        return "invalid_input"
+    if any(ord(character) < 32 or ord(character) == 127 for character in target):
+        return "invalid_input"
+    if "\x00" in content or not _valid_identifier(account_id) or not _valid_identifier(message_id):
+        return "invalid_input"
+    if len(target) > MAX_TARGET_BYTES or len(content) > MAX_CONTENT_BYTES:
+        return "input_too_large"
+    target_size = _utf8_size(target)
+    content_size = _utf8_size(content)
+    if target_size is None or content_size is None:
+        return "invalid_input"
+    if target_size > MAX_TARGET_BYTES or content_size > MAX_CONTENT_BYTES:
+        return "input_too_large"
+    return _validate_metadata(metadata)
+
+
+def _safe_provider_message_id(value: Any) -> Optional[str]:
+    if not isinstance(value, str) or not value:
+        return None
+    if len(value) > MAX_IDENTIFIER_BYTES:
+        return None
+    size = _utf8_size(value)
+    if size is None or size > MAX_IDENTIFIER_BYTES:
+        return None
+    if any(ord(character) < 32 or ord(character) == 127 for character in value):
+        return None
+    return value
+
+
+def send_ucc_adapter(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> BackendResult:
+    effective_account = account_id or WECHAT_ACCOUNT_ID
+    effective_metadata: Any = {} if metadata is None else metadata
+    error_code = _validate_payload(
+        target,
+        content,
+        effective_account,
+        message_id,
+        effective_metadata,
+        dry_run,
+    )
+    if error_code:
+        return BackendResult(False, "ucc-adapter", "failed", {"code": error_code})
+
+    payload = {
+        "target": target,
+        "content": content,
+        "account_id": effective_account,
+        "message_id": message_id,
+        "metadata": effective_metadata,
+        "dry_run": dry_run,
+    }
+    last_endpoint: Optional[str] = None
+    for path in _ucc_send_paths():
+        last_endpoint = path
+        try:
+            http_status, data = _post_ucc(path, payload)
+        except urllib.error.HTTPError as exc:
+            if exc.code in {404, 405}:
+                continue
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_http_error", "http_status": int(exc.code), "endpoint": path},
+            )
+        except (OSError, ValueError, urllib.error.URLError, _BackendProtocolError):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_unavailable", "endpoint": path},
+            )
+        except Exception:
+            # Fail closed without returning exception text, type, or chained cause.
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_unavailable", "endpoint": path},
+            )
+
+        if not (200 <= http_status < 300):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_http_error", "http_status": http_status, "endpoint": path},
+            )
+        if not _ucc_result_is_ok(data, dry_run):
+            return BackendResult(
+                False,
+                "ucc-adapter",
+                "failed",
+                {"code": "ucc_adapter_rejected", "http_status": http_status, "endpoint": path},
+            )
+        confirmed_status = "dry_run" if dry_run else "sent"
+        result_data: Dict[str, Any] = {
+            "code": confirmed_status,
+            "http_status": http_status,
+            "endpoint": path,
+        }
+        provider_message_id = _safe_provider_message_id(data.get("provider_message_id"))
+        if provider_message_id:
+            result_data["provider_message_id"] = provider_message_id
+        return BackendResult(True, "ucc-adapter", confirmed_status, result_data)
+
+    data: Dict[str, Any] = {"code": "ucc_adapter_unavailable"}
+    if last_endpoint:
+        data["endpoint"] = last_endpoint
+    return BackendResult(False, "ucc-adapter", "failed", data)
+
+
+def send_openclaw_channel(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> BackendResult:
+    del target, content, account_id, message_id, metadata, dry_run
+    return BackendResult(
+        False,
+        "openclaw-channel",
+        "failed",
+        {"code": "unsupported_backend"},
+    )
+
+
+def send_message(
+    target: str,
+    content: str,
+    account_id: str = "",
+    message_id: str = "",
+    metadata: Optional[Dict[str, Any]] = None,
+    dry_run: bool = False,
+) -> Dict[str, Any]:
+    selected = select_backend()
+    if selected.get("backend") == "ucc-adapter":
+        return send_ucc_adapter(
+            target, content, account_id, message_id, metadata, dry_run
+        ).to_dict()
+    code = "unsupported_backend" if selected.get("reason") == "unsupported_backend" else "wechat_outbound_backend_unavailable"
+    return BackendResult(False, "unavailable", "failed", {"code": code}).to_dict()
+
+
+if __name__ == "__main__":
+    print(json.dumps(select_backend(), ensure_ascii=False, separators=(",", ":")))
@@ -0,0 +1 @@
+"""AIOS services namespace."""
@@ -0,0 +1,5 @@
+"""Reliable, fail-closed Codex command watcher."""
+
+from .watcher import CommandWatcher, Config, load_config
+
+__all__ = ["CommandWatcher", "Config", "load_config"]
@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import argparse
+import sys
+
+from .watcher import CommandWatcher, load_config
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Claim and execute Codex work units")
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--once", action="store_true", help="process one scan and exit")
+    args = parser.parse_args()
+    watcher = CommandWatcher(load_config(args.config))
+    return watcher.run_once() if args.once else watcher.run_forever()
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,480 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+import stat
+import subprocess
+import time
+from typing import Any, Callable
+import urllib.error
+import urllib.parse
+import urllib.request
+import uuid
+
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+TERMINAL = {"COMPLETED", "DEAD_LETTER", "REJECTED"}
+
+
+class WatcherError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class Config:
+    active_dir: Path
+    completed_dir: Path
+    dead_letter_dir: Path
+    database_path: Path
+    audit_log_path: Path
+    gateway_endpoint: str
+    gateway_token_file: Path
+    contract_commit: str
+    lease_seconds: int
+    retry_seconds: int
+    max_attempts: int
+    poll_seconds: float
+    executor_path: Path | None
+    executor_args: tuple[str, ...]
+    executor_timeout_seconds: int
+
+
+def _absolute(value: Any, key: str) -> Path:
+    if not isinstance(value, str) or not value or not Path(value).is_absolute():
+        raise WatcherError(f"invalid_{key}")
+    return Path(value)
+
+
+def _secure_regular(path: Path, *, writable_mask: int = 0o022) -> os.stat_result:
+    info = os.lstat(path)
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise WatcherError(f"unsafe_file:{path}")
+    if stat.S_IMODE(info.st_mode) & writable_mask:
+        raise WatcherError(f"unsafe_mode:{path}")
+    return info
+
+
+def _read_secure_text(
+    path: Path,
+    *,
+    writable_mask: int = 0o022,
+    max_bytes: int,
+    allowed_owners: set[int] | None = None,
+) -> str:
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    fd = os.open(path, flags)
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) & writable_mask:
+            raise WatcherError(f"unsafe_file:{path}")
+        if allowed_owners is not None and info.st_uid not in allowed_owners:
+            raise WatcherError(f"owner_mismatch:{path}")
+        if info.st_size > max_bytes:
+            raise WatcherError(f"file_too_large:{path}")
+        chunks: list[bytes] = []
+        total = 0
+        while True:
+            chunk = os.read(fd, min(65_536, max_bytes + 1 - total))
+            if not chunk:
+                break
+            chunks.append(chunk)
+            total += len(chunk)
+            if total > max_bytes:
+                raise WatcherError(f"file_too_large:{path}")
+        return b"".join(chunks).decode("utf-8")
+    finally:
+        os.close(fd)
+
+
+def load_config(path_value: str) -> Config:
+    path = _absolute(path_value, "config_path")
+    raw = json.loads(_read_secure_text(path, max_bytes=131_072, allowed_owners={0, os.geteuid()}))
+    expected = {
+        "active_dir", "completed_dir", "dead_letter_dir", "database_path",
+        "audit_log_path", "gateway_endpoint", "gateway_token_file",
+        "contract_commit", "lease_seconds", "retry_seconds", "max_attempts",
+        "poll_seconds", "executor_path", "executor_args",
+        "executor_timeout_seconds",
+    }
+    if not isinstance(raw, dict) or set(raw) != expected:
+        raise WatcherError("invalid_config_fields")
+    contract = raw["contract_commit"]
+    if not isinstance(contract, str) or not OID_RE.fullmatch(contract):
+        raise WatcherError("invalid_contract_commit")
+    endpoint = raw["gateway_endpoint"]
+    if not isinstance(endpoint, str) or not endpoint.startswith("http://127.0.0.1:") or not endpoint.endswith("/v1/orchestrator/events"):
+        raise WatcherError("gateway_endpoint_must_be_loopback")
+    executor_raw = raw["executor_path"]
+    executor = None if executor_raw is None else _absolute(executor_raw, "executor_path")
+    executor_args = raw["executor_args"]
+    if not isinstance(executor_args, list) or any(not isinstance(x, str) or not x for x in executor_args):
+        raise WatcherError("invalid_executor_args")
+    ints = ("lease_seconds", "retry_seconds", "max_attempts", "executor_timeout_seconds")
+    if any(not isinstance(raw[x], int) or raw[x] <= 0 for x in ints):
+        raise WatcherError("invalid_positive_integer")
+    if not isinstance(raw["poll_seconds"], (int, float)) or raw["poll_seconds"] <= 0:
+        raise WatcherError("invalid_poll_seconds")
+    return Config(
+        active_dir=_absolute(raw["active_dir"], "active_dir"),
+        completed_dir=_absolute(raw["completed_dir"], "completed_dir"),
+        dead_letter_dir=_absolute(raw["dead_letter_dir"], "dead_letter_dir"),
+        database_path=_absolute(raw["database_path"], "database_path"),
+        audit_log_path=_absolute(raw["audit_log_path"], "audit_log_path"),
+        gateway_endpoint=endpoint,
+        gateway_token_file=_absolute(raw["gateway_token_file"], "gateway_token_file"),
+        contract_commit=contract,
+        lease_seconds=raw["lease_seconds"], retry_seconds=raw["retry_seconds"],
+        max_attempts=raw["max_attempts"], poll_seconds=float(raw["poll_seconds"]),
+        executor_path=executor, executor_args=tuple(executor_args),
+        executor_timeout_seconds=raw["executor_timeout_seconds"],
+    )
+
+
+def canonical_digest(value: dict[str, Any]) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def validate_command(value: Any, expected_contract: str) -> dict[str, Any]:
+    expected = {
+        "schema_version", "command_id", "project_key", "work_unit_id",
+        "contract_commit", "state_version", "execution",
+    }
+    if not isinstance(value, dict) or set(value) != expected or value.get("schema_version") != 1:
+        raise WatcherError("invalid_command_fields")
+    for key in ("command_id", "project_key", "work_unit_id"):
+        if not isinstance(value[key], str) or not ID_RE.fullmatch(value[key]):
+            raise WatcherError(f"invalid_{key}")
+    if value["contract_commit"] != expected_contract:
+        raise WatcherError("contract_commit_mismatch")
+    if not isinstance(value["state_version"], int) or value["state_version"] < 0:
+        raise WatcherError("invalid_state_version")
+    execution = value["execution"]
+    if not isinstance(execution, dict) or set(execution) != {"kind", "input"}:
+        raise WatcherError("invalid_execution")
+    if execution["kind"] not in {"acceptance_probe", "codex"}:
+        raise WatcherError("invalid_execution_kind")
+    if not isinstance(execution["input"], str) or len(execution["input"].encode()) > 65_536:
+        raise WatcherError("invalid_execution_input")
+    return value
+
+
+class CommandWatcher:
+    def __init__(
+        self,
+        config: Config,
+        *,
+        clock: Callable[[], float] = time.time,
+        event_poster: Callable[[dict[str, Any]], None] | None = None,
+        executor: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
+        worker_id: str | None = None,
+    ) -> None:
+        self.config = config
+        self.clock = clock
+        self.worker_id = worker_id or f"{os.uname().nodename}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
+        self._uses_gateway = event_poster is None
+        self.event_poster = event_poster or self._post_event
+        self.executor = executor or self._execute
+        self._prepare()
+
+    def _prepare(self) -> None:
+        for directory in (self.config.active_dir, self.config.completed_dir, self.config.dead_letter_dir):
+            info = os.lstat(directory)
+            if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) & 0o002:
+                raise WatcherError(f"unsafe_directory:{directory}")
+        self.config.database_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.config.audit_log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        with self._db() as db:
+            db.execute("PRAGMA journal_mode=WAL")
+            db.execute("""
+                CREATE TABLE IF NOT EXISTS commands (
+                    command_id TEXT PRIMARY KEY, work_unit_id TEXT NOT NULL UNIQUE,
+                    payload_sha256 TEXT NOT NULL,
+                    source_path TEXT NOT NULL, status TEXT NOT NULL,
+                    phase TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
+                    lease_owner TEXT, lease_until REAL, next_attempt REAL NOT NULL DEFAULT 0,
+                    result_json TEXT, start_event_json TEXT, end_event_json TEXT,
+                    last_error TEXT, updated_at REAL NOT NULL
+                )
+            """)
+
+    def _db(self) -> sqlite3.Connection:
+        db = sqlite3.connect(self.config.database_path, timeout=5, isolation_level=None)
+        db.row_factory = sqlite3.Row
+        return db
+
+    def _audit(self, marker: str, command_id: str, **fields: Any) -> None:
+        safe = {k: v for k, v in fields.items() if k not in {"input", "token", "secret"}}
+        record = {"time": int(self.clock()), "marker": marker, "command_id": command_id, **safe}
+        fd = os.open(self.config.audit_log_path, os.O_APPEND | os.O_CREAT | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
+        try:
+            os.write(fd, (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode())
+        finally:
+            os.close(fd)
+
+    def _read_command(self, path: Path) -> tuple[dict[str, Any], str]:
+        raw = _read_secure_text(path, max_bytes=131_072, allowed_owners={0, os.geteuid()})
+        value = validate_command(json.loads(raw), self.config.contract_commit)
+        return value, canonical_digest(value)
+
+    def _register(self, command: dict[str, Any], digest: str, path: Path) -> str:
+        now = self.clock()
+        with self._db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT payload_sha256,status FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
+            if row:
+                db.execute("COMMIT")
+                if row["payload_sha256"] != digest:
+                    raise WatcherError("duplicate_command_payload_conflict")
+                return "duplicate_terminal" if row["status"] in TERMINAL else "existing"
+            prior_work_unit = db.execute(
+                "SELECT command_id FROM commands WHERE work_unit_id=?",
+                (command["work_unit_id"],),
+            ).fetchone()
+            if prior_work_unit:
+                db.execute("COMMIT")
+                raise WatcherError("duplicate_work_unit")
+            db.execute(
+                "INSERT INTO commands(command_id,work_unit_id,payload_sha256,source_path,status,phase,updated_at) VALUES(?,?,?,?,?,?,?)",
+                (command["command_id"], command["work_unit_id"], digest, str(path), "PENDING", "START", now),
+            )
+            db.execute("COMMIT")
+        return "new"
+
+    def _claim(self, command_id: str) -> sqlite3.Row | None:
+        now = self.clock()
+        with self._db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
+            if not row or row["status"] in TERMINAL or row["next_attempt"] > now:
+                db.execute("COMMIT")
+                return None
+            if row["lease_until"] is not None and row["lease_until"] > now and row["lease_owner"] != self.worker_id:
+                db.execute("COMMIT")
+                return None
+            attempts = row["attempts"] + 1
+            db.execute(
+                "UPDATE commands SET status='CLAIMED',attempts=?,lease_owner=?,lease_until=?,updated_at=? WHERE command_id=?",
+                (attempts, self.worker_id, now + self.config.lease_seconds, now, command_id),
+            )
+            row = db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
+            db.execute("COMMIT")
+        assert row is not None
+        self._audit("CODEX_AUTO_CLAIM", command_id, attempt=row["attempts"], worker=self.worker_id)
+        return row
+
+    def _update(self, command_id: str, *, status: str, phase: str, result: dict[str, Any] | None = None, error: str | None = None) -> None:
+        with self._db() as db:
+            release = status in TERMINAL
+            db.execute(
+                "UPDATE commands SET status=?,phase=?,result_json=COALESCE(?,result_json),last_error=?,lease_owner=CASE WHEN ? THEN NULL ELSE lease_owner END,lease_until=CASE WHEN ? THEN NULL ELSE lease_until END,updated_at=? WHERE command_id=?",
+                (status, phase, json.dumps(result, sort_keys=True) if result is not None else None, error, release, release, self.clock(), command_id),
+            )
+
+    def _retry_or_dead(self, command_id: str, path: Path, error: Exception) -> None:
+        message = f"{type(error).__name__}:{str(error)[:240]}"
+        with self._db() as db:
+            row = db.execute("SELECT attempts,phase FROM commands WHERE command_id=?", (command_id,)).fetchone()
+            attempts = row["attempts"] if row else self.config.max_attempts
+            if str(error) == "gateway_stale_state_version" and row:
+                column = "start_event_json" if row["phase"] == "START" else "end_event_json"
+                db.execute(f"UPDATE commands SET {column}=NULL WHERE command_id=?", (command_id,))
+            if attempts >= self.config.max_attempts:
+                db.execute("UPDATE commands SET status='DEAD_LETTER',last_error=?,lease_owner=NULL,lease_until=NULL,updated_at=? WHERE command_id=?", (message, self.clock(), command_id))
+                self._move(path, self.config.dead_letter_dir)
+                self._audit("DEAD_LETTER", command_id, error=message)
+            else:
+                db.execute("UPDATE commands SET status='RETRY',last_error=?,next_attempt=?,lease_owner=NULL,lease_until=NULL,updated_at=? WHERE command_id=?", (message, self.clock() + self.config.retry_seconds, self.clock(), command_id))
+                self._audit("RETRY_SCHEDULED", command_id, attempt=attempts, error=message)
+
+    def _event(self, command: dict[str, Any], event_type: str, status: str, suffix: str, result: dict[str, Any] | None = None) -> dict[str, Any]:
+        return {
+            "schema_version": 1, "project_key": command["project_key"],
+            "contract_commit": command["contract_commit"],
+            "event_id": f"{command['command_id']}:{suffix}", "source_agent": "codex",
+            "event_type": event_type, "work_unit_id": command["work_unit_id"],
+            "status": status, "state_version": command["state_version"],
+            "implementation_commit": None, "result_commit": None, "blocker": None,
+            "evidence_refs": [], "payload": {"command_id": command["command_id"], "result": result or {}},
+        }
+
+    def _gateway_token(self) -> str:
+        token = _read_secure_text(
+            self.config.gateway_token_file,
+            writable_mask=0o027,
+            max_bytes=16_384,
+            allowed_owners={0, os.geteuid()},
+        ).strip()
+        if not token or "\n" in token or "\r" in token:
+            raise WatcherError("invalid_gateway_token")
+        return token
+
+    def _gateway_state_version(self, project_key: str, fallback: int) -> int:
+        state_url = self.config.gateway_endpoint.removesuffix("/events") + "/state/" + urllib.parse.quote(project_key, safe="")
+        request = urllib.request.Request(
+            state_url,
+            headers={"Authorization": f"Bearer {self._gateway_token()}"},
+            method="GET",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=15) as response:
+                payload = json.loads(response.read(262_145))
+                version = payload.get("state_version")
+                if not isinstance(version, int) or version < 0:
+                    raise WatcherError("invalid_gateway_state")
+                return version
+        except urllib.error.HTTPError as exc:
+            if exc.code == 404:
+                return fallback
+            raise WatcherError(f"gateway_state_http_{exc.code}") from exc
+        except urllib.error.URLError as exc:
+            raise WatcherError("gateway_state_unavailable") from exc
+
+    def _durable_event(self, command: dict[str, Any], column: str, event: dict[str, Any]) -> dict[str, Any]:
+        with self._db() as db:
+            row = db.execute(f"SELECT {column} FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
+            if row and row[column]:
+                return json.loads(row[column])
+        if self._uses_gateway:
+            event["state_version"] = self._gateway_state_version(command["project_key"], command["state_version"])
+        encoded = json.dumps(event, sort_keys=True, separators=(",", ":"))
+        with self._db() as db:
+            db.execute(f"UPDATE commands SET {column}=? WHERE command_id=? AND {column} IS NULL", (encoded, command["command_id"]))
+            saved = db.execute(f"SELECT {column} FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
+        if not saved or not saved[column]:
+            raise WatcherError("event_persistence_failed")
+        return json.loads(saved[column])
+
+    def _post_event(self, event: dict[str, Any]) -> None:
+        request = urllib.request.Request(
+            self.config.gateway_endpoint,
+            data=json.dumps(event, separators=(",", ":")).encode(),
+            headers={"Authorization": f"Bearer {self._gateway_token()}", "Content-Type": "application/json"},
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=15) as response:
+                if response.status not in {200, 201, 202}:
+                    raise WatcherError(f"gateway_http_{response.status}")
+        except urllib.error.HTTPError as exc:
+            body = exc.read(16_384)
+            try:
+                reason = json.loads(body).get("error")
+            except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
+                reason = None
+            if exc.code == 409 and reason == "stale_state_version":
+                raise WatcherError("gateway_stale_state_version") from exc
+            raise WatcherError(f"gateway_http_{exc.code}") from exc
+        except urllib.error.URLError as exc:
+            raise WatcherError("gateway_unavailable") from exc
+
+    def _execute(self, command: dict[str, Any]) -> dict[str, Any]:
+        execution = command["execution"]
+        if execution["kind"] == "acceptance_probe":
+            return {"kind": "acceptance_probe", "input_sha256": hashlib.sha256(execution["input"].encode()).hexdigest(), "exit_code": 0}
+        assert self.config.executor_path is not None
+        proc = subprocess.run(
+            [str(self.config.executor_path), *self.config.executor_args],
+            input=execution["input"], text=True, stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL, timeout=self.config.executor_timeout_seconds,
+            check=False, env={"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"},
+        )
+        if proc.returncode != 0:
+            raise WatcherError(f"codex_executor_failed:{proc.returncode}")
+        return {"kind": "codex", "exit_code": 0}
+
+    def _preflight_execution(self, command: dict[str, Any]) -> None:
+        if command["execution"]["kind"] == "acceptance_probe":
+            return
+        if self.config.executor_path is None:
+            raise WatcherError("codex_executor_unavailable")
+        info = _secure_regular(self.config.executor_path)
+        if not info.st_mode & stat.S_IXUSR:
+            raise WatcherError("codex_executor_not_executable")
+
+    def _move(self, source: Path, target_dir: Path) -> None:
+        if not source.exists() and not source.is_symlink():
+            return
+        target = target_dir / source.name
+        if target.exists() or target.is_symlink():
+            target = target_dir / f"{source.stem}-{uuid.uuid4().hex[:8]}{source.suffix}"
+        os.replace(source, target)
+
+    def _reject_unregistered(self, path: Path, error: Exception) -> None:
+        command_id = path.stem[:128] if ID_RE.fullmatch(path.stem[:128]) else f"invalid-{uuid.uuid4().hex[:12]}"
+        self._audit("COMMAND_REJECTED", command_id, error=f"{type(error).__name__}:{str(error)[:240]}")
+        self._move(path, self.config.dead_letter_dir)
+
+    def process_path(self, path: Path) -> bool:
+        try:
+            command, digest = self._read_command(path)
+            registration = self._register(command, digest, path)
+            if registration == "duplicate_terminal":
+                self._move(path, self.config.completed_dir)
+                return False
+            row = self._claim(command["command_id"])
+            if row is None:
+                return False
+            phase = row["phase"]
+            result = json.loads(row["result_json"]) if row["result_json"] else None
+            if phase == "START":
+                start_event = self._durable_event(
+                    command,
+                    "start_event_json",
+                    self._event(command, "work_unit_started", "RUNNING", "started"),
+                )
+                self.event_poster(start_event)
+                self._audit("TASK_STARTED", command["command_id"])
+                self._update(command["command_id"], status="PENDING", phase="EXECUTE")
+                phase = "EXECUTE"
+            if phase == "EXECUTE":
+                self._preflight_execution(command)
+                self._update(command["command_id"], status="CLAIMED", phase="EXECUTING")
+                phase = "EXECUTING"
+                result = self.executor(command)
+                self._audit("TASK_EXECUTED", command["command_id"], result=result)
+                self._update(command["command_id"], status="PENDING", phase="END", result=result)
+                phase = "END"
+            elif phase == "EXECUTING":
+                raise WatcherError("ambiguous_execution_not_repeated")
+            if phase == "END":
+                assert result is not None
+                end_event = self._durable_event(
+                    command,
+                    "end_event_json",
+                    self._event(command, "work_unit_ended", "PASS", "ended", result),
+                )
+                self.event_poster(end_event)
+                self._audit("TASK_ENDED", command["command_id"])
+                self._audit("EVENT_POSTED", command["command_id"], event="work_unit_ended")
+                self._update(command["command_id"], status="COMPLETED", phase="DONE", result=result)
+                self._move(path, self.config.completed_dir)
+            return True
+        except Exception as exc:
+            try:
+                if isinstance(exc, WatcherError) and str(exc) in {"duplicate_command_payload_conflict", "duplicate_work_unit"}:
+                    self._reject_unregistered(path, exc)
+                elif "command" in locals():
+                    self._retry_or_dead(command["command_id"], path, exc)
+                else:
+                    self._reject_unregistered(path, exc)
+            except Exception:
+                pass
+            return False
+
+    def run_once(self) -> int:
+        failures = 0
+        for path in sorted(self.config.active_dir.glob("*.json")):
+            if not self.process_path(path):
+                failures += 1
+        return 1 if failures else 0
+
+    def run_forever(self) -> int:
+        while True:
+            self.run_once()
+            time.sleep(self.config.poll_seconds)
@@ -0,0 +1,3 @@
+"""Reliable central orchestration gateway for AIOS."""
+
+__version__ = "0.2.0"
@@ -0,0 +1,34 @@
+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())
@@ -0,0 +1,178 @@
+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")
@@ -0,0 +1,203 @@
+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
@@ -0,0 +1,89 @@
+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}")
@@ -0,0 +1,26 @@
+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."""
@@ -0,0 +1,346 @@
+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"]},
+    },
+}
@@ -0,0 +1,229 @@
+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)
@@ -0,0 +1,642 @@
+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()
@@ -0,0 +1,826 @@
+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
@@ -0,0 +1,108 @@
+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,
+    }
@@ -0,0 +1,35 @@
+[Unit]
+Description=AIOS Codex automatic command watcher
+After=network-online.target aios-orchestrator-gateway.service
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+User=aios-orchestrator
+Group=aios-orchestrator
+UMask=0077
+Environment=PYTHONPATH=/opt/aios-codex-watcher/current
+ExecStart=/usr/bin/python3 -m services.aios_codex_watcher --config /etc/aios-codex-watcher/config.json --once
+NoNewPrivileges=yes
+PrivateDevices=yes
+PrivateTmp=yes
+ProtectClock=yes
+ProtectControlGroups=yes
+ProtectHome=yes
+ProtectHostname=yes
+ProtectKernelLogs=yes
+ProtectKernelModules=yes
+ProtectKernelTunables=yes
+ProtectSystem=strict
+ReadOnlyPaths=/etc/aios-orchestrator-gateway/inbound.token
+ReadWritePaths=/data/aios-gitops/workspace/agents/commands /var/lib/aios-codex-watcher /var/log/aios-codex-watcher
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+SystemCallArchitectures=native
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1 @@
+u aios-orchestrator - "AIOS Orchestrator" /var/lib/aios-orchestrator-gateway
@@ -0,0 +1,12 @@
+[Unit]
+Description=Poll AIOS Codex command queue
+
+[Timer]
+OnBootSec=30s
+OnUnitActiveSec=5s
+AccuracySec=1s
+Persistent=true
+Unit=aios-codex-watcher.service
+
+[Install]
+WantedBy=timers.target
@@ -0,0 +1,2 @@
+d /var/lib/aios-codex-watcher 0700 aios-orchestrator aios-orchestrator -
+d /var/log/aios-codex-watcher 0700 aios-orchestrator aios-orchestrator -
@@ -0,0 +1,37 @@
+[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
@@ -0,0 +1,12 @@
+[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
@@ -0,0 +1,52 @@
+[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
@@ -0,0 +1,2 @@
+g aios-orchestrator -
+u aios-orchestrator - "AIOS Orchestrator Gateway" /var/lib/aios-orchestrator-gateway /usr/sbin/nologin
@@ -0,0 +1,19 @@
+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 -
@@ -0,0 +1,196 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+
+from services.aios_codex_watcher.watcher import CommandWatcher, Config, WatcherError, validate_command
+
+
+CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
+
+
+class Clock:
+    def __init__(self) -> None:
+        self.value = 1000.0
+
+    def __call__(self) -> float:
+        return self.value
+
+
+class WatcherTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        for name in ("active", "completed", "dead", "state", "log"):
+            (self.root / name).mkdir(mode=0o700)
+        self.token = self.root / "token"
+        self.token.write_text("not-a-real-secret\n", encoding="utf-8")
+        self.token.chmod(0o600)
+        self.clock = Clock()
+        self.events: list[dict] = []
+        self.exec_count = 0
+        self.config = Config(
+            active_dir=self.root / "active", completed_dir=self.root / "completed",
+            dead_letter_dir=self.root / "dead", database_path=self.root / "state/state.sqlite3",
+            audit_log_path=self.root / "log/audit.jsonl",
+            gateway_endpoint="http://127.0.0.1:18920/v1/orchestrator/events",
+            gateway_token_file=self.token, contract_commit=CONTRACT,
+            lease_seconds=10, retry_seconds=2, max_attempts=3, poll_seconds=1,
+            executor_path=None, executor_args=(), executor_timeout_seconds=10,
+        )
+
+    def tearDown(self) -> None:
+        self.temp.cleanup()
+
+    def command(self, command_id: str = "cmd-1", *, contract: str = CONTRACT, kind: str = "acceptance_probe", text: str = "no-op") -> dict:
+        return {
+            "schema_version": 1, "command_id": command_id,
+            "project_key": "a0-codex-watcher-nonproduction",
+            "work_unit_id": "A0-CODEX-WATCHER-ACCEPTANCE",
+            "contract_commit": contract, "state_version": 0,
+            "execution": {"kind": kind, "input": text},
+        }
+
+    def write(self, value: dict, name: str = "command.json") -> Path:
+        path = self.config.active_dir / name
+        path.write_text(json.dumps(value), encoding="utf-8")
+        path.chmod(0o600)
+        return path
+
+    def watcher(self, **kwargs) -> CommandWatcher:
+        def execute(command: dict) -> dict:
+            self.exec_count += 1
+            return {"kind": command["execution"]["kind"], "exit_code": 0}
+        return CommandWatcher(
+            self.config, clock=self.clock,
+            event_poster=kwargs.get("event_poster", self.events.append),
+            executor=kwargs.get("executor", execute), worker_id=kwargs.get("worker_id"),
+        )
+
+    def row(self, command_id: str = "cmd-1") -> sqlite3.Row:
+        db = sqlite3.connect(self.config.database_path)
+        db.row_factory = sqlite3.Row
+        try:
+            return db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
+        finally:
+            db.close()
+
+    def test_happy_path_posts_lifecycle_once_and_archives(self) -> None:
+        path = self.write(self.command())
+        self.assertTrue(self.watcher().process_path(path))
+        self.assertEqual([e["event_type"] for e in self.events], ["work_unit_started", "work_unit_ended"])
+        self.assertEqual(self.exec_count, 1)
+        self.assertEqual(self.row()["status"], "COMPLETED")
+        self.assertTrue((self.config.completed_dir / path.name).is_file())
+        markers = [json.loads(line)["marker"] for line in self.config.audit_log_path.read_text().splitlines()]
+        self.assertEqual(markers, ["CODEX_AUTO_CLAIM", "TASK_STARTED", "TASK_EXECUTED", "TASK_ENDED", "EVENT_POSTED"])
+
+    def test_contract_mismatch_is_rejected_without_registration(self) -> None:
+        path = self.write(self.command(contract="0" * 40))
+        self.assertFalse(self.watcher().process_path(path))
+        self.assertTrue((self.config.dead_letter_dir / path.name).is_file())
+        db = sqlite3.connect(self.config.database_path)
+        self.assertEqual(db.execute("SELECT count(*) FROM commands").fetchone()[0], 0)
+
+    def test_duplicate_id_with_different_payload_is_dead_lettered(self) -> None:
+        first = self.write(self.command(), "first.json")
+        self.assertTrue(self.watcher().process_path(first))
+        second = self.write(self.command(text="different"), "second.json")
+        self.assertFalse(self.watcher().process_path(second))
+        self.assertEqual(self.row()["status"], "COMPLETED")
+        self.assertTrue((self.config.dead_letter_dir / second.name).is_file())
+
+    def test_same_work_unit_with_different_command_id_is_not_executed(self) -> None:
+        first = self.write(self.command(), "first.json")
+        self.assertTrue(self.watcher().process_path(first))
+        second_command = self.command(command_id="cmd-2")
+        second = self.write(second_command, "second.json")
+        self.assertFalse(self.watcher().process_path(second))
+        self.assertEqual(self.exec_count, 1)
+        self.assertTrue((self.config.dead_letter_dir / second.name).is_file())
+
+    def test_two_workers_cannot_hold_same_lease(self) -> None:
+        path = self.write(self.command())
+        one = self.watcher(worker_id="worker-one")
+        command, digest = one._read_command(path)
+        one._register(command, digest, path)
+        self.assertIsNotNone(one._claim("cmd-1"))
+        two = self.watcher(worker_id="worker-two")
+        self.assertIsNone(two._claim("cmd-1"))
+
+    def test_stale_lease_is_recovered(self) -> None:
+        path = self.write(self.command())
+        one = self.watcher(worker_id="worker-one")
+        command, digest = one._read_command(path)
+        one._register(command, digest, path)
+        one._claim("cmd-1")
+        self.clock.value += 11
+        self.assertIsNotNone(self.watcher(worker_id="worker-two")._claim("cmd-1"))
+
+    def test_end_post_retry_does_not_execute_twice(self) -> None:
+        calls = 0
+        attempted_end_events: list[dict] = []
+        def poster(event: dict) -> None:
+            nonlocal calls
+            calls += 1
+            if event["event_type"] == "work_unit_ended":
+                attempted_end_events.append(event)
+                if calls == 2:
+                    raise WatcherError("temporary")
+            self.events.append(event)
+        path = self.write(self.command())
+        watcher = self.watcher(event_poster=poster)
+        self.assertFalse(watcher.process_path(path))
+        self.assertEqual(self.row()["phase"], "END")
+        self.assertEqual(self.exec_count, 1)
+        self.clock.value += 3
+        self.assertTrue(watcher.process_path(path))
+        self.assertEqual(self.exec_count, 1)
+        self.assertEqual(len(attempted_end_events), 2)
+        self.assertEqual(attempted_end_events[0], attempted_end_events[1])
+        self.assertEqual(self.row()["status"], "COMPLETED")
+
+    def test_failure_exhaustion_moves_to_dead_letter(self) -> None:
+        executions = 0
+        def fail(_command: dict) -> dict:
+            nonlocal executions
+            executions += 1
+            raise WatcherError("executor failure")
+        path = self.write(self.command())
+        watcher = self.watcher(executor=fail)
+        for _ in range(3):
+            self.assertFalse(watcher.process_path(path))
+            self.clock.value += 3
+        self.assertEqual(self.row()["status"], "DEAD_LETTER")
+        self.assertEqual(executions, 1)
+        self.assertTrue((self.config.dead_letter_dir / path.name).is_file())
+
+    def test_symlink_command_is_rejected_without_following(self) -> None:
+        outside = self.root / "outside.json"
+        outside.write_text(json.dumps(self.command()), encoding="utf-8")
+        link = self.config.active_dir / "link.json"
+        os.symlink(outside, link)
+        self.assertFalse(self.watcher().process_path(link))
+        self.assertTrue(outside.is_file())
+        self.assertTrue((self.config.dead_letter_dir / link.name).is_symlink())
+
+    def test_codex_mode_fails_closed_when_executor_absent(self) -> None:
+        path = self.write(self.command(kind="codex"))
+        watcher = CommandWatcher(self.config, clock=self.clock, event_poster=self.events.append)
+        self.assertFalse(watcher.process_path(path))
+        self.assertEqual(self.row()["phase"], "EXECUTE")
+        self.assertIn("codex_executor_unavailable", self.row()["last_error"])
+
+    def test_validation_rejects_unknown_fields(self) -> None:
+        command = self.command()
+        command["secret"] = "must-not-be-accepted"
+        with self.assertRaisesRegex(WatcherError, "invalid_command_fields"):
+            validate_command(command, CONTRACT)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,774 @@
+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()
@@ -0,0 +1,59 @@
+#!/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,11 +1,20 @@
 from __future__ import annotations

 import importlib.util
+import io
+import os
+import signal
+import subprocess
 import sys
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from types import SimpleNamespace
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import Mock, call, patch


 BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
@@ -80,6 +89,450 @@ class WechatSendBackendTests(TestCase):
        self.assertEqual(args[1]["content"], "hello")
        self.assertFalse(args[1]["dry_run"])

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

 if __name__ == "__main__":
    import unittest
@@ -0,0 +1,31 @@
+# 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`.
@@ -0,0 +1,114 @@
+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	4e63b1559a	docs(codex): record A0 automation closure result	2026-07-18 03:48:54 +08:00
Codex	c3984ca2b4	fix(codex): isolate watcher runtime directory ownership	2026-07-18 03:39:45 +08:00
Codex	0c7e2d3b89	feat(codex): add persistent automatic command watcher	2026-07-18 03:36:46 +08:00
Codex	1b3b74fd62	merge(codex): integrate AIOS gateway with baseline remediation	2026-07-18 03:01:30 +08:00
Codex	378cc4d56a	fix(codex): restore hardened WeChat send backend lineage	2026-07-18 03:00:32 +08:00
Codex	95fc4d37f0	feat(codex): add reliable AIOS central orchestration gateway	2026-07-18 02:26:14 +08:00
Codex	9195470ceb	chore(orchestration): dispatch adaptive OpenClaw and Codex work units	2026-07-17 22:13:38 +08:00
Codex	cbe33ecab8	chore(governance): refresh adaptive dual-agent production contract	2026-07-17 22:13:38 +08:00
Codex	e3fc48e5b9	chore(governance): refresh adaptive dual-agent production contract	2026-07-17 22:04:03 +08:00
Codex	ed3bcf76c7	chore(governance): add shared V2R1 production execution contract	2026-07-17 21:24:20 +08:00
				`@@ -0,0 +1 @@`
				`u aios-orchestrator - "AIOS Orchestrator" /var/lib/aios-orchestrator-gateway`