feat(codex): harden callback transport and pending recovery

fix(codex): harden workspace trigger protocol
fix(codex): make central GPT health probing single-flight
2026-07-18 15:50:42 +08:00 · 2026-07-18 14:51:57 +08:00 · 2026-07-18 13:44:06 +08:00 · 2026-07-18 11:59:27 +08:00 · 2026-07-18 11:57:52 +08:00 · 2026-07-18 11:56:31 +08:00
57 changed files with 5338 additions and 167 deletions
@@ -1,32 +1,76 @@
 {
  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
-  "created_at": "2026-07-17T14:13:38+00:00",
-  "schema_version": 1,
-  "state_writer": "openclaw",
+  "created_at": "2026-07-18T04:58:00+00:00",
+  "schema_version": 2,
+  "state_writer": "gateway",
  "work_units": [
    {
-      "fallback": "codex",
-      "gate": "R0_RUNNER_LIVE_RECOVERY",
      "id": "R0-DISCOVER-SECRET-PROVIDER",
+      "track": "R_RUNNER",
+      "gate": "R0_RUNNER_LIVE_RECOVERY",
      "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"
+      "quorum": "requires_gpt_plus_one_executor"
    },
    {
-      "fallback": "openclaw",
-      "gate": "P1_REPAIR3_CODE",
-      "id": "P1-POST-CONTRACT-REACK",
-      "primary": "codex",
+      "id": "P2-INVENTORY-REAL-EVIDENCE",
+      "track": "P_PERSONAL_WEIXIN",
+      "gate": "P2_REAL_B3_EVIDENCE",
+      "primary": "openclaw",
+      "fallback": "codex",
      "status": "READY",
-      "track": "P_PERSONAL_WEIXIN"
+      "quorum": "requires_gpt_plus_one_executor"
+    },
+    {
+      "id": "P1-POST-CONTRACT-REACK",
+      "track": "P_PERSONAL_WEIXIN",
+      "gate": "P1_REPAIR3_CODE",
+      "primary": "codex",
+      "fallback": "openclaw",
+      "status": "READY",
+      "quorum": "requires_gpt_plus_one_executor"
+    },
+    {
+      "id": "Q0-GPT-HEALTH-INIT",
+      "track": "_QUORUM",
+      "gate": "Q0_GPT_AND_EXECUTOR_HEALTH",
+      "primary": "gateway",
+      "fallback": null,
+      "status": "READY",
+      "quorum": "gateway_initiated"
+    },
+    {
+      "id": "Q1-ACTIVE-EXECUTOR-ELECTION",
+      "track": "_QUORUM",
+      "gate": "Q1_ELECT_ACTIVE_EXECUTOR",
+      "primary": "gateway",
+      "fallback": null,
+      "status": "PENDING",
+      "depends_on": "Q0-GPT-HEALTH-INIT",
+      "quorum": "gateway_initiated"
+    },
+    {
+      "id": "Q2-FAILOVER-HEALTHCHECK",
+      "track": "_QUORUM",
+      "gate": "Q2_STANDBY_HEALTH_MONITOR",
+      "primary": "gateway",
+      "fallback": null,
+      "status": "PENDING",
+      "depends_on": "Q1-ACTIVE-EXECUTOR-ELECTION",
+      "quorum": "log_only_if_active_ok"
+    },
+    {
+      "id": "Q3-UNATTENDED-ACCEPTANCE-MATRIX",
+      "track": "_QUORUM",
+      "gate": "A0_AUTOMATION_CLOSED",
+      "primary": "gateway",
+      "fallback": null,
+      "status": "PENDING_LIVE_EVIDENCE",
+      "depends_on": "Q2-FAILOVER-HEALTHCHECK",
+      "quorum": "requires_gpt_plus_one_executor",
+      "required_scenarios": 16,
+      "may_mark_pass_without_live_evidence": false
    }
  ]
 }
@@ -8,11 +8,13 @@
      "strong_model_materially_better"
    ],
    "default_orchestrator": "openclaw",
-    "dual_required": [
-      "release_gate",
-      "crypto_or_identity_change",
-      "production_migration_or_rollback"
-    ],
+    "dynamic_executor_selection": {
+      "large_repetitive_infrastructure_monitoring": "openclaw",
+      "complex_code_security_architecture": "codex",
+      "both_healthy_active": "assign_by_preference",
+      "one_healthy_active": "use_healthy_only",
+      "both_unhealthy_blocked": true
+    },
    "max_attempts_per_work_unit": 2,
    "on_timeout": [
      "mark_TIMEOUT",
@@ -28,6 +30,7 @@
    ],
    "single_state_writer_lease_required": true,
    "strict_role_partition": false,
+    "state_writer": "gateway",
    "timeout_can_pass": false
  },
  "authoritative_state": {
@@ -51,7 +54,65 @@
      "result": "e789632e359e9232d8ae081e7c4f23ed7b615a9d"
    }
  },
+  "central_gpt": {
+    "required": "HEALTHY",
+    "health_definition": {
+      "live_probe_required": true,
+      "workspace_agent_validated_live_or_responses_validated_live": true,
+      "configuration_or_credential_presence_is_insufficient": true,
+      "unknown_is_not_healthy": true
+    },
+    "independent_review": {
+      "single_executor_mode": true,
+      "required_evidence": [
+        "committed_git_tree",
+        "full_commit_oid",
+        "test_logs_with_rc",
+        "gitleaks",
+        "artifact_digest",
+        "worktree_bare_gitea_oid",
+        "live_runtime_evidence"
+      ],
+      "model_judgment_insufficient": true
+    }
+  },
  "contract_id": "V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT",
+  "executor_quorum": {
+    "minimum_healthy": 1,
+    "candidates": [
+      "openclaw",
+      "codex"
+    ],
+    "quorum_pass": [
+      "GPT_HEALTHY_AND_OPENCLAW_HEALTHY",
+      "GPT_HEALTHY_AND_CODEX_HEALTHY"
+    ],
+    "quorum_block": [
+      "GPT_UNHEALTHY",
+      "GPT_HEALTHY_BUT_BOTH_EXECUTORS_UNAVAILABLE",
+      "GPT_OR_SOLE_EXECUTOR_UNKNOWN"
+    ]
+  },
+  "automation_closed_success_conditions": [
+    "CENTRAL_GPT_HEALTH=PASS",
+    "EXECUTOR_QUORUM=PASS",
+    "ACTIVE_EXECUTOR=openclaw|codex",
+    "EVENT_TRIGGER=PASS",
+    "POLLING=PASS",
+    "FAILOVER=PASS",
+    "LIFECYCLE_SYNC=PASS",
+    "TWO_CYCLE_UNATTENDED_LOOP=PASS",
+    "RESTART_RECOVERY=PASS",
+    "AUTOMATION_CLOSED=PASS"
+  ],
+  "gateway_lifecycle_events": [
+    "TASK_DISPATCHED",
+    "TASK_STARTED",
+    "TASK_ENDED",
+    "DECISION_APPLIED",
+    "PROJECT_SUMMARY",
+    "PROJECT_COMPLETED"
+  ],
  "facts_policy": {
    "accept": [
      "committed_git_tree",
@@ -84,7 +145,7 @@
    "reuse_accident_worktree",
    "delete_live_worktree_to_clear_collision"
  ],
-  "generated_at": "2026-07-17T14:13:38+00:00",
+  "generated_at": "2026-07-18T04:58:00+00:00",
  "handoff": {
    "ack": "CONTRACT_ACK=<full_contract_commit_oid>",
    "advance": "independent_verification_required",
@@ -92,6 +153,7 @@
  },
  "join_gate": {
    "requires": [
+      "AUTOMATION_CLOSED",
      "RUNNER_RECOVERED",
      "B3_PASS",
      "B4_PASS",
@@ -114,17 +176,46 @@
    "user_step_by_step_confirmation_required": false
  },
  "release_gates": [
+    "AUTOMATION_CLOSED",
    "MAIN_MERGE",
    "STAGING_AND_CANARY",
    "PRODUCTION"
  ],
-  "schema_version": 3,
+  "schema_version": 4,
+  "self_healing": {
+    "priorities": {
+      "p0": "central_gpt_failure",
+      "p1": "both_executors_unavailable",
+      "p2": "active_executor_failed_with_healthy_standby",
+      "p3": "standby_executor_alone_failed"
+    },
+    "rules": {
+      "standby_failure_when_active_ok": {
+        "action": "log_only",
+        "block_business": false,
+        "create_blocking_gate": false,
+        "pause_release": false
+      },
+      "active_executor_failure": {
+        "action": "failover_to_standby",
+        "inherit_contract_commit": true,
+        "inherit_event_id": true,
+        "inherit_state_version": true,
+        "mark_original_task": "INTERRUPTED_AGENT_FAILOVER",
+        "new_executor_continues_from_checkpoint": true,
+        "block_duplicate_side_effects": true
+      },
+      "both_unavailable_block": "BLOCKED_NO_HEALTHY_EXECUTOR"
+    }
+  },
  "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"
+    "production_remains_V1R4C6_paused_until_production_pass",
+    "gateway_persists_authoritative_state",
+    "lifecycle_events_persisted_regardless_of_openclaw_online"
  ],
  "status": "ACTIVE",
  "tracks": {
@@ -1,55 +0,0 @@
---
-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.
@@ -5,11 +5,40 @@
  "contract_branch": "task/governance/v2r1-personal-weixin-production-contract-20260717",
  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
  "dispatch_commit": null,
+  "executor_quorum": {
+    "candidates": ["openclaw", "codex"],
+    "minimum_healthy": 1,
+    "openclaw": "UNKNOWN",
+    "codex": "UNKNOWN"
+  },
+  "active_executor": null,
+  "standby_executor": null,
+  "central_gpt": "UNKNOWN",
+  "automation_closed": {
+    "central_gpt_health": "UNKNOWN",
+    "executor_quorum": "UNKNOWN",
+    "active_executor": null,
+    "event_trigger": "UNKNOWN",
+    "polling": "UNKNOWN",
+    "failover": "UNKNOWN",
+    "lifecycle_sync": "UNKNOWN",
+    "two_cycle_unattended_loop": "UNKNOWN",
+    "restart_recovery": "UNKNOWN",
+    "result": "NOT_STARTED"
+  },
+  "lifecycle_events": [],
  "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"
+  "schema_version": 2,
+  "single_state_writer": "gateway",
+  "single_state_writer_lease_required": true,
+  "quorum_controller": {
+    "implementation": "CODE_COMMITTED_TESTED_PENDING_LIVE_ACCEPTANCE",
+    "live_acceptance": "NOT_STARTED",
+    "health_source": "ROOT_OWNED_METADATA_ONLY_LIVE_PROBE_RECEIPTS",
+    "unknown_counts_as_healthy": false
+  },
+  "updated_at": "2026-07-18T04:58:00+00:00"
 }
@@ -0,0 +1,119 @@
+# OpenClaw A0 Three-Party Self-Healing and Release Supervision
+
+CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
+
+## Mission
+
+Close the GPT/OpenClaw/Codex unattended loop before any main merge or production release. OpenClaw is the default state writer and long-running supervisor, but work may be reassigned dynamically under the shared contract.
+
+## Authoritative inputs
+
+- Governance contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
+- Governance dispatch head before this task: `9195470ceb78fd57ffface3dcc38f40b75efa9d8`
+- Gateway integration implementation: `c3984ca2b4388875d2875b4fdfdf95a2984f6ab3`
+- Gateway integration RESULT: `4e63b1559aa3fc07b1384154c97b2c90b84353ed`
+- Three-party self-healer candidate branch: `patch/a0-three-party-self-healing-20260718-20260718T043256`
+- Production baseline: `V1R4C6`
+- Release status: `paused`
+
+## Priority invariant
+
+Continuously evaluate exactly three parties:
+
+1. `central_gpt` — Orchestrator Gateway plus a valid Workspace Agent or Responses API decision channel.
+2. `openclaw` — OpenClaw watcher, lifecycle sync, durable task execution and heartbeat.
+3. `codex` — Codex watcher, auto-claim, lifecycle sync, durable task execution and heartbeat.
+
+When exactly two parties are HEALTHY and the third is explicitly UNHEALTHY, create a P0 repair work unit for the unique unhealthy party before every other task. UNKNOWN is not UNHEALTHY and must fail closed. Do not repair when fewer than two are healthy or any health result is ambiguous.
+
+## A0-SH1: supervise implementation completion
+
+Codex owns the high-risk code changes. Independently verify that the final implementation contains:
+
+- strict root-owned configuration validation and no symlink chain;
+- fixed built-in party adapters rather than arbitrary untrusted repair commands;
+- bounded stdout/stderr redaction so tokens and secrets cannot enter audit logs;
+- exclusive lease, cooldown, circuit breaker and stale lease recovery;
+- event trigger plus periodic polling;
+- OpenClaw lifecycle events: `TASK_DISPATCHED`, `TASK_STARTED`, `TASK_ENDED`, `DECISION_APPLIED`, `PROJECT_SUMMARY`, `PROJECT_COMPLETED`;
+- durable file outbox fallback when HTTP callback is unavailable;
+- systemd service and timer;
+- health and repair adapters for central GPT, OpenClaw and Codex;
+- install, verify and rollback scripts;
+- deterministic tests for all health combinations and failures.
+
+Reject the current candidate if it only contains `automation/three_party_self_healer.py` without these controls.
+
+## A0-SH2: live three-party validation
+
+After Codex implementation RESULT passes, deploy only to a non-production acceptance context and run:
+
+1. All-three healthy steady state for at least three poll intervals.
+2. GPT unhealthy while OpenClaw and Codex healthy: GPT repair becomes P0 and recovers.
+3. OpenClaw unhealthy while GPT and Codex healthy: OpenClaw repair becomes P0 and recovers.
+4. Codex unhealthy while GPT and OpenClaw healthy: Codex repair becomes P0 and recovers.
+5. One healthy plus two unhealthy: no repair; fail closed.
+6. Any UNKNOWN health: no repair; fail closed.
+7. Repair timeout: retry with backoff, then circuit open.
+8. Service restart: state, lease and pending work recover.
+9. HTTP event path unavailable: file inbox polling recovers.
+10. Git webhook unavailable: Git ref polling detects change.
+11. Duplicate event ID: no duplicate dispatch.
+12. Main Workspace Agent channel unavailable: Responses fallback takes over only with a valid approved credential.
+13. OpenClaw receives every task start/end and project summary notification.
+14. Two consecutive unattended decision/execution/result cycles complete with no chat message or manual dispatch.
+
+Do not use dummy API keys or environment-variable presence as proof of a valid central decision channel.
+
+## A0-SH3: authoritative closure
+
+Only after independent verification of the above create one append-only state update and an exact-one RESULT file. Required output:
+
+```text
+AUTOMATION_CLOSED=PASS
+CENTRAL_GPT_HEALTH=PASS
+OPENCLAW_AUTO_CLAIM=PASS
+CODEX_AUTO_CLAIM=PASS
+THREE_PARTY_SELF_HEALING=PASS
+EVENT_TRIGGER=PASS
+POLLING=PASS
+OPENCLAW_LIFECYCLE_SYNC=PASS
+TWO_CYCLE_UNATTENDED_LOOP=PASS
+RESTART_RECOVERY=PASS
+ROLLBACK=PASS
+```
+
+Any missing item keeps `AUTOMATION_CLOSED=NO`.
+
+## Release continuation
+
+After and only after `AUTOMATION_CLOSED=PASS`, automatically resume the governed release sequence:
+
+1. Real B3 evidence intake and independent review.
+2. B4 implementation and validation.
+3. Full integration, regression, security and migration tests.
+4. Normal PR/GitOps merge to main; never direct-push main.
+5. Immutable release artifact, digest, SBOM and signature.
+6. Staging and rollback exercise.
+7. Canary and monitoring validation.
+8. Owner/admin GitOps production deployment using existing approved release controls.
+9. Production smoke tests and production RESULT-only commit.
+
+External facts that cannot be fabricated — real WeChat credentials/raw-wire, valid secret references and owner-only release approval — remain fail-closed blockers. The loop must monitor them and resume automatically when they become available.
+
+## Forbidden
+
+- reset, amend, rebase, force push;
+- direct push to main;
+- controller mutation;
+- secret disclosure;
+- dummy credentials presented as production credentials;
+- fabricated B3 evidence;
+- timeout or skipped checks recorded as PASS;
+- production release before `AUTOMATION_CLOSED=PASS`.
+
+Progress format:
+
+```text
+CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> AGENT=openclaw STATUS=<status> COMMIT=<oid-or-NONE> BLOCKER=<stable-reason-or-NONE>
+```
@@ -10,11 +10,22 @@ from orchestrator.action_monitor import ActionMonitor

 ROOT = Path(os.getenv('AIOS_GITOPS_ROOT', '/data/aios-gitops'))
 WS = Path(os.getenv('AIOS_GITOPS_WORKSPACE', str(ROOT / 'workspace')))
+sys.path.insert(0, str(WS))
+from services.aios_orchestrator_gateway.action_bridge import (
+    CallbackBridgeError,
+    submit_orchestration_decision,
+)
+
 SECRETS = Path(os.getenv('AIOS_GITOPS_SECRETS_DIR', str(ROOT / 'runtime' / 'secrets')))
 TOKEN_FILE = Path(os.getenv('AIOS_GITOPS_TOKEN_FILE', str(SECRETS / 'gitops.tokens')))
 HOST = os.getenv('AIOS_GITOPS_HOST', '127.0.0.1')
 PORT = int(os.getenv('AIOS_GITOPS_PORT', '18890'))
 OPENAPI = WS / 'docs' / 'gitops' / 'openapi-chatgpt-actions.yaml'
+ORCHESTRATOR_OPENAPI = WS / 'workspace-agent' / 'orchestrator-decision-action.openapi.yaml'
+ORCHESTRATOR_CONTRACT_COMMIT = os.getenv('AIOS_ORCHESTRATOR_CONTRACT_COMMIT', '')
+ORCHESTRATOR_CALLBACK_SECRET_REF = os.getenv('AIOS_ORCHESTRATOR_CALLBACK_SECRET_REF', '')
+_resolver = os.getenv('AIOS_ORCHESTRATOR_SECRET_REF_RESOLVER', '')
+ORCHESTRATOR_SECRET_REF_RESOLVER = Path(_resolver) if _resolver else None
 BARE_REPO = Path(os.getenv('AIOS_GITOPS_BARE_REPO', str(ROOT / 'runtime' / 'gitea-bin' / 'data' / 'repositories' / 'aios' / 'aios.git')))
 monitor = ActionMonitor()

@@ -184,6 +195,8 @@ class H(BaseHTTPRequestHandler):
            if path in ('/health','/api/health'): return self.sendj({'status':'ok','service':'chatgpt-gitops-adapter','workspace':str(WS)})
            if path in ('/api/openapi.yaml','/api/openapi-chatgpt-actions.yaml','/api/schema'):
                return self.sendt(OPENAPI.read_text(encoding='utf-8'), 'application/yaml; charset=utf-8') if OPENAPI.exists() else self.sendj({'error':'schema_missing'},404)
+            if path == '/api/orchestrator-decision.openapi.yaml':
+                return self.sendt(ORCHESTRATOR_OPENAPI.read_text(encoding='utf-8'), 'application/yaml; charset=utf-8') if ORCHESTRATOR_OPENAPI.exists() else self.sendj({'error':'schema_missing'},404)
            if path in ('/api','/api/status'): return self.sendj(api_spec())
            if path in ('/api/base','/api/baseline','/base'):
                data=jload(WS/'baseline'/'current.json', {}); data.setdefault('git', sh(['rev-parse','--short','HEAD'])['stdout']); return self.sendj(data)
@@ -216,6 +229,7 @@ class H(BaseHTTPRequestHandler):
    def do_POST(self):
        path=urllib.parse.urlparse(self.path).path.rstrip('/'); data=self.body()
        try:
+            if path == '/v1/orchestrator/decisions': return self.orchestration_decision(data)
            if path in ('/api/branch','/api/create-branch'): return self.branch(data)
            if path in ('/api/file','/api/commit-file'): return self.file(data)
            if path == '/api/commit': return self.commit(data)
@@ -229,6 +243,25 @@ class H(BaseHTTPRequestHandler):
        except Exception as e:
            monitor.record_failure('do_POST', str(e), {'path':path})
            return self.sendj({'error':str(e)},500)
+    def orchestration_decision(self,data):
+        a=self.need('orchestrator:decision')
+        if not a: return
+        if not ORCHESTRATOR_CONTRACT_COMMIT or ORCHESTRATOR_SECRET_REF_RESOLVER is None:
+            return self.sendj({'error':'CALLBACK_CONFIGURATION_UNAVAILABLE'},503)
+        try:
+            result=submit_orchestration_decision(
+                data,
+                contract_commit=ORCHESTRATOR_CONTRACT_COMMIT,
+                secret_ref=ORCHESTRATOR_CALLBACK_SECRET_REF,
+                resolver=ORCHESTRATOR_SECRET_REF_RESOLVER,
+            )
+        except CallbackBridgeError as exc:
+            return self.sendj({'error':exc.category},exc.status_code)
+        except Exception:
+            # The outer request handler records exception strings. Keep this
+            # secret-bearing boundary metadata-only even for unexpected faults.
+            return self.sendj({'error':'CALLBACK_INTERNAL_ERROR'},500)
+        return self.sendj(result,202)
    def branch(self,data):
        a=self.need('branch:create');
        if not a: return
@@ -390,9 +423,9 @@ return{vs,cmds,ar,cmd,cdet,tot,sok,sfail,runs,baseline,refreshMsg,loadCmd,selVer


 def api_spec():
-    return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /admin/ai-evolution'],'write':['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl'],'policy':{'bearer_required_for_write':True,'direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True}}
+    return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','orchestrator_openapi':'/api/orchestrator-decision.openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /admin/ai-evolution'],'write':['POST /v1/orchestrator/decisions','POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl'],'policy':{'bearer_required_for_write':True,'orchestrator_decision_scope':'orchestrator:decision','orchestrator_callback_credential':'secret_ref_only','direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True}}

 def main():
    print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
    ThreadingHTTPServer((HOST,PORT), H).serve_forever()
