Compare commits

..
27 changed files with 2222 additions and 526 deletions
@@ -0,0 +1,94 @@
{
"contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
"created_at": "2026-07-18T09:37:00+00:00",
"schema_version": 3,
"state_writer": "gateway",
"work_units": [
{
"id": "R0-DISCOVER-SECRET-PROVIDER",
"track": "R_RUNNER",
"gate": "R0_RUNNER_LIVE_RECOVERY",
"primary": "openclaw",
"fallback": "codex",
"status": "READY",
"quorum": "requires_gpt_plus_one_executor"
},
{
"id": "P2-INVENTORY-REAL-EVIDENCE",
"track": "P_PERSONAL_WEIXIN",
"gate": "P2_REAL_B3_EVIDENCE",
"primary": "openclaw",
"fallback": "codex",
"status": "READY",
"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,
"scenarios": [
"GPT+OpenClaw-healthy-Codex-stopped-two-cycles",
"GPT+Codex-healthy-OpenClaw-stopped-two-cycles",
"Both-healthy-route-by-workload",
"Active-stops-standby-auto-failover-no-duplicates",
"Standby-alone-stopped-no-block",
"GPT-stopped-block-even-with-both-executors",
"GPT-healthy-both-executors-stopped-block",
"UNKNOWN-never-counts-as-healthy",
"HTTP-failure-then-file-polling-recovery",
"Webhook-missed-then-git-ref-polling-recovery",
"Gateway-restart-recovers-event-lease-outbox",
"Same-event-id-no-redispatch",
"Same-blocker-threshold-dead-letter",
"Workspace-Agent-fallback-to-Responses-with-valid-credentials",
"OpenClaw-offline-Codex-lifecycle-persisted",
"Codex-offline-OpenClaw-lifecycle-persisted"
]
}
]
}
@@ -0,0 +1,269 @@
{
"adaptive_dispatch": {
"codex_first": [
"complex_code",
"security_boundary",
"architecture",
"high_risk_review",
"strong_model_materially_better"
],
"default_orchestrator": "openclaw",
"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",
"preserve_logs",
"switch_agent"
],
"openclaw_first": [
"quality_close",
"large_or_repetitive",
"long_running",
"infra_or_git_inventory",
"cost_sensitive"
],
"single_state_writer_lease_required": true,
"strict_role_partition": false,
"state_writer": "gateway",
"timeout_can_pass": false
},
"authoritative_state": {
"b3": "BLOCKED_EVIDENCE_REQUIRED",
"b4": "NOT_STARTED",
"baseline": "V1R4C6",
"production_release": "NOT_STARTED",
"release": "paused",
"repair3": {
"base": "8d9c6da90fc8caa69b691d243c6fefb40b6b0575",
"candidate": "PASS_CANDIDATE_PREPARED",
"commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4",
"exact4": "PASS",
"result": "49a9c2e0cc898ecf2bbd45f454e007644d7f3045"
},
"runner": {
"artifact_sha256": "85c472c05a1e55a5899b215d3cd41df53912787de529ac9cb3d5308da4158130",
"commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
"live": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
"offline": "PASS",
"result": "e789632e359e9232d8ae081e7c4f23ed7b615a9d"
}
},
"central_gpt": {
"required": "HEALTHY",
"health_definition": {
"live_probe_required": true,
"workspace_agent_validated_live_or_responses_validated_live": true,
"required_real_channels": true,
"rejects": [
"dummy_acceptance_key",
"config_presence_only",
"credential_presence_only"
],
"configuration_or_credential_presence_is_insufficient": true,
"unknown_is_not_healthy": true,
"no_health_without_channel": 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,
"both_executors_healthy_extra_reviewer_available": true,
"extra_reviewer_not_mandatory_gate": 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"
]
},
"gateway_lifecycle_events": [
"TASK_DISPATCHED",
"TASK_STARTED",
"TASK_ENDED",
"DECISION_APPLIED",
"PROJECT_SUMMARY",
"PROJECT_COMPLETED"
],
"facts_policy": {
"accept": [
"committed_git_tree",
"full_oid",
"worktree_bare_gitea_oid_match",
"test_log_with_rc",
"immutable_digest",
"result_only_commit",
"live_evidence"
],
"reject": [
"verbal_claim",
"readme_only",
"local_only_commit",
"environment_variable_as_secret_ref",
"compile_only",
"timeout_as_pass",
"unrelated_repo",
"fabricated_evidence"
]
},
"forbidden_git": [
"reset",
"amend",
"rebase",
"force_push",
"direct_main_push",
"direct_bare_ref_mutation",
"unapproved_gitea_db_mutation",
"reuse_accident_worktree",
"delete_live_worktree_to_clear_collision"
],
"generated_at": "2026-07-18T09:37:00+00:00",
"handoff": {
"ack": "CONTRACT_ACK=<full_contract_commit_oid>",
"advance": "independent_verification_required",
"all_agents_match": true
},
"join_gate": {
"requires": [
"AUTOMATION_CLOSED",
"RUNNER_RECOVERED",
"B3_PASS",
"B4_PASS",
"INTEGRATION_PASS"
]
},
"model_policy": {
"codex": "strongest_model_actually_available_in_runtime",
"gpt_5_6": "use_only_if_exposed",
"no_unverified_model_claims": true,
"openclaw_default": "DeepSeek V4 Flash"
},
"owner_authorization": {
"adaptive_agent_assignment_authorized": true,
"agent_substitution_authorized": true,
"autonomous_execution": true,
"may_bypass_release_controls": false,
"may_bypass_secret_controls": false,
"may_fabricate_evidence": false,
"user_step_by_step_confirmation_required": false
},
"release_gates": [
"AUTOMATION_CLOSED",
"MAIN_MERGE",
"STAGING_AND_CANARY",
"PRODUCTION"
],
"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",
"gateway_persists_authoritative_state",
"lifecycle_events_persisted_regardless_of_openclaw_online"
],
"status": "ACTIVE",
"tracks": {
"P_PERSONAL_WEIXIN": {
"b3": "BLOCKED_EVIDENCE_REQUIRED",
"b4": "NOT_STARTED",
"parallel": true,
"repair3": "PASS_CANDIDATE_PREPARED",
"repair3_commit": "e110cda9968cadc5c7b7179ec5003b05aad11ce4"
},
"R_RUNNER": {
"commit": "bbd27cf06acf1f589eef291eab54d4b127aa3aea",
"current": "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE",
"fallback": "codex",
"parallel": true,
"primary": "openclaw",
"required": [
"artifact_verified",
"approved_secret_provider_contract",
"owner_probe_verified",
"service_healthy",
"heartbeat_healthy",
"gateway_auth",
"durable_task_create_read_cancel",
"execution_channel",
"live_rollback"
]
}
},
"automation_closed": {
"success_criteria": {
"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"
},
"no_longer_require": [
"OPENCLAW_AUTO_CLAIM=PASS",
"CODEX_AUTO_CLAIM=PASS"
],
"only_current_active_executor_claim_required": true
},
"lifecycle_ownership": {
"authoritative_persister": "gateway",
"openclaw_online_mirror": true,
"openclaw_offline_behavior": "gateway_persists_directly"
}
}
@@ -0,0 +1,47 @@
{
"b3": "BLOCKED_EVIDENCE_REQUIRED",
"b4": "NOT_STARTED",
"baseline": "V1R4C6",
"contract_branch": "task/governance/v2r1-personal-weixin-production-contract-20260717",
"contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
"dispatch_commit": null,
"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": 3,
"single_state_writer": "gateway",
"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-18T09:37:00+00:00",
"gateway_persists_authoritative_state": true
}
@@ -0,0 +1,13 @@
# Codex V2R1 adaptive execution
CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
Execute P1-POST-CONTRACT-REACK now:
- read-only verify Repair3 e110cda9968cadc5c7b7179ec5003b05aad11ce4
- read-only verify RESULT 49a9c2e0cc898ecf2bbd45f454e007644d7f3045
- verify lineage from 8d9c6da90fc8caa69b691d243c6fefb40b6b0575, exact4, tests, Gitleaks and RESULT-only scope
- do not modify Repair3 code in this unit
Act as fallback for OpenClaw only on timeout/unavailability. Use GPT-5.6 only if actually exposed; otherwise strongest visible model. Prefer OpenClaw when quality is comparable or volume is high; prefer Codex for complex/security-critical work.
Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
Report:
CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=codex STATUS=<status> IMPLEMENTATION=<oid-or-NONE> RESULT=<oid-or-NONE> OID_MATCH=<YES-or-NO> BLOCKER=<reason-or-NONE>
Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.
@@ -1,267 +0,0 @@
# 执行 GitOps 独立 Worktree 与统一运行身份迁移
```yaml
task_id: TASK-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8
created_at: 2026-07-14T23:37:42+08:00
timezone: Asia/Shanghai
owner_approved: true
user_action_required: false
risk: high
production_business_change: false
```
## 目标
根治共享 worktree 与多身份写入导致的 ACL、checkout 和未跟踪文件覆盖问题。
实施:
1. 统一 GitOps/OpenClaw 写入身份为 `aiman:gitops`;
2. 统一 `UMask=0002`;
3. 为 GitOps 根目录设置 setgid 与默认 ACL;
4. 建立 bare repo 和每任务独立 worktree;
5. 启用任务启动前自检;
6. 验证 OpenClaw 与 ChatGPT 并发任务互不干扰;
7. 保留原共享 workspace 作为只读回滚源,禁止智能体继续在其中切换分支。
## 强制通知
收到任务后立即微信通知 Owner:
```text
【任务已接收】GitOps独立worktree与统一运行身份迁移开始。将暂停写入服务、建立独立任务工作树并验证并发执行;生产业务版本不变。
```
任务结束后无论 PASS、BLOCKED 或 ROLLED_BACK,立即微信汇报结果并记录 message_id。
## 1. 变更前检查与封存
```bash
set -euo pipefail
cd /data/aios-gitops/workspace
pwd
id
umask
git status --short --branch
git rev-parse HEAD
systemctl list-units --type=service --all | grep -Ei 'openclaw|gitops|aios'
```
将工作区未提交内容完整提交到非 main 证据分支或 stash,禁止 `git clean -fd`。
记录当前运行服务、用户、组、umask、PID、工作区 HEAD 和未跟踪文件清单。
## 2. 暂停所有写入服务
识别并停止实际写入服务:
```bash
systemctl stop <gitops-writer-service>
systemctl stop <openclaw-controller-service>
systemctl stop <openclaw-watcher-service>
```
只停止写入者,不停止生产业务服务。记录每条命令和退出码。
## 3. 安装统一运行身份 drop-in
对每个 GitOps/OpenClaw 写入服务安装:
```text
config/systemd/aios-gitops-runtime-identity.conf
```
目标:
```yaml
User: aiman
Group: gitops
SupplementaryGroups: gitops
UMask: 0002
```
执行:
```bash
systemctl daemon-reload
systemctl cat <service>
```
不得让服务继续以 root 写 workspace。
## 4. 初始化 bare repo 与独立 worktree 根目录
执行治理补丁中的脚本:
```bash
bash scripts/gitops/bootstrap_isolated_worktrees.sh
```
验收:
```yaml
bare_repo: /data/aios-gitops/repo.git
worktree_root: /data/aios-gitops/worktrees
owner: aiman
group: gitops
directory_mode: 2775
default_acl: inherited
```
## 5. 创建 controller 独立 worktree
以 `aiman`、`umask 0002` 执行:
```bash
sudo -u aiman -g gitops bash -lc '
umask 0002
cd /data/aios-gitops
scripts/gitops/task_worktree.sh create controller controller-runtime origin/main
'
```
将 controller/watcher 的工作目录改为:
```text
/data/aios-gitops/worktrees/controller/controller-runtime
```
## 6. 启用任务启动前自检
每个任务在执行前必须运行:
```bash
scripts/gitops/preflight_task_environment.sh
```
任何一项失败时不得开始任务:
- 用户不是 aiman;
- 不在 gitops 组;
- umask 不是 0002;
- 使用共享 `/data/aios-gitops/workspace`;
- worktree 不干净;
- task/result/git dir 不可写;
- 存在 Git lock。
## 7. 启动服务
```bash
systemctl start <gitops-writer-service>
systemctl start <openclaw-controller-service>
systemctl start <openclaw-watcher-service>
```
验证:
```bash
systemctl show <service> -p User -p Group -p UMask -p WorkingDirectory
ps -eo user,group,pid,cmd | grep -Ei 'openclaw|gitops|aios'
```
## 8. 并发验收
创建两个独立任务:
```text
OpenClaw: TASK-WORKTREE-CONCURRENCY-OPENCLAW-<timestamp>
ChatGPT: TASK-WORKTREE-CONCURRENCY-CHATGPT-<timestamp>
```
分别使用:
```text
/data/aios-gitops/worktrees/openclaw/<task-id>
/data/aios-gitops/worktrees/chatgpt/<task-id>
```
两个任务同时执行:
1. 写各自 `.agent/tasks/<agent>/`;
2. 写各自 `.agent/results/<agent>/`;
3. 创建深层子目录,验证默认 ACL;
4. commit;
5. push;
6. 不切换共享 workspace 分支。
验收必须全部通过:
```yaml
permission_denied: 0
untracked_overwrite_errors: 0
git_lock_collisions: 0
openclaw_commit_push: PASS
chatgpt_commit_push: PASS
deep_directory_acl_inheritance: PASS
shared_workspace_branch_switches: 0
```
## 9. 回滚
若迁移失败:
1. 停止新服务;
2. 保留失败 worktree 供审计;
3. 恢复 systemd drop-in 备份;
4. `daemon-reload`;
5. 恢复原服务状态;
6. 不删除 `/data/aios-gitops/workspace`;
7. 生产业务保持不变。
## 10. 结果文件
提交:
```text
.agent/results/openclaw/RESULT-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8.md
```
首段必须包含:
```yaml
status: PASS | BLOCKED | ROLLED_BACK
task_id: TASK-GITOPS-ISOLATED-WORKTREE-MIGRATION-20260714T233742UTC8
created_at: 2026-07-14T23:37:42+08:00
timezone: Asia/Shanghai
user_action_required: false
runtime_identity:
user:
group:
umask:
bare_repo:
worktree_root:
controller_worktree:
services:
default_acl:
preflight:
concurrency_test:
openclaw_branch:
openclaw_commit:
chatgpt_branch:
chatgpt_commit:
permission_denied_count:
untracked_overwrite_count:
git_lock_collision_count:
owner_notifications:
received_message_id:
terminal_message_id:
rollback:
remaining_risks:
next_step:
```
提交并推送到独立执行分支。禁止直接推 main。
## 完成微信模板
PASS:
```text
【任务完成】GitOps独立worktree迁移已通过。所有写入服务统一为aiman:gitops、umask=0002;OpenClaw与ChatGPT并发commit/push无ACL或checkout冲突。生产业务版本未变化。
```
阻塞:
```text
【任务阻塞】GitOps独立worktree迁移未通过。原因=<具体服务/权限/并发验收>;已保留原workspace和回滚能力,生产业务未变化。
```
@@ -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>
```
@@ -0,0 +1,13 @@
# OpenClaw V2R1 adaptive orchestration
CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
You are default orchestrator and sole authoritative state writer.
Execute immediately:
1. R0-DISCOVER-SECRET-PROVIDER: discover approved provider metadata and owner-probe source without reading or outputting secrets. Environment variables alone are not approved secret refs.
2. P2-INVENTORY-REAL-EVIDENCE: inventory approved restricted B3 evidence locations; validate existing facts; never synthesize evidence.
3. Independently verify Codex results before advancing.
Prefer OpenClaw for comparable-quality, large, repetitive, long-running, infrastructure and cost-sensitive work. Prefer Codex strongest actually available model for complex code, security boundaries, architecture and high-risk review. On timeout mark TIMEOUT and switch agent; never infer PASS.
Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
Report:
CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=openclaw STATUS=<status> COMMIT=<oid-or-NONE> BLOCKER=<reason-or-NONE>
Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.
+820
View File
@@ -0,0 +1,820 @@
#!/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
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 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:
completed = subprocess.run(
["git", "-C", str(repo), "rev-parse", "--verify", ref_name],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
timeout=10,
)
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 post_json(url: str, payload: Mapping[str, Any], timeout: float = 5.0) -> None:
validate_metadata_only(payload)
request = urllib.request.Request(
url,
data=canonical_json(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=timeout) as response:
if response.status < 200 or response.status >= 300:
raise urllib.error.HTTPError(url, response.status, "unexpected status", response.headers, None)
def run_once(config_path: Path) -> int:
config = read_secure_json(config_path, expected_uid=0 if os.name == "posix" else None)
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")
)
store.enqueue(
f"quorum-{time.time_ns()}",
"gateway",
{
"event_type": "DECISION_APPLIED",
"allowed": decision.allowed,
"priority": decision.priority,
"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(config["gateway_event_url"], payload),
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,21 @@
{
"$schema": "../schemas/automation-quorum-controller.schema.json",
"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:8765/v1/events"
}
-59
View File
@@ -1,59 +0,0 @@
version: 1
policy_id: gitops-isolated-worktree-governance-20260714T232948UTC8
runtime_identity:
user: aiman
group: gitops
umask: "0002"
prohibit_root_workspace_writes: true
repository:
source_worktree: /data/aios-gitops/workspace
bare_repo: /data/aios-gitops/repo.git
worktree_root: /data/aios-gitops/worktrees
remote: origin
worktrees:
controller: /data/aios-gitops/worktrees/controller
task_pattern: /data/aios-gitops/worktrees/{agent}/{task_id}
allowed_agents:
- openclaw
- chatgpt
- controller
require_clean_start: true
remove_after_successful_push: true
retain_on_failure: true
permissions:
owner: aiman
group: gitops
directory_mode: "2775"
file_mode: "0664"
default_acl:
user:aiman: rwx
group:gitops: rwx
apply_to:
- /data/aios-gitops
- /data/aios-gitops/worktrees
preflight:
required_user: aiman
required_group: gitops
required_umask: "0002"
require_writable_git_dir: true
require_writable_task_dir: true
require_writable_result_dir: true
reject_shared_source_worktree: true
reject_existing_git_lock: true
notifications:
owner_on_received: required
owner_on_terminal: required
preferred_channel: weixin
redact_secrets: true
rollback:
preserve_source_worktree: true
remove_created_worktrees: true
restore_service_dropins: true
restore_service_state: true
@@ -1,14 +0,0 @@
# Install as a systemd drop-in for GitOps/OpenClaw writer services.
# Example: /etc/systemd/system/<service>.service.d/10-runtime-identity.conf
[Service]
User=aiman
Group=gitops
SupplementaryGroups=gitops
UMask=0002
Environment=AIOS_GITOPS_ROOT=/data/aios-gitops
Environment=AIOS_BARE_REPO=/data/aios-gitops/repo.git
Environment=AIOS_WORKTREE_ROOT=/data/aios-gitops/worktrees
Environment=AIOS_RUNTIME_USER=aiman
Environment=AIOS_RUNTIME_GROUP=gitops
Environment=AIOS_RUNTIME_UMASK=0002
+101
View File
@@ -0,0 +1,101 @@
# 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.
## 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
```
Run deterministic verification before enablement:
```bash
scripts/verify_automation_quorum_controller.sh "$PWD"
```
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
```
@@ -1,9 +0,0 @@
{
"id": "patch-1784043007770-e97206c3",
"patch_name": "gitops-isolated-worktree-governance-20260714T232948UTC8",
"branch": "patch/gitops-isolated-worktree-governance-20260714T232948UTC8",
"risk": "high",
"description": "Root-cause remediation for recurring ACL and shared-worktree failures: unified aiman:gitops runtime identity, UMask=0002, inherited default ACLs, per-task isolated Git worktrees, startup preflight, migration/rollback scripts, systemd drop-in templates, audit evidence, and an OpenClaw execution task. No production business deployment in this patch.",
"created_by": "chatgpt-gitops-bot",
"created_at": "2026-07-14T15:30:07Z"
}
@@ -1,9 +0,0 @@
{
"type": "branch",
"branch": "task/codex/gitops-isolated-worktree-migration-20260715T001652UTC8",
"base": "patch/gitops-isolated-worktree-governance-20260714T232948UTC8",
"created_by": "chatgpt-gitops-bot",
"created_at": "2026-07-14T16:17:28Z",
"direct_main_push": false,
"owner_approval_required": true
}
@@ -0,0 +1,55 @@
{
"$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": [
"database_path",
"file_inbox_path",
"health_receipts",
"blocker_retry_threshold",
"max_health_age_seconds",
"lease_ttl_seconds",
"gateway_event_url"
],
"properties": {
"$schema": {"type": "string"},
"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", "pattern": "^https?://"}
}
}
@@ -1,47 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT="${AIOS_GITOPS_ROOT:-/data/aios-gitops}"
SOURCE="${AIOS_SOURCE_WORKTREE:-$ROOT/workspace}"
BARE="${AIOS_BARE_REPO:-$ROOT/repo.git}"
WORKTREES="${AIOS_WORKTREE_ROOT:-$ROOT/worktrees}"
RUNTIME_USER="${AIOS_RUNTIME_USER:-aiman}"
RUNTIME_GROUP="${AIOS_RUNTIME_GROUP:-gitops}"
if [[ "$(id -u)" -ne 0 ]]; then
echo "ERROR: bootstrap must run as root" >&2
exit 2
fi
getent passwd "$RUNTIME_USER" >/dev/null
getent group "$RUNTIME_GROUP" >/dev/null
[[ -d "$SOURCE/.git" || -f "$SOURCE/.git" ]] || {
echo "ERROR: source worktree not found: $SOURCE" >&2
exit 3
}
install -d -o "$RUNTIME_USER" -g "$RUNTIME_GROUP" -m 2775 "$ROOT" "$WORKTREES"
install -d -o "$RUNTIME_USER" -g "$RUNTIME_GROUP" -m 2775 \
"$WORKTREES/openclaw" "$WORKTREES/chatgpt" "$WORKTREES/controller"
if [[ ! -d "$BARE" ]]; then
sudo -u "$RUNTIME_USER" -g "$RUNTIME_GROUP" git clone --bare "$SOURCE" "$BARE"
fi
chown -R "$RUNTIME_USER:$RUNTIME_GROUP" "$BARE" "$WORKTREES"
find "$BARE" "$WORKTREES" -type d -exec chmod 2775 {} +
find "$BARE" "$WORKTREES" -type f -exec chmod ug+rw {} +
setfacl -R -m "u:${RUNTIME_USER}:rwX,g:${RUNTIME_GROUP}:rwX,m::rwX" "$BARE" "$WORKTREES"
find "$BARE" "$WORKTREES" -type d -exec \
setfacl -m "d:u:${RUNTIME_USER}:rwx,d:g:${RUNTIME_GROUP}:rwx,d:m::rwx" {} +
cat <<EOF
bootstrap_status: PASS
root: $ROOT
source_worktree: $SOURCE
bare_repo: $BARE
worktree_root: $WORKTREES
runtime_user: $RUNTIME_USER
runtime_group: $RUNTIME_GROUP
EOF
@@ -1,48 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
EXPECTED_USER="${AIOS_RUNTIME_USER:-aiman}"
EXPECTED_GROUP="${AIOS_RUNTIME_GROUP:-gitops}"
EXPECTED_UMASK="${AIOS_RUNTIME_UMASK:-0002}"
TASK_DIR="${AIOS_TASK_DIR:-.agent/tasks}"
RESULT_DIR="${AIOS_RESULT_DIR:-.agent/results}"
fail() {
echo "preflight_status: FAIL"
echo "reason: $1"
exit 1
}
USER_NOW="$(id -un)"
GROUPS_NOW="$(id -Gn)"
UMASK_NOW="$(umask)"
ROOT="$(git rev-parse --show-toplevel 2>/dev/null || true)"
GIT_DIR="$(git rev-parse --git-dir 2>/dev/null || true)"
BRANCH="$(git branch --show-current 2>/dev/null || true)"
[[ "$USER_NOW" == "$EXPECTED_USER" ]] || fail "runtime user is $USER_NOW, expected $EXPECTED_USER"
[[ " $GROUPS_NOW " == *" $EXPECTED_GROUP "* ]] || fail "runtime user is not in $EXPECTED_GROUP"
[[ "$UMASK_NOW" == "$EXPECTED_UMASK" || "$UMASK_NOW" == "002" ]] || fail "umask is $UMASK_NOW, expected $EXPECTED_UMASK"
[[ -n "$ROOT" ]] || fail "not inside a git worktree"
[[ "$ROOT" == /data/aios-gitops/worktrees/* ]] || fail "shared source worktree is prohibited: $ROOT"
[[ -w "$ROOT" ]] || fail "worktree root is not writable"
[[ -w "$GIT_DIR" ]] || fail "git dir is not writable: $GIT_DIR"
[[ ! -e "$GIT_DIR/index.lock" ]] || fail "git index lock exists"
install -d -m 2775 "$TASK_DIR" "$RESULT_DIR"
[[ -w "$TASK_DIR" ]] || fail "task directory is not writable: $TASK_DIR"
[[ -w "$RESULT_DIR" ]] || fail "result directory is not writable: $RESULT_DIR"
[[ -z "$(git status --porcelain)" ]] || fail "worktree is not clean at task start"
cat <<EOF
preflight_status: PASS
runtime_user: $USER_NOW
runtime_groups: $GROUPS_NOW
umask: $UMASK_NOW
worktree: $ROOT
git_dir: $GIT_DIR
branch: $BRANCH
workspace_clean: true
task_path_writable: true
result_path_writable: true
git_lock_present: false
EOF
-73
View File
@@ -1,73 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
ACTION="${1:-}"
AGENT="${2:-}"
TASK_ID="${3:-}"
BASE_REF="${4:-origin/main}"
ROOT="${AIOS_GITOPS_ROOT:-/data/aios-gitops}"
BARE="${AIOS_BARE_REPO:-$ROOT/repo.git}"
WORKTREES="${AIOS_WORKTREE_ROOT:-$ROOT/worktrees}"
RUNTIME_USER="${AIOS_RUNTIME_USER:-aiman}"
RUNTIME_GROUP="${AIOS_RUNTIME_GROUP:-gitops}"
usage() {
echo "usage: $0 create|remove <openclaw|chatgpt|controller> <task-id> [base-ref]" >&2
exit 64
}
[[ "$ACTION" =~ ^(create|remove)$ ]] || usage
[[ "$AGENT" =~ ^(openclaw|chatgpt|controller)$ ]] || usage
[[ "$TASK_ID" =~ ^[A-Za-z0-9._-]+$ ]] || {
echo "ERROR: invalid task id" >&2
exit 65
}
[[ "$(id -un)" == "$RUNTIME_USER" ]] || {
echo "ERROR: must run as $RUNTIME_USER" >&2
exit 66
}
[[ "$(id -gn)" == "$RUNTIME_GROUP" || " $(id -Gn) " == *" $RUNTIME_GROUP "* ]] || {
echo "ERROR: user must be in $RUNTIME_GROUP" >&2
exit 67
}
[[ "$(umask)" == "0002" || "$(umask)" == "002" ]] || {
echo "ERROR: umask must be 0002" >&2
exit 68
}
[[ -d "$BARE" ]] || {
echo "ERROR: bare repository missing: $BARE" >&2
exit 69
}
PATHNAME="$WORKTREES/$AGENT/$TASK_ID"
BRANCH="task/$AGENT/$TASK_ID"
case "$ACTION" in
create)
[[ ! -e "$PATHNAME" ]] || {
echo "ERROR: worktree already exists: $PATHNAME" >&2
exit 70
}
git --git-dir="$BARE" fetch --all --prune
git --git-dir="$BARE" worktree add -b "$BRANCH" "$PATHNAME" "$BASE_REF"
install -d -m 2775 "$PATHNAME/.agent/tasks/$AGENT" "$PATHNAME/.agent/results/$AGENT"
setfacl -m "u:${RUNTIME_USER}:rwx,g:${RUNTIME_GROUP}:rwx,m::rwx" "$PATHNAME"
find "$PATHNAME" -type d -exec chmod g+s {} +
find "$PATHNAME" -type d -exec \
setfacl -m "d:u:${RUNTIME_USER}:rwx,d:g:${RUNTIME_GROUP}:rwx,d:m::rwx" {} +
printf 'worktree: %s\nbranch: %s\nbase: %s\n' "$PATHNAME" "$BRANCH" "$BASE_REF"
;;
remove)
[[ -d "$PATHNAME" ]] || {
echo "ERROR: worktree not found: $PATHNAME" >&2
exit 71
}
[[ -z "$(git -C "$PATHNAME" status --porcelain)" ]] || {
echo "ERROR: worktree is not clean; retain for investigation" >&2
exit 72
}
git --git-dir="$BARE" worktree remove "$PATHNAME"
printf 'removed: %s\n' "$PATHNAME"
;;
esac
+82
View File
@@ -0,0 +1,82 @@
#!/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
schemas/automation-quorum-controller.schema.json
config/aios-automation-quorum-controller.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
)
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}/schemas/automation-quorum-controller.schema.json" \
"${VERSION_DIR}/share/automation-quorum-controller.schema.json"
python3 -m py_compile "${VERSION_DIR}/bin/executor-quorum-controller"
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-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
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}"
+29
View File
@@ -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")"
+99
View File
@@ -0,0 +1,99 @@
#!/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 = {
"database_path",
"file_inbox_path",
"health_receipts",
"blocker_retry_threshold",
"max_health_age_seconds",
"lease_ttl_seconds",
"gateway_event_url",
}
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")
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 not in {"http", "https"} or not parsed.hostname:
fail("gateway_event_url must be an HTTP(S) URL with a host")
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())
+17
View File
@@ -0,0 +1,17 @@
#!/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 unittest discover -s "${ROOT}/tests" -p 'test_executor_quorum_controller.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"
@@ -0,0 +1,37 @@
[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
ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/executor-quorum-controller --once --config /etc/aios-automation-quorum/controller.json
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
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 -
+390
View File
@@ -0,0 +1,390 @@
from __future__ import annotations
import importlib.util
import json
import os
import sqlite3
import subprocess
import sys
import tempfile
import time
import unittest
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_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()
@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)