-if __name__ == '__main__': main()
+if __name__ == '__main__': main()
@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""Prove two real central-decision to Gateway-dispatch to Codex lifecycle cycles."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import sqlite3
+import time
+import uuid
+
+from health_probe_common import (
+    ProbeError, gateway_base, load_config, read_token, request_json, state_version,
+)
+
+
+REQUIRED_MARKERS = {"CODEX_AUTO_CLAIM", "TASK_STARTED", "TASK_EXECUTED", "TASK_ENDED", "EVENT_POSTED"}
+
+
+def marker_counts(path: Path, command_id: str) -> dict[str, int]:
+    counts = {marker: 0 for marker in REQUIRED_MARKERS}
+    with path.open("r", encoding="utf-8") as handle:
+        for line in handle:
+            try:
+                record = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            marker = record.get("marker")
+            if record.get("command_id") == command_id and marker in counts:
+                counts[marker] += 1
+    return counts
+
+
+def watcher_status(database: Path, command_id: str) -> str | None:
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=2) as db:
+        row = db.execute("SELECT status FROM commands WHERE command_id=?", (command_id,)).fetchone()
+    return None if row is None else str(row[0])
+
+
+def wait_for_cycle(config: dict[str, object], token: str, cycle: int) -> tuple[str, str]:
+    endpoint = str(config["gateway_event_url"])
+    project = str(config["project_key"])
+    event_id = f"a0-cycle-{cycle}-" + uuid.uuid4().hex
+    work_unit_id = f"A0-CODEX-CYCLE-{cycle}-" + uuid.uuid4().hex[:16]
+    event = {
+        "schema_version": 1,
+        "project_key": project,
+        "contract_commit": config["contract_commit"],
+        "event_id": event_id,
+        "source_agent": "system",
+        "event_type": "a0_unattended_acceptance_requested",
+        "work_unit_id": work_unit_id,
+        "status": "READY",
+        "state_version": state_version(endpoint, project, token),
+        "implementation_commit": None,
+        "result_commit": None,
+        "blocker": None,
+        "evidence_refs": [],
+        "payload": {
+            "acceptance_contract": {
+                "verdict": "DISPATCH",
+                "work_unit_id": work_unit_id,
+                "track_id": "A0_AUTOMATION",
+                "gate_id": "A0_AUTOMATION_CLOSED",
+                "primary_agent": "codex",
+                "fallback_agent": "openclaw",
+                "clean_base_commit": config["probe_clean_base_commit"],
+                "allowed_paths": [],
+                "required_checks": ["acceptance_probe"],
+                "forbidden_operations": ["production_mutation", "git_mutation"],
+                "instruction": "Run the fixed side-effect-free acceptance probe only.",
+            }
+        },
+    }
+    request_json(endpoint, token, method="POST", payload=event)
+    deadline = time.monotonic() + int(config["cycle_timeout_seconds"])
+    channel = None
+    while time.monotonic() < deadline:
+        _, status = request_json(endpoint + "/" + event_id, token)
+        if status.get("status") == "TRIGGER_FAILED":
+            raise ProbeError("central_trigger_failed")
+        if status.get("status") == "DECIDED" and status.get("decision_applied") is True:
+            channel = status.get("trigger_channel")
+            break
+        time.sleep(float(config["poll_interval_seconds"]))
+    if channel not in {"workspace_agent", "responses_api"}:
+        raise ProbeError("central_decision_timeout")
+
+    database = Path(str(config["codex_database_path"]))
+    audit = Path(str(config["codex_audit_log"]))
+    while time.monotonic() < deadline:
+        if database.exists() and audit.exists() and watcher_status(database, work_unit_id) == "COMPLETED":
+            counts = marker_counts(audit, work_unit_id)
+            if all(counts[marker] == 1 for marker in REQUIRED_MARKERS):
+                for suffix in ("started", "ended"):
+                    _, lifecycle = request_json(endpoint + "/" + work_unit_id + ":" + suffix, token)
+                    if lifecycle.get("event_id") != work_unit_id + ":" + suffix:
+                        raise ProbeError("gateway_lifecycle_missing")
+                return event_id, str(channel)
+        time.sleep(float(config["poll_interval_seconds"]))
+    raise ProbeError("gateway_dispatch_or_codex_lifecycle_timeout")
+
+
+def project_summaries_since(database: Path, project: str, since: int) -> int:
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=2) as db:
+        row = db.execute(
+            "SELECT COUNT(*) FROM lifecycle WHERE project_key=? AND kind='PROJECT_SUMMARY' AND created_at>=?",
+            (project, since),
+        ).fetchone()
+    return 0 if row is None else int(row[0])
+
+
+def run(config_path: Path) -> int:
+    config = load_config(config_path)
+    endpoint = str(config["gateway_event_url"])
+    gateway_base(endpoint)
+    token = read_token(Path(str(config["gateway_token_file"])))
+    started = int(time.time())
+    results = [wait_for_cycle(config, token, cycle) for cycle in (1, 2)]
+    if len({item[0] for item in results}) != 2:
+        raise ProbeError("duplicate_cycle_event")
+    if project_summaries_since(
+        Path(str(config["gateway_database_path"])), str(config["project_key"]), started
+    ) < 2:
+        raise ProbeError("project_summary_not_persisted")
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        return run(args.config)
+    except (KeyError, OSError, ProbeError, sqlite3.Error, ValueError):
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Prove a real central-model decision through the authenticated Gateway."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import stat
+import sys
+import time
+from pathlib import Path
+import uuid
+
+from health_probe_common import (
+    ProbeError, atomic_receipt, gateway_base, load_config, read_token,
+    request_json, state_version,
+)
+
+PENDING_PATH = Path("/run/aios-health/central-gpt.pending.json")
+
+
+def _load_pending(
+    path: Path, contract: str, *, owners: set[int] | None = None
+) -> tuple[str, int] | None:
+    allowed_owners = {0} if owners is None else owners
+    if not path.exists():
+        return None
+    info = path.lstat()
+    if (
+        not path.is_absolute()
+        or stat.S_ISLNK(info.st_mode)
+        or not stat.S_ISREG(info.st_mode)
+        or stat.S_IMODE(info.st_mode) & ~0o640
+        or (os.name == "posix" and info.st_uid not in allowed_owners)
+        or path.resolve(strict=True) != Path(os.path.abspath(path))
+    ):
+        raise ProbeError("unsafe_pending_marker")
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict) or set(value) != {"contract_commit", "event_id", "state_version"}:
+        raise ProbeError("invalid_pending_marker")
+    event_id = value["event_id"]
+    version = value["state_version"]
+    if (
+        value["contract_commit"] != contract
+        or not isinstance(event_id, str)
+        or not event_id.startswith("health-central-")
+        or len(event_id) != 47
+        or not isinstance(version, int)
+        or version < 0
+    ):
+        raise ProbeError("invalid_pending_marker")
+    return event_id, version
+
+
+def _clear_pending(path: Path) -> None:
+    if not path.exists():
+        return
+    info = path.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise ProbeError("unsafe_pending_marker")
+    path.unlink()
+
+
+def _safe_reason(exc: BaseException) -> str:
+    reason = str(exc)
+    if reason and len(reason) <= 64 and all(char.islower() or char.isdigit() or char == "_" for char in reason):
+        return reason
+    return "probe_error"
+
+
+def run(config_path: Path, pending_path: Path = PENDING_PATH) -> int:
+    config = load_config(config_path)
+    endpoint = config["gateway_event_url"]
+    gateway_base(endpoint)
+    token = read_token(Path(config["gateway_token_file"]))
+    project = config["project_key"]
+    contract = config["contract_commit"]
+    pending = _load_pending(pending_path, contract)
+    is_new = pending is None
+    if is_new:
+        event_id = "health-central-" + uuid.uuid4().hex
+        version = state_version(endpoint, project, token)
+        atomic_receipt(
+            pending_path,
+            {"contract_commit": contract, "event_id": event_id, "state_version": version},
+        )
+    else:
+        event_id, version = pending
+    event = {
+        "schema_version": 1,
+        "project_key": project,
+        "contract_commit": contract,
+        "event_id": event_id,
+        "source_agent": "system",
+        "event_type": "central_gpt_health_probe",
+        "work_unit_id": "CENTRAL-GPT-HEALTH",
+        "status": "READY",
+        "state_version": version,
+        "implementation_commit": None,
+        "result_commit": None,
+        "blocker": None,
+        "evidence_refs": [],
+        "payload": {"probe": "side_effect_free", "nonce": event_id},
+    }
+    should_post = is_new
+    if not is_new:
+        try:
+            _, observed = request_json(endpoint + "/" + event_id, token)
+        except ProbeError as exc:
+            if str(exc) != "gateway_http_404":
+                raise
+            should_post = True
+    else:
+        observed = None
+    if should_post:
+        try:
+            status, _ = request_json(endpoint, token, method="POST", payload=event)
+        except ProbeError as exc:
+            if str(exc) == "gateway_fail_closed_409":
+                _clear_pending(pending_path)
+            raise
+        if status not in {200, 202}:
+            raise ProbeError("event_not_accepted")
+        observed = None
+    deadline = time.monotonic() + int(config["central_timeout_seconds"])
+    while time.monotonic() < deadline:
+        _, observed = request_json(endpoint + "/" + event_id, token)
+        if observed.get("status") in {"DECIDED", "TRIGGER_FAILED", "DEAD_LETTER"}:
+            break
+        time.sleep(float(config["poll_interval_seconds"]))
+    if isinstance(observed, dict) and observed.get("status") == "DEAD_LETTER":
+        _clear_pending(pending_path)
+        raise ProbeError("central_event_dead_letter")
+    if not isinstance(observed, dict) or observed.get("status") != "DECIDED" or observed.get("decision_applied") is not True:
+        raise ProbeError("central_decision_not_proven")
+    channel = observed.get("trigger_channel")
+    if channel not in {"workspace_agent", "responses_api"}:
+        raise ProbeError("central_channel_not_proven")
+    _clear_pending(pending_path)
+    now = int(time.time())
+    atomic_receipt(
+        Path(config["central_receipt_path"]),
+        {
+            "subject": "central_gpt",
+            "health": "HEALTHY",
+            "live_probe_succeeded": True,
+            "observed_at_epoch": now,
+            "event_id": event_id,
+            "decision_applied": True,
+            "state_version_after": observed["state_version_after"],
+            "workspace_agent": {"validated_live": channel == "workspace_agent"},
+            "responses_api": {"validated_live": channel == "responses_api"},
+        },
+    )
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        return run(args.config)
+    except (KeyError, OSError, ProbeError, ValueError) as exc:
+        print(f"CENTRAL_GPT_PROBE=BLOCKED reason={_safe_reason(exc)}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""Queue one fixed no-side-effect command and prove the watcher lifecycle."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import sqlite3
+import stat
+import time
+import uuid
+
+from health_probe_common import (
+    ProbeError, atomic_receipt, gateway_base, load_config, read_token,
+    request_json, state_version,
+)
+
+
+MARKERS = {"CODEX_AUTO_CLAIM", "TASK_STARTED", "TASK_EXECUTED", "TASK_ENDED", "EVENT_POSTED"}
+
+
+def _write_command(path: Path, value: dict[str, object]) -> None:
+    directory = path.parent
+    info = directory.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) & 0o002:
+        raise ProbeError("unsafe_command_queue")
+    data = (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o640)
+    try:
+        os.write(fd, data)
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def _status(database: Path, command_id: str) -> str | None:
+    with sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=2) as db:
+        row = db.execute("SELECT status FROM commands WHERE command_id=?", (command_id,)).fetchone()
+    return None if row is None else str(row[0])
+
+
+def _markers(path: Path, command_id: str) -> dict[str, int]:
+    result = {name: 0 for name in MARKERS}
+    with path.open("r", encoding="utf-8") as handle:
+        for line in handle:
+            try:
+                item = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if item.get("command_id") == command_id and item.get("marker") in result:
+                result[item["marker"]] += 1
+    return result
+
+
+def run(config_path: Path) -> int:
+    config = load_config(config_path)
+    endpoint = config["gateway_event_url"]
+    gateway_base(endpoint)
+    token = read_token(Path(config["gateway_token_file"]))
+    command_id = "health-codex-" + uuid.uuid4().hex
+    version = state_version(endpoint, config["project_key"], token)
+    command = {
+        "schema_version": 1,
+        "command_id": command_id,
+        "project_key": config["project_key"],
+        "work_unit_id": command_id,
+        "contract_commit": config["contract_commit"],
+        "state_version": version,
+        "execution": {"kind": "acceptance_probe", "input": "fixed-side-effect-free-health-probe-v1"},
+    }
+    _write_command(Path(config["codex_command_queue"]) / f"{command_id}.json", command)
+    deadline = time.monotonic() + int(config["codex_timeout_seconds"])
+    database = Path(config["codex_database_path"])
+    audit = Path(config["codex_audit_log"])
+    while time.monotonic() < deadline:
+        if database.exists() and audit.exists() and _status(database, command_id) == "COMPLETED":
+            counts = _markers(audit, command_id)
+            if all(counts[name] == 1 for name in MARKERS):
+                for suffix in ("started", "ended"):
+                    _, event = request_json(endpoint + "/" + command_id + ":" + suffix, token)
+                    if event.get("event_id") != command_id + ":" + suffix:
+                        raise ProbeError("gateway_lifecycle_missing")
+                atomic_receipt(
+                    Path(config["codex_receipt_path"]),
+                    {
+                        "subject": "codex",
+                        "health": "HEALTHY",
+                        "live_probe_succeeded": True,
+                        "observed_at_epoch": int(time.time()),
+                        "command_id": command_id,
+                        "lifecycle_complete": True,
+                        "duplicate_side_effects": False,
+                    },
+                )
+                return 0
+        time.sleep(float(config["poll_interval_seconds"]))
+    raise ProbeError("codex_lifecycle_not_proven")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        return run(args.config)
+    except (KeyError, OSError, ProbeError, sqlite3.Error, ValueError):
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,940 @@
+#!/usr/bin/env python3
+"""Durable GPT + one-executor quorum and failover controller.
+
+The controller persists decisions before dispatch and never stores credentials.
+Health is accepted only from metadata-only, live-probe receipts.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import sqlite3
+import stat
+import subprocess
+import tempfile
+import time
+import urllib.error
+import urllib.request
+import urllib.parse
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+from typing import Any, Callable, Iterable, Mapping
+
+
+ALLOWED_LIFECYCLE_EVENTS = frozenset(
+    {
+        "TASK_DISPATCHED",
+        "TASK_STARTED",
+        "TASK_ENDED",
+        "DECISION_APPLIED",
+        "PROJECT_SUMMARY",
+        "PROJECT_COMPLETED",
+    }
+)
+EXECUTORS = ("openclaw", "codex")
+SECRET_KEYS = frozenset(
+    {
+        "secret",
+        "token",
+        "password",
+        "api_key",
+        "authorization",
+        "credential",
+        "private_key",
+        "environment",
+        "env",
+        "argv",
+        "stdout",
+        "stderr",
+    }
+)
+
+
+class Health(str, Enum):
+    HEALTHY = "HEALTHY"
+    UNHEALTHY = "UNHEALTHY"
+    UNKNOWN = "UNKNOWN"
+
+
+class ControllerError(RuntimeError):
+    """Fail-closed controller error."""
+
+
+class EventConflict(ControllerError):
+    """An event id was reused with a different payload."""
+
+
+class StaleStateVersion(ControllerError):
+    """A dispatch attempted to overwrite a newer task state."""
+
+
+@dataclass(frozen=True)
+class QuorumDecision:
+    allowed: bool
+    priority: str
+    status: str
+    active_executor: str | None
+    standby_executor: str | None
+    repair_required: bool
+
+
+def canonical_json(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+
+
+def digest_json(value: Any) -> str:
+    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def _contains_forbidden_key(value: Any) -> bool:
+    if isinstance(value, Mapping):
+        for key, child in value.items():
+            normalized = str(key).strip().lower().replace("-", "_")
+            if normalized in SECRET_KEYS or _contains_forbidden_key(child):
+                return True
+    elif isinstance(value, list):
+        return any(_contains_forbidden_key(item) for item in value)
+    return False
+
+
+def validate_metadata_only(value: Mapping[str, Any]) -> None:
+    if _contains_forbidden_key(value):
+        raise ControllerError("secret-bearing or output-bearing fields are forbidden")
+
+
+def atomic_json_write(path: Path, value: Mapping[str, Any], mode: int = 0o600) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    validate_secure_directory(path.parent)
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
+    try:
+        os.fchmod(fd, mode)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(canonical_json(value))
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def validate_secure_directory(path: Path, *, expected_uid: int | None = None) -> None:
+    info = path.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
+        raise ControllerError(f"unsafe directory type rejected: {path}")
+    if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
+        raise ControllerError(f"symlinked directory component rejected: {path}")
+    if os.name == "posix" and info.st_mode & 0o022:
+        raise ControllerError(f"group/world-writable directory rejected: {path}")
+    if expected_uid is not None and info.st_uid != expected_uid:
+        raise ControllerError(f"unexpected owner for directory {path}")
+
+
+def read_secure_json(path: Path, *, expected_uid: int | None = None) -> dict[str, Any]:
+    info = path.lstat()
+    if stat.S_ISLNK(info.st_mode):
+        raise ControllerError(f"symlink rejected: {path}")
+    if not stat.S_ISREG(info.st_mode):
+        raise ControllerError(f"not a regular file: {path}")
+    if os.name == "posix" and info.st_mode & 0o022:
+        raise ControllerError(f"group/world-writable file rejected: {path}")
+    validate_secure_directory(path.parent, expected_uid=expected_uid)
+    if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
+        raise ControllerError(f"symlinked file component rejected: {path}")
+    if expected_uid is not None and info.st_uid != expected_uid:
+        raise ControllerError(f"unexpected owner for {path}")
+    with path.open("r", encoding="utf-8") as handle:
+        value = json.load(handle)
+    if not isinstance(value, dict):
+        raise ControllerError(f"JSON object required: {path}")
+    validate_metadata_only(value)
+    return value
+
+
+def read_gateway_token(path: Path) -> str:
+    if not path.is_absolute():
+        raise ControllerError("gateway token path must be absolute")
+    try:
+        info = path.lstat()
+    except FileNotFoundError as exc:
+        raise ControllerError("gateway token unavailable") from exc
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise ControllerError("gateway token must be a regular non-symlink file")
+    if stat.S_IMODE(info.st_mode) & ~0o640:
+        raise ControllerError("gateway token mode exceeds 0640")
+    if os.name == "posix" and info.st_uid not in {0, os.geteuid()}:
+        raise ControllerError("gateway token owner is not approved")
+    if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
+        raise ControllerError("gateway token has a symlinked path component")
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    fd = os.open(path, flags)
+    try:
+        opened = os.fstat(fd)
+        if not stat.S_ISREG(opened.st_mode) or opened.st_ino != info.st_ino or opened.st_dev != info.st_dev:
+            raise ControllerError("gateway token changed during validation")
+        data = os.read(fd, 16_385)
+        if os.read(fd, 1):
+            raise ControllerError("gateway token too large")
+    finally:
+        os.close(fd)
+    if not data or len(data) > 16_384 or b"\n" in data or b"\r" in data:
+        raise ControllerError("gateway token format rejected")
+    try:
+        token = data.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ControllerError("gateway token encoding rejected") from exc
+    if not token:
+        raise ControllerError("gateway token unavailable")
+    return token
+
+
+def validate_health_receipt(
+    receipt: Mapping[str, Any],
+    subject: str,
+    *,
+    now: float | None = None,
+    max_age_seconds: int = 60,
+) -> Health:
+    validate_metadata_only(receipt)
+    if receipt.get("subject") != subject:
+        return Health.UNKNOWN
+    try:
+        health = Health(str(receipt.get("health", "UNKNOWN")))
+    except ValueError:
+        return Health.UNKNOWN
+    if health is not Health.HEALTHY:
+        return health
+    if receipt.get("live_probe_succeeded") is not True:
+        return Health.UNKNOWN
+    observed_at = receipt.get("observed_at_epoch")
+    if not isinstance(observed_at, (int, float)):
+        return Health.UNKNOWN
+    now = time.time() if now is None else now
+    if observed_at > now + 5 or now - observed_at > max_age_seconds:
+        return Health.UNKNOWN
+    if subject == "central_gpt":
+        workspace = receipt.get("workspace_agent")
+        responses = receipt.get("responses_api")
+        workspace_ok = isinstance(workspace, Mapping) and workspace.get("validated_live") is True
+        responses_ok = isinstance(responses, Mapping) and responses.get("validated_live") is True
+        if not (workspace_ok or responses_ok):
+            return Health.UNKNOWN
+    return Health.HEALTHY
+
+
+def choose_executor(workload: str) -> str:
+    if workload in {"complex_code", "security_boundary", "architecture", "complex_code_security_architecture"}:
+        return "codex"
+    return "openclaw"
+
+
+def evaluate_quorum(
+    central_gpt: Health,
+    openclaw: Health,
+    codex: Health,
+    workload: str,
+) -> QuorumDecision:
+    if central_gpt is not Health.HEALTHY:
+        status = "BLOCKED_CENTRAL_GPT_UNKNOWN" if central_gpt is Health.UNKNOWN else "BLOCKED_CENTRAL_GPT_UNHEALTHY"
+        return QuorumDecision(False, "P0", status, None, None, True)
+
+    healthy = [name for name, health in (("openclaw", openclaw), ("codex", codex)) if health is Health.HEALTHY]
+    if not healthy:
+        has_unknown = Health.UNKNOWN in {openclaw, codex}
+        status = "BLOCKED_EXECUTOR_HEALTH_UNKNOWN" if has_unknown else "BLOCKED_NO_HEALTHY_EXECUTOR"
+        return QuorumDecision(False, "P1", status, None, None, True)
+
+    if len(healthy) == 1:
+        active = healthy[0]
+        standby = "codex" if active == "openclaw" else "openclaw"
+        return QuorumDecision(True, "P3", "EXECUTOR_QUORUM_PASS", active, standby, False)
+
+    active = choose_executor(workload)
+    standby = "codex" if active == "openclaw" else "openclaw"
+    return QuorumDecision(True, "NONE", "EXECUTOR_QUORUM_PASS", active, standby, False)
+
+
+class QuorumStore:
+    def __init__(self, path: Path):
+        self.path = path
+        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        expected_uid = os.geteuid() if os.name == "posix" else None
+        validate_secure_directory(path.parent, expected_uid=expected_uid)
+        if path.exists() or path.is_symlink():
+            info = path.lstat()
+            if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
+                raise ControllerError(f"unsafe database path rejected: {path}")
+            if expected_uid is not None and info.st_uid != expected_uid:
+                raise ControllerError(f"unexpected database owner: {path}")
+        self.db = sqlite3.connect(str(path), isolation_level=None, timeout=30)
+        if os.name == "posix":
+            path.chmod(0o600)
+        self.db.row_factory = sqlite3.Row
+        self.db.execute("PRAGMA journal_mode=WAL")
+        self.db.execute("PRAGMA synchronous=FULL")
+        self._initialize()
+
+    def close(self) -> None:
+        self.db.close()
+
+    def _initialize(self) -> None:
+        self.db.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS events (
+                event_id TEXT PRIMARY KEY,
+                payload_digest TEXT NOT NULL,
+                payload_json TEXT NOT NULL,
+                source TEXT NOT NULL,
+                status TEXT NOT NULL,
+                created_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS tasks (
+                work_unit TEXT PRIMARY KEY,
+                contract_commit TEXT NOT NULL,
+                event_id TEXT NOT NULL,
+                state_version INTEGER NOT NULL,
+                workload TEXT NOT NULL,
+                status TEXT NOT NULL,
+                active_executor TEXT,
+                standby_executor TEXT,
+                checkpoint TEXT,
+                lease_owner TEXT,
+                lease_until REAL,
+                interrupted_executor TEXT,
+                recovery_priority TEXT NOT NULL,
+                updated_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS lifecycle (
+                sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+                work_unit TEXT NOT NULL,
+                event_type TEXT NOT NULL,
+                executor TEXT,
+                state_version INTEGER NOT NULL,
+                metadata_json TEXT NOT NULL,
+                created_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS effects (
+                effect_id TEXT PRIMARY KEY,
+                work_unit TEXT NOT NULL,
+                payload_digest TEXT NOT NULL,
+                status TEXT NOT NULL,
+                created_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS outbox (
+                outbox_id TEXT PRIMARY KEY,
+                destination TEXT NOT NULL,
+                payload_digest TEXT NOT NULL,
+                payload_json TEXT NOT NULL,
+                status TEXT NOT NULL,
+                attempts INTEGER NOT NULL DEFAULT 0,
+                next_attempt REAL NOT NULL,
+                created_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS blockers (
+                blocker_key TEXT PRIMARY KEY,
+                occurrences INTEGER NOT NULL,
+                status TEXT NOT NULL,
+                updated_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS observed_refs (
+                ref_name TEXT PRIMARY KEY,
+                oid TEXT NOT NULL,
+                updated_at REAL NOT NULL
+            );
+            CREATE TABLE IF NOT EXISTS summaries (
+                work_unit TEXT PRIMARY KEY,
+                summary_digest TEXT NOT NULL,
+                summary_json TEXT NOT NULL,
+                created_at REAL NOT NULL
+            );
+            CREATE UNIQUE INDEX IF NOT EXISTS lifecycle_once
+            ON lifecycle(work_unit, event_type, COALESCE(executor, ''), state_version);
+            """
+        )
+
+    def ingest_event(self, event_id: str, payload: Mapping[str, Any], source: str) -> bool:
+        validate_metadata_only(payload)
+        encoded = canonical_json(payload)
+        digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+        existing = self.db.execute("SELECT payload_digest FROM events WHERE event_id=?", (event_id,)).fetchone()
+        if existing:
+            if existing["payload_digest"] != digest:
+                self.db.execute("UPDATE events SET status='DEAD_LETTER_CONFLICT' WHERE event_id=?", (event_id,))
+                raise EventConflict(f"event id payload conflict: {event_id}")
+            return False
+        self.db.execute(
+            "INSERT INTO events VALUES(?,?,?,?,?,?)",
+            (event_id, digest, encoded, source, "ACCEPTED", time.time()),
+        )
+        return True
+
+    def upsert_task(
+        self,
+        *,
+        work_unit: str,
+        contract_commit: str,
+        event_id: str,
+        state_version: int,
+        workload: str,
+        decision: QuorumDecision,
+        checkpoint: str | None = None,
+    ) -> str:
+        now = time.time()
+        self.db.execute("BEGIN IMMEDIATE")
+        try:
+            current = self.db.execute("SELECT * FROM tasks WHERE work_unit=?", (work_unit,)).fetchone()
+            if current and state_version < current["state_version"]:
+                raise StaleStateVersion(f"stale state version for {work_unit}")
+            interrupted: str | None = None
+            recovery_priority = decision.priority
+            status = "BLOCKED"
+            if decision.allowed:
+                status = "READY"
+                if current and current["active_executor"] and current["active_executor"] != decision.active_executor:
+                    interrupted = current["active_executor"]
+                    recovery_priority = "P2"
+                    status = "INTERRUPTED_AGENT_FAILOVER"
+                    if checkpoint is None:
+                        checkpoint = current["checkpoint"]
+            if current:
+                if contract_commit != current["contract_commit"] or event_id != current["event_id"]:
+                    raise ControllerError("failover must preserve contract_commit and event_id")
+                if (
+                    current["state_version"] == state_version
+                    and current["active_executor"] == decision.active_executor
+                    and current["status"] in {"RUNNING", "COMPLETED"}
+                ):
+                    self.db.execute("COMMIT")
+                    return current["status"]
+                self.db.execute(
+                    """UPDATE tasks SET state_version=?, workload=?, status=?, active_executor=?,
+                    standby_executor=?, checkpoint=?, interrupted_executor=?, recovery_priority=?, updated_at=? WHERE work_unit=?""",
+                    (
+                        state_version,
+                        workload,
+                        status,
+                        decision.active_executor,
+                        decision.standby_executor,
+                        checkpoint,
+                        interrupted,
+                        recovery_priority,
+                        now,
+                        work_unit,
+                    ),
+                )
+            else:
+                self.db.execute(
+                    "INSERT INTO tasks VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        work_unit,
+                        contract_commit,
+                        event_id,
+                        state_version,
+                        workload,
+                        status,
+                        decision.active_executor,
+                        decision.standby_executor,
+                        checkpoint,
+                        None,
+                        None,
+                        interrupted,
+                        recovery_priority,
+                        now,
+                    ),
+                )
+            self.db.execute("COMMIT")
+            return status
+        except Exception:
+            self.db.execute("ROLLBACK")
+            raise
+
+    def claim_lease(self, work_unit: str, worker: str, ttl_seconds: int, now: float | None = None) -> bool:
+        now = time.time() if now is None else now
+        self.db.execute("BEGIN IMMEDIATE")
+        try:
+            task = self.db.execute(
+                "SELECT active_executor, status, lease_owner, lease_until FROM tasks WHERE work_unit=?",
+                (work_unit,),
+            ).fetchone()
+            if not task:
+                raise ControllerError(f"unknown work unit: {work_unit}")
+            if worker != task["active_executor"]:
+                self.db.execute("ROLLBACK")
+                return False
+            if task["status"] in {"BLOCKED", "COMPLETED"}:
+                self.db.execute("ROLLBACK")
+                return False
+            lease_live = task["lease_owner"] and task["lease_until"] and task["lease_until"] > now
+            if lease_live and task["lease_owner"] != worker:
+                self.db.execute("ROLLBACK")
+                return False
+            self.db.execute(
+                "UPDATE tasks SET lease_owner=?, lease_until=?, status='RUNNING', updated_at=? WHERE work_unit=?",
+                (worker, now + ttl_seconds, now, work_unit),
+            )
+            self.db.execute("COMMIT")
+            return True
+        except Exception:
+            if self.db.in_transaction:
+                self.db.execute("ROLLBACK")
+            raise
+
+    def dispatch_task(
+        self,
+        *,
+        work_unit: str,
+        contract_commit: str,
+        event_id: str,
+        state_version: int,
+        workload: str,
+        decision: QuorumDecision,
+        checkpoint: str | None = None,
+    ) -> str:
+        status = self.upsert_task(
+            work_unit=work_unit,
+            contract_commit=contract_commit,
+            event_id=event_id,
+            state_version=state_version,
+            workload=workload,
+            decision=decision,
+            checkpoint=checkpoint,
+        )
+        if decision.allowed and status != "COMPLETED":
+            metadata = {
+                "contract_commit": contract_commit,
+                "event_id": event_id,
+                "status": status,
+            }
+            self.append_lifecycle(work_unit, "TASK_DISPATCHED", decision.active_executor, state_version, metadata)
+            self.enqueue(
+                f"dispatch:{work_unit}:{state_version}:{decision.active_executor}",
+                "gateway",
+                {"event_type": "TASK_DISPATCHED", "work_unit": work_unit, **metadata},
+            )
+        return status
+
+    def start_task(self, work_unit: str, executor: str, state_version: int, lease_ttl_seconds: int) -> bool:
+        if not self.claim_lease(work_unit, executor, lease_ttl_seconds):
+            return False
+        self.append_lifecycle(work_unit, "TASK_STARTED", executor, state_version, {"status": "RUNNING"})
+        self.enqueue(
+            f"started:{work_unit}:{state_version}:{executor}",
+            "gateway",
+            {"event_type": "TASK_STARTED", "work_unit": work_unit, "executor": executor, "state_version": state_version},
+        )
+        return True
+
+    def end_task(
+        self,
+        work_unit: str,
+        executor: str,
+        state_version: int,
+        result: str,
+        summary: Mapping[str, Any],
+        *,
+        project_completed: bool = False,
+    ) -> None:
+        validate_metadata_only(summary)
+        self.db.execute("BEGIN IMMEDIATE")
+        try:
+            task = self.db.execute("SELECT active_executor,state_version FROM tasks WHERE work_unit=?", (work_unit,)).fetchone()
+            if not task or task["active_executor"] != executor:
+                raise ControllerError("only the active executor may end a task")
+            if task["state_version"] != state_version:
+                raise StaleStateVersion(f"stale state version for {work_unit}")
+            self.db.execute(
+                "UPDATE tasks SET status='COMPLETED', lease_owner=NULL, lease_until=NULL, updated_at=? WHERE work_unit=?",
+                (time.time(), work_unit),
+            )
+            self.append_lifecycle(work_unit, "TASK_ENDED", executor, state_version, {"result": result})
+            self.append_lifecycle(work_unit, "DECISION_APPLIED", executor, state_version, {"result": result})
+            self.append_lifecycle(work_unit, "PROJECT_SUMMARY", executor, state_version, summary)
+            if project_completed:
+                self.append_lifecycle(work_unit, "PROJECT_COMPLETED", executor, state_version, {"result": result})
+            self.persist_summary(work_unit, summary)
+            outbound = (
+                ("TASK_ENDED", {"result": result}),
+                ("DECISION_APPLIED", {"result": result}),
+                ("PROJECT_SUMMARY", dict(summary)),
+            )
+            for event_type, metadata in outbound:
+                self.enqueue(
+                    f"{event_type.lower()}:{work_unit}:{state_version}:{executor}",
+                    "gateway",
+                    {
+                        "event_type": event_type,
+                        "work_unit": work_unit,
+                        "executor": executor,
+                        "state_version": state_version,
+                        **metadata,
+                    },
+                )
+            if project_completed:
+                self.enqueue(
+                    f"project_completed:{work_unit}:{state_version}:{executor}",
+                    "gateway",
+                    {
+                        "event_type": "PROJECT_COMPLETED",
+                        "work_unit": work_unit,
+                        "executor": executor,
+                        "state_version": state_version,
+                        "result": result,
+                    },
+                )
+            self.db.execute("COMMIT")
+        except Exception:
+            if self.db.in_transaction:
+                self.db.execute("ROLLBACK")
+            raise
+
+    def record_effect(self, effect_id: str, work_unit: str, payload: Mapping[str, Any]) -> bool:
+        validate_metadata_only(payload)
+        digest = digest_json(payload)
+        existing = self.db.execute("SELECT payload_digest FROM effects WHERE effect_id=?", (effect_id,)).fetchone()
+        if existing:
+            if existing["payload_digest"] != digest:
+                raise EventConflict(f"effect id payload conflict: {effect_id}")
+            return False
+        self.db.execute(
+            "INSERT INTO effects VALUES(?,?,?,?,?)",
+            (effect_id, work_unit, digest, "COMPLETED", time.time()),
+        )
+        return True
+
+    def append_lifecycle(
+        self,
+        work_unit: str,
+        event_type: str,
+        executor: str | None,
+        state_version: int,
+        metadata: Mapping[str, Any],
+    ) -> None:
+        if event_type not in ALLOWED_LIFECYCLE_EVENTS:
+            raise ControllerError(f"unsupported lifecycle event: {event_type}")
+        validate_metadata_only(metadata)
+        self.db.execute(
+            "INSERT OR IGNORE INTO lifecycle(work_unit,event_type,executor,state_version,metadata_json,created_at) VALUES(?,?,?,?,?,?)",
+            (work_unit, event_type, executor, state_version, canonical_json(metadata), time.time()),
+        )
+
+    def persist_summary(self, work_unit: str, summary: Mapping[str, Any]) -> None:
+        validate_metadata_only(summary)
+        encoded = canonical_json(summary)
+        digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+        self.db.execute(
+            "INSERT OR REPLACE INTO summaries VALUES(?,?,?,?)",
+            (work_unit, digest, encoded, time.time()),
+        )
+
+    def enqueue(self, outbox_id: str, destination: str, payload: Mapping[str, Any]) -> bool:
+        validate_metadata_only(payload)
+        encoded = canonical_json(payload)
+        digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+        existing = self.db.execute("SELECT payload_digest FROM outbox WHERE outbox_id=?", (outbox_id,)).fetchone()
+        if existing:
+            if existing["payload_digest"] != digest:
+                raise EventConflict(f"outbox id payload conflict: {outbox_id}")
+            return False
+        self.db.execute(
+            "INSERT INTO outbox VALUES(?,?,?,?,?,?,?,?)",
+            (outbox_id, destination, digest, encoded, "PENDING", 0, time.time(), time.time()),
+        )
+        return True
+
+    def record_blocker(self, blocker_key: str, threshold: int) -> str:
+        now = time.time()
+        row = self.db.execute("SELECT occurrences FROM blockers WHERE blocker_key=?", (blocker_key,)).fetchone()
+        occurrences = (row["occurrences"] if row else 0) + 1
+        status = "DEAD_LETTER" if occurrences > threshold else "RETRY"
+        self.db.execute(
+            "INSERT INTO blockers VALUES(?,?,?,?) ON CONFLICT(blocker_key) DO UPDATE SET occurrences=excluded.occurrences,status=excluded.status,updated_at=excluded.updated_at",
+            (blocker_key, occurrences, status, now),
+        )
+        return status
+
+    def deliver_outbox(self, sender: Callable[[Mapping[str, Any]], None], max_attempts: int, now: float | None = None) -> tuple[int, int]:
+        now = time.time() if now is None else now
+        sent = 0
+        failed = 0
+        rows = self.db.execute(
+            "SELECT * FROM outbox WHERE status='PENDING' AND next_attempt<=? ORDER BY created_at,outbox_id LIMIT 100",
+            (now,),
+        ).fetchall()
+        for row in rows:
+            payload = json.loads(row["payload_json"])
+            validate_metadata_only(payload)
+            try:
+                sender(payload)
+            except Exception:
+                attempts = row["attempts"] + 1
+                status = "DEAD_LETTER" if attempts >= max_attempts else "PENDING"
+                delay = min(300, 2 ** min(attempts, 8))
+                self.db.execute(
+                    "UPDATE outbox SET attempts=?,status=?,next_attempt=? WHERE outbox_id=?",
+                    (attempts, status, now + delay, row["outbox_id"]),
+                )
+                failed += 1
+            else:
+                self.db.execute(
+                    "UPDATE outbox SET attempts=attempts+1,status='SENT',next_attempt=? WHERE outbox_id=?",
+                    (now, row["outbox_id"]),
+                )
+                sent += 1
+        return sent, failed
+
+    def observe_ref(self, ref_name: str, oid: str) -> bool:
+        row = self.db.execute("SELECT oid FROM observed_refs WHERE ref_name=?", (ref_name,)).fetchone()
+        changed = row is not None and row["oid"] != oid
+        self.db.execute(
+            "INSERT INTO observed_refs VALUES(?,?,?) ON CONFLICT(ref_name) DO UPDATE SET oid=excluded.oid,updated_at=excluded.updated_at",
+            (ref_name, oid, time.time()),
+        )
+        return changed
+
+
+def spool_after_http_failure(
+    sender: Callable[[Mapping[str, Any]], None],
+    payload: Mapping[str, Any],
+    spool_dir: Path,
+    event_id: str,
+) -> bool:
+    validate_metadata_only(payload)
+    try:
+        sender(payload)
+        return True
+    except Exception:
+        atomic_json_write(spool_dir / f"{event_id}.json", {"event_id": event_id, "payload": dict(payload)})
+        return False
+
+
+def poll_file_inbox(store: QuorumStore, inbox_dir: Path) -> int:
+    if not inbox_dir.exists():
+        return 0
+    validate_secure_directory(inbox_dir, expected_uid=os.geteuid() if os.name == "posix" else None)
+    count = 0
+    for path in sorted(inbox_dir.glob("*.json")):
+        if path.is_symlink():
+            raise ControllerError(f"symlink inbox entry rejected: {path}")
+        item = read_secure_json(path, expected_uid=os.geteuid() if os.name == "posix" else None)
+        event_id = item.get("event_id")
+        payload = item.get("payload")
+        if not isinstance(event_id, str) or not isinstance(payload, dict):
+            raise ControllerError(f"invalid inbox envelope: {path}")
+        if store.ingest_event(event_id, payload, "file-inbox"):
+            count += 1
+        path.unlink()
+    return count
+
+
+def resolve_git_ref(repo: Path, ref_name: str) -> str:
+    if not repo.is_absolute():
+        raise ControllerError("Git repository path must be absolute")
+    info = repo.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
+        raise ControllerError("Git repository must be a non-symlink directory")
+    if os.name == "posix" and repo.resolve(strict=True) != Path(os.path.abspath(repo)):
+        raise ControllerError("Git repository has a symlinked path component")
+    if os.name == "posix" and info.st_mode & 0o002:
+        raise ControllerError("world-writable Git repository rejected")
+    try:
+        completed = subprocess.run(
+            [
+                "git",
+                "-c",
+                f"safe.directory={repo}",
+                "-C",
+                str(repo),
+                "rev-parse",
+                "--verify",
+                ref_name,
+            ],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            text=True,
+            timeout=10,
+        )
+    except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
+        raise ControllerError("Git ref unavailable") from exc
+    oid = completed.stdout.strip()
+    if len(oid) != 40 or any(character not in "0123456789abcdef" for character in oid.lower()):
+        raise ControllerError("invalid Git object id")
+    return oid
+
+
+def poll_git_refs(
+    store: QuorumStore,
+    repo: Path,
+    refs: Iterable[str],
+    resolver: Callable[[Path, str], str] = resolve_git_ref,
+) -> int:
+    changed = 0
+    for ref_name in refs:
+        if not ref_name.startswith("refs/") or ".." in ref_name or ref_name.endswith("/"):
+            raise ControllerError(f"invalid Git ref: {ref_name}")
+        oid = resolver(repo, ref_name)
+        if store.observe_ref(ref_name, oid):
+            event_id = f"gitref:{ref_name}:{oid}"
+            if store.ingest_event(event_id, {"ref": ref_name, "oid": oid}, "git-ref-poll"):
+                changed += 1
+    return changed
+
+
+def responses_fallback_allowed(
+    workspace_agent_ok: bool,
+    receipt: Mapping[str, Any],
+    *,
+    now: float | None = None,
+) -> bool:
+    validate_metadata_only(receipt)
+    if workspace_agent_ok:
+        return False
+    responses = receipt.get("responses_api")
+    return (
+        validate_health_receipt(receipt, "central_gpt", now=now) is Health.HEALTHY
+        and isinstance(responses, Mapping)
+        and responses.get("validated_live") is True
+    )
+
+
+def _validate_gateway_url(url: str) -> None:
+    parsed = urllib.parse.urlparse(url)
+    if (
+        parsed.scheme != "http"
+        or parsed.hostname != "127.0.0.1"
+        or parsed.port != 18920
+        or parsed.path != "/v1/orchestrator/events"
+        or parsed.params
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise ControllerError("gateway endpoint must be the fixed loopback endpoint")
+
+
+def get_gateway_state(url: str, project_key: str, token: str, timeout: float = 5.0) -> int:
+    _validate_gateway_url(url)
+    state_url = url.removesuffix("/events") + "/state/" + urllib.parse.quote(project_key, safe="")
+    request = urllib.request.Request(
+        state_url,
+        headers={"Authorization": f"Bearer {token}"},
+        method="GET",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            payload = json.loads(response.read(262_145))
+    except urllib.error.HTTPError as exc:
+        if exc.code == 404:
+            return 0
+        raise ControllerError(f"gateway state rejected HTTP {exc.code}") from exc
+    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
+        raise ControllerError("gateway state unavailable") from exc
+    version = payload.get("state_version") if isinstance(payload, Mapping) else None
+    if not isinstance(version, int) or version < 0:
+        raise ControllerError("gateway state version invalid")
+    return version
+
+
+def post_json(url: str, payload: Mapping[str, Any], token: str, timeout: float = 5.0) -> None:
+    validate_metadata_only(payload)
+    _validate_gateway_url(url)
+    request = urllib.request.Request(
+        url,
+        data=canonical_json(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            if response.status < 200 or response.status >= 300:
+                raise ControllerError(f"gateway rejected HTTP {response.status}")
+    except urllib.error.HTTPError as exc:
+        raise ControllerError(f"gateway rejected HTTP {exc.code}") from exc
+    except (urllib.error.URLError, TimeoutError) as exc:
+        raise ControllerError("gateway unavailable") from exc
+
+
+def run_once(config_path: Path) -> int:
+    config = read_secure_json(config_path, expected_uid=0 if os.name == "posix" else None)
+    gateway_url = config["gateway_event_url"]
+    token = read_gateway_token(Path(config["gateway_token_file"]))
+    state_version = get_gateway_state(gateway_url, config["project_key"], token)
+    store = QuorumStore(Path(config["database_path"]))
+    try:
+        receipts: dict[str, Health] = {}
+        for subject in ("central_gpt", "openclaw", "codex"):
+            receipt_path = Path(config["health_receipts"][subject])
+            try:
+                receipt = read_secure_json(receipt_path, expected_uid=0 if os.name == "posix" else None)
+                receipts[subject] = validate_health_receipt(
+                    receipt,
+                    subject,
+                    max_age_seconds=int(config.get("max_health_age_seconds", 60)),
+                )
+            except (FileNotFoundError, ControllerError, json.JSONDecodeError):
+                receipts[subject] = Health.UNKNOWN
+        decision = evaluate_quorum(
+            receipts["central_gpt"], receipts["openclaw"], receipts["codex"], config.get("default_workload", "large_repetitive_infrastructure_monitoring")
+        )
+        quorum_event_id = f"quorum-{time.time_ns()}"
+        store.enqueue(
+            quorum_event_id,
+            "gateway",
+            {
+                "schema_version": 1,
+                "project_key": config["project_key"],
+                "contract_commit": config["contract_commit"],
+                "event_id": quorum_event_id,
+                "source_agent": "system",
+                "event_type": "executor_quorum_evaluated",
+                "work_unit_id": "A0-AUTOMATION-QUORUM",
+                "status": "READY" if decision.allowed else "BLOCKED",
+                "state_version": state_version,
+                "implementation_commit": None,
+                "result_commit": None,
+                "blocker": None if decision.allowed else decision.status,
+                "evidence_refs": [],
+                "payload": {
+                    "allowed": decision.allowed,
+                    "priority": decision.priority,
+                    "quorum_status": decision.status,
+                    "active_executor": decision.active_executor,
+                    "standby_executor": decision.standby_executor,
+                },
+            },
+        )
+        poll_file_inbox(store, Path(config["file_inbox_path"]))
+        git_polling = config.get("git_ref_polling")
+        if isinstance(git_polling, Mapping):
+            poll_git_refs(store, Path(git_polling["repository"]), git_polling["refs"])
+        store.deliver_outbox(
+            lambda payload: post_json(gateway_url, payload, token),
+            max_attempts=int(config["blocker_retry_threshold"]),
+        )
+        return 0 if decision.allowed else 2
+    finally:
+        store.close()
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--config", type=Path, required=True)
+    parser.add_argument("--once", action="store_true", required=True)
+    return parser
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        return run_once(args.config)
+    except (ControllerError, KeyError, OSError, sqlite3.Error, json.JSONDecodeError) as error:
+        print(f"automation-quorum-controller: fail-closed: {type(error).__name__}", file=os.sys.stderr)
+        return 3
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Strict metadata-only helpers shared by root health probes."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import stat
+import tempfile
+from typing import Any
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+class ProbeError(RuntimeError):
+    pass
+
+
+def secure_regular(path: Path, *, max_mode: int, owners: set[int]) -> os.stat_result:
+    if not path.is_absolute():
+        raise ProbeError("path_not_absolute")
+    info = path.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise ProbeError("unsafe_file_type")
+    if stat.S_IMODE(info.st_mode) & ~max_mode:
+        raise ProbeError("unsafe_file_mode")
+    if os.name == "posix" and info.st_uid not in owners:
+        raise ProbeError("unsafe_file_owner")
+    if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
+        raise ProbeError("symlinked_path_component")
+    return info
+
+
+def load_config(path: Path) -> dict[str, Any]:
+    secure_regular(path, max_mode=0o600, owners={0})
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ProbeError("config_not_object")
+    return value
+
+
+def read_token(path: Path) -> str:
+    info = secure_regular(path, max_mode=0o640, owners={0, os.geteuid()})
+    if info.st_size <= 0 or info.st_size > 16_384:
+        raise ProbeError("invalid_token_size")
+    fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+    try:
+        opened = os.fstat(fd)
+        if not stat.S_ISREG(opened.st_mode) or opened.st_ino != info.st_ino or opened.st_dev != info.st_dev:
+            raise ProbeError("token_changed_during_validation")
+        data = os.read(fd, 16_385)
+        if os.read(fd, 1):
+            raise ProbeError("invalid_token_size")
+    finally:
+        os.close(fd)
+    if b"\n" in data or b"\r" in data:
+        raise ProbeError("invalid_token_format")
+    try:
+        return data.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ProbeError("invalid_token_encoding") from exc
+
+
+def gateway_base(endpoint: str) -> str:
+    parsed = urllib.parse.urlparse(endpoint)
+    if (
+        parsed.scheme != "http"
+        or parsed.hostname != "127.0.0.1"
+        or parsed.port != 18920
+        or parsed.path != "/v1/orchestrator/events"
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise ProbeError("invalid_gateway_endpoint")
+    return endpoint.removesuffix("/events")
+
+
+def request_json(
+    url: str,
+    token: str,
+    *,
+    method: str = "GET",
+    payload: dict[str, Any] | None = None,
+    timeout: float = 10.0,
+) -> tuple[int, dict[str, Any]]:
+    data = None if payload is None else json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
+    headers = {"Authorization": f"Bearer {token}"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(url, data=data, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            body = json.loads(response.read(262_145))
+            if not isinstance(body, dict):
+                raise ProbeError("gateway_response_not_object")
+            return response.status, body
+    except urllib.error.HTTPError as exc:
+        if exc.code in {401, 403, 409, 422}:
+            raise ProbeError(f"gateway_fail_closed_{exc.code}") from exc
+        raise ProbeError(f"gateway_http_{exc.code}") from exc
+    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
+        raise ProbeError("gateway_unavailable") from exc
+
+
+def state_version(endpoint: str, project_key: str, token: str) -> int:
+    url = gateway_base(endpoint) + "/state/" + urllib.parse.quote(project_key, safe="")
+    try:
+        _, value = request_json(url, token)
+    except ProbeError as exc:
+        if str(exc) == "gateway_http_404":
+            return 0
+        raise
+    version = value.get("state_version")
+    if not isinstance(version, int) or version < 0:
+        raise ProbeError("invalid_state_version")
+    return version
+
+
+def atomic_receipt(path: Path, value: dict[str, Any]) -> None:
+    if not path.is_absolute() or path.parent != Path("/run/aios-health"):
+        raise ProbeError("invalid_receipt_path")
+    parent = path.parent
+    info = parent.lstat()
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
+        raise ProbeError("unsafe_receipt_directory")
+    if os.name == "posix" and (info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o750):
+        raise ProbeError("unsafe_receipt_directory_metadata")
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=parent)
+    try:
+        # Receipts contain availability metadata only. Give the dedicated
+        # aios-quorum group read access so its unprivileged controller can
+        # consume them, while retaining root ownership and no public access.
+        if os.name == "posix":
+            os.fchown(fd, 0, info.st_gid)
+        os.fchmod(fd, 0o640)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(value, handle, sort_keys=True, separators=(",", ":"))
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
@@ -0,0 +1,24 @@
+{
+  "$schema": "../schemas/automation-quorum-controller.schema.json",
+  "project_key": "v2r1-personal-weixin-production",
+  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
+  "database_path": "/var/lib/aios-automation-quorum/controller.sqlite3",
+  "file_inbox_path": "/var/lib/aios-automation-quorum/inbox",
+  "default_workload": "large_repetitive_infrastructure_monitoring",
+  "health_receipts": {
+    "central_gpt": "/run/aios-health/central-gpt.json",
+    "openclaw": "/run/aios-health/openclaw.json",
+    "codex": "/run/aios-health/codex.json"
+  },
+  "git_ref_polling": {
+    "repository": "/data/aios-gitops/repo.git",
+    "refs": [
+      "refs/heads/task/governance/v2r1-personal-weixin-production-contract-20260717"
+    ]
+  },
+  "blocker_retry_threshold": 3,
+  "max_health_age_seconds": 60,
+  "lease_ttl_seconds": 120,
+  "gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
+  "gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token"
+}
@@ -0,0 +1,17 @@
+{
+  "project_key": "v2r1-personal-weixin-production",
+  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
+  "gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
+  "gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
+  "gateway_database_path": "/var/lib/aios-orchestrator-gateway/state.sqlite3",
+  "probe_clean_base_commit": "c3984ca2b4388875d2875b4fdfdf95a2984f6ab3",
+  "central_receipt_path": "/run/aios-health/central-gpt.json",
+  "codex_receipt_path": "/run/aios-health/codex.json",
+  "codex_command_queue": "/data/aios-gitops/workspace/agents/commands/active",
+  "codex_database_path": "/var/lib/aios-codex-watcher/state.sqlite3",
+  "codex_audit_log": "/var/log/aios-codex-watcher/audit.jsonl",
+  "central_timeout_seconds": 90,
+  "codex_timeout_seconds": 90,
+  "cycle_timeout_seconds": 180,
+  "poll_interval_seconds": 1
+}
@@ -13,6 +13,7 @@ Environment=AIOS_GITOPS_TOKEN_FILE=/data/aios-gitops/runtime/secrets/gitops.toke
 Environment=AIOS_GITOPS_BARE_REPO=/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git
 Environment=AIOS_GITOPS_HOST=127.0.0.1
 Environment=AIOS_GITOPS_PORT=18890
+EnvironmentFile=-/etc/aios-gitops/orchestrator-callback.env
 ExecStart=/usr/bin/python3 /data/aios-gitops/workspace/active-runtime/app/aios_chatgpt_actions_api.py
 Restart=always
 RestartSec=3
@@ -0,0 +1,137 @@
+# AIOS GPT + one-executor automation quorum
+
+## Authority and release boundary
+
+The Gateway is the only authoritative state writer. It must hold the SQLite writer lease before changing task state. A model response, service process, configured credential, or agent presence does not establish health. Release remains paused until all live acceptance rows are backed by logs and exit codes.
+
+Minimum quorum:
+
+```text
+central_gpt=HEALTHY AND (openclaw=HEALTHY OR codex=HEALTHY)
+```
+
+`UNKNOWN` never counts as healthy. If one executor is proven healthy, an unhealthy or unknown standby is P3 telemetry only and cannot block the active gate. If no executor is proven healthy, the controller blocks. Central GPT `UNKNOWN` or `UNHEALTHY` always blocks.
+
+## Live health receipts
+
+The controller reads three root-owned metadata-only JSON receipts from `/run/aios-health`. Files must be regular files, must not be symlinks, and neither files nor parent directories may be group/world writable. The controller rejects fields named `secret`, `token`, `password`, `api_key`, `authorization`, `credential`, `private_key`, `environment`, `env`, `argv`, `stdout`, or `stderr` at any depth.
+
+A receipt claiming `HEALTHY` must contain:
+
+```json
+{
+  "subject": "codex",
+  "health": "HEALTHY",
+  "live_probe_succeeded": true,
+  "observed_at_epoch": 1784347200
+}
+```
+
+The central GPT receipt additionally requires either `workspace_agent.validated_live=true` or `responses_api.validated_live=true`. Responses fallback is allowed only after a real Responses decision probe succeeded; credential/configuration presence is insufficient.
+
+The controller and probes use only the fixed loopback endpoint
+`http://127.0.0.1:18920/v1/orchestrator/events`. The bearer is read from
+`/etc/aios-orchestrator-gateway/inbound.token`; the file must be absolute,
+root-owned, regular, non-symlinked, no wider than `0640`, nonempty, and contain
+no CR/LF. It is never accepted through argv, environment, logs, receipts, or
+Git. HTTP `401`, `403`, `409`, and `422` are terminal fail-closed results for a
+probe attempt.
+
+The authenticated metadata query
+`GET /v1/orchestrator/events/<event_id>` returns only the event ID, public
+status, selected central channel, whether the decision was applied, and the
+resulting state version. It never returns the submitted event, model output,
+credential, or work-unit instruction.
+
+## Durable behavior
+
+SQLite persists ingress event digests, optimistic task state versions, leases, checkpoints, lifecycle events, effect IDs, outbox messages, blocker counts, observed Git refs, and project summaries. Reusing an event/effect/outbox ID with different content fails closed. An expired lease may be reclaimed; a live lease cannot be stolen.
+
+On active-executor failure, a healthy standby receives the same `contract_commit`, `event_id`, state version, and checkpoint. The former executor is recorded as `INTERRUPTED_AGENT_FAILOVER`. Unique effect IDs prevent replay of completed side effects.
+
+The Gateway persists these events even when OpenClaw is offline:
+
+```text
+TASK_DISPATCHED
+TASK_STARTED
+TASK_ENDED
+DECISION_APPLIED
+PROJECT_SUMMARY
+PROJECT_COMPLETED
+```
+
+OpenClaw mirroring is optional and uses the durable outbox with bounded retry and dead-letter. HTTP ingress failure spools a metadata-only file for polling; the configured local Git refs are also polled and changes missed by webhooks become idempotent events. Repeated blockers beyond the configured threshold enter `DEAD_LETTER`.
+
+## Install, verify, and rollback
+
+Installation is versioned and does not overwrite an existing release:
+
+```bash
+sudo scripts/install_automation_quorum_controller.sh "$PWD" 20260718.1
+sudo systemctl enable --now aios-automation-quorum-controller.timer
+sudo systemctl enable --now aios-central-gpt-health-probe.timer
+sudo systemctl enable --now aios-codex-executor-health-probe.timer
+```
+
+Run deterministic verification before enablement:
+
+```bash
+scripts/verify_automation_quorum_controller.sh "$PWD"
+```
+
+The two root probes are fixed, bounded, side-effect-free programs. They write
+receipts atomically with mode `0600` below the root-owned `0750`
+`/run/aios-health`. The Codex probe may write only one `acceptance_probe`
+command into the controlled watcher queue; it cannot accept arbitrary command
+text from configuration.
+
+Run the strict two-cycle closure only after both services are installed from an
+immutable committed release:
+
+```bash
+sudo /opt/aios-automation-quorum/current/bin/a0-unattended-loop-probe \
+  --config /etc/aios-automation-quorum/health-probes.json
+```
+
+Each cycle requires a live central decision that dispatches the named Codex
+acceptance work unit through the Gateway. The verifier then requires exactly
+one claim/start/execute/end/post marker, both lifecycle events in the Gateway,
+and a persisted `PROJECT_SUMMARY`. Directly queuing a command cannot satisfy
+this two-cycle closure.
+
+Rollback accepts only an existing non-symlink directory below `/opt/aios-automation-quorum/releases`. Repeating rollback to the current target is a no-op:
+
+```bash
+sudo scripts/rollback_automation_quorum_controller.sh \
+  /opt/aios-automation-quorum/releases/20260718.0
+```
+
+## Live acceptance matrix
+
+Each row requires captured logs, timestamps, exit codes, committed OIDs, Gitleaks output, and where applicable live Gateway DB evidence. Static unit tests are necessary but cannot close the gate.
+
+1. GPT + OpenClaw healthy, Codex stopped: two unattended cycles.
+2. GPT + Codex healthy, OpenClaw stopped: two unattended cycles.
+3. Both executors healthy: OpenClaw gets repetitive infrastructure; Codex gets complex code/security.
+4. Active stops mid-task: standby continues from checkpoint with no duplicate effect.
+5. Standby alone stops: active flow continues and no repair gate is created.
+6. GPT stops: execution blocks even with two healthy executors.
+7. GPT healthy and both executors stop: `BLOCKED_NO_HEALTHY_EXECUTOR`.
+8. `UNKNOWN` is never counted as healthy.
+9. HTTP ingress failure recovers through file polling.
+10. Missed webhook recovers through Git ref polling.
+11. Gateway restart restores events, leases, and outbox.
+12. Repeated event ID does not create a second dispatch; changed payload dead-letters.
+13. Repeated blocker beyond threshold dead-letters.
+14. Workspace Agent failure falls back only with a live-validated Responses channel.
+15. OpenClaw offline: Codex lifecycle and project summary remain complete in Gateway.
+16. Codex offline: OpenClaw lifecycle and project summary remain complete in Gateway.
+
+Until all rows pass against the live service, retain:
+
+```text
+AUTOMATION_CLOSED=NOT_STARTED
+BASELINE=V1R4C6
+RELEASE=paused
+PRODUCTION_RELEASE=NOT_STARTED
+```
@@ -0,0 +1,65 @@
+# A0 callback transport and pending-decision recovery
+
+This runbook separates code acceptance from live callback acceptance. Passing
+unit tests does not prove that a cloud Workspace Agent can reach the installed
+tool endpoint.
+
+## Transport configuration
+
+Choose exactly one mode in the owner-managed remote-tool configuration:
+
+- `remote_https_action`: `endpoint` must be the real public HTTPS Action URL.
+- `secure_mcp_tunnel`: `endpoint` must exactly match an entry in
+  `approved_mcp_tunnel_endpoints`.
+
+The configuration stores only `access_secret_ref` and `callback_secret_ref`.
+It must not contain resolved credentials. Startup rejects HTTP, loopback,
+unspecified/private literal addresses, URL credentials, query/fragment data,
+unapproved tunnel endpoints, and known placeholder markers.
+
+Generate the cloud-facing manifest only after owner configuration exists:
+
+```bash
+PYTHONPATH=/opt/aios-orchestrator-gateway/current \
+python3 /opt/aios-orchestrator-gateway/current/scripts/generate_orchestrator_tool_manifest.py \
+  --transport-config /etc/aios-orchestrator-gateway/remote-tool.json \
+  --output /var/lib/aios-orchestrator-gateway/orchestrator-tool-manifest.json
+```
+
+The generated Action manifest exposes only
+`submit_orchestration_decision`. The MCP manifest lists the same single tool
+with write-side-effect and idempotency annotations.
+
+## Recovery
+
+Plan mode is read-only:
+
+```bash
+PYTHONPATH=/opt/aios-orchestrator-gateway/current \
+python3 /opt/aios-orchestrator-gateway/current/scripts/recover_pending_decisions.py \
+  --config /etc/aios-orchestrator-gateway/config.json --plan
+```
+
+Application handles at most one leased record and never resends a model prompt
+or replays a decision:
+
+```bash
+PYTHONPATH=/opt/aios-orchestrator-gateway/current \
+python3 /opt/aios-orchestrator-gateway/current/scripts/recover_pending_decisions.py \
+  --config /etc/aios-orchestrator-gateway/config.json --apply-one
+```
+
+## Required live acceptance
+
+Do not mark the live Gate PASS until metadata-only evidence proves, in order:
+
+1. Workspace trigger returned HTTP 202.
+2. The remote callback tool received one authenticated request.
+3. The bridge returned HTTP 202 from the loopback 18920 decision endpoint.
+4. The event reached `DECIDED`, `callback_state=APPLIED`, and
+   `decision_applied=true`.
+5. Recovery planning and one apply cycle did not create a second outbox side
+   effect for the same event.
+
+Never include Authorization, token values, resolved SecretRefs, prompts, or raw
+model output in logs or RESULT files.
@@ -0,0 +1,89 @@
+openapi: 3.0.3
+info:
+  title: AIOS Orchestrator Gateway metadata API
+  version: 0.2.0
+servers:
+  - url: http://127.0.0.1:18920
+paths:
+  /v1/orchestrator/events:
+    post:
+      security:
+        - bearerAuth: []
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+      responses:
+        '202':
+          description: Event accepted
+        '409':
+          description: Idempotency or state conflict
+        '422':
+          description: Invalid event
+  /v1/orchestrator/events/{event_id}:
+    get:
+      security:
+        - bearerAuth: []
+      parameters:
+        - in: path
+          name: event_id
+          required: true
+          schema:
+            type: string
+            pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$'
+      responses:
+        '200':
+          description: Metadata-only event status
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: false
+                required: [event_id, status, trigger_http_status, trigger_error_category, trigger_channel, remote_id, callback_state, callback_deadline, callback_attempts, decision_applied, state_version_after]
+                properties:
+                  event_id:
+                    type: string
+                  status:
+                    enum: [RECEIVED, AWAITING_DECISION, DECIDED, TRIGGER_FAILED, DEAD_LETTER]
+                  trigger_channel:
+                    type: string
+                    nullable: true
+                    enum: [workspace_agent, responses_api]
+                  trigger_http_status:
+                    type: integer
+                    nullable: true
+                    minimum: 100
+                    maximum: 599
+                  trigger_error_category:
+                    type: string
+                    nullable: true
+                    enum: [UNAUTHORIZED, FORBIDDEN, NOT_FOUND, NOT_RUNNABLE, NETWORK_ERROR]
+                  remote_id:
+                    type: string
+                    nullable: true
+                  callback_state:
+                    enum: [PENDING, SENT, AWAITING_CALLBACK, CALLBACK_RECEIVED, APPLIED, STALE_SUPERSEDED, DEAD_LETTER]
+                  callback_deadline:
+                    type: integer
+                    nullable: true
+                    minimum: 0
+                  callback_attempts:
+                    type: integer
+                    minimum: 0
+                  decision_applied:
+                    type: boolean
+                  state_version_after:
+                    type: integer
+                    nullable: true
+                    minimum: 0
+        '401':
+          description: Missing or invalid bearer
+        '404':
+          description: Event not found
+components:
+  securitySchemes:
+    bearerAuth:
+      type: http
+      scheme: bearer
@@ -0,0 +1,61 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://aios.local/schemas/automation-quorum-controller.schema.json",
+  "title": "AIOS GPT plus one-executor quorum controller configuration",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "project_key",
+    "contract_commit",
+    "database_path",
+    "file_inbox_path",
+    "health_receipts",
+    "blocker_retry_threshold",
+    "max_health_age_seconds",
+    "lease_ttl_seconds",
+    "gateway_event_url",
+    "gateway_token_file"
+  ],
+  "properties": {
+    "$schema": {"type": "string"},
+    "project_key": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
+    "contract_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "database_path": {"type": "string", "pattern": "^/"},
+    "file_inbox_path": {"type": "string", "pattern": "^/"},
+    "default_workload": {
+      "enum": [
+        "large_repetitive_infrastructure_monitoring",
+        "complex_code_security_architecture"
+      ]
+    },
+    "health_receipts": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["central_gpt", "openclaw", "codex"],
+      "properties": {
+        "central_gpt": {"type": "string", "pattern": "^/"},
+        "openclaw": {"type": "string", "pattern": "^/"},
+        "codex": {"type": "string", "pattern": "^/"}
+      }
+    },
+    "git_ref_polling": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["repository", "refs"],
+      "properties": {
+        "repository": {"type": "string", "pattern": "^/"},
+        "refs": {
+          "type": "array",
+          "minItems": 1,
+          "uniqueItems": true,
+          "items": {"type": "string", "pattern": "^refs/"}
+        }
+      }
+    },
+    "blocker_retry_threshold": {"type": "integer", "minimum": 1, "maximum": 20},
+    "max_health_age_seconds": {"type": "integer", "minimum": 5, "maximum": 300},
+    "lease_ttl_seconds": {"type": "integer", "minimum": 15, "maximum": 3600},
+    "gateway_event_url": {"type": "string", "const": "http://127.0.0.1:18920/v1/orchestrator/events"},
+    "gateway_token_file": {"type": "string", "pattern": "^/"}
+  }
+}
@@ -0,0 +1,22 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "AIOS Callback Transport",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["mode", "endpoint", "approved_mcp_tunnel_endpoints"],
+  "properties": {
+    "mode": {
+      "enum": ["remote_https_action", "secure_mcp_tunnel"]
+    },
+    "endpoint": {
+      "type": "string",
+      "format": "uri",
+      "pattern": "^https://"
+    },
+    "approved_mcp_tunnel_endpoints": {
+      "type": "array",
+      "uniqueItems": true,
+      "items": {"type": "string", "format": "uri", "pattern": "^https://"}
+    }
+  }
+}
@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import tempfile
+
+from services.aios_orchestrator_gateway.callback_transport import CallbackTransport
+
+
+def atomic_write(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(value, handle, indent=2, sort_keys=True)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.chmod(temporary, 0o600)
+        os.replace(temporary, path)
+        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        try:
+            os.unlink(temporary)
+        except FileNotFoundError:
+            pass
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--transport-config", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    value = json.loads(args.transport_config.read_text(encoding="utf-8"))
+    if "callback_transport" in value:
+        value = value["callback_transport"]
+    transport = CallbackTransport.from_dict(value)
+    manifest = (
+        transport.action_openapi()
+        if transport.mode == "remote_https_action"
+        else {"endpoint": transport.endpoint, "tools": [transport.mcp_tool()]}
+    )
+    atomic_write(args.output, manifest)
+    print(f"TOOL_MANIFEST=PASS mode={transport.mode}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -37,6 +37,7 @@ for path in \
  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer" \
  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers" \
  "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles" \
+  "$SOURCE_ROOT/systemd/aios-orchestrator-remote-tool.service" \
  "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json"; do
  [[ -e "$path" ]] || fail "MISSING_SOURCE_${path}"
 done
@@ -56,6 +57,7 @@ systemd-analyze verify \

 SOURCE_FILES=(
  README-AIOS-ORCHESTRATOR-GATEWAY.md
+  docs/runbooks/A0-CALLBACK-TRANSPORT-AND-PENDING-RECOVERY.md
  config/aios-orchestrator-gateway.example.json
  schemas/decision.schema.json
  schemas/event.schema.json
@@ -69,10 +71,16 @@ SOURCE_FILES=(
  systemd/aios-orchestrator-gateway.service
  systemd/aios-orchestrator-gateway.sysusers
  systemd/aios-orchestrator-gateway.tmpfiles
+  systemd/aios-orchestrator-remote-tool.service
+  schemas/callback-transport.schema.json
+  scripts/generate_orchestrator_tool_manifest.py
+  scripts/recover_pending_decisions.py
  tests/test_aios_orchestrator_gateway.py
  tests/test_aios_orchestrator_gateway_rollback.sh
+  tests/test_aios_callback_recovery.py
  workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md
  workspace-agent/orchestrator-decision-action.openapi.yaml
+  openapi/aios-orchestrator-gateway.openapi.yaml
 )
 while IFS= read -r path; do SOURCE_FILES+=("$path"); done < <(
  cd "$SOURCE_ROOT"
@@ -97,11 +105,12 @@ if [[ "$DRY_RUN" -eq 1 ]]; then
  exit 0
 fi

+systemd-sysusers "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers"
+
 install -d -m 0755 "$INSTALL_ROOT/releases"
-install -d -m 0700 "$CONFIG_ROOT"
+install -d -o root -g aios-orchestrator -m 0750 "$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
@@ -121,9 +130,6 @@ 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" \
@@ -134,11 +140,13 @@ 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"
+install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-remote-tool.service" \
+  "$UNIT_ROOT/aios-orchestrator-remote-tool.service"

 if [[ ! -f "$CONFIG_ROOT/inbound.token" ]]; then
  python3 - <<'PY' > "$CONFIG_ROOT/inbound.token"
-import secrets
-print(secrets.token_urlsafe(48))
+import secrets, sys
+sys.stdout.write(secrets.token_urlsafe(48))
 PY
 fi
 [[ ! -L "$CONFIG_ROOT/inbound.token" ]] || fail INBOUND_TOKEN_SYMLINK
@@ -147,14 +155,49 @@ 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))
+import secrets, sys
+sys.stdout.write(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"

+# Historical installers appended one LF. Remove only that exact legacy byte in
+# place; reject every other newline form without ever copying or printing data.
+python3 - "$CONFIG_ROOT/config.json" "$CONFIG_ROOT/inbound.token" "$CONFIG_ROOT/callback.token" <<'PY'
+import json, os, pathlib, stat, sys
+config_path = pathlib.Path(sys.argv[1])
+names = list(sys.argv[2:])
+if config_path.is_file() and not config_path.is_symlink():
+    config = json.loads(config_path.read_text())
+    candidates = [
+        config.get("inbound_token_file"),
+        config.get("callback_token_file"),
+        (config.get("workspace_agent") or {}).get("access_token_file"),
+        (config.get("responses_fallback") or {}).get("api_key_file"),
+        (config.get("openclaw_sync") or {}).get("bearer_token_file"),
+    ]
+    names.extend(name for name in candidates if isinstance(name, str) and name)
+for name in dict.fromkeys(names):
+    path = pathlib.Path(name)
+    if not path.is_absolute() or path.is_symlink() or path.resolve() != path:
+        raise SystemExit("invalid token path")
+    fd = os.open(name, os.O_RDWR | getattr(os, "O_NOFOLLOW", 0))
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_size <= 0 or info.st_size > 16384:
+            raise SystemExit("invalid token metadata")
+        data = os.read(fd, info.st_size)
+        if data.endswith(b"\n") and b"\n" not in data[:-1] and b"\r" not in data:
+            os.ftruncate(fd, info.st_size - 1)
+            os.fsync(fd)
+        elif b"\n" in data or b"\r" in data:
+            raise SystemExit("invalid token format")
+    finally:
+        os.close(fd)
+PY
+
 if [[ ! -f "$CONFIG_ROOT/config.json" ]]; then
  install -m 0600 \
    "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json" \
@@ -170,7 +213,7 @@ 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" \
+PYTHONPATH="$RELEASE_DIR" \
 python3 - "$CONFIG_ROOT/config.json" <<'PY'
 import sys
 from pathlib import Path
@@ -187,6 +230,11 @@ if cfg.openclaw_sync_endpoint:
 print("CONFIG_VALIDATE=PASS")
 PY

+# Commit the release pointer only after every candidate/config/secret check has
+# succeeded. A failed validation therefore leaves the running release intact.
+ln -s "$RELEASE_DIR" "$INSTALL_ROOT/.current-new-$$"
+mv -Tf "$INSTALL_ROOT/.current-new-$$" "$INSTALL_ROOT/current"
+
 python3 - "$STATE_ROOT/install-receipt.json" "$SOURCE_DIGEST" "$RELEASE_DIR" "$PREVIOUS_TARGET" <<'PY'
 import json, os, sys, tempfile
 from pathlib import Path
@@ -219,12 +267,39 @@ finally:
    except FileNotFoundError:
        pass
 PY
-chown aios-orchestrator:aios-orchestrator "$STATE_ROOT/install-receipt.json"
+chown root:root "$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 enable aios-orchestrator-gateway.service
+systemctl enable aios-orchestrator-gateway-poll.timer
+systemctl stop aios-orchestrator-gateway-poll.timer || true
+systemctl stop aios-orchestrator-gateway-poll.service || true
+
+rollback_activation() {
+  if [[ -n "$PREVIOUS_TARGET" && -d "$PREVIOUS_TARGET" ]]; then
+    bash "$RELEASE_DIR/scripts/rollback_aios_orchestrator_gateway.sh" || true
+  else
+    systemctl stop aios-orchestrator-gateway.service || true
+  fi
+  fail ACTIVATION_FAILED_ROLLED_BACK
+}
+
+systemctl restart aios-orchestrator-gateway.service || rollback_activation
+python3 - <<'PY' || rollback_activation
+import json, time, urllib.request
+for _ in range(30):
+    try:
+        with urllib.request.urlopen("http://127.0.0.1:18920/healthz", timeout=1) as response:
+            body = json.loads(response.read(65536))
+            if response.status == 200 and body.get("status") == "active":
+                raise SystemExit(0)
+    except Exception:
+        pass
+    time.sleep(0.5)
+raise SystemExit(1)
+PY
+systemctl start aios-orchestrator-gateway-poll.timer || rollback_activation
 systemctl is-active --quiet aios-orchestrator-gateway.service
 systemctl is-active --quiet aios-orchestrator-gateway-poll.timer

@@ -233,7 +308,27 @@ 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
+systemctl start aios-orchestrator-gateway-poll.service || rollback_activation
+
+if [[ -f "$CONFIG_ROOT/remote-tool.json" ]]; then
+  [[ ! -L "$CONFIG_ROOT/remote-tool.json" ]] || fail REMOTE_TOOL_CONFIG_SYMLINK
+  chown root:aios-orchestrator "$CONFIG_ROOT/remote-tool.json"
+  chmod 0640 "$CONFIG_ROOT/remote-tool.json"
+  PYTHONPATH="$RELEASE_DIR" python3 - "$CONFIG_ROOT/remote-tool.json" <<'PY'
+import sys
+from pathlib import Path
+from services.aios_orchestrator_gateway.remote_tool_server import RemoteToolServerConfig
+RemoteToolServerConfig.from_file(Path(sys.argv[1]))
+print("REMOTE_TOOL_CONFIG_VALIDATE=PASS")
+PY
+  PYTHONPATH="$RELEASE_DIR" python3 \
+    "$RELEASE_DIR/scripts/generate_orchestrator_tool_manifest.py" \
+    --transport-config "$CONFIG_ROOT/remote-tool.json" \
+    --output "$STATE_ROOT/orchestrator-tool-manifest.json"
+  systemctl enable --now aios-orchestrator-remote-tool.service
+else
+  printf 'REMOTE_TOOL_CONFIGURATION=BLOCKED reason=REMOTE_CALLBACK_ENDPOINT_CONFIGURATION\n'
+fi

 printf 'AIOS_ORCHESTRATOR_INSTALL=PASS digest=%s previous=%s\n' \
  "$SOURCE_DIGEST" "${PREVIOUS_TARGET:-NONE}"
@@ -0,0 +1,112 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ ${EUID} -ne 0 ]]; then
+  echo "root required" >&2
+  exit 1
+fi
+
+SOURCE_ROOT=${1:?usage: install_automation_quorum_controller.sh SOURCE_ROOT VERSION}
+VERSION=${2:?usage: install_automation_quorum_controller.sh SOURCE_ROOT VERSION}
+[[ ${VERSION} =~ ^[0-9A-Za-z._-]+$ ]] || { echo "invalid version" >&2; exit 1; }
+SOURCE_REAL=$(readlink -f -- "${SOURCE_ROOT}" 2>/dev/null || true)
+[[ ${SOURCE_ROOT} = /* && -n ${SOURCE_REAL} && ${SOURCE_ROOT} == "${SOURCE_REAL}" && -d ${SOURCE_ROOT} && ! -L ${SOURCE_ROOT} ]] || {
+  echo "unsafe source root" >&2
+  exit 1
+}
+
+SOURCE_FILES=(
+  automation/executor_quorum_controller.py
+  automation/health_probe_common.py
+  automation/central_gpt_health_probe.py
+  automation/codex_executor_health_probe.py
+  automation/a0_unattended_loop_probe.py
+  schemas/automation-quorum-controller.schema.json
+  config/aios-automation-quorum-controller.example.json
+  config/aios-health-probes.example.json
+  systemd/aios-automation-quorum-controller.service
+  systemd/aios-automation-quorum-controller.timer
+  systemd/aios-automation-quorum-controller.sysusers
+  systemd/aios-automation-quorum-controller.tmpfiles
+  systemd/aios-central-gpt-health-probe.service
+  systemd/aios-central-gpt-health-probe.timer
+  systemd/aios-codex-executor-health-probe.service
+  systemd/aios-codex-executor-health-probe.timer
+)
+for relative in "${SOURCE_FILES[@]}"; do
+  [[ -f ${SOURCE_ROOT}/${relative} && ! -L ${SOURCE_ROOT}/${relative} ]] || {
+    echo "unsafe source file: ${relative}" >&2
+    exit 1
+  }
+done
+
+INSTALL_ROOT=/opt/aios-automation-quorum
+VERSION_DIR=${INSTALL_ROOT}/releases/${VERSION}
+[[ ! -e ${VERSION_DIR} ]] || { echo "version already installed" >&2; exit 1; }
+
+install -d -o root -g root -m 0755 "${VERSION_DIR}/bin" "${VERSION_DIR}/share"
+install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/executor_quorum_controller.py" \
+  "${VERSION_DIR}/bin/executor-quorum-controller"
+install -o root -g root -m 0644 "${SOURCE_ROOT}/automation/health_probe_common.py" \
+  "${VERSION_DIR}/bin/health_probe_common.py"
+install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/central_gpt_health_probe.py" \
+  "${VERSION_DIR}/bin/central-gpt-health-probe"
+install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/codex_executor_health_probe.py" \
+  "${VERSION_DIR}/bin/codex-executor-health-probe"
+install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/a0_unattended_loop_probe.py" \
+  "${VERSION_DIR}/bin/a0-unattended-loop-probe"
+install -o root -g root -m 0644 "${SOURCE_ROOT}/schemas/automation-quorum-controller.schema.json" \
+  "${VERSION_DIR}/share/automation-quorum-controller.schema.json"
+python3 -m py_compile "${VERSION_DIR}/bin/executor-quorum-controller" \
+  "${VERSION_DIR}/bin/health_probe_common.py" \
+  "${VERSION_DIR}/bin/central-gpt-health-probe" \
+  "${VERSION_DIR}/bin/codex-executor-health-probe"
+python3 -m py_compile "${VERSION_DIR}/bin/a0-unattended-loop-probe"
+
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.service" /etc/systemd/system/
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.timer" /etc/systemd/system/
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-central-gpt-health-probe.service" /etc/systemd/system/
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-central-gpt-health-probe.timer" /etc/systemd/system/
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-codex-executor-health-probe.service" /etc/systemd/system/
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-codex-executor-health-probe.timer" /etc/systemd/system/
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.sysusers" /usr/lib/sysusers.d/aios-automation-quorum-controller.conf
+install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.tmpfiles" /usr/lib/tmpfiles.d/aios-automation-quorum-controller.conf
+
+systemd-sysusers /usr/lib/sysusers.d/aios-automation-quorum-controller.conf
+systemd-tmpfiles --create /usr/lib/tmpfiles.d/aios-automation-quorum-controller.conf
+install -d -o root -g aios-quorum -m 0750 /etc/aios-automation-quorum
+if [[ ! -e /etc/aios-automation-quorum/controller.json ]]; then
+  install -o root -g aios-quorum -m 0640 \
+    "${SOURCE_ROOT}/config/aios-automation-quorum-controller.example.json" \
+    /etc/aios-automation-quorum/controller.json
+fi
+if [[ ! -e /etc/aios-automation-quorum/health-probes.json ]]; then
+  install -o root -g root -m 0600 \
+    "${SOURCE_ROOT}/config/aios-health-probes.example.json" \
+    /etc/aios-automation-quorum/health-probes.json
+fi
+
+PREVIOUS_TARGET=$(readlink -f "${INSTALL_ROOT}/current" 2>/dev/null || true)
+RECEIPT=${VERSION_DIR}/install-receipt.json
+python3 - "${VERSION_DIR}" "${VERSION}" "${PREVIOUS_TARGET}" "${RECEIPT}" <<'PY'
+import hashlib, json, os, pathlib, sys, tempfile
+root, version, previous, receipt = map(pathlib.Path, (sys.argv[1], sys.argv[2], sys.argv[3] or ".", sys.argv[4]))
+files = {}
+for item in sorted(root.rglob("*")):
+    if item.is_file() and not item.is_symlink():
+        files[str(item.relative_to(root))] = hashlib.sha256(item.read_bytes()).hexdigest()
+payload = {"version": str(version), "previous_target": "" if str(previous) == "." else str(previous), "files": files}
+fd, temp = tempfile.mkstemp(dir=str(receipt.parent), prefix=".receipt-")
+with os.fdopen(fd, "w", encoding="utf-8") as handle:
+    json.dump(payload, handle, sort_keys=True, separators=(",", ":"))
+    handle.write("\n")
+os.chmod(temp, 0o600)
+os.replace(temp, receipt)
+PY
+
+ln -sfn "${VERSION_DIR}" "${INSTALL_ROOT}/.current-${VERSION}"
+mv -Tf "${INSTALL_ROOT}/.current-${VERSION}" "${INSTALL_ROOT}/current"
+
+systemctl daemon-reload
+echo "installed=${VERSION_DIR}"
+echo "previous=${PREVIOUS_TARGET}"
@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from services.aios_orchestrator_gateway.config import Config
+from services.aios_orchestrator_gateway.recovery import (
+    PendingDecisionRecovery,
+    RecoverySettings,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", type=Path, required=True)
+    mode = parser.add_mutually_exclusive_group(required=True)
+    mode.add_argument("--plan", action="store_true")
+    mode.add_argument("--apply-one", action="store_true")
+    args = parser.parse_args()
+    config = Config.from_file(args.config)
+    recovery = PendingDecisionRecovery(RecoverySettings(
+        database_path=config.database_path,
+        contract_commit=config.contract_commit,
+        lease_seconds=config.lease_seconds,
+        max_retries=config.max_retries,
+        retry_after=config.retry_base_seconds,
+    ))
+    if args.plan:
+        print(json.dumps(recovery.plan(), separators=(",", ":"), sort_keys=True))
+    else:
+        print(json.dumps({"processed_event_id": recovery.apply_one()}, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ ${EUID} -ne 0 ]]; then
+  echo "root required" >&2
+  exit 1
+fi
+
+TARGET=${1:?usage: rollback_automation_quorum_controller.sh ABSOLUTE_RELEASE_PATH}
+INSTALL_ROOT=/opt/aios-automation-quorum
+TARGET_REAL=$(readlink -f -- "${TARGET}" 2>/dev/null || true)
+[[ -n ${TARGET_REAL} && ${TARGET} == "${TARGET_REAL}" ]] || { echo "non-canonical or missing rollback target" >&2; exit 1; }
+case ${TARGET_REAL} in
+  "${INSTALL_ROOT}"/releases/*) ;;
+  *) echo "target outside release root" >&2; exit 1 ;;
+esac
+[[ -d ${TARGET_REAL} && ! -L ${TARGET} ]] || { echo "invalid rollback target" >&2; exit 1; }
+
+CURRENT=$(readlink -f "${INSTALL_ROOT}/current" 2>/dev/null || true)
+if [[ ${CURRENT} == "${TARGET_REAL}" ]]; then
+  echo "already_current=${CURRENT}"
+  exit 0
+fi
+
+ln -sfn "${TARGET_REAL}" "${INSTALL_ROOT}/.rollback-current"
+mv -Tf "${INSTALL_ROOT}/.rollback-current" "${INSTALL_ROOT}/current"
+systemctl daemon-reload
+systemctl try-restart aios-automation-quorum-controller.timer
+echo "rolled_back_to=$(readlink -f "${INSTALL_ROOT}/current")"
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Dependency-free, deterministic validation for the shipped controller config."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+REQUIRED = {
+    "project_key",
+    "contract_commit",
+    "database_path",
+    "file_inbox_path",
+    "health_receipts",
+    "blocker_retry_threshold",
+    "max_health_age_seconds",
+    "lease_ttl_seconds",
+    "gateway_event_url",
+    "gateway_token_file",
+}
+OPTIONAL = {"$schema", "default_workload", "git_ref_polling"}
+WORKLOADS = {
+    "large_repetitive_infrastructure_monitoring",
+    "complex_code_security_architecture",
+}
+
+
+def fail(message: str) -> None:
+    raise ValueError(message)
+
+
+def absolute_path(value: object, name: str) -> None:
+    if not isinstance(value, str) or not value.startswith("/") or "\x00" in value:
+        fail(f"{name} must be an absolute POSIX path")
+
+
+def bounded_integer(value: object, name: str, minimum: int, maximum: int) -> None:
+    if isinstance(value, bool) or not isinstance(value, int) or not minimum <= value <= maximum:
+        fail(f"{name} must be an integer in [{minimum}, {maximum}]")
+
+
+def validate(config: object, schema: object) -> None:
+    if not isinstance(config, dict) or not isinstance(schema, dict):
+        fail("config and schema must be JSON objects")
+    if schema.get("$schema") != "https://json-schema.org/draft/2020-12/schema":
+        fail("unexpected JSON Schema dialect")
+    if schema.get("additionalProperties") is not False:
+        fail("schema must reject additional properties")
+    if set(schema.get("required", [])) != REQUIRED:
+        fail("schema required fields drifted from validator")
+    keys = set(config)
+    if not REQUIRED <= keys or keys - REQUIRED - OPTIONAL:
+        fail("missing or unexpected top-level config fields")
+    absolute_path(config["database_path"], "database_path")
+    absolute_path(config["file_inbox_path"], "file_inbox_path")
+    absolute_path(config["gateway_token_file"], "gateway_token_file")
+    project_key = config["project_key"]
+    if not isinstance(project_key, str) or not project_key or len(project_key) > 128:
+        fail("project_key is invalid")
+    commit = config["contract_commit"]
+    if not isinstance(commit, str) or len(commit) != 40 or any(c not in "0123456789abcdef" for c in commit):
+        fail("contract_commit must be a full lowercase Git OID")
+    receipts = config["health_receipts"]
+    if not isinstance(receipts, dict) or set(receipts) != {"central_gpt", "openclaw", "codex"}:
+        fail("health_receipts must contain exactly central_gpt, openclaw, and codex")
+    for subject, value in receipts.items():
+        absolute_path(value, f"health_receipts.{subject}")
+    git_polling = config.get("git_ref_polling")
+    if git_polling is not None:
+        if not isinstance(git_polling, dict) or set(git_polling) != {"repository", "refs"}:
+            fail("git_ref_polling must contain exactly repository and refs")
+        absolute_path(git_polling["repository"], "git_ref_polling.repository")
+        refs = git_polling["refs"]
+        if not isinstance(refs, list) or not refs or len(refs) != len(set(refs)):
+            fail("git_ref_polling.refs must be a non-empty unique list")
+        if any(not isinstance(ref, str) or not ref.startswith("refs/") or ".." in ref for ref in refs):
+            fail("git_ref_polling.refs contains an invalid ref")
+    if config.get("default_workload", "large_repetitive_infrastructure_monitoring") not in WORKLOADS:
+        fail("unsupported default_workload")
+    bounded_integer(config["blocker_retry_threshold"], "blocker_retry_threshold", 1, 20)
+    bounded_integer(config["max_health_age_seconds"], "max_health_age_seconds", 5, 300)
+    bounded_integer(config["lease_ttl_seconds"], "lease_ttl_seconds", 15, 3600)
+    url = config["gateway_event_url"]
+    parsed = urlparse(url) if isinstance(url, str) else None
+    if (
+        not parsed
+        or parsed.scheme != "http"
+        or parsed.hostname != "127.0.0.1"
+        or parsed.port != 18920
+        or parsed.path != "/v1/orchestrator/events"
+        or parsed.query
+        or parsed.fragment
+    ):
+        fail("gateway_event_url must be the fixed loopback orchestrator endpoint")
+
+
+def main() -> int:
+    if len(sys.argv) != 3:
+        print(f"usage: {Path(sys.argv[0]).name} CONFIG SCHEMA", file=sys.stderr)
+        return 2
+    with Path(sys.argv[1]).open("r", encoding="utf-8") as handle:
+        config = json.load(handle)
+    with Path(sys.argv[2]).open("r", encoding="utf-8") as handle:
+        schema = json.load(handle)
+    try:
+        validate(config, schema)
+    except ValueError as error:
+        print(f"schema validation failed: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT=${1:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}
+
+python3 -m py_compile "${ROOT}/automation/executor_quorum_controller.py"
+python3 -m py_compile \
+  "${ROOT}/automation/health_probe_common.py" \
+  "${ROOT}/automation/central_gpt_health_probe.py" \
+  "${ROOT}/automation/codex_executor_health_probe.py" \
+  "${ROOT}/automation/a0_unattended_loop_probe.py"
+python3 -m unittest discover -s "${ROOT}/tests" -p 'test_executor_quorum_controller.py' -v
+python3 -m unittest discover -s "${ROOT}/tests" -p 'test_aios_health_probes.py' -v
+python3 -m json.tool "${ROOT}/config/aios-automation-quorum-controller.example.json" >/dev/null
+python3 -m json.tool "${ROOT}/schemas/automation-quorum-controller.schema.json" >/dev/null
+python3 "${ROOT}/scripts/validate_automation_quorum_schema.py" \
+  "${ROOT}/config/aios-automation-quorum-controller.example.json" \
+  "${ROOT}/schemas/automation-quorum-controller.schema.json"
+python3 -m py_compile "${ROOT}/scripts/validate_automation_quorum_schema.py"
+bash -n "${ROOT}/scripts/install_automation_quorum_controller.sh"
+bash -n "${ROOT}/scripts/rollback_automation_quorum_controller.sh"
+bash -n "${ROOT}/scripts/verify_automation_quorum_controller.sh"
+systemd-analyze verify \
+  "${ROOT}/systemd/aios-automation-quorum-controller.service" \
+  "${ROOT}/systemd/aios-automation-quorum-controller.timer" \
+  "${ROOT}/systemd/aios-central-gpt-health-probe.service" \
+  "${ROOT}/systemd/aios-central-gpt-health-probe.timer" \
+  "${ROOT}/systemd/aios-codex-executor-health-probe.service" \
+  "${ROOT}/systemd/aios-codex-executor-health-probe.timer"
@@ -309,7 +309,7 @@ class CommandWatcher:
            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
@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+from typing import Any, Callable
+import urllib.error
+import urllib.request
+
+from .models import Decision
+
+
+GATEWAY_DECISION_URL = "http://127.0.0.1:18920/v1/orchestrator/decisions"
+SECRET_REF_RE = re.compile(r"^secret://[A-Za-z0-9._/-]{1,480}$")
+
+
+class CallbackBridgeError(RuntimeError):
+    def __init__(self, category: str, status_code: int) -> None:
+        super().__init__(category)
+        self.category = category
+        self.status_code = status_code
+
+
+def _trusted_resolver(path: Path) -> Path:
+    if not path.is_absolute():
+        raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503)
+    try:
+        info = path.lstat()
+        resolved = path.resolve(strict=True)
+    except OSError as exc:
+        raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503) from exc
+    if (
+        resolved != path
+        or stat.S_ISLNK(info.st_mode)
+        or not stat.S_ISREG(info.st_mode)
+        or info.st_uid != 0
+        or stat.S_IMODE(info.st_mode) & 0o022
+        or not stat.S_IMODE(info.st_mode) & 0o111
+    ):
+        raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503)
+    return resolved
+
+
+def resolve_secret_ref(secret_ref: str, resolver: Path) -> str:
+    if not isinstance(secret_ref, str) or not SECRET_REF_RE.fullmatch(secret_ref):
+        raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503)
+    executable = _trusted_resolver(resolver)
+    request = json.dumps(
+        {"operation": "resolve", "secret_ref": secret_ref},
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode()
+    try:
+        result = subprocess.run(
+            [str(executable)],
+            input=request,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
+            timeout=5,
+            check=False,
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        raise CallbackBridgeError("CALLBACK_CREDENTIAL_UNAVAILABLE", 503) from exc
+    token = result.stdout.removesuffix(b"\n")
+    if (
+        result.returncode != 0
+        or not token
+        or len(token) > 16_384
+        or b"\n" in token
+        or b"\r" in token
+    ):
+        raise CallbackBridgeError("CALLBACK_CREDENTIAL_UNAVAILABLE", 503)
+    try:
+        return token.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise CallbackBridgeError("CALLBACK_CREDENTIAL_UNAVAILABLE", 503) from exc
+
+
+def submit_orchestration_decision(
+    payload: dict[str, Any],
+    *,
+    contract_commit: str,
+    secret_ref: str,
+    resolver: Path,
+    request_open: Callable[..., Any] = urllib.request.urlopen,
+) -> dict[str, Any]:
+    try:
+        decision = Decision.from_dict(payload)
+    except Exception as exc:
+        raise CallbackBridgeError("INVALID_DECISION", 422) from exc
+    if decision.contract_commit != contract_commit:
+        raise CallbackBridgeError("CONTRACT_MISMATCH", 409)
+    token = resolve_secret_ref(secret_ref, resolver)
+    request = urllib.request.Request(
+        GATEWAY_DECISION_URL,
+        data=json.dumps(
+            decision.to_dict(), separators=(",", ":"), sort_keys=True
+        ).encode(),
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Idempotency-Key": decision.event_id,
+            "User-Agent": "aios-chatgpt-actions/decision-bridge-v1",
+        },
+    )
+    try:
+        with request_open(request, timeout=10) as response:
+            raw = response.read(65_537)
+            if response.status != 202 or len(raw) > 65_536:
+                raise CallbackBridgeError("CALLBACK_HTTP_ERROR", 502)
+    except urllib.error.HTTPError as exc:
+        if exc.code == 409:
+            raise CallbackBridgeError("STALE_STATE", 409) from exc
+        raise CallbackBridgeError("CALLBACK_HTTP_ERROR", 502) from exc
+    except (urllib.error.URLError, TimeoutError, OSError) as exc:
+        raise CallbackBridgeError("CALLBACK_NETWORK_ERROR", 502) from exc
+    try:
+        value = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise CallbackBridgeError("CALLBACK_RESPONSE_INVALID", 502) from exc
+    if (
+        not isinstance(value, dict)
+        or value.get("accepted") is not True
+        or not isinstance(value.get("state_version"), int)
+    ):
+        raise CallbackBridgeError("CALLBACK_RESPONSE_INVALID", 502)
+    return {"accepted": True, "state_version": value["state_version"]}
@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import ipaddress
+from typing import Any
+from urllib.parse import urlparse
+
+from .errors import ConfigurationError
+from .models import DECISION_SCHEMA
+
+
+TRANSPORT_MODES = {"remote_https_action", "secure_mcp_tunnel"}
+FORBIDDEN_HOSTS = {"localhost", "localhost.localdomain"}
+FORBIDDEN_MARKERS = {"replace_with_actions_host", "placeholder"}
+
+
+def _https_endpoint(value: Any, *, public: bool) -> str:
+    if not isinstance(value, str) or not value or any(
+        marker in value.lower() for marker in FORBIDDEN_MARKERS
+    ):
+        raise ConfigurationError("invalid_callback_endpoint")
+    parsed = urlparse(value)
+    host = (parsed.hostname or "").lower()
+    if (
+        parsed.scheme != "https"
+        or not host
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.query
+        or parsed.fragment
+        or host in FORBIDDEN_HOSTS
+    ):
+        raise ConfigurationError("invalid_callback_endpoint")
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        address = None
+    if address is not None and (
+        address.is_loopback
+        or address.is_private
+        or address.is_unspecified
+        or address.is_link_local
+        or address.is_multicast
+        or address.is_reserved
+    ):
+        raise ConfigurationError("invalid_callback_endpoint")
+    if public and (address is None and ("." not in host or host.endswith((".local", ".internal")))):
+        raise ConfigurationError("callback_endpoint_not_public")
+    return value.rstrip("/")
+
+
+@dataclass(frozen=True)
+class CallbackTransport:
+    mode: str
+    endpoint: str
+    approved_mcp_tunnel_endpoints: tuple[str, ...]
+
+    @classmethod
+    def from_dict(cls, value: dict[str, Any]) -> "CallbackTransport":
+        expected = {"mode", "endpoint", "approved_mcp_tunnel_endpoints"}
+        if not isinstance(value, dict) or set(value) != expected:
+            raise ConfigurationError("invalid_callback_transport_fields")
+        mode = value.get("mode")
+        if mode not in TRANSPORT_MODES:
+            raise ConfigurationError("invalid_callback_transport_mode")
+        approved = value.get("approved_mcp_tunnel_endpoints")
+        if not isinstance(approved, list) or any(not isinstance(x, str) for x in approved):
+            raise ConfigurationError("invalid_approved_mcp_tunnel_endpoints")
+        normalized_approved = tuple(_https_endpoint(item, public=False) for item in approved)
+        endpoint = _https_endpoint(value.get("endpoint"), public=mode == "remote_https_action")
+        if mode == "secure_mcp_tunnel" and endpoint not in normalized_approved:
+            raise ConfigurationError("unapproved_secure_mcp_tunnel_endpoint")
+        return cls(mode, endpoint, normalized_approved)
+
+    def action_openapi(self) -> dict[str, Any]:
+        if self.mode != "remote_https_action":
+            raise ConfigurationError("action_manifest_for_wrong_transport")
+        return {
+            "openapi": "3.1.0",
+            "info": {"title": "AIOS Orchestrator Decision Callback", "version": "1.1.0"},
+            "servers": [{"url": self.endpoint}],
+            "paths": {
+                "/v1/orchestrator/decisions": {
+                    "post": {
+                        "operationId": "submit_orchestration_decision",
+                        "summary": "Submit one strict orchestration decision",
+                        "security": [{"bearerAuth": []}],
+                        "requestBody": {
+                            "required": True,
+                            "content": {"application/json": {"schema": DECISION_SCHEMA}},
+                        },
+                        "responses": {"202": {"description": "Accepted"}},
+                    }
+                }
+            },
+            "components": {
+                "securitySchemes": {
+                    "bearerAuth": {"type": "http", "scheme": "bearer"}
+                }
+            },
+        }
+
+    def mcp_tool(self) -> dict[str, Any]:
+        if self.mode != "secure_mcp_tunnel":
+            raise ConfigurationError("mcp_manifest_for_wrong_transport")
+        return {
+            "name": "submit_orchestration_decision",
+            "description": "Validate and submit one AIOS orchestration decision.",
+            "inputSchema": DECISION_SCHEMA,
+            "annotations": {
+                "title": "Submit orchestration decision",
+                "readOnlyHint": False,
+                "destructiveHint": False,
+                "idempotentHint": True,
+                "openWorldHint": False,
+            },
+        }
@@ -1,6 +1,7 @@
 from __future__ import annotations

 import json
+from dataclasses import dataclass
 import urllib.error
 import urllib.request
 from typing import Any
@@ -9,6 +10,20 @@ from .errors import TriggerError
 from .models import DECISION_SCHEMA, Decision, Event


+@dataclass(frozen=True)
+class WorkspaceTriggerResult:
+    http_status: int
+    remote_id: None = None
+
+
+WORKSPACE_HTTP_CATEGORIES = {
+    401: "UNAUTHORIZED",
+    403: "FORBIDDEN",
+    404: "NOT_FOUND",
+    409: "NOT_RUNNABLE",
+}
+
+
 class WorkspaceAgentClient:
    """Asynchronous Workspace Agent trigger.

@@ -21,7 +36,7 @@ class WorkspaceAgentClient:
        self.token = token
        self.timeout_seconds = timeout_seconds

-    def trigger(self, event: Event) -> None:
+    def trigger(self, event: Event) -> WorkspaceTriggerResult:
        url = f"https://api.chatgpt.com/v1/workspace_agents/{self.trigger_id}/trigger"
        prompt = (
            "Process this AIOS orchestration event under the committed contract. "
@@ -48,13 +63,24 @@ class WorkspaceAgentClient:
        )
        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}")
+                    raise TriggerError(
+                        "workspace_agent_unexpected_http_status",
+                        http_status=response.status,
+                    )
+                return WorkspaceTriggerResult(http_status=202)
        except urllib.error.HTTPError as exc:
-            raise TriggerError(f"workspace_agent_http_{exc.code}") from exc
+            category = WORKSPACE_HTTP_CATEGORIES.get(exc.code)
+            raise TriggerError(
+                "workspace_agent_http_error",
+                http_status=exc.code,
+                category=category,
+            ) from exc
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
-            raise TriggerError("workspace_agent_trigger_failed") from exc
+            raise TriggerError(
+                "workspace_agent_network_error",
+                category="NETWORK_ERROR",
+            ) from exc


 class ResponsesClient:
@@ -3,10 +3,11 @@ from __future__ import annotations
 from dataclasses import dataclass
 import os
 from pathlib import Path
+import stat
 from typing import Any

 from .errors import ConfigurationError
-from .securefs import read_secure_json, read_secure_text
+from .securefs import read_secure_bytes, read_secure_json, verify_secure_file


@dataclass(frozen=True)
@@ -185,7 +186,16 @@ class Config:

    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)
+        info = verify_secure_file(path, expected_uid=owner_uid)
+        if stat.S_IMODE(info.st_mode) & ~0o640:
+            raise ConfigurationError(f"permissive_secret_mode:{path}")
+        data = read_secure_bytes(path, expected_uid=owner_uid, max_bytes=64_000)
+        if not data or b"\n" in data or b"\r" in data:
+            raise ConfigurationError(f"invalid_secret_format:{path}")
+        try:
+            return data.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise ConfigurationError(f"secret_not_utf8:{path}") from exc

    def inbound_token(self) -> str:
        return self._secret(self.inbound_token_file)
@@ -49,6 +49,9 @@ class DispatchSink:
        except json.JSONDecodeError as exc:
            raise DispatchError("invalid_dispatch_payload") from exc

+        project_key = record.get("project_key")
+        state_version = record.get("state_version")
+
        envelope = {
            "schema_version": 1,
            "contract_commit": self.contract_commit,
@@ -62,13 +65,43 @@ class DispatchSink:
            },
        }
        outbox_path = self.outbox_dir / "dispatched" / f"{work_unit_id}.json"
-        queue_path = self.command_queue_dir / agent / "pending" / f"{work_unit_id}.json"
+        if agent == "codex":
+            if (
+                not isinstance(project_key, str)
+                or not SAFE_ID.fullmatch(project_key)
+                or not isinstance(state_version, int)
+                or state_version < 0
+            ):
+                raise DispatchError("invalid_dispatch_context")
+            required_checks = work_unit.get("required_checks", [])
+            execution_kind = (
+                "acceptance_probe"
+                if isinstance(required_checks, list) and "acceptance_probe" in required_checks
+                else "codex"
+            )
+            queue_value = {
+                "schema_version": 1,
+                "command_id": work_unit_id,
+                "project_key": project_key,
+                "work_unit_id": work_unit_id,
+                "contract_commit": self.contract_commit,
+                "state_version": state_version,
+                "execution": {
+                    "kind": execution_kind,
+                    "input": work_unit.get("instruction", ""),
+                },
+            }
+            # The Codex watcher polls the configured active directory itself.
+            queue_path = self.command_queue_dir / f"{work_unit_id}.json"
+        else:
+            queue_value = envelope
+            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)
+        self._write_idempotent_json(queue_path, queue_value, shared=True)
        return outbox_path, queue_path

    @staticmethod
@@ -21,6 +21,17 @@ class ConflictError(OrchestratorError):
 class TriggerError(OrchestratorError):
    """Central model trigger failed."""

+    def __init__(
+        self,
+        message: str,
+        *,
+        http_status: int | None = None,
+        category: str | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.http_status = http_status
+        self.category = category
+

 class DispatchError(OrchestratorError):
    """Dispatch or synchronization output failed."""
@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import socket
+
+from .store import Store
+
+
+@dataclass(frozen=True)
+class RecoverySettings:
+    database_path: Path
+    contract_commit: str
+    lease_seconds: int
+    max_retries: int
+    retry_after: int
+
+
+class PendingDecisionRecovery:
+    def __init__(self, settings: RecoverySettings):
+        self.settings = settings
+        self.store = Store(settings.database_path)
+        self.owner = f"recovery-{socket.gethostname()}"
+
+    def plan(self) -> dict[str, int]:
+        return self.store.pending_decision_plan(
+            self.settings.contract_commit, self.settings.max_retries
+        )
+
+    def apply_one(self) -> str | None:
+        # Recovery never calls a model or replays a decision. It only advances
+        # durable callback bookkeeping under a lease.
+        return self.store.recover_one_pending(
+            owner=self.owner,
+            contract_commit=self.settings.contract_commit,
+            lease_seconds=self.settings.lease_seconds,
+            max_retries=self.settings.max_retries,
+            retry_after=self.settings.retry_after,
+        )
@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import hmac
+import json
+import os
+from pathlib import Path
+import secrets
+import stat
+import time
+from typing import Any, Callable
+
+from .callback_transport import CallbackTransport
+from .errors import AuthenticationError, ValidationError
+from .models import Decision
+from .models import ID_RE, OID_RE
+from .securefs import ensure_secure_directory
+
+
+AUDIT_FIELDS = (
+    "callback_request_id", "event_id", "contract_commit", "received_at",
+    "schema_valid", "auth_valid", "forward_status", "state_version_after",
+)
+
+
+class CallbackAudit:
+    def __init__(self, path: Path):
+        self.path = path
+
+    def write(self, record: dict[str, Any]) -> None:
+        if set(record) != set(AUDIT_FIELDS):
+            raise ValueError("invalid_callback_audit_fields")
+        ensure_secure_directory(self.path.parent)
+        flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
+        fd = os.open(self.path, flags, 0o600)
+        try:
+            info = os.fstat(fd)
+            if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) & 0o077:
+                raise ValueError("unsafe_callback_audit_file")
+            data = (json.dumps(record, separators=(",", ":"), sort_keys=True) + "\n").encode()
+            os.write(fd, data)
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+
+
+class MinimalDecisionTool:
+    """Transport-neutral single-tool boundary; no generic GitOps operations."""
+
+    def __init__(
+        self,
+        transport: CallbackTransport,
+        *,
+        contract_commit: str,
+        access_token: str,
+        forward: Callable[[dict[str, Any]], dict[str, Any]],
+        audit: CallbackAudit,
+    ):
+        self.transport = transport
+        self.contract_commit = contract_commit
+        self.access_token = access_token
+        self.forward = forward
+        self.audit = audit
+
+    def tools(self) -> list[dict[str, Any]]:
+        return [self.transport.mcp_tool()] if self.transport.mode == "secure_mcp_tunnel" else []
+
+    def authenticate(self, authorization: str) -> None:
+        if not hmac.compare_digest(authorization, f"Bearer {self.access_token}"):
+            raise AuthenticationError("invalid_bearer")
+
+    def submit(self, payload: dict[str, Any], authorization: str) -> dict[str, Any]:
+        request_id = secrets.token_hex(16)
+        received_at = int(time.time())
+        schema_valid = False
+        auth_valid = False
+        raw_event_id = payload.get("event_id") if isinstance(payload, dict) else None
+        raw_contract = payload.get("contract_commit") if isinstance(payload, dict) else None
+        event_id = raw_event_id if isinstance(raw_event_id, str) and ID_RE.fullmatch(raw_event_id) else None
+        contract = raw_contract if isinstance(raw_contract, str) and OID_RE.fullmatch(raw_contract) else None
+        forward_status = "REJECTED"
+        state_version = None
+        try:
+            self.authenticate(authorization)
+            auth_valid = True
+            decision = Decision.from_dict(payload)
+            schema_valid = True
+            if decision.contract_commit != self.contract_commit:
+                raise ValidationError("contract_version_mismatch")
+            result = self.forward(decision.to_dict())
+            if result.get("accepted") is not True or not isinstance(result.get("state_version"), int):
+                raise ValidationError("invalid_forward_response")
+            forward_status = "202"
+            state_version = result["state_version"]
+            return result
+        finally:
+            self.audit.write({
+                "callback_request_id": request_id,
+                "event_id": event_id,
+                "contract_commit": contract,
+                "received_at": received_at,
+                "schema_valid": schema_valid,
+                "auth_valid": auth_valid,
+                "forward_status": forward_status,
+                "state_version_after": state_version,
+            })
+
+    def handle_mcp(self, request: dict[str, Any], authorization: str) -> dict[str, Any]:
+        self.authenticate(authorization)
+        method = request.get("method")
+        request_id = request.get("id")
+        if method == "tools/list":
+            return {"jsonrpc": "2.0", "id": request_id, "result": {"tools": self.tools()}}
+        if method != "tools/call":
+            raise ValidationError("unsupported_mcp_method")
+        params = request.get("params")
+        if not isinstance(params, dict) or params.get("name") != "submit_orchestration_decision":
+            raise ValidationError("unsupported_mcp_tool")
+        arguments = params.get("arguments")
+        if not isinstance(arguments, dict):
+            raise ValidationError("invalid_mcp_arguments")
+        result = self.submit(arguments, authorization)
+        return {"jsonrpc": "2.0", "id": request_id, "result": {"structuredContent": result}}
@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import argparse
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+from .action_bridge import resolve_secret_ref, submit_orchestration_decision
+from .callback_transport import CallbackTransport
+from .errors import AuthenticationError, ConfigurationError, OrchestratorError
+from .remote_tool import CallbackAudit, MinimalDecisionTool
+from .securefs import read_secure_json
+
+
+@dataclass(frozen=True)
+class RemoteToolServerConfig:
+    listen_host: str
+    listen_port: int
+    contract_commit: str
+    transport: CallbackTransport
+    access_secret_ref: str
+    callback_secret_ref: str
+    secret_ref_resolver: Path
+    audit_path: Path
+    manifest_path: Path
+
+    @classmethod
+    def from_file(cls, path: Path) -> "RemoteToolServerConfig":
+        value = read_secure_json(path, expected_uid=None)
+        expected = {
+            "listen_host", "listen_port", "contract_commit", "callback_transport",
+            "access_secret_ref", "callback_secret_ref", "secret_ref_resolver",
+            "audit_path", "manifest_path",
+        }
+        if set(value) != expected:
+            raise ConfigurationError("invalid_remote_tool_config_fields")
+        if value["listen_host"] not in {"127.0.0.1", "::1"}:
+            raise ConfigurationError("remote_tool_must_listen_loopback")
+        if not isinstance(value["listen_port"], int) or not 1 <= value["listen_port"] <= 65535:
+            raise ConfigurationError("invalid_remote_tool_port")
+        contract = value["contract_commit"]
+        if not isinstance(contract, str) or len(contract) != 40:
+            raise ConfigurationError("invalid_contract_commit")
+        paths = [Path(value[name]) for name in ("secret_ref_resolver", "audit_path", "manifest_path")]
+        if any(not path_value.is_absolute() for path_value in paths):
+            raise ConfigurationError("remote_tool_path_not_absolute")
+        for name in ("access_secret_ref", "callback_secret_ref"):
+            if not isinstance(value[name], str) or not value[name].startswith("secret://"):
+                raise ConfigurationError("remote_tool_secret_ref_required")
+        return cls(
+            value["listen_host"], value["listen_port"], contract,
+            CallbackTransport.from_dict(value["callback_transport"]),
+            value["access_secret_ref"], value["callback_secret_ref"], paths[0],
+            paths[1], paths[2],
+        )
+
+
+class Handler(BaseHTTPRequestHandler):
+    app: MinimalDecisionTool
+    config: RemoteToolServerConfig
+
+    def log_message(self, _format: str, *_args: object) -> None:
+        return
+
+    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("Cache-Control", "no-store")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def body(self) -> dict[str, Any]:
+        length = int(self.headers.get("Content-Length", "0"))
+        if length <= 0 or length > 1_048_576:
+            raise ValueError("invalid_body_size")
+        value = json.loads(self.rfile.read(length))
+        if not isinstance(value, dict):
+            raise ValueError("invalid_json_root")
+        return value
+
+    def do_GET(self) -> None:
+        path = urlparse(self.path).path
+        if path == "/healthz":
+            self.reply(200, {"status": "active", "mode": self.config.transport.mode})
+            return
+        if path == "/orchestrator-tool-manifest":
+            value = json.loads(self.config.manifest_path.read_text(encoding="utf-8"))
+            self.reply(200, value)
+            return
+        self.reply(404, {"error": "not_found"})
+
+    def do_POST(self) -> None:
+        path = urlparse(self.path).path.rstrip("/")
+        try:
+            if self.config.transport.mode == "remote_https_action" and path == "/v1/orchestrator/decisions":
+                result = self.app.submit(self.body(), self.headers.get("Authorization", ""))
+            elif self.config.transport.mode == "secure_mcp_tunnel" and path == "/mcp":
+                result = self.app.handle_mcp(self.body(), self.headers.get("Authorization", ""))
+            else:
+                self.reply(404, {"error": "not_found"})
+                return
+            self.reply(202, result)
+        except AuthenticationError:
+            self.reply(401, {"error": "UNAUTHORIZED"})
+        except OrchestratorError as exc:
+            self.reply(422, {"error": type(exc).__name__})
+        except Exception:
+            self.reply(500, {"error": "INTERNAL_ERROR"})
+
+
+def serve(config_path: Path) -> None:
+    config = RemoteToolServerConfig.from_file(config_path)
+    access_token = resolve_secret_ref(config.access_secret_ref, config.secret_ref_resolver)
+
+    def forward(payload: dict[str, Any]) -> dict[str, Any]:
+        return submit_orchestration_decision(
+            payload, contract_commit=config.contract_commit,
+            secret_ref=config.callback_secret_ref,
+            resolver=config.secret_ref_resolver,
+        )
+
+    Handler.config = config
+    Handler.app = MinimalDecisionTool(
+        config.transport, contract_commit=config.contract_commit,
+        access_token=access_token, forward=forward, audit=CallbackAudit(config.audit_path),
+    )
+    ThreadingHTTPServer((config.listen_host, config.listen_port), Handler).serve_forever()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("config", type=Path)
+    args = parser.parse_args()
+    serve(args.config)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -168,10 +168,9 @@ def ensure_shared_directory(path: Path, *, group_writable: bool = False) -> None
    reject_symlink_chain(path.parent)
    mode = 0o2770 if group_writable else 0o2750
    try:
+        os.lstat(path)
+    except FileNotFoundError:
        path.mkdir(parents=True, mode=mode)
-    except FileExistsError:
-        pass
-    else:
        os.chmod(path, mode)
    reject_symlink_chain(path)
    st = os.lstat(path)
@@ -25,7 +25,7 @@ from .errors import (
    TriggerError,
    ValidationError,
 )
-from .models import Decision, Event
+from .models import Decision, Event, ID_RE
 from .securefs import ensure_secure_directory, ensure_shared_directory, read_secure_json
 from .store import Store
 from .sync import OpenClawSynchronizer, lifecycle_record
@@ -97,11 +97,16 @@ class Runtime:
        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,
-        )
+        try:
+            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,
+            )
+        except ConflictError as exc:
+            if str(exc) == "stale_decision_state_version":
+                self.store.mark_callback_state(decision.event_id, "STALE_SUPERSEDED")
+            raise
        self._enqueue_decision_lifecycle(decision, version)
        return version

@@ -235,7 +240,7 @@ class Runtime:
        rows = self.store.claim_events(
            owner=self.worker_id,
            statuses=("RECEIVED", "TRIGGER_FAILED"),
-            limit=32,
+            limit=1,
            lease_seconds=self.config.lease_seconds,
        )
        for row in rows:
@@ -243,13 +248,34 @@ class Runtime:
            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,
-                    )
+                    try:
+                        self.store.update_event_trigger(
+                            event.event_id,
+                            status="RECEIVED",
+                            channel="workspace_agent",
+                            remote_id=None,
+                            callback_state="SENT",
+                        )
+                        accepted = self.workspace.trigger(event)
+                    except TriggerError as exc:
+                        self._retry_or_dead_letter(
+                            event,
+                            attempts,
+                            str(exc),
+                            trigger_channel="workspace_agent",
+                            trigger_http_status=exc.http_status,
+                            trigger_error_category=exc.category,
+                        )
+                    else:
+                        self.store.update_event_trigger(
+                            event.event_id,
+                            status="AWAITING_DECISION",
+                            channel="workspace_agent",
+                            remote_id=accepted.remote_id,
+                            trigger_http_status=accepted.http_status,
+                            callback_state="AWAITING_CALLBACK",
+                            callback_deadline_after=self.config.callback_timeout_seconds,
+                        )
                elif self.responses is not None:
                    response_id = self.responses.start(event)
                    self.store.update_event_trigger(
@@ -265,20 +291,31 @@ class Runtime:
                self._retry_or_dead_letter(event, attempts, str(exc))

    def process_workspace_timeouts(self) -> None:
-        cutoff = int(time.time()) - self.config.callback_timeout_seconds
+        cutoff = int(time.time())
        for row in self.store.claim_workspace_timeouts(
            owner=self.worker_id,
            older_than=cutoff,
-            limit=32,
+            limit=1,
            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",
-                )
+                attempts = int(row["callback_attempts"])
+                if attempts + 1 > self.config.max_retries:
+                    self.store.update_event_trigger(
+                        event.event_id, status="DEAD_LETTER",
+                        channel="workspace_agent", remote_id=None,
+                        callback_state="DEAD_LETTER",
+                        increment_callback_attempt=True,
+                    )
+                else:
+                    self.store.update_event_trigger(
+                        event.event_id, status="AWAITING_DECISION",
+                        channel="workspace_agent", remote_id=None,
+                        callback_state="AWAITING_CALLBACK",
+                        callback_deadline_after=self._backoff(attempts),
+                        increment_callback_attempt=True,
+                    )
                continue
            try:
                response_id = self.responses.start(event)
@@ -289,6 +326,8 @@ class Runtime:
                    remote_id=response_id,
                    retry_after=self.config.poll_interval_seconds,
                    increment_attempt=True,
+                    callback_state="AWAITING_CALLBACK",
+                    callback_deadline_after=self.config.callback_timeout_seconds,
                )
            except TriggerError as exc:
                self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
@@ -298,7 +337,7 @@ class Runtime:
            return
        for row in self.store.claim_response_polls(
            owner=self.worker_id,
-            limit=32,
+            limit=1,
            lease_seconds=self.config.lease_seconds,
        ):
            event = Event.from_dict(json.loads(row["event_json"]))
@@ -484,15 +523,26 @@ class Runtime:
            raise ConflictError("event_not_found")
        return str(rows["project_key"])

-    def _retry_or_dead_letter(self, event: Event, attempts: int, reason: str) -> None:
+    def _retry_or_dead_letter(
+        self,
+        event: Event,
+        attempts: int,
+        reason: str,
+        *,
+        trigger_channel: str | None = None,
+        trigger_http_status: int | None = None,
+        trigger_error_category: str | None = None,
+    ) -> 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,
+                channel=trigger_channel,
                remote_id=None,
                increment_attempt=True,
+                trigger_http_status=trigger_http_status,
+                trigger_error_category=trigger_error_category,
            )
            record = lifecycle_record(
                contract_commit=self.config.contract_commit,
@@ -517,10 +567,12 @@ class Runtime:
        self.store.update_event_trigger(
            event.event_id,
            status="TRIGGER_FAILED",
-            channel=None,
+            channel=trigger_channel,
            remote_id=None,
            retry_after=self._backoff(attempts),
            increment_attempt=True,
+            trigger_http_status=trigger_http_status,
+            trigger_error_category=trigger_error_category,
        )

    def _backoff(self, attempts: int) -> int:
@@ -584,6 +636,14 @@ class Handler(BaseHTTPRequestHandler):
                state = self.runtime.store.project_summary(key)
                self._reply(200 if state else 404, state or {"error": "project_not_found"})
                return
+            if path.startswith("/v1/orchestrator/events/"):
+                self._authenticate(self.runtime.config.inbound_token())
+                event_id = path.rsplit("/", 1)[-1]
+                if not ID_RE.fullmatch(event_id):
+                    raise ValidationError("invalid_event_id")
+                status = self.runtime.store.event_status(event_id)
+                self._reply(200 if status else 404, status or {"error": "event_not_found"})
+                return
            self._reply(404, {"error": "not_found"})
        except AuthenticationError as exc:
            self._reply(401, {"error": str(exc)})
@@ -1,6 +1,7 @@
 from __future__ import annotations

 import json
+import os
 from pathlib import Path
 import sqlite3
 import threading
@@ -12,6 +13,15 @@ from .models import Decision, Event
 from .securefs import ensure_secure_directory


+TRIGGER_ERROR_CATEGORIES = frozenset(
+    {"UNAUTHORIZED", "FORBIDDEN", "NOT_FOUND", "NOT_RUNNABLE", "NETWORK_ERROR"}
+)
+CALLBACK_STATES = frozenset({
+    "PENDING", "SENT", "AWAITING_CALLBACK", "CALLBACK_RECEIVED",
+    "APPLIED", "STALE_SUPERSEDED", "DEAD_LETTER",
+})
+
+
 class Store:
    """Durable state machine with idempotency, optimistic versioning, and leases."""

@@ -67,6 +77,13 @@ class Store:
                    status TEXT NOT NULL,
                    channel TEXT,
                    remote_id TEXT,
+                    trigger_http_status INTEGER,
+                    trigger_error_category TEXT,
+                    callback_state TEXT NOT NULL DEFAULT 'PENDING',
+                    callback_deadline INTEGER,
+                    callback_attempts INTEGER NOT NULL DEFAULT 0,
+                    callback_received_at INTEGER,
+                    applied_at INTEGER,
                    attempts INTEGER NOT NULL DEFAULT 0,
                    next_attempt_at INTEGER NOT NULL DEFAULT 0,
                    lease_owner TEXT,
@@ -148,6 +165,24 @@ class Store:
                );
                """
            )
+            columns = {
+                str(row["name"])
+                for row in conn.execute("PRAGMA table_info(events)").fetchall()
+            }
+            if "trigger_http_status" not in columns:
+                conn.execute("ALTER TABLE events ADD COLUMN trigger_http_status INTEGER")
+            if "trigger_error_category" not in columns:
+                conn.execute("ALTER TABLE events ADD COLUMN trigger_error_category TEXT")
+            migrations = {
+                "callback_state": "TEXT NOT NULL DEFAULT 'PENDING'",
+                "callback_deadline": "INTEGER",
+                "callback_attempts": "INTEGER NOT NULL DEFAULT 0",
+                "callback_received_at": "INTEGER",
+                "applied_at": "INTEGER",
+            }
+            for name, declaration in migrations.items():
+                if name not in columns:
+                    conn.execute(f"ALTER TABLE events ADD COLUMN {name} {declaration}")
        finally:
            conn.close()

@@ -230,7 +265,16 @@ class Store:
                WHERE status IN ({placeholders})
                  AND next_attempt_at <= ?
                  AND (lease_until IS NULL OR lease_until < ?)
-                ORDER BY created_at
+                ORDER BY
+                  CASE json_extract(event_json, '$.event_type')
+                    WHEN 'central_gpt_health_probe' THEN 0
+                    WHEN 'a0_unattended_acceptance_requested' THEN 0
+                    ELSE 1
+                  END,
+                  CASE WHEN json_extract(event_json, '$.event_type') IN (
+                    'central_gpt_health_probe', 'a0_unattended_acceptance_requested'
+                  ) THEN created_at END DESC,
+                  created_at ASC
                LIMIT ?
                """,
                (*statuses, now, now, limit),
@@ -251,13 +295,14 @@ class Store:
        if not ids:
            return []
        placeholders = ",".join("?" for _ in ids)
-        return [
-            dict(row)
+        claimed = {
+            str(row["event_id"]): dict(row)
            for row in conn.execute(
                f"SELECT * FROM events WHERE event_id IN ({placeholders})",
                ids,
            ).fetchall()
-        ]
+        }
+        return [claimed[str(event_id)] for event_id in ids]

    def update_event_trigger(
        self,
@@ -268,12 +313,35 @@ class Store:
        remote_id: str | None,
        retry_after: int = 0,
        increment_attempt: bool = False,
+        trigger_http_status: int | None = None,
+        trigger_error_category: str | None = None,
+        callback_state: str | None = None,
+        callback_deadline_after: int | None = None,
+        increment_callback_attempt: bool = False,
    ) -> None:
+        if trigger_http_status is not None and (
+            not isinstance(trigger_http_status, int)
+            or not 100 <= trigger_http_status <= 599
+        ):
+            raise ValueError("invalid_trigger_http_status")
+        if (
+            trigger_error_category is not None
+            and trigger_error_category not in TRIGGER_ERROR_CATEGORIES
+        ):
+            raise ValueError("invalid_trigger_error_category")
+        if callback_state is not None and callback_state not in CALLBACK_STATES:
+            raise ValueError("invalid_callback_state")
+        if callback_deadline_after is not None and callback_deadline_after < 0:
+            raise ValueError("invalid_callback_deadline")
        now = int(time.time())
        self._conn().execute(
            """
            UPDATE events
            SET status=?, channel=?, remote_id=?,
+                trigger_http_status=?, trigger_error_category=?,
+                callback_state=COALESCE(?, callback_state),
+                callback_deadline=?,
+                callback_attempts=callback_attempts + ?,
                attempts=attempts + ?,
                next_attempt_at=?,
                lease_owner=NULL, lease_until=NULL, updated_at=?
@@ -283,6 +351,11 @@ class Store:
                status,
                channel,
                remote_id,
+                trigger_http_status,
+                trigger_error_category,
+                callback_state,
+                None if callback_deadline_after is None else now + callback_deadline_after,
+                1 if increment_callback_attempt else 0,
                1 if increment_attempt else 0,
                now + max(0, retry_after),
                now,
@@ -300,7 +373,7 @@ class Store:
    ) -> list[dict[str, Any]]:
        return self._claim_event_subset(
            owner=owner,
-            where_sql="status='AWAITING_DECISION' AND channel='workspace_agent' AND updated_at <= ?",
+            where_sql="status='AWAITING_DECISION' AND channel='workspace_agent' AND callback_state='AWAITING_CALLBACK' AND callback_deadline <= ?",
            where_params=(older_than,),
            limit=limit,
            lease_seconds=lease_seconds,
@@ -332,7 +405,7 @@ class Store:
        lease_seconds: int,
    ) -> list[dict[str, Any]]:
        allowed = {
-            "status='AWAITING_DECISION' AND channel='workspace_agent' AND updated_at <= ?",
+            "status='AWAITING_DECISION' AND channel='workspace_agent' AND callback_state='AWAITING_CALLBACK' AND callback_deadline <= ?",
            "status='POLLING_RESPONSE' AND next_attempt_at <= ?",
        }
        if where_sql not in allowed:
@@ -402,9 +475,22 @@ class Store:
            if existing:
                if json.loads(existing["decision_json"]) != decision.to_dict():
                    raise ConflictError("decision_idempotency_conflict")
+                conn.execute(
+                    "UPDATE events SET callback_state='APPLIED', callback_deadline=NULL WHERE event_id=?",
+                    (decision.event_id,),
+                )
                conn.execute("COMMIT")
                return int(existing["state_version_after"])

+            conn.execute(
+                """
+                UPDATE events SET callback_state='CALLBACK_RECEIVED',
+                    callback_received_at=?, callback_deadline=NULL
+                WHERE event_id=?
+                """,
+                (now, decision.event_id),
+            )
+
            state = conn.execute(
                "SELECT * FROM project_state WHERE project_key=?",
                (decision.project_key,),
@@ -512,11 +598,12 @@ class Store:
                )
            conn.execute(
                """
-                UPDATE events SET status='DECIDED', lease_owner=NULL,
+                UPDATE events SET status='DECIDED', callback_state='APPLIED',
+                    applied_at=?, callback_deadline=NULL, lease_owner=NULL,
                    lease_until=NULL, updated_at=?
                WHERE event_id=?
                """,
-                (now, decision.event_id),
+                (now, now, decision.event_id),
            )
            conn.execute("COMMIT")
            return next_version
@@ -525,7 +612,7 @@ class Store:
            raise

    def claim_outbox(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
-        return self._claim_table(
+        rows = self._claim_table(
            table="outbox",
            key="work_unit_id",
            owner=owner,
@@ -533,6 +620,182 @@ class Store:
            limit=limit,
            lease_seconds=lease_seconds,
        )
+        for row in rows:
+            context = self._conn().execute(
+                """
+                SELECT e.project_key, d.state_version_after
+                FROM events e JOIN decisions d ON d.event_id=e.event_id
+                WHERE e.event_id=?
+                """,
+                (row["event_id"],),
+            ).fetchone()
+            if context is not None:
+                row["project_key"] = str(context["project_key"])
+                row["state_version"] = int(context["state_version_after"])
+        return rows
+
+    def event_status(self, event_id: str) -> dict[str, Any] | None:
+        row = self._conn().execute(
+            """
+            SELECT e.event_id, e.status, e.channel, e.remote_id,
+                   e.trigger_http_status, e.trigger_error_category,
+                   e.callback_state, e.callback_deadline, e.callback_attempts,
+                   d.state_version_after
+            FROM events e LEFT JOIN decisions d ON d.event_id=e.event_id
+            WHERE e.event_id=?
+            """,
+            (event_id,),
+        ).fetchone()
+        if row is None:
+            return None
+        raw = str(row["status"])
+        if raw == "DECIDED":
+            public_status = "DECIDED"
+        elif raw == "DEAD_LETTER":
+            public_status = "DEAD_LETTER"
+        elif raw == "TRIGGER_FAILED":
+            public_status = "TRIGGER_FAILED"
+        elif raw in {"AWAITING_DECISION", "POLLING_RESPONSE"}:
+            public_status = "AWAITING_DECISION"
+        else:
+            public_status = "RECEIVED"
+        channel = row["channel"]
+        if channel not in {"workspace_agent", "responses_api"}:
+            channel = None
+        version = row["state_version_after"]
+        return {
+            "event_id": str(row["event_id"]),
+            "status": public_status,
+            "trigger_channel": channel,
+            "trigger_http_status": (
+                None if row["trigger_http_status"] is None
+                else int(row["trigger_http_status"])
+            ),
+            "trigger_error_category": row["trigger_error_category"],
+            "remote_id": row["remote_id"],
+            "callback_state": row["callback_state"],
+            "callback_deadline": row["callback_deadline"],
+            "callback_attempts": int(row["callback_attempts"]),
+            "decision_applied": raw == "DECIDED" and version is not None,
+            "state_version_after": None if version is None else int(version),
+        }
+
+    def mark_callback_state(self, event_id: str, state: str) -> None:
+        if state not in CALLBACK_STATES:
+            raise ValueError("invalid_callback_state")
+        now = int(time.time())
+        received = now if state == "CALLBACK_RECEIVED" else None
+        applied = now if state == "APPLIED" else None
+        self._conn().execute(
+            """
+            UPDATE events SET callback_state=?,
+                callback_received_at=COALESCE(?, callback_received_at),
+                applied_at=COALESCE(?, applied_at),
+                callback_deadline=NULL, lease_owner=NULL, lease_until=NULL,
+                updated_at=? WHERE event_id=?
+            """,
+            (state, received, applied, now, event_id),
+        )
+
+    def pending_decision_plan(self, contract_commit: str, max_retries: int) -> dict[str, int]:
+        counts = {key: 0 for key in (
+            "eligible", "stale", "duplicate", "contract_mismatch",
+            "state_version_mismatch", "dead_letter_candidate",
+        )}
+        now = int(time.time())
+        rows = self._conn().execute(
+            """
+            SELECT e.*, p.contract_commit AS current_contract,
+                   p.state_version AS current_state_version,
+                   d.event_id AS decision_exists
+            FROM events e
+            JOIN project_state p ON p.project_key=e.project_key
+            LEFT JOIN decisions d ON d.event_id=e.event_id
+            WHERE e.callback_state IN ('PENDING','SENT','AWAITING_CALLBACK','CALLBACK_RECEIVED')
+            """
+        ).fetchall()
+        for row in rows:
+            event = json.loads(row["event_json"])
+            if row["decision_exists"] is not None:
+                counts["duplicate"] += 1
+            elif event["contract_commit"] != contract_commit or row["current_contract"] != contract_commit:
+                counts["contract_mismatch"] += 1
+            elif int(event["state_version"]) != int(row["current_state_version"]):
+                counts["state_version_mismatch"] += 1
+            elif int(row["callback_attempts"]) >= max_retries:
+                counts["dead_letter_candidate"] += 1
+            elif row["callback_deadline"] is None or int(row["callback_deadline"]) <= now:
+                counts["eligible"] += 1
+            else:
+                counts["stale"] += 1
+        return counts
+
+    def recover_one_pending(self, *, owner: str, contract_commit: str,
+                            lease_seconds: int, max_retries: int,
+                            retry_after: int) -> str | None:
+        now = int(time.time())
+        conn = self._conn()
+        conn.execute("BEGIN IMMEDIATE")
+        try:
+            row = conn.execute(
+                """
+                SELECT e.event_id, e.event_json, e.callback_attempts,
+                       p.contract_commit AS current_contract,
+                       p.state_version AS current_state_version,
+                       d.event_id AS decision_exists
+                FROM events e JOIN project_state p ON p.project_key=e.project_key
+                LEFT JOIN decisions d ON d.event_id=e.event_id
+                WHERE e.callback_state IN ('PENDING','SENT','AWAITING_CALLBACK','CALLBACK_RECEIVED')
+                  AND (e.lease_until IS NULL OR e.lease_until < ?)
+                  AND (
+                    e.callback_deadline IS NULL OR e.callback_deadline <= ?
+                    OR d.event_id IS NOT NULL
+                    OR p.contract_commit != ?
+                    OR CAST(json_extract(e.event_json, '$.state_version') AS INTEGER)
+                       != p.state_version
+                    OR e.callback_attempts >= ?
+                  )
+                ORDER BY e.created_at LIMIT 1
+                """,
+                (now, now, contract_commit, max_retries),
+            ).fetchone()
+            if row is None:
+                conn.execute("COMMIT")
+                return None
+            event = json.loads(row["event_json"])
+            state = "AWAITING_CALLBACK"
+            if row["decision_exists"] is not None:
+                state = "APPLIED"
+            elif event["contract_commit"] != contract_commit or row["current_contract"] != contract_commit:
+                state = "STALE_SUPERSEDED"
+            elif int(event["state_version"]) != int(row["current_state_version"]):
+                state = "STALE_SUPERSEDED"
+            elif int(row["callback_attempts"]) >= max_retries:
+                state = "DEAD_LETTER"
+            conn.execute(
+                """
+                UPDATE events SET lease_owner=?, lease_until=?, callback_state=?,
+                    callback_attempts=callback_attempts+1,
+                    callback_deadline=?, updated_at=? WHERE event_id=?
+                """,
+                (owner, now + lease_seconds, state,
+                 now + retry_after if state == "AWAITING_CALLBACK" else None,
+                 now, row["event_id"]),
+            )
+            conn.execute("COMMIT")
+            self.fsync_state()
+            return str(row["event_id"])
+        except Exception:
+            conn.execute("ROLLBACK")
+            raise
+
+    def fsync_state(self) -> None:
+        self._conn().execute("PRAGMA wal_checkpoint(FULL)")
+        fd = os.open(self.path, os.O_RDONLY)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)

    def mark_outbox(
        self,
@@ -0,0 +1,41 @@
+[Unit]
+Description=AIOS GPT plus one-executor quorum controller
+After=network-online.target
+Wants=network-online.target
+ConditionPathExists=/etc/aios-automation-quorum/controller.json
+
+[Service]
+Type=oneshot
+User=aios-quorum
+Group=aios-quorum
+SupplementaryGroups=aios-orchestrator gitops
+ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/executor-quorum-controller --once --config /etc/aios-automation-quorum/controller.json
+# Exit 2 means the controller successfully persisted a fail-closed BLOCKED
+# decision. Runtime/configuration errors use exit 3 and remain unit failures.
+SuccessExitStatus=2
+UMask=0077
+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-automation-quorum /run/aios-health /data/aios-gitops/repo.git
+ReadWritePaths=/var/lib/aios-automation-quorum
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1 @@
+u aios-quorum - "AIOS automation quorum controller" /var/lib/aios-automation-quorum /usr/sbin/nologin
@@ -0,0 +1,12 @@
+[Unit]
+Description=Poll AIOS executor quorum and recovery sources
+
+[Timer]
+OnBootSec=30s
+OnUnitActiveSec=15s
+AccuracySec=1s
+Persistent=true
+Unit=aios-automation-quorum-controller.service
+
+[Install]
+WantedBy=timers.target
@@ -0,0 +1,3 @@
+d /var/lib/aios-automation-quorum 0700 aios-quorum aios-quorum -
+d /var/lib/aios-automation-quorum/inbox 0700 aios-quorum aios-quorum -
+d /run/aios-health 0750 root aios-quorum -
@@ -0,0 +1,38 @@
+[Unit]
+Description=AIOS central GPT live health probe
+After=network-online.target aios-orchestrator-gateway.service
+Wants=network-online.target
+ConditionPathExists=/etc/aios-automation-quorum/health-probes.json
+
+[Service]
+Type=oneshot
+User=root
+Group=root
+ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/central-gpt-health-probe --config /etc/aios-automation-quorum/health-probes.json
+TimeoutStartSec=120
+UMask=0077
+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-automation-quorum /etc/aios-orchestrator-gateway
+ReadWritePaths=/run/aios-health
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,12 @@
+[Unit]
+Description=Run AIOS central GPT health probe periodically
+
+[Timer]
+OnBootSec=2min
+OnUnitActiveSec=1min
+AccuracySec=5s
+Persistent=true
+Unit=aios-central-gpt-health-probe.service
+
+[Install]
+WantedBy=timers.target
@@ -0,0 +1,42 @@
+[Unit]
+Description=AIOS Codex watcher unattended lifecycle health probe
+After=network-online.target aios-orchestrator-gateway.service aios-codex-watcher.timer
+Wants=network-online.target
+ConditionPathExists=/etc/aios-automation-quorum/health-probes.json
+
+[Service]
+Type=oneshot
+User=root
+Group=aios-orchestrator
+ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/codex-executor-health-probe --config /etc/aios-automation-quorum/health-probes.json
+TimeoutStartSec=120
+# The probe creates commands as 0640. Preserve group-read access so the
+# aios-orchestrator watcher can consume them without making the queue public.
+UMask=0027
+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-automation-quorum /etc/aios-orchestrator-gateway /var/log/aios-codex-watcher
+# SQLite read-only connections still need the dedicated WAL directory writable
+# for shared-memory/locking state. The probe opens the database with mode=ro.
+ReadWritePaths=/run/aios-health /data/aios-gitops/workspace/agents/commands/active /var/lib/aios-codex-watcher
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,12 @@
+[Unit]
+Description=Run AIOS Codex executor health probe periodically
+
+[Timer]
+OnBootSec=3min
+OnUnitActiveSec=1min
+AccuracySec=5s
+Persistent=true
+Unit=aios-codex-executor-health-probe.service
+
+[Install]
+WantedBy=timers.target
@@ -5,6 +5,7 @@ ConditionPathExists=/etc/aios-orchestrator-gateway/config.json

 [Service]
 Type=oneshot
+TimeoutStartSec=120
 User=aios-orchestrator
 Group=aios-orchestrator
 SupplementaryGroups=gitops
@@ -32,6 +33,6 @@ 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
+ReadWritePaths=/var/lib/aios-orchestrator-gateway /data/aios-gitops/workspace/agents/commands/active
 SystemCallFilter=@system-service
 SystemCallErrorNumber=EPERM
@@ -42,7 +42,7 @@ 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
+ReadWritePaths=/var/lib/aios-orchestrator-gateway /data/aios-gitops/workspace/agents/commands/active
 SystemCallFilter=@system-service
 SystemCallErrorNumber=EPERM
 StandardOutput=journal
@@ -1,4 +1,4 @@
-d /etc/aios-orchestrator-gateway 0700 root root -
+d /etc/aios-orchestrator-gateway 0750 root aios-orchestrator -
 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 -
@@ -0,0 +1,22 @@
+[Unit]
+Description=AIOS minimal orchestration decision tool
+After=network-online.target aios-orchestrator-gateway.service
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=aios-orchestrator
+Group=aios-orchestrator
+ExecStart=/usr/bin/python3 -m services.aios_orchestrator_gateway.remote_tool_server /etc/aios-orchestrator-gateway/remote-tool.json
+WorkingDirectory=/opt/aios-orchestrator-gateway/current
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadWritePaths=/var/lib/aios-orchestrator-gateway
+Restart=on-failure
+RestartSec=3
+
+[Install]
+WantedBy=multi-user.target
@@ -0,0 +1,201 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+import tempfile
+import unittest
+from unittest import mock
+
+from services.aios_orchestrator_gateway.callback_transport import CallbackTransport
+from services.aios_orchestrator_gateway.errors import ConfigurationError, ConflictError
+from services.aios_orchestrator_gateway.models import Decision, Event
+from services.aios_orchestrator_gateway.recovery import PendingDecisionRecovery, RecoverySettings
+from services.aios_orchestrator_gateway.remote_tool import CallbackAudit, MinimalDecisionTool
+from services.aios_orchestrator_gateway.service import Runtime
+from services.aios_orchestrator_gateway.store import Store
+from tests.test_aios_orchestrator_gateway import CONTRACT, decision_payload, event_payload
+
+
+def transport(mode: str, endpoint: str, approved=None) -> CallbackTransport:
+    return CallbackTransport.from_dict({
+        "mode": mode,
+        "endpoint": endpoint,
+        "approved_mcp_tunnel_endpoints": approved or [],
+    })
+
+
+class CallbackTransportTests(unittest.TestCase):
+    def test_rejects_placeholder_action_host(self):
+        with self.assertRaises(ConfigurationError):
+            transport("remote_https_action", "https://REPLACE_WITH_ACTIONS_HOST")
+
+    def test_rejects_loopback_cloud_tool_url(self):
+        for endpoint in (
+            "https://localhost/tool", "https://127.0.0.1/tool",
+            "https://0.0.0.0/tool", "http://actions.example.org/tool",
+        ):
+            with self.subTest(endpoint=endpoint), self.assertRaises(ConfigurationError):
+                transport("remote_https_action", endpoint)
+
+    def test_accepts_approved_https_action_url(self):
+        value = transport("remote_https_action", "https://actions.example.org/aios")
+        self.assertEqual(value.action_openapi()["servers"][0]["url"], value.endpoint)
+        self.assertEqual(
+            list(value.action_openapi()["paths"]), ["/v1/orchestrator/decisions"]
+        )
+
+    def test_accepts_secure_mcp_tunnel_endpoint(self):
+        endpoint = "https://mcp-tunnel.corp.example/aios"
+        value = transport("secure_mcp_tunnel", endpoint, [endpoint])
+        tool = value.mcp_tool()
+        self.assertEqual(tool["name"], "submit_orchestration_decision")
+        self.assertFalse(tool["annotations"]["readOnlyHint"])
+        self.assertTrue(tool["annotations"]["idempotentHint"])
+
+    def test_unapproved_secure_mcp_tunnel_is_rejected(self):
+        with self.assertRaises(ConfigurationError):
+            transport("secure_mcp_tunnel", "https://other.corp.example/mcp", [])
+
+
+class CallbackStateRecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.store = Store(self.root / "state.sqlite3")
+
+    def tearDown(self):
+        self.store.close()
+        self.temp.cleanup()
+
+    def test_callback_deadline_transitions(self):
+        self.store.register_event(Event.from_dict(event_payload()))
+        self.assertEqual(self.store.event_status("evt-001")["callback_state"], "PENDING")
+        self.store.update_event_trigger(
+            "evt-001", status="RECEIVED", channel="workspace_agent", remote_id=None,
+            callback_state="SENT",
+        )
+        self.assertEqual(self.store.event_status("evt-001")["callback_state"], "SENT")
+        self.store.update_event_trigger(
+            "evt-001", status="AWAITING_DECISION", channel="workspace_agent",
+            remote_id=None, callback_state="AWAITING_CALLBACK", callback_deadline_after=0,
+        )
+        claimed = self.store.claim_workspace_timeouts(
+            owner="worker", older_than=10**10, limit=1, lease_seconds=30
+        )
+        self.assertEqual([row["event_id"] for row in claimed], ["evt-001"])
+
+    def test_stale_state_is_superseded(self):
+        self.store.register_event(Event.from_dict(event_payload()))
+        self.store._conn().execute(
+            "UPDATE project_state SET state_version=2 WHERE project_key=?",
+            ("v2r1-personal-weixin",),
+        )
+        runtime = object.__new__(Runtime)
+        runtime.config = SimpleNamespace(
+            contract_commit=CONTRACT, max_hops_per_gate=20,
+            max_same_blocker_retries=3,
+        )
+        runtime.store = self.store
+        with self.assertRaisesRegex(ConflictError, "stale_decision_state_version"):
+            runtime.submit_decision(decision_payload())
+        self.assertEqual(
+            self.store.event_status("evt-001")["callback_state"], "STALE_SUPERSEDED"
+        )
+
+    def test_pending_recovery_is_one_at_a_time(self):
+        for event_id in ("evt-001", "evt-002"):
+            self.store.register_event(Event.from_dict(event_payload(event_id=event_id)))
+            self.store.update_event_trigger(
+                event_id, status="AWAITING_DECISION", channel="workspace_agent",
+                remote_id=None, callback_state="AWAITING_CALLBACK", callback_deadline_after=0,
+            )
+        recovery = PendingDecisionRecovery(RecoverySettings(
+            self.store.path, CONTRACT, 30, 3, 10
+        ))
+        recovery.store.close()
+        recovery.store = self.store
+        processed = recovery.apply_one()
+        attempts = self.store._conn().execute(
+            "SELECT event_id, callback_attempts FROM events ORDER BY event_id"
+        ).fetchall()
+        self.assertIn(processed, {"evt-001", "evt-002"})
+        self.assertEqual(sum(int(row["callback_attempts"]) for row in attempts), 1)
+
+    def test_pending_recovery_plan_has_required_categories(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, callback_state="AWAITING_CALLBACK", callback_deadline_after=0,
+        )
+        plan = self.store.pending_decision_plan(CONTRACT, 3)
+        self.assertEqual(set(plan), {
+            "eligible", "stale", "duplicate", "contract_mismatch",
+            "state_version_mismatch", "dead_letter_candidate",
+        })
+        self.assertEqual(plan["eligible"], 1)
+
+    def test_completed_side_effect_not_replayed(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
+        )
+        count = self.store._conn().execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
+        self.assertEqual(first, second)
+        self.assertEqual(count, 1)
+
+
+class MinimalToolTests(unittest.TestCase):
+    def test_callback_audit_redacts_credentials(self):
+        with tempfile.TemporaryDirectory() as directory:
+            audit_path = Path(directory) / "callback-audit.jsonl"
+            app = MinimalDecisionTool(
+                transport("remote_https_action", "https://actions.example.org"),
+                contract_commit=CONTRACT,
+                access_token="credential-never-persisted",
+                forward=lambda _payload: {"accepted": True, "state_version": 1},
+                audit=CallbackAudit(audit_path),
+            )
+            app.submit(decision_payload(), "Bearer credential-never-persisted")
+            value = audit_path.read_text(encoding="utf-8")
+            self.assertNotIn("credential-never-persisted", value)
+            self.assertNotIn("Authorization", value)
+            self.assertNotIn("secret_ref", value)
+            self.assertEqual(set(json.loads(value)), {
+                "callback_request_id", "event_id", "contract_commit", "received_at",
+                "schema_valid", "auth_valid", "forward_status", "state_version_after",
+            })
+
+    def test_workspace_tool_calls_decision_once(self):
+        forward = mock.Mock(return_value={"accepted": True, "state_version": 1})
+        with tempfile.TemporaryDirectory() as directory:
+            app = MinimalDecisionTool(
+                transport(
+                    "secure_mcp_tunnel", "https://mcp-tunnel.corp.example/aios",
+                    ["https://mcp-tunnel.corp.example/aios"],
+                ),
+                contract_commit=CONTRACT,
+                access_token="access",
+                forward=forward,
+                audit=CallbackAudit(Path(directory) / "audit.jsonl"),
+            )
+            tools = app.handle_mcp(
+                {"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, "Bearer access"
+            )
+            self.assertEqual([tool["name"] for tool in tools["result"]["tools"]], [
+                "submit_orchestration_decision"
+            ])
+            app.handle_mcp({
+                "jsonrpc": "2.0", "id": 2, "method": "tools/call",
+                "params": {"name": "submit_orchestration_decision", "arguments": decision_payload()},
+            }, "Bearer access")
+        forward.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,244 @@
+from __future__ import annotations
+
+import json
+import importlib.util
+import os
+from pathlib import Path
+import subprocess
+import unittest
+import urllib.error
+from unittest import mock
+
+from services.aios_orchestrator_gateway.action_bridge import (
+    CallbackBridgeError,
+    resolve_secret_ref,
+    submit_orchestration_decision,
+)
+from tests.test_aios_orchestrator_gateway import CONTRACT, decision_payload
+
+
+class DecisionActionBridgeTests(unittest.TestCase):
+    def test_18890_openapi_and_handler_expose_only_decision_callback_operation(self):
+        root = Path(__file__).resolve().parents[1]
+        openapi = (root / "workspace-agent/orchestrator-decision-action.openapi.yaml").read_text(
+            encoding="utf-8"
+        )
+        source = (root / "active-runtime/app/aios_chatgpt_actions_api.py").read_text(
+            encoding="utf-8"
+        )
+        self.assertIn("operationId: submit_orchestration_decision", openapi)
+        self.assertIn("if path == '/v1/orchestrator/decisions'", source)
+        self.assertIn("self.need('orchestrator:decision')", source)
+        self.assertIn("submit_orchestration_decision(", source)
+
+    def test_18890_handler_calls_dedicated_callback_bridge(self):
+        root = Path(__file__).resolve().parents[1]
+        module_path = root / "active-runtime/app/aios_chatgpt_actions_api.py"
+        spec = importlib.util.spec_from_file_location("actions_api_under_test", module_path)
+        module = importlib.util.module_from_spec(spec)
+        previous = os.environ.get("AIOS_GITOPS_WORKSPACE")
+        os.environ["AIOS_GITOPS_WORKSPACE"] = str(root)
+        try:
+            assert spec.loader is not None
+            spec.loader.exec_module(module)
+        finally:
+            if previous is None:
+                os.environ.pop("AIOS_GITOPS_WORKSPACE", None)
+            else:
+                os.environ["AIOS_GITOPS_WORKSPACE"] = previous
+
+        class FakeHandler:
+            def __init__(self):
+                self.sent = None
+
+            def need(self, scope):
+                self.scope = scope
+                return {"account": "central-gpt"}
+
+            def sendj(self, payload, status=200):
+                self.sent = (payload, status)
+                return self.sent
+
+        handler = FakeHandler()
+        with (
+            mock.patch.object(module, "ORCHESTRATOR_CONTRACT_COMMIT", CONTRACT),
+            mock.patch.object(
+                module,
+                "ORCHESTRATOR_SECRET_REF_RESOLVER",
+                Path("/usr/libexec/aios-secret-resolver"),
+            ),
+            mock.patch.object(
+                module,
+                "submit_orchestration_decision",
+                return_value={"accepted": True, "state_version": 2},
+            ) as submit,
+        ):
+            module.H.orchestration_decision(handler, decision_payload())
+        self.assertEqual(handler.scope, "orchestrator:decision")
+        self.assertEqual(handler.sent, ({"accepted": True, "state_version": 2}, 202))
+        submit.assert_called_once()
+
+    def test_18890_handler_redacts_unexpected_bridge_exception(self):
+        root = Path(__file__).resolve().parents[1]
+        module_path = root / "active-runtime/app/aios_chatgpt_actions_api.py"
+        spec = importlib.util.spec_from_file_location("actions_api_error_test", module_path)
+        module = importlib.util.module_from_spec(spec)
+        previous = os.environ.get("AIOS_GITOPS_WORKSPACE")
+        os.environ["AIOS_GITOPS_WORKSPACE"] = str(root)
+        try:
+            assert spec.loader is not None
+            spec.loader.exec_module(module)
+        finally:
+            if previous is None:
+                os.environ.pop("AIOS_GITOPS_WORKSPACE", None)
+            else:
+                os.environ["AIOS_GITOPS_WORKSPACE"] = previous
+
+        class FakeHandler:
+            def need(self, _scope):
+                return {"account": "central-gpt"}
+
+            def sendj(self, payload, status=200):
+                self.sent = (payload, status)
+                return self.sent
+
+        handler = FakeHandler()
+        with (
+            mock.patch.object(module, "ORCHESTRATOR_CONTRACT_COMMIT", CONTRACT),
+            mock.patch.object(
+                module,
+                "ORCHESTRATOR_SECRET_REF_RESOLVER",
+                Path("/usr/libexec/aios-secret-resolver"),
+            ),
+            mock.patch.object(
+                module,
+                "submit_orchestration_decision",
+                side_effect=RuntimeError("sensitive-token-value"),
+            ),
+        ):
+            module.H.orchestration_decision(handler, decision_payload())
+        self.assertEqual(handler.sent, ({"error": "CALLBACK_INTERNAL_ERROR"}, 500))
+
+    def _response(self, *, version=1):
+        response = mock.MagicMock(status=202)
+        response.read.return_value = json.dumps(
+            {"accepted": True, "state_version": version}
+        ).encode()
+        response.__enter__.return_value = response
+        response.__exit__.return_value = False
+        return response
+
+    def test_strict_schema_rejects_unknown_field_before_secret_resolution(self):
+        payload = decision_payload()
+        payload["unexpected"] = True
+        with mock.patch(
+            "services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref"
+        ) as resolver:
+            with self.assertRaisesRegex(CallbackBridgeError, "INVALID_DECISION"):
+                submit_orchestration_decision(
+                    payload,
+                    contract_commit=CONTRACT,
+                    secret_ref="secret://callback/gateway",
+                    resolver=Path("/usr/libexec/aios-secret-resolver"),
+                )
+        resolver.assert_not_called()
+
+    def test_contract_mismatch_rejected_before_secret_resolution(self):
+        with mock.patch(
+            "services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref"
+        ) as resolver:
+            with self.assertRaisesRegex(CallbackBridgeError, "CONTRACT_MISMATCH"):
+                submit_orchestration_decision(
+                    decision_payload(),
+                    contract_commit="f" * 40,
+                    secret_ref="secret://callback/gateway",
+                    resolver=Path("/usr/libexec/aios-secret-resolver"),
+                )
+        resolver.assert_not_called()
+
+    def test_valid_decision_forwards_to_fixed_gateway_with_secret_ref_token(self):
+        opened = mock.Mock(return_value=self._response(version=4))
+        with mock.patch(
+            "services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref",
+            return_value="resolved-token-not-logged",
+        ):
+            result = submit_orchestration_decision(
+                decision_payload(),
+                contract_commit=CONTRACT,
+                secret_ref="secret://callback/gateway",
+                resolver=Path("/usr/libexec/aios-secret-resolver"),
+                request_open=opened,
+            )
+        request = opened.call_args.args[0]
+        self.assertEqual(
+            request.full_url,
+            "http://127.0.0.1:18920/v1/orchestrator/decisions",
+        )
+        self.assertEqual(request.headers["Idempotency-key"], "evt-001")
+        self.assertEqual(result, {"accepted": True, "state_version": 4})
+
+    def test_stale_state_409_fails_closed_without_response_body(self):
+        error = urllib.error.HTTPError(
+            "http://127.0.0.1:18920/v1/orchestrator/decisions",
+            409,
+            "redacted",
+            {},
+            None,
+        )
+        with mock.patch(
+            "services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref",
+            return_value="resolved-token-not-logged",
+        ):
+            with self.assertRaisesRegex(CallbackBridgeError, "STALE_STATE") as raised:
+                submit_orchestration_decision(
+                    decision_payload(),
+                    contract_commit=CONTRACT,
+                    secret_ref="secret://callback/gateway",
+                    resolver=Path("/usr/libexec/aios-secret-resolver"),
+                    request_open=mock.Mock(side_effect=error),
+                )
+        self.assertEqual(raised.exception.status_code, 409)
+
+    def test_duplicate_callback_reuses_same_idempotency_key(self):
+        opened = mock.Mock(side_effect=[self._response(), self._response()])
+        with mock.patch(
+            "services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref",
+            return_value="resolved-token-not-logged",
+        ):
+            for _ in range(2):
+                submit_orchestration_decision(
+                    decision_payload(),
+                    contract_commit=CONTRACT,
+                    secret_ref="secret://callback/gateway",
+                    resolver=Path("/usr/libexec/aios-secret-resolver"),
+                    request_open=opened,
+                )
+        keys = [call.args[0].headers["Idempotency-key"] for call in opened.call_args_list]
+        self.assertEqual(keys, ["evt-001", "evt-001"])
+
+    def test_secret_ref_is_stdin_not_argv_or_environment(self):
+        completed = subprocess.CompletedProcess(
+            ["/usr/libexec/aios-secret-resolver"], 0, b"resolved-token\n", b""
+        )
+        with (
+            mock.patch(
+                "services.aios_orchestrator_gateway.action_bridge._trusted_resolver",
+                return_value=Path("/usr/libexec/aios-secret-resolver"),
+            ),
+            mock.patch(
+                "services.aios_orchestrator_gateway.action_bridge.subprocess.run",
+                return_value=completed,
+            ) as run,
+        ):
+            token = resolve_secret_ref(
+                "secret://callback/gateway",
+                Path("/usr/libexec/aios-secret-resolver"),
+            )
+        self.assertEqual(token, "resolved-token")
+        self.assertEqual(run.call_args.args[0], ["/usr/libexec/aios-secret-resolver"])
+        self.assertIn(b"secret://callback/gateway", run.call_args.kwargs["input"])
+        self.assertNotIn("secret", " ".join(run.call_args.kwargs["env"].values()).lower())
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import importlib.util
+import os
+from pathlib import Path
+import json
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+
+MODULE = Path(__file__).resolve().parents[1] / "automation" / "health_probe_common.py"
+SPEC = importlib.util.spec_from_file_location("health_probe_common", MODULE)
+assert SPEC and SPEC.loader
+common = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(common)
+sys.modules["health_probe_common"] = common
+
+CENTRAL_MODULE = Path(__file__).resolve().parents[1] / "automation" / "central_gpt_health_probe.py"
+CENTRAL_SPEC = importlib.util.spec_from_file_location("central_gpt_health_probe", CENTRAL_MODULE)
+assert CENTRAL_SPEC and CENTRAL_SPEC.loader
+central = importlib.util.module_from_spec(CENTRAL_SPEC)
+CENTRAL_SPEC.loader.exec_module(central)
+
+CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
+
+
+class HealthProbeSecurityTests(unittest.TestCase):
+    def test_existing_pending_event_is_rechecked_without_duplicate_post(self):
+        config = {
+            "gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
+            "gateway_token_file": "/run/aios-orchestrator-gateway/token",
+            "project_key": "v2r1-personal-weixin",
+            "contract_commit": CONTRACT,
+            "central_timeout_seconds": 1,
+            "poll_interval_seconds": 0.01,
+            "central_receipt_path": "/run/aios-health/central-gpt.json",
+        }
+        event_id = "health-central-" + "a" * 32
+        with (
+            mock.patch.object(central, "load_config", return_value=config),
+            mock.patch.object(central, "read_token", return_value="not-a-real-token"),
+            mock.patch.object(central, "_load_pending", return_value=(event_id, 3)),
+            mock.patch.object(
+                central,
+                "request_json",
+                return_value=(200, {"status": "TRIGGER_FAILED"}),
+            ) as request,
+        ):
+            with self.assertRaisesRegex(common.ProbeError, "central_decision_not_proven"):
+                central.run(Path("/not/read"))
+        self.assertTrue(request.call_args_list)
+        self.assertTrue(all(call.kwargs.get("method", "GET") == "GET" for call in request.call_args_list))
+
+    def test_central_pending_marker_is_metadata_only_and_reusable(self):
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "central-gpt.pending.json"
+            path.write_text(
+                json.dumps(
+                    {
+                        "contract_commit": CONTRACT,
+                        "event_id": "health-central-" + "a" * 32,
+                        "state_version": 7,
+                    }
+                ),
+                encoding="utf-8",
+            )
+            os.chmod(path, 0o640)
+            self.assertEqual(
+                central._load_pending(path, CONTRACT, owners={os.geteuid()}),
+                ("health-central-" + "a" * 32, 7),
+            )
+
+    def test_central_pending_marker_wrong_contract_fails_closed(self):
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "central-gpt.pending.json"
+            path.write_text(
+                json.dumps(
+                    {
+                        "contract_commit": "0" * 40,
+                        "event_id": "health-central-" + "a" * 32,
+                        "state_version": 0,
+                    }
+                ),
+                encoding="utf-8",
+            )
+            os.chmod(path, 0o640)
+            with self.assertRaises(common.ProbeError):
+                central._load_pending(path, CONTRACT, owners={os.geteuid()})
+
+    def test_central_pending_marker_symlink_fails_closed(self):
+        if not hasattr(os, "symlink"):
+            self.skipTest("symlink unavailable")
+        with tempfile.TemporaryDirectory() as temp:
+            target = Path(temp) / "target"
+            target.write_text("{}", encoding="utf-8")
+            link = Path(temp) / "central-gpt.pending.json"
+            try:
+                link.symlink_to(target)
+            except OSError:
+                self.skipTest("symlink privilege unavailable")
+            with self.assertRaises(common.ProbeError):
+                central._load_pending(link, CONTRACT, owners={os.geteuid()})
+
+    def test_probe_error_reason_is_metadata_only(self):
+        self.assertEqual(central._safe_reason(common.ProbeError("gateway_unavailable")), "gateway_unavailable")
+        self.assertEqual(central._safe_reason(ValueError("secret value leaked!")), "probe_error")
+
+    def test_availability_receipts_are_readable_only_by_quorum_group(self):
+        source = MODULE.read_text(encoding="utf-8")
+        self.assertIn("os.fchown(fd, 0, info.st_gid)", source)
+        self.assertIn("os.fchmod(fd, 0o640)", source)
+        self.assertNotIn("os.fchmod(fd, 0o644)", source)
+
+    def test_codex_probe_commands_are_group_readable_by_watcher(self):
+        unit = (
+            Path(__file__).resolve().parents[1]
+            / "systemd"
+            / "aios-codex-executor-health-probe.service"
+        ).read_text(encoding="utf-8")
+        self.assertIn("User=root\n", unit)
+        self.assertIn("Group=aios-orchestrator\n", unit)
+        self.assertIn("UMask=0027\n", unit)
+        self.assertIn(
+            "ReadOnlyPaths=/etc/aios-automation-quorum "
+            "/etc/aios-orchestrator-gateway /var/log/aios-codex-watcher\n",
+            unit,
+        )
+        self.assertIn(
+            "ReadWritePaths=/run/aios-health "
+            "/data/aios-gitops/workspace/agents/commands/active "
+            "/var/lib/aios-codex-watcher\n",
+            unit,
+        )
+        self.assertNotIn("Group=root\n", unit)
+
+    def test_gateway_endpoint_is_exact_loopback(self):
+        self.assertEqual(
+            common.gateway_base("http://127.0.0.1:18920/v1/orchestrator/events"),
+            "http://127.0.0.1:18920/v1/orchestrator",
+        )
+        for url in (
+            "http://127.0.0.1:8765/v1/orchestrator/events",
+            "http://localhost:18920/v1/orchestrator/events",
+            "https://127.0.0.1:18920/v1/orchestrator/events",
+            "http://127.0.0.1:18920/v1/orchestrator/events?token=x",
+        ):
+            with self.assertRaises(common.ProbeError):
+                common.gateway_base(url)
+
+    def test_token_newline_is_rejected(self):
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "token"
+            path.write_bytes(b"not-a-secret-value\n")
+            os.chmod(path, 0o600)
+            with self.assertRaises(common.ProbeError):
+                common.read_token(path)
+
+    def test_token_symlink_is_rejected(self):
+        if not hasattr(os, "symlink"):
+            self.skipTest("symlink unavailable")
+        with tempfile.TemporaryDirectory() as temp:
+            target = Path(temp) / "target"
+            target.write_bytes(b"not-a-secret-value")
+            os.chmod(target, 0o600)
+            link = Path(temp) / "link"
+            try:
+                link.symlink_to(target)
+            except OSError:
+                self.skipTest("symlink privilege unavailable")
+            with self.assertRaises(common.ProbeError):
+                common.read_token(link)
+
+    def test_token_wide_mode_is_rejected(self):
+        if os.name != "posix":
+            self.skipTest("POSIX mode semantics required")
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "token"
+            path.write_bytes(b"not-a-secret-value")
+            os.chmod(path, 0o644)
+            with self.assertRaises(common.ProbeError):
+                common.read_token(path)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -2,6 +2,7 @@ from __future__ import annotations

 import json
 import os
+import sqlite3
 import stat
 from pathlib import Path
 import tempfile
@@ -10,12 +11,17 @@ import unittest
 import urllib.error
 from unittest import mock

-from services.aios_orchestrator_gateway.clients import ResponsesClient, WorkspaceAgentClient
+from services.aios_orchestrator_gateway.clients import (
+    ResponsesClient,
+    WorkspaceAgentClient,
+    WorkspaceTriggerResult,
+)
 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.errors import ConflictError, DispatchError, TriggerError, ValidationError
 from services.aios_orchestrator_gateway.models import Decision, Event
 from services.aios_orchestrator_gateway.service import Runtime
+from services.aios_orchestrator_gateway.securefs import ensure_shared_directory
 from services.aios_orchestrator_gateway.store import Store
 from services.aios_orchestrator_gateway.sync import OpenClawSynchronizer, lifecycle_record
 from scripts import report_orchestrator_event
@@ -110,7 +116,12 @@ class ModelTests(unittest.TestCase):


 class ClientTests(unittest.TestCase):
-    def test_workspace_trigger_accepts_202_without_run_id(self):
+    def _workspace_http_error(self, code):
+        return urllib.error.HTTPError(
+            "https://api.chatgpt.com/redacted", code, "redacted", {}, None
+        )
+
+    def test_workspace_202_empty_body_is_accepted(self):
        event = Event.from_dict(event_payload())
        response = mock.MagicMock()
        response.status = 202
@@ -121,9 +132,74 @@ class ClientTests(unittest.TestCase):
            "services.aios_orchestrator_gateway.clients.urllib.request.urlopen",
            return_value=response,
        ) as call:
-            WorkspaceAgentClient("agtch_test", "token").trigger(event)
+            result = WorkspaceAgentClient("agtch_test", "token").trigger(event)
        request = call.call_args.args[0]
        self.assertEqual(request.headers["Idempotency-key"], "evt-001")
+        self.assertEqual(result.http_status, 202)
+
+    def test_workspace_202_requires_no_remote_id(self):
+        response = mock.MagicMock(status=202)
+        response.__enter__.return_value = response
+        response.__exit__.return_value = False
+        with mock.patch(
+            "services.aios_orchestrator_gateway.clients.urllib.request.urlopen",
+            return_value=response,
+        ):
+            result = WorkspaceAgentClient("agtch_test", "token").trigger(
+                Event.from_dict(event_payload())
+            )
+        self.assertIsNone(result.remote_id)
+
+    def _assert_workspace_http_category(self, code, category):
+        with mock.patch(
+            "services.aios_orchestrator_gateway.clients.urllib.request.urlopen",
+            side_effect=self._workspace_http_error(code),
+        ):
+            with self.assertRaises(TriggerError) as raised:
+                WorkspaceAgentClient("agtch_test", "token").trigger(
+                    Event.from_dict(event_payload())
+                )
+        self.assertEqual(raised.exception.http_status, code)
+        self.assertEqual(raised.exception.category, category)
+
+    def test_workspace_401_wrong_token_type(self):
+        self._assert_workspace_http_category(401, "UNAUTHORIZED")
+
+    def test_workspace_403_agent_not_shared(self):
+        self._assert_workspace_http_category(403, "FORBIDDEN")
+
+    def test_workspace_404_trigger_not_visible(self):
+        self._assert_workspace_http_category(404, "NOT_FOUND")
+
+    def test_workspace_409_channel_not_runnable(self):
+        self._assert_workspace_http_category(409, "NOT_RUNNABLE")
+
+    def test_workspace_network_failure_classified(self):
+        with mock.patch(
+            "services.aios_orchestrator_gateway.clients.urllib.request.urlopen",
+            side_effect=urllib.error.URLError("redacted"),
+        ):
+            with self.assertRaises(TriggerError) as raised:
+                WorkspaceAgentClient("agtch_test", "token").trigger(
+                    Event.from_dict(event_payload())
+                )
+        self.assertIsNone(raised.exception.http_status)
+        self.assertEqual(raised.exception.category, "NETWORK_ERROR")
+
+    def test_workspace_retry_reuses_idempotency_key(self):
+        event = Event.from_dict(event_payload())
+        response = mock.MagicMock(status=202)
+        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:
+            client = WorkspaceAgentClient("agtch_test", "token")
+            client.trigger(event)
+            client.trigger(event)
+        keys = [item.args[0].headers["Idempotency-key"] for item in call.call_args_list]
+        self.assertEqual(keys, ["evt-001", "evt-001"])

    def test_responses_background_start_and_poll(self):
        event = Event.from_dict(event_payload())
@@ -208,6 +284,69 @@ class StoreTests(unittest.TestCase):
        self.assertTrue(created)
        self.assertFalse(duplicate)

+    def test_legacy_event_table_migrates_trigger_metadata_columns(self):
+        self.store.close()
+        database = Path(self.temp.name) / "legacy.sqlite3"
+        with sqlite3.connect(database) as connection:
+            connection.execute(
+                """
+                CREATE TABLE 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
+                )
+                """
+            )
+        migrated = Store(database)
+        try:
+            columns = {
+                row["name"]
+                for row in migrated._conn().execute("PRAGMA table_info(events)")
+            }
+        finally:
+            migrated.close()
+        self.assertIn("trigger_http_status", columns)
+        self.assertIn("trigger_error_category", columns)
+        self.assertIn("callback_state", columns)
+        self.assertIn("callback_deadline", columns)
+
+    def test_event_status_is_metadata_only_and_maps_internal_states(self):
+        self.store.register_event(Event.from_dict(event_payload()))
+        received = self.store.event_status("evt-001")
+        self.assertEqual(
+            set(received),
+            {
+                "event_id", "status", "trigger_channel", "trigger_http_status",
+                "trigger_error_category", "remote_id", "decision_applied",
+                "state_version_after", "callback_state", "callback_deadline",
+                "callback_attempts",
+            },
+        )
+        self.assertEqual(received["status"], "RECEIVED")
+        self.store.update_event_trigger(
+            "evt-001", status="POLLING_RESPONSE", channel="responses_api", remote_id="resp-1"
+        )
+        waiting = self.store.event_status("evt-001")
+        self.assertEqual(waiting["status"], "AWAITING_DECISION")
+        self.assertEqual(waiting["trigger_channel"], "responses_api")
+
+    def test_event_status_decided_has_applied_version(self):
+        self.store.register_event(Event.from_dict(event_payload()))
+        decision = Decision.from_dict(
+            decision_payload(verdict="NOOP", next_work_units=[])
+        )
+        version = self.store.apply_decision(
+            decision, max_hops_per_gate=20, max_same_blocker_retries=3
+        )
+        status = self.store.event_status("evt-001")
+        self.assertEqual(status["status"], "DECIDED")
+        self.assertTrue(status["decision_applied"])
+        self.assertEqual(status["state_version_after"], version)
+
    def test_event_payload_conflict_rejected(self):
        self.store.register_event(Event.from_dict(event_payload()))
        with self.assertRaises(ConflictError):
@@ -235,12 +374,58 @@ class StoreTests(unittest.TestCase):
        self.assertEqual(len(first), 1)
        self.assertEqual(second, [])

+    def test_health_probe_claim_has_priority_over_older_normal_event(self):
+        self.store.register_event(Event.from_dict(event_payload(event_id="evt-normal")))
+        self.store.register_event(
+            Event.from_dict(
+                event_payload(
+                    event_id="health-central-priority",
+                    event_type="central_gpt_health_probe",
+                )
+            )
+        )
+        claimed = self.store.claim_events(
+            owner="health-worker", statuses=("RECEIVED",), limit=1, lease_seconds=60
+        )
+        self.assertEqual([row["event_id"] for row in claimed], ["health-central-priority"])
+
+    def test_newest_health_probe_claims_before_stale_health_backlog(self):
+        for event_id in ("health-central-old", "health-central-new"):
+            self.store.register_event(
+                Event.from_dict(
+                    event_payload(event_id=event_id, event_type="central_gpt_health_probe")
+                )
+            )
+        self.store._conn().execute(
+            "UPDATE events SET created_at=1 WHERE event_id='health-central-old'"
+        )
+        self.store._conn().execute(
+            "UPDATE events SET created_at=2 WHERE event_id='health-central-new'"
+        )
+        claimed = self.store.claim_events(
+            owner="health-worker", statuses=("RECEIVED",), limit=1, lease_seconds=60
+        )
+        self.assertEqual([row["event_id"] for row in claimed], ["health-central-new"])
+
+    def test_event_status_distinguishes_dead_letter_from_retryable_failure(self):
+        self.store.register_event(Event.from_dict(event_payload()))
+        self.store.update_event_trigger(
+            "evt-001", status="TRIGGER_FAILED", channel=None, remote_id=None
+        )
+        self.assertEqual(self.store.event_status("evt-001")["status"], "TRIGGER_FAILED")
+        self.store.update_event_trigger(
+            "evt-001", status="DEAD_LETTER", channel=None, remote_id=None
+        )
+        self.assertEqual(self.store.event_status("evt-001")["status"], "DEAD_LETTER")
+
    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.store.update_event_trigger(
+            "evt-001", status="AWAITING_DECISION", channel="workspace_agent",
+            remote_id=None, callback_state="AWAITING_CALLBACK", callback_deadline_after=0,
+        )
+        first = self.store.claim_workspace_timeouts(owner="one", older_than=10**10, limit=5, lease_seconds=60)
+        second = self.store.claim_workspace_timeouts(owner="two", older_than=10**10, limit=5, lease_seconds=60)
        self.assertEqual(len(first), 1)
        self.assertEqual(second, [])

@@ -374,6 +559,44 @@ class StoreTests(unittest.TestCase):


 class DispatchTests(unittest.TestCase):
+    def test_existing_shared_directory_is_verified_without_mkdir(self):
+        with tempfile.TemporaryDirectory() as temp:
+            queue = Path(temp) / "queue"
+            queue.mkdir(mode=0o2750)
+            queue.chmod(0o2750)
+            with mock.patch.object(Path, "mkdir", side_effect=PermissionError("sandbox")):
+                ensure_shared_directory(queue)
+
+    def test_codex_dispatch_uses_watcher_command_schema_at_active_root(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)
+            work_unit = decision_payload()["next_work_units"][0]
+            work_unit.update(
+                primary_agent="codex",
+                fallback_agent="openclaw",
+                required_checks=["acceptance_probe"],
+            )
+            _, queue_path = sink.dispatch(
+                {
+                    "work_unit_id": work_unit["id"],
+                    "event_id": "evt-001",
+                    "assigned_agent": "codex",
+                    "payload_json": json.dumps(work_unit),
+                    "project_key": "v2r1-personal-weixin",
+                    "state_version": 1,
+                }
+            )
+            self.assertEqual(queue_path.parent, queue)
+            command = json.loads(queue_path.read_text())
+            self.assertEqual(command["execution"]["kind"], "acceptance_probe")
+            self.assertEqual(command["state_version"], 1)
+            self.assertEqual(command["contract_commit"], CONTRACT)
+
    def test_dispatch_is_idempotent(self):
        with tempfile.TemporaryDirectory() as temp:
            outbox = Path(temp) / "outbox"
@@ -640,18 +863,39 @@ class RuntimePollingTests(unittest.TestCase):
        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:
+        with mock.patch.object(
+            self.runtime.workspace,
+            "trigger",
+            return_value=WorkspaceTriggerResult(202),
+        ) 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_accepted_event_waits_for_callback(self):
+        self.runtime.submit_event(event_payload())
+        self.runtime.workspace = mock.Mock()
+        self.runtime.workspace.trigger.return_value = WorkspaceTriggerResult(202)
+        fallback = mock.Mock()
+        self.runtime.responses = fallback
+        self.runtime.process_new_events()
+        status = self.runtime.store.event_status("evt-001")
+        self.assertEqual(status["status"], "AWAITING_DECISION")
+        self.assertEqual(status["trigger_http_status"], 202)
+        self.assertEqual(status["trigger_channel"], "workspace_agent")
+        self.assertIsNone(status["remote_id"])
+        self.assertFalse(status["decision_applied"])
+        fallback.start.assert_not_called()
+        self.runtime.process_new_events()
+        self.runtime.workspace.trigger.assert_called_once()
+
    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
+            "evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None,
+            callback_state="AWAITING_CALLBACK", callback_deadline_after=0,
        )
-        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
@@ -662,18 +906,51 @@ class RuntimePollingTests(unittest.TestCase):
        self.assertEqual(tuple(row), ("POLLING_RESPONSE", "responses_api", "resp_fallback"))
        fallback.start.assert_called_once()

+    def test_workspace_start_failure_does_not_use_responses_before_timeout(self):
+        self.runtime.submit_event(event_payload())
+        self.runtime.workspace = mock.Mock()
+        self.runtime.workspace.trigger.side_effect = TriggerError(
+            "workspace_agent_http_error", http_status=403, category="FORBIDDEN"
+        )
+        fallback = mock.Mock()
+        fallback.start.return_value = "resp_start_fallback"
+        self.runtime.responses = fallback
+        self.runtime.process_new_events()
+        row = self.runtime.store._conn().execute(
+            """SELECT status, channel, remote_id, attempts,
+                      trigger_http_status, trigger_error_category
+               FROM events WHERE event_id='evt-001'"""
+        ).fetchone()
+        self.assertEqual(
+            tuple(row),
+            ("TRIGGER_FAILED", "workspace_agent", None, 1, 403, "FORBIDDEN"),
+        )
+        fallback.start.assert_not_called()
+
+    def test_workspace_start_failure_without_responses_fails_closed(self):
+        self.runtime.submit_event(event_payload())
+        self.runtime.workspace = mock.Mock()
+        self.runtime.workspace.trigger.side_effect = TriggerError("workspace_unavailable")
+        self.assertIsNone(self.runtime.responses)
+        self.runtime.process_new_events()
+        row = self.runtime.store._conn().execute(
+            "SELECT status, channel, remote_id, attempts FROM events WHERE event_id='evt-001'"
+        ).fetchone()
+        self.assertEqual(tuple(row), ("TRIGGER_FAILED", "workspace_agent", None, 1))
+
    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
+            "evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None,
+            callback_state="AWAITING_CALLBACK", callback_deadline_after=0,
        )
        self.runtime.store._conn().execute(
-            "UPDATE events SET attempts=2, updated_at=0 WHERE event_id='evt-001'"
+            "UPDATE events SET callback_attempts=2 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'"
+            "SELECT status, callback_attempts FROM events WHERE event_id='evt-001'"
        ).fetchone()
        self.assertEqual(tuple(row), ("DEAD_LETTER", 3))

@@ -0,0 +1,445 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from unittest import mock
+from pathlib import Path
+
+
+MODULE_PATH = Path(__file__).resolve().parents[1] / "automation" / "executor_quorum_controller.py"
+SPEC = importlib.util.spec_from_file_location("executor_quorum_controller", MODULE_PATH)
+assert SPEC and SPEC.loader
+gateway = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = gateway
+SPEC.loader.exec_module(gateway)
+
+
+class QuorumDecisionTests(unittest.TestCase):
+    def test_gpt_and_openclaw_two_unattended_cycles(self):
+        for _ in range(2):
+            decision = gateway.evaluate_quorum(
+                gateway.Health.HEALTHY,
+                gateway.Health.HEALTHY,
+                gateway.Health.UNHEALTHY,
+                "large_repetitive_infrastructure_monitoring",
+            )
+            self.assertTrue(decision.allowed)
+            self.assertEqual("openclaw", decision.active_executor)
+            self.assertFalse(decision.repair_required)
+
+    def test_gpt_and_codex_two_unattended_cycles(self):
+        for _ in range(2):
+            decision = gateway.evaluate_quorum(
+                gateway.Health.HEALTHY,
+                gateway.Health.UNHEALTHY,
+                gateway.Health.HEALTHY,
+                "complex_code_security_architecture",
+            )
+            self.assertTrue(decision.allowed)
+            self.assertEqual("codex", decision.active_executor)
+
+    def test_both_healthy_route_by_workload(self):
+        infrastructure = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.HEALTHY,
+            "large_repetitive_infrastructure_monitoring",
+        )
+        code = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.HEALTHY,
+            "complex_code_security_architecture",
+        )
+        self.assertEqual("openclaw", infrastructure.active_executor)
+        self.assertEqual("codex", code.active_executor)
+
+    def test_standby_failure_does_not_block_or_request_repair(self):
+        decision = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.UNHEALTHY, "architecture"
+        )
+        self.assertTrue(decision.allowed)
+        self.assertEqual("P3", decision.priority)
+        self.assertFalse(decision.repair_required)
+
+    def test_gpt_unhealthy_blocks_even_with_two_executors(self):
+        decision = gateway.evaluate_quorum(
+            gateway.Health.UNHEALTHY, gateway.Health.HEALTHY, gateway.Health.HEALTHY, "architecture"
+        )
+        self.assertFalse(decision.allowed)
+        self.assertEqual("P0", decision.priority)
+
+    def test_gpt_unknown_blocks(self):
+        decision = gateway.evaluate_quorum(
+            gateway.Health.UNKNOWN, gateway.Health.HEALTHY, gateway.Health.HEALTHY, "architecture"
+        )
+        self.assertFalse(decision.allowed)
+        self.assertIn("UNKNOWN", decision.status)
+
+    def test_both_executors_unhealthy_block(self):
+        decision = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.UNHEALTHY, gateway.Health.UNHEALTHY, "architecture"
+        )
+        self.assertFalse(decision.allowed)
+        self.assertEqual("BLOCKED_NO_HEALTHY_EXECUTOR", decision.status)
+
+    def test_unknown_executor_does_not_count_as_healthy(self):
+        blocked = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.UNKNOWN, gateway.Health.UNHEALTHY, "architecture"
+        )
+        allowed = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.UNKNOWN, gateway.Health.HEALTHY, "architecture"
+        )
+        self.assertFalse(blocked.allowed)
+        self.assertTrue(allowed.allowed)
+        self.assertEqual("codex", allowed.active_executor)
+
+    def test_health_receipt_requires_live_decision_channel(self):
+        base = {
+            "subject": "central_gpt",
+            "health": "HEALTHY",
+            "live_probe_succeeded": True,
+            "observed_at_epoch": 1,
+        }
+        self.assertEqual(gateway.Health.UNKNOWN, gateway.validate_health_receipt(base, "central_gpt", now=1))
+        base["workspace_agent"] = {"validated_live": True}
+        self.assertEqual(gateway.Health.HEALTHY, gateway.validate_health_receipt(base, "central_gpt", now=1))
+
+    def test_stale_health_receipt_is_unknown(self):
+        receipt = {
+            "subject": "codex", "health": "HEALTHY", "live_probe_succeeded": True,
+            "observed_at_epoch": 1,
+        }
+        self.assertEqual(
+            gateway.Health.UNKNOWN,
+            gateway.validate_health_receipt(receipt, "codex", now=100, max_age_seconds=60),
+        )
+
+    def test_workspace_fallback_requires_validated_responses(self):
+        receipt = {
+            "subject": "central_gpt", "health": "HEALTHY", "live_probe_succeeded": True,
+            "observed_at_epoch": 10, "responses_api": {"validated_live": True},
+        }
+        unvalidated = dict(receipt, responses_api={"configured": True})
+        self.assertFalse(gateway.responses_fallback_allowed(False, unvalidated, now=10))
+        self.assertTrue(gateway.responses_fallback_allowed(False, receipt, now=10))
+        self.assertFalse(gateway.responses_fallback_allowed(True, receipt, now=10))
+
+
+class StoreTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        os.chmod(self.root, 0o700)
+        self.database = self.root / "controller.sqlite3"
+        self.store = gateway.QuorumStore(self.database)
+        self.decision_openclaw = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.UNHEALTHY,
+            "large_repetitive_infrastructure_monitoring",
+        )
+        self.decision_codex = gateway.evaluate_quorum(
+            gateway.Health.HEALTHY, gateway.Health.UNHEALTHY, gateway.Health.HEALTHY,
+            "complex_code_security_architecture",
+        )
+
+    def tearDown(self):
+        self.store.close()
+        self.temporary.cleanup()
+
+    def create_task(self):
+        self.store.ingest_event("event-1", {"work_unit": "WU-1"}, "http")
+        return self.store.upsert_task(
+            work_unit="WU-1",
+            contract_commit="c" * 40,
+            event_id="event-1",
+            state_version=1,
+            workload="large_repetitive_infrastructure_monitoring",
+            decision=self.decision_openclaw,
+            checkpoint="checkpoint-7",
+        )
+
+    def test_same_event_id_same_payload_is_idempotent(self):
+        self.assertTrue(self.store.ingest_event("event", {"value": 1}, "http"))
+        self.assertFalse(self.store.ingest_event("event", {"value": 1}, "file-inbox"))
+        count = self.store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"]
+        self.assertEqual(1, count)
+
+    def test_same_dispatch_is_not_emitted_twice(self):
+        self.store.ingest_event("event-3", {"work_unit": "WU-3"}, "http")
+        kwargs = dict(
+            work_unit="WU-3", contract_commit="e" * 40, event_id="event-3", state_version=1,
+            workload="large_repetitive_infrastructure_monitoring", decision=self.decision_openclaw,
+        )
+        self.store.dispatch_task(**kwargs)
+        self.store.dispatch_task(**kwargs)
+        lifecycle = self.store.db.execute(
+            "SELECT COUNT(*) AS n FROM lifecycle WHERE work_unit='WU-3' AND event_type='TASK_DISPATCHED'"
+        ).fetchone()["n"]
+        outbox = self.store.db.execute(
+            "SELECT COUNT(*) AS n FROM outbox WHERE outbox_id LIKE 'dispatch:WU-3:%'"
+        ).fetchone()["n"]
+        self.assertEqual(1, lifecycle)
+        self.assertEqual(1, outbox)
+
+    def test_same_event_id_different_payload_dead_letters(self):
+        self.store.ingest_event("event", {"value": 1}, "http")
+        with self.assertRaises(gateway.EventConflict):
+            self.store.ingest_event("event", {"value": 2}, "file-inbox")
+        status = self.store.db.execute("SELECT status FROM events WHERE event_id='event'").fetchone()["status"]
+        self.assertEqual("DEAD_LETTER_CONFLICT", status)
+
+    def test_stale_state_version_rejected(self):
+        self.create_task()
+        with self.assertRaises(gateway.StaleStateVersion):
+            self.store.upsert_task(
+                work_unit="WU-1", contract_commit="c" * 40, event_id="event-1", state_version=0,
+                workload="architecture", decision=self.decision_codex,
+            )
+
+    def test_two_workers_compete_for_lease(self):
+        self.create_task()
+        self.assertTrue(self.store.claim_lease("WU-1", "openclaw", 30, now=100))
+        self.assertFalse(self.store.claim_lease("WU-1", "codex", 30, now=101))
+
+    def test_stale_lease_is_recovered(self):
+        self.create_task()
+        self.assertTrue(self.store.claim_lease("WU-1", "openclaw", 5, now=100))
+        self.store.upsert_task(
+            work_unit="WU-1", contract_commit="c" * 40, event_id="event-1", state_version=1,
+            workload="complex_code_security_architecture", decision=self.decision_codex,
+        )
+        self.assertTrue(self.store.claim_lease("WU-1", "codex", 5, now=106))
+
+    def test_active_failover_preserves_identity_checkpoint_and_effect(self):
+        self.create_task()
+        self.assertTrue(self.store.record_effect("effect-1", "WU-1", {"kind": "write_ref", "oid": "a" * 40}))
+        status = self.store.upsert_task(
+            work_unit="WU-1", contract_commit="c" * 40, event_id="event-1", state_version=1,
+            workload="complex_code_security_architecture", decision=self.decision_codex,
+        )
+        task = self.store.db.execute("SELECT * FROM tasks WHERE work_unit='WU-1'").fetchone()
+        self.assertEqual("INTERRUPTED_AGENT_FAILOVER", status)
+        self.assertEqual("openclaw", task["interrupted_executor"])
+        self.assertEqual("codex", task["active_executor"])
+        self.assertEqual("P2", task["recovery_priority"])
+        self.assertEqual("event-1", task["event_id"])
+        self.assertEqual("c" * 40, task["contract_commit"])
+        self.assertEqual("checkpoint-7", task["checkpoint"])
+        self.assertFalse(self.store.record_effect("effect-1", "WU-1", {"kind": "write_ref", "oid": "a" * 40}))
+
+    def test_lifecycle_and_summary_persist_without_openclaw(self):
+        self.create_task()
+        for event_type in (
+            "TASK_DISPATCHED", "TASK_STARTED", "TASK_ENDED", "DECISION_APPLIED",
+            "PROJECT_SUMMARY", "PROJECT_COMPLETED",
+        ):
+            self.store.append_lifecycle("WU-1", event_type, "codex", 1, {"result": "PASS"})
+        self.store.persist_summary("WU-1", {"executor": "codex", "result": "PASS"})
+        events = self.store.db.execute("SELECT COUNT(*) AS n FROM lifecycle").fetchone()["n"]
+        summaries = self.store.db.execute("SELECT COUNT(*) AS n FROM summaries").fetchone()["n"]
+        self.assertEqual(6, events)
+        self.assertEqual(1, summaries)
+
+    def test_lifecycle_and_summary_persist_without_codex(self):
+        self.create_task()
+        self.store.append_lifecycle("WU-1", "TASK_STARTED", "openclaw", 1, {"result": "RUNNING"})
+        self.store.append_lifecycle("WU-1", "TASK_ENDED", "openclaw", 1, {"result": "PASS"})
+        self.store.persist_summary("WU-1", {"executor": "openclaw", "result": "PASS"})
+        self.assertEqual(
+            "openclaw",
+            json.loads(self.store.db.execute("SELECT summary_json FROM summaries").fetchone()["summary_json"])["executor"],
+        )
+
+    def test_restart_recovers_event_lease_and_outbox(self):
+        self.create_task()
+        self.store.claim_lease("WU-1", "openclaw", 5, now=100)
+        self.store.enqueue("outbox-1", "gateway", {"event_type": "TASK_STARTED"})
+        self.store.close()
+        self.store = gateway.QuorumStore(self.database)
+        self.assertEqual(1, self.store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"])
+        self.assertEqual(1, self.store.db.execute("SELECT COUNT(*) AS n FROM outbox").fetchone()["n"])
+        self.assertTrue(self.store.claim_lease("WU-1", "openclaw", 5, now=106))
+
+    def test_lifecycle_helpers_enforce_active_executor_and_complete(self):
+        self.store.ingest_event("event-2", {"work_unit": "WU-2"}, "http")
+        self.store.dispatch_task(
+            work_unit="WU-2", contract_commit="d" * 40, event_id="event-2", state_version=3,
+            workload="complex_code_security_architecture", decision=self.decision_codex,
+        )
+        self.assertFalse(self.store.start_task("WU-2", "openclaw", 3, 30))
+        self.assertTrue(self.store.start_task("WU-2", "codex", 3, 30))
+        self.store.end_task(
+            "WU-2", "codex", 3, "PASS", {"executor": "codex", "result": "PASS"}, project_completed=True,
+        )
+        task = self.store.db.execute("SELECT status,lease_owner FROM tasks WHERE work_unit='WU-2'").fetchone()
+        self.assertEqual("COMPLETED", task["status"])
+        self.assertIsNone(task["lease_owner"])
+        event_types = [row["event_type"] for row in self.store.db.execute("SELECT event_type FROM lifecycle ORDER BY sequence")]
+        self.assertEqual(
+            ["TASK_DISPATCHED", "TASK_STARTED", "TASK_ENDED", "DECISION_APPLIED", "PROJECT_SUMMARY", "PROJECT_COMPLETED"],
+            event_types,
+        )
+        self.assertEqual(6, self.store.db.execute("SELECT COUNT(*) AS n FROM outbox").fetchone()["n"])
+
+    def test_blocker_threshold_enters_dead_letter(self):
+        self.assertEqual("RETRY", self.store.record_blocker("same", 2))
+        self.assertEqual("RETRY", self.store.record_blocker("same", 2))
+        self.assertEqual("DEAD_LETTER", self.store.record_blocker("same", 2))
+
+    def test_outbox_retries_then_dead_letters_and_success_is_persisted(self):
+        self.store.enqueue("failure", "gateway", {"event_type": "TASK_STARTED"})
+        self.store.enqueue("success", "gateway", {"event_type": "TASK_ENDED"})
+
+        def sender(payload):
+            if payload["event_type"] == "TASK_STARTED":
+                raise OSError("gateway unavailable")
+
+        now = time.time() + 1
+        self.assertEqual((1, 1), self.store.deliver_outbox(sender, max_attempts=2, now=now))
+        self.assertEqual((0, 1), self.store.deliver_outbox(sender, max_attempts=2, now=now + 3))
+        statuses = {
+            row["outbox_id"]: row["status"] for row in self.store.db.execute("SELECT outbox_id,status FROM outbox")
+        }
+        self.assertEqual("DEAD_LETTER", statuses["failure"])
+        self.assertEqual("SENT", statuses["success"])
+
+    def test_git_ref_polling_detects_missed_webhook(self):
+        self.assertFalse(self.store.observe_ref("refs/heads/main", "a" * 40))
+        self.assertTrue(self.store.observe_ref("refs/heads/main", "b" * 40))
+        self.assertFalse(self.store.observe_ref("refs/heads/main", "b" * 40))
+
+    def test_git_ref_poller_creates_one_event_per_change(self):
+        values = iter(("a" * 40, "b" * 40, "b" * 40))
+
+        def resolver(_repo, _ref):
+            return next(values)
+
+        self.assertEqual(0, gateway.poll_git_refs(self.store, Path("/repo"), ["refs/heads/main"], resolver))
+        self.assertEqual(1, gateway.poll_git_refs(self.store, Path("/repo"), ["refs/heads/main"], resolver))
+        self.assertEqual(0, gateway.poll_git_refs(self.store, Path("/repo"), ["refs/heads/main"], resolver))
+        self.assertEqual(1, self.store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"])
+
+
+class RecoveryAndSafetyTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        os.chmod(self.root, 0o700)
+
+    def tearDown(self):
+        self.temporary.cleanup()
+
+    def test_controller_git_authority_is_read_only_in_systemd_sandbox(self):
+        unit = (
+            Path(__file__).resolve().parents[1]
+            / "systemd"
+            / "aios-automation-quorum-controller.service"
+        ).read_text(encoding="utf-8")
+        self.assertIn("SupplementaryGroups=aios-orchestrator gitops\n", unit)
+        self.assertIn("SuccessExitStatus=2\n", unit)
+        self.assertIn(
+            "ReadOnlyPaths=/etc/aios-automation-quorum /run/aios-health "
+            "/data/aios-gitops/repo.git\n",
+            unit,
+        )
+        self.assertNotIn(
+            "ReadWritePaths=/var/lib/aios-automation-quorum /data/aios-gitops/repo.git",
+            unit,
+        )
+
+    def test_http_failure_spools_then_file_poll_recovers(self):
+        inbox = self.root / "inbox"
+        inbox.mkdir(mode=0o700)
+        store = gateway.QuorumStore(self.root / "store.sqlite3")
+        try:
+            def fail(_payload):
+                raise OSError("network unavailable")
+
+            self.assertFalse(gateway.spool_after_http_failure(fail, {"kind": "dispatch"}, inbox, "event-1"))
+            self.assertEqual(1, gateway.poll_file_inbox(store, inbox))
+            self.assertEqual(1, store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"])
+        finally:
+            store.close()
+
+    def test_git_ref_resolution_scopes_safe_directory_to_validated_repo(self):
+        repo = self.root / "repo.git"
+        repo.mkdir(mode=0o700)
+        completed = subprocess.CompletedProcess([], 0, stdout="a" * 40 + "\n")
+        with mock.patch.object(gateway.subprocess, "run", return_value=completed) as run:
+            self.assertEqual("a" * 40, gateway.resolve_git_ref(repo, "refs/heads/main"))
+        self.assertEqual(
+            [
+                "git",
+                "-c",
+                f"safe.directory={repo}",
+                "-C",
+                str(repo),
+                "rev-parse",
+                "--verify",
+                "refs/heads/main",
+            ],
+            run.call_args.args[0],
+        )
+
+    def test_git_ref_resolution_rejects_symlink_and_world_writable_repo(self):
+        target = self.root / "target.git"
+        target.mkdir(mode=0o700)
+        link = self.root / "link.git"
+        try:
+            link.symlink_to(target, target_is_directory=True)
+        except OSError:
+            link = None
+        if link is not None:
+            with self.assertRaises(gateway.ControllerError):
+                gateway.resolve_git_ref(link, "refs/heads/main")
+        if os.name == "posix":
+            target.chmod(0o707)
+            with self.assertRaises(gateway.ControllerError):
+                gateway.resolve_git_ref(target, "refs/heads/main")
+
+    @unittest.skipIf(os.name == "nt", "POSIX mode semantics required")
+    def test_symlink_and_writable_parent_are_fail_closed(self):
+        safe = self.root / "safe"
+        safe.mkdir(mode=0o700)
+        receipt = safe / "receipt.json"
+        receipt.write_text('{"subject":"codex"}', encoding="utf-8")
+        receipt.chmod(0o600)
+        link = safe / "link.json"
+        link.symlink_to(receipt)
+        with self.assertRaises(gateway.ControllerError):
+            gateway.read_secure_json(link)
+        safe.chmod(0o777)
+        with self.assertRaises(gateway.ControllerError):
+            gateway.read_secure_json(receipt)
+
+    @unittest.skipIf(os.name == "nt", "POSIX ownership and symlink semantics required")
+    def test_database_symlink_is_fail_closed(self):
+        safe = self.root / "safe-db"
+        safe.mkdir(mode=0o700)
+        target = safe / "target.sqlite3"
+        target.write_bytes(b"")
+        target.chmod(0o600)
+        link = safe / "controller.sqlite3"
+        link.symlink_to(target)
+        with self.assertRaises(gateway.ControllerError):
+            gateway.QuorumStore(link)
+
+    def test_secret_and_process_output_fields_rejected(self):
+        for payload in ({"token": "x"}, {"nested": {"stderr": "x"}}, {"env": {"X": "x"}}):
+            with self.assertRaises(gateway.ControllerError):
+                gateway.validate_metadata_only(payload)
+
+    def test_invalid_lifecycle_event_rejected(self):
+        store = gateway.QuorumStore(self.root / "store.sqlite3")
+        try:
+            with self.assertRaises(gateway.ControllerError):
+                store.append_lifecycle("WU", "UNAPPROVED", None, 1, {})
+        finally:
+            store.close()
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
@@ -2,13 +2,18 @@ openapi: 3.1.0
 info:
  title: AIOS Orchestrator Decision Callback
  version: 1.0.0
-servers:
-  - url: https://REPLACE_WITH_GATEWAY_HOST
+servers: []
+x-aios-install-generated-server-required: true
 paths:
  /v1/orchestrator/decisions:
    post:
      operationId: submit_orchestration_decision
      summary: Submit one strict orchestration decision for a triggered event
+      description: >-
+        The 18890 Actions API validates the decision strictly and forwards it
+        to the loopback Gateway with a callback credential resolved only from
+        an approved SecretRef. This operation cannot perform arbitrary GitOps
+        mutations.
      security:
        - bearerAuth: []
      requestBody:
@@ -21,9 +26,17 @@ paths:
        '202':
          description: Accepted
        '409':
-          description: Contract, idempotency, or state version conflict
+          description: Contract mismatch or stale state version
        '422':
          description: Schema validation failed
+        '401':
+          description: Missing or invalid Actions API bearer credential
+        '403':
+          description: Credential lacks orchestrator:decision scope
+        '502':
+          description: Metadata-only Gateway forwarding failure
+        '503':
+          description: SecretRef provider or callback configuration unavailable
 components:
  securitySchemes:
    bearerAuth:
Author	SHA1	Message	Date
Codex	59edc85930	feat(codex): harden callback transport and pending recovery	2026-07-18 15:50:42 +08:00
Codex	815e39f36c	fix(codex): harden workspace trigger protocol	2026-07-18 14:51:57 +08:00
Codex	c679c8394f	fix(codex): make central GPT health probing single-flight	2026-07-18 13:44:06 +08:00
Codex	9ec60c162a	fix(quorum): treat persisted blocked state as healthy cycle	2026-07-18 11:59:27 +08:00
Codex	415d736f88	fix(quorum): read authoritative bare refs in sandbox	2026-07-18 11:57:52 +08:00
Codex	20bfb3e773	fix(quorum): restore controller executable mode	2026-07-18 11:56:31 +08:00
Codex	cf211ac4ac	fix(quorum): validate and trust exact Git ref source	2026-07-18 11:56:18 +08:00
Codex	a76153c637	fix(quorum): make health receipts group-readable	2026-07-18 11:53:50 +08:00
Codex	3ddab59f65	fix(codex): allow read-only probe WAL access	2026-07-18 11:47:02 +08:00
Codex	4b88d90531	fix(codex): preserve watcher access to probe commands	2026-07-18 11:42:37 +08:00
Codex	2a80c8ad7a	fix(codex): bound gateway polling work units	2026-07-18 10:42:51 +08:00
Codex	1e00541e50	fix(codex): bound central health event scheduling	2026-07-18 10:36:22 +08:00
Codex	cb4f13601c	fix(codex): fail over model trigger channel safely	2026-07-18 10:32:12 +08:00
Codex	6170627542	fix(codex): verify existing shared queue without mutation	2026-07-18 10:26:16 +08:00
Codex	56827247f0	fix(codex): align gateway sandbox with durable queue	2026-07-18 10:24:10 +08:00
Codex	84714c712a	fix(codex): recover failed gateway activation safely	2026-07-18 10:20:09 +08:00
Codex	87fe8aea91	fix(codex): make gateway activation transactional	2026-07-18 10:17:41 +08:00
Codex	e850db7437	feat(codex): close GPT plus executor unattended automation loop	2026-07-18 10:15:05 +08:00
Codex	e116146341	merge: integrate GPT plus one-executor governance and quorum	2026-07-18 09:43:06 +08:00
Codex	df78b26437	feat(orchestration): implement GPT plus one-executor quorum controller	2026-07-18 05:40:50 +08:00
Codex	87a5fac40d	chore(governance): adopt GPT plus one-executor automation quorum	2026-07-18 05:01:34 +08:00
Codex	08fe81efee	chore(orchestration): dispatch OpenClaw three-party self-healing and release supervision	2026-07-18 04:54:21 +08:00
				`@@ -0,0 +1 @@`
				`u aios-quorum - "AIOS automation quorum controller" /var/lib/aios-automation-quorum /usr/sbin/nologin`