Add autonomous operation command center deployment command

Add command stall detector tests
Add unified operations command center model
2026-07-09 15:57:18 +08:00 · 2026-07-09 15:57:13 +08:00 · 2026-07-09 15:57:11 +08:00 · 2026-07-09 15:57:05 +08:00 · 2026-07-09 15:53:00 +08:00 · 2026-07-09 15:52:46 +08:00
259 changed files with 1041 additions and 24681 deletions
@@ -1,76 +0,0 @@
 {
  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
  "created_at": "2026-07-18T04:58:00+00:00",
  "schema_version": 2,
  "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
    }
  ]
 }
@@ -1,248 +0,0 @@
 {
  "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,
      "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",
      "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-18T04:58: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"
      ]
    }
  }
 }
@@ -1,156 +0,0 @@
 # V2R1 Gate 1 — Clean-Room Final Evidence
 ## Summary
 | Field | Value |
 |---|---|
 | Date | 2026-07-14 |
 | Source commit | `502ef30642597fb0b70b90ae2e8a0b6af217d80e` |
 | Source branch | `origin/work/v2r1-integration-quality-2000` |
 | Result branch | `fix/v2r1-g1-cleanroom-final-evidence-r2-20260714` |
 | Result commit reference | `HEAD` |
 | Gate | `G1` |
 | Variant | `openclaw-cleanroom` |
 | Status | `PASS` |
 The immutable report uses `HEAD` for post-commit verification. The resolved commit object ID is recorded by the external acceptance result after this file is committed; a Git commit cannot reliably embed its own object ID because changing the file changes the tree and commit identity.
 ## Machine-Readable Result
 ```yaml
 gate: G1
 variant: openclaw-cleanroom
 date: 2026-07-14
 status: PASS
 source_commit: 502ef30642597fb0b70b90ae2e8a0b6af217d80e
 source_branch: origin/work/v2r1-integration-quality-2000
 result_branch: fix/v2r1-g1-cleanroom-final-evidence-r2-20260714
 result_commit_ref: HEAD
 tests_total: 32
 tests_passed: 32
 tests_failed: 0
 cleanup_scan: PASS
 secret_scan: PASS
 rollback_status: PASS
 recovery_verified: true
 workspace_clean_at_test_completion: true
 merge_performed: false
 tag_created: false
 baseline_promoted: false
 deployment_performed: false
 ```
 ## Evidence Provenance
 The test, cleanup, secret-scan, rollback, and recovery results below are execution evidence supplied by OpenClaw for the Gate 1 run. This correction commit changes only the evidence report and does not claim that the GitOps API reran the suites.
 ## HMAC Test-Key Length Verification
 | Test file | Variable | Test-only value | Length |
 |---|---|---|---:|
 | `tests/v2r1/test_runtime_wiring.py:24` | `internal_key` | `b"runtime-internal-context-key-is-32bytes"` | 39 bytes |
 | `tests/v2r1/test_runtime_wiring.py:25` | `ingress_key` | `b"runtime-external-ingress-key-is-32bytes"` | 39 bytes |
 Both values are test fixtures. The supplied evidence reports that no production secret was introduced.
 ## Test Results Supplied by OpenClaw
 ### V2R1 suite — 29 of 29 passed
 ```shell
 PYTHONPATH=active-runtime/app python3 -m pytest \
  tests/v2r1/test_hermes_fabric.py \
  tests/v2r1/test_message_ledger_append.py \
  tests/v2r1/test_router_isolation.py \
  tests/v2r1/test_runtime_identity_binding.py \
  tests/v2r1/test_runtime_wiring.py \
  tests/v2r1/test_tenant_context.py -v
 # Reported result: 29 passed, exit code 0
 ```
 ### Integration suite — 3 of 3 passed
 ```shell
 PYTHONPATH=active-runtime/app python3 -m pytest tests/integration/test_runtime_ingress_flow.py -v
 # Reported result: 3 passed, exit code 0
 ```
 **Reported total: 32 of 32 passed.**
 ## Cleanup, Security, and Recovery Evidence
 | Check | Supplied result |
 |---|---|
 | Conflict-marker scan | PASS |
 | Cleanup scan | PASS |
 | Secret scan | PASS |
 | Legitimate hardcoded test keys | Two 39-byte fixtures in `test_runtime_wiring.py` |
 | Rollback rehearsal | PASS; 13 of 13 core tests reported after rollback to `6077331` |
 | Recovery verification | PASS |
 ## Deployment State
 | Operation | Performed |
 |---|---|
 | Merge | No |
 | Tag creation | No |
 | Baseline promotion | No |
 | Staging deployment | No |
 | Production deployment | No |
 ## Post-Commit Verification
 Run from the repository root after the commit is created:
 ```shell
 set -eu
 SOURCE=502ef30642597fb0b70b90ae2e8a0b6af217d80e
 RESULT_FILE=.agent/results/openclaw/RESULT-V2R1-G1-OPENCLAW-FINAL-EVIDENCE-20260714.md
 git rev-parse HEAD
 test "$(git diff --name-only "$SOURCE" HEAD | wc -l | tr -d ' ')" = "1"
 test "$(git diff --name-only "$SOURCE" HEAD)" = "$RESULT_FILE"
 git ls-tree -r HEAD -- "$RESULT_FILE"
 python3 - "$RESULT_FILE" <<'PY'
 from pathlib import Path
 import re
 import sys
 path = Path(sys.argv[1])
 text = path.read_text(encoding="utf-8")
 encoded_tokens = [
    [84, 66, 68],
    [80, 69, 78, 68, 73, 78, 71],
    [84, 79, 68, 79],
    [70, 73, 88, 77, 69],
    [91, 123, 99, 111, 109, 109, 105, 116, 125, 93],
 ]
 tokens = ["".join(chr(value) for value in item) for item in encoded_tokens]
 angle_token = re.compile(chr(60) + r"[^>\n]+" + chr(62))
 hits = [token for token in tokens if token in text]
 if angle_token.search(text):
    hits.append("angle-delimited-token")
 if hits:
    print("TOKEN_SCAN=FAIL", hits)
    raise SystemExit(1)
 print("TOKEN_SCAN=PASS")
 PY
 ```
 ## Acceptance Conditions
 Gate 1 is eligible to proceed only when the external acceptance record confirms all of the following against the resolved `HEAD` commit:
 - the commit exists and has source commit `502ef30642597fb0b70b90ae2e8a0b6af217d80e` as its parent;
 - exactly this one report file changed;
 - the committed report contains no unresolved legacy token;
 - the result tree is not identical to rejected commit `71239b4d187e683e05fd22fc9d5df8ae324e3390`;
 - no merge, tag, baseline promotion, or deployment occurred.
@@ -1,44 +0,0 @@
 {
  "b3": "BLOCKED_EVIDENCE_REQUIRED",
  "b4": "NOT_STARTED",
  "baseline": "V1R4C6",
  "contract_branch": "task/governance/v2r1-personal-weixin-production-contract-20260717",
  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
  "dispatch_commit": null,
  "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": 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"
 }
@@ -1,69 +0,0 @@
 # Codex task: apply and verify AIOS Orchestrator Gateway v1
 Contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
 Use the strongest actually available Codex model for the initial security review.
 If unavailable or timed out, record TIMEOUT/NOT_EXECUTED and let OpenClaw run
 the deterministic test path. Do not count a timeout as PASS.
 ## Apply
 1. Create a new clean append-only branch and isolated worktree.
 2. Validate the delivery tarball and `SHA256SUMS`.
 3. Run the external `scripts/apply_gateway_patch.sh` with `SOURCE_DIR` and
   `WORKTREE`; do not alter unrelated paths.
 4. Review all added code, especially:
   - exact contract and optimistic state-version checks;
   - event/decision idempotency;
   - SQLite leases and retry behavior;
   - Workspace Agent `202` asynchronous callback flow;
   - Responses background creation and polling;
   - file-inbox and Git-ref polling;
   - traversal/symlink protections;
   - OpenClaw lifecycle synchronization;
   - systemd sandbox;
   - install and rollback transactions.
 ## Tests
 Run from the final source tree:
 - Python compile for all gateway and helper files;
 - unit tests;
 - JSON schema/config parsing;
 - Bash syntax;
 - `systemd-analyze verify`;
 - event duplicate and payload-conflict tests;
 - stale state-version rejection;
 - lease/double-claim test;
 - Workspace trigger failure and Responses fallback;
 - response queued/in-progress/completed polling;
 - file-inbox recovery after HTTP failure;
 - Git-ref initial baseline and subsequent-change event;
 - task start/end and project summary synchronization;
 - dispatch traversal and symlink rejection;
 - retry/backoff/dead-letter;
 - install dry-run;
 - rollback test in an isolated fixture;
 - `git diff --check`;
 - full repository tests;
 - final Gitleaks.
 ## Commit and review
 1. Create an append-only implementation commit.
 2. Push without force.
 3. Confirm `WORKTREE_OID == BARE_OID == GITEA_OID`.
 4. Create an independent RESULT-only branch and exact-one result commit.
 5. Record the source bundle SHA-256, implementation OID, RESULT OID, tests,
   Gitleaks, systemd validation and remaining external configuration blockers.
 ## Forbidden
 No reset, amend, rebase, force push, direct main push, controller mutation,
 secret disclosure, fabricated test result, production deployment or B3 evidence
 fabrication.
 Success:
 `GATEWAY_CODE_REVIEW=PASS IMPLEMENTATION=<oid> RESULT=<oid> OID_MATCH=YES`
@@ -1,13 +0,0 @@
 # Codex V2R1 adaptive execution
 CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
 Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
 Execute P1-POST-CONTRACT-REACK now:
 - read-only verify Repair3 e110cda9968cadc5c7b7179ec5003b05aad11ce4
 - read-only verify RESULT 49a9c2e0cc898ecf2bbd45f454e007644d7f3045
 - verify lineage from 8d9c6da90fc8caa69b691d243c6fefb40b6b0575, exact4, tests, Gitleaks and RESULT-only scope
 - do not modify Repair3 code in this unit
 Act as fallback for OpenClaw only on timeout/unavailability. Use GPT-5.6 only if actually exposed; otherwise strongest visible model. Prefer OpenClaw when quality is comparable or volume is high; prefer Codex for complex/security-critical work.
 Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
 Report:
 CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=codex STATUS=<status> IMPLEMENTATION=<oid-or-NONE> RESULT=<oid-or-NONE> OID_MATCH=<YES-or-NO> BLOCKER=<reason-or-NONE>
 Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.
@@ -1,119 +0,0 @@
 # 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>
 ```
@@ -1,68 +0,0 @@
 # OpenClaw task: deploy and supervise AIOS Orchestrator Gateway v1
 Contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
 You are the default orchestrator and deployment supervisor.
 ## Preconditions
 - Codex has applied and independently reviewed this source bundle on a clean,
  append-only branch.
 - Implementation and RESULT OIDs are identical across worktree, bare repo and
  Gitea.
 - Final Gitleaks and repository tests are PASS.
 - No force push, reset, amend or rebase occurred.
 ## Responsibilities
 1. Validate the source bundle SHA-256 and `SHA256SUMS`.
 2. Review the generated systemd service, poll timer, tmpfiles and sysusers files.
 3. Install through `scripts/install_aios_orchestrator_gateway.sh`.
 4. Configure one or both central-model channels:
   - published Workspace Agent API channel with `agtch_...` trigger ID and a
     Workspace Agent access token stored outside Git; or
   - Responses API fallback with an approved model and API key stored outside Git.
 5. Publish the callback endpoint securely and configure the Workspace Agent
   action from `workspace-agent/orchestrator-decision-action.openapi.yaml`.
 6. Confirm that the Workspace Agent instructions match
   `workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md`.
 7. Enable both:
   - the continuously running gateway service;
   - the independent `aios-orchestrator-gateway-poll.timer`.
 8. Verify HTTP event ingestion, file-inbox polling, Git-ref polling,
   Workspace Agent trigger acceptance, callback decision, Responses background
   polling fallback, dispatch queue generation, retry leases and loop guards.
 9. Verify OpenClaw receives and synchronizes:
   - TASK_DISPATCHED
   - TASK_STARTED
   - TASK_ENDED
   - PROJECT_SUMMARY
   - PROJECT_COMPLETED
 10. Run a harmless bootstrap work unit through OpenClaw and Codex and confirm
    duplicate event submission is idempotent.
 11. Run rollback, reinstall and heartbeat/health verification.
 12. Create one RESULT-only review recording exact commits, artifact digest,
    tests, systemd security, polling recovery, lifecycle synchronization and
    rollback.
 ## Fail-closed boundaries
 - Do not put Workspace Agent tokens, API keys, callback tokens or gateway
  credentials in Git, logs, command arguments or RESULT files.
 - Do not claim the gateway resolves the existing B3 credential/raw-wire blocker.
 - Do not advance B4 or production merely because this gateway is healthy.
 - Do not bypass owner/admin production release approvals.
 - Do not use reset, amend, rebase, force push, direct main push or bare-ref edits.
 ## Required final status
 Success:
 `ORCHESTRATOR_GATEWAY_LIVE=PASS POLLING=PASS OPENCLAW_SYNC=PASS ROLLBACK=PASS`
 Blocker:
 `ORCHESTRATOR_GATEWAY_LIVE=BLOCKED reason=<stable_reason>`
 Keep `BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED` until the
 existing production contract advances all release gates.
@@ -1,13 +0,0 @@
 # OpenClaw V2R1 adaptive orchestration
 CONTRACT_ACK=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
 Read `.agent/policies/V2R1-PERSONAL-WEIXIN-PRODUCTION-EXECUTION-CONTRACT.yaml`.
 You are default orchestrator and sole authoritative state writer.
 Execute immediately:
 1. R0-DISCOVER-SECRET-PROVIDER: discover approved provider metadata and owner-probe source without reading or outputting secrets. Environment variables alone are not approved secret refs.
 2. P2-INVENTORY-REAL-EVIDENCE: inventory approved restricted B3 evidence locations; validate existing facts; never synthesize evidence.
 3. Independently verify Codex results before advancing.
 Prefer OpenClaw for comparable-quality, large, repetitive, long-running, infrastructure and cost-sensitive work. Prefer Codex strongest actually available model for complex code, security boundaries, architecture and high-risk review. On timeout mark TIMEOUT and switch agent; never infer PASS.
 Forbidden: reset, amend, rebase, force push, direct main push, controller mutation, secret disclosure, fabricated evidence, skipped gates.
 Report:
 CONTRACT=cbe33ecab8e9794ba5f318bea43aeaf447a56fbc WORK_UNIT=<id> TRACK=<id> GATE=<id> AGENT=openclaw STATUS=<status> COMMIT=<oid-or-NONE> BLOCKER=<reason-or-NONE>
 Keep BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED until production gate passes.
@@ -1,11 +0,0 @@
 # AIOS Codex automatic command watcher
 The watcher polls the governed `commands/active` queue without a chat trigger. It uses SQLite transactions for exclusive leases, records the exact command digest, rejects contract drift, retries transient failures, dead-letters exhausted work, and resumes from the last durable phase so an end-event retry does not execute a work unit twice.
 Commands are JSON documents validated by `schemas/codex-command.schema.json`. `acceptance_probe` is the only built-in action and has no external side effect. Production `codex` work requires an administrator-provisioned absolute executor path; the instruction is passed on standard input, never argv or environment. The executor receives a sanitized environment and must obtain credentials through its approved file or exec provider. If the executor is absent, the task fails closed and follows the retry/dead-letter policy.
 The watcher provides at-most-once execution. It writes the `EXECUTING` phase before launching an external executor; if the host dies after an external effect but before a result is durably recorded, recovery does not replay the work and instead dead-letters the ambiguity for human review. Start and end event JSON is persisted before transmission, so response-loss retries reuse the exact event ID and payload and are deduplicated by the Gateway.
 Lifecycle evidence is appended to `/var/log/aios-codex-watcher/audit.jsonl` as `CODEX_AUTO_CLAIM`, `TASK_STARTED`, `TASK_EXECUTED`, `TASK_ENDED`, and `EVENT_POSTED`. Inputs, tokens, stdout, and stderr are not logged; only result digests are retained.
 Install from a reviewed commit with `sudo scripts/install_aios_codex_watcher.sh`. The installer creates a content-addressed release, atomically updates `current`, writes a receipt, and enables the five-second systemd timer. The existing Gateway inbound token file is referenced in place and never copied.
@@ -1,107 +0,0 @@
 # AIOS Orchestrator Gateway v0.2
 This component turns the current passive GPT orchestration pattern into a
 durable, unattended loop:
 ```text
 Gitea/OpenClaw/Codex event
  -> HTTP event endpoint or polled file inbox
  -> durable SQLite state machine
  -> Workspace Agent API trigger
  -> strict callback decision
  -> Responses API background fallback and polling
  -> OpenClaw/Codex dispatch queues
  -> start/end/result events
  -> next orchestration cycle
 ```
 ## Reliability mechanisms
 The gateway does not depend on one timer, webhook or model call:
 - HTTP event ingestion.
 - Periodic file-inbox polling for missed webhooks and offline producers.
 - Local bare-Git ref polling for `main`, governance, patch and result refs.
 - Workspace Agent retries with a stable event `Idempotency-Key`.
 - Workspace Agent callback timeout and Responses API background fallback.
 - Responses status polling until completion or fail-closed terminal status.
 - SQLite optimistic state versions, idempotency records and processing leases.
 - Exponential retry with a maximum retry count and dead-letter state.
 - Same-blocker and maximum-hop guards to stop non-progress loops.
 - A systemd timer runs `--poll-once` as an independent reliability watchdog.
 Workspace Agent trigger runs are asynchronous and do not expose a retrievable
 final response. The central Workspace Agent must therefore call the
 `submit_orchestration_decision` action.
 ## OpenClaw synchronization
 Every lifecycle stage is durably written to the OpenClaw synchronization queue:
 - `TASK_DISPATCHED`
 - `TASK_STARTED`
 - `TASK_ENDED`
 - `DECISION_APPLIED`
 - `PROJECT_SUMMARY`
 - `PROJECT_COMPLETED`
 An optional authenticated OpenClaw HTTP endpoint can receive the same records.
 A failed HTTP delivery remains in the lifecycle retry queue. OpenClaw remains
 the default project-status synchronizer and authoritative progress presenter.
 ## Agent execution contract
 Each dispatched JSON envelope contains:
 - exact contract commit;
 - source event;
 - clean base commit;
 - primary and fallback agent;
 - allowed paths;
 - required checks;
 - forbidden operations;
 - completion event callback format.
 OpenClaw and Codex must send a `work_unit_started` event before execution and a
 `work_unit_completed`, `work_unit_blocked`, or `work_unit_failed` event when
 they stop. This guarantees that OpenClaw can synchronize every task start,
 finish, and overall project summary.
 ## Security
 - Model and callback credentials are read only from root-owned secret files.
 - Credentials never enter Git, dispatch payloads or logs.
 - The gateway runs as a dedicated unprivileged user.
 - Dispatch and sync paths are protected against traversal and symlink targets.
 - All state changes require the exact contract commit and current state version.
 - Production evidence, credentials and approvals stay fail-closed.
 - Git history operations remain controlled by the committed governance contract.
 ## Deployment sequence
 1. Apply this source bundle to a clean append-only branch.
 2. Run unit, repository, shell, JSON, systemd and Gitleaks checks.
 3. Configure one or both model channels:
   - published Workspace Agent API channel and Workspace Agent access token;
   - Responses API key and approved fallback model.
 4. Publish the decision callback as a Workspace Agent action or MCP tool.
 5. Install the service through `scripts/install_aios_orchestrator_gateway.sh`.
 6. Run `scripts/verify_aios_orchestrator_gateway.sh`.
 7. Submit a harmless event and verify:
   - trigger acceptance;
   - strict decision callback;
   - dispatch file;
   - OpenClaw task-start/task-end/project-summary synchronization;
   - duplicate idempotency;
   - polling recovery after suppressing the HTTP webhook.
 The installer copies an explicit gateway-only allowlist into a release directory
 named by its SHA-256 digest, atomically updates `/opt/aios-orchestrator-gateway/current`,
 and persists `/var/lib/aios-orchestrator-gateway/install-receipt.json` with the
 current and previous release. `--dry-run` performs source validation and emits
 `AIOS_ORCHESTRATOR_INSTALL_DRY_RUN=PASS` without changing installed state.
 Rollback reads that root-owned `0600` receipt, verifies both release directory
 names and `SOURCE_DIGEST` files, coordinates the gateway and poll units, and
 atomically switches `current` to the previous release. Repeating the same
 rollback is a no-op and emits `NOOP_ALREADY_ROLLED_BACK`.
@@ -1 +1 @@
-V1R4C6
+V1R4C2
@@ -10,22 +10,11 @@ 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()
@@ -195,8 +184,6 @@ 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)
@@ -229,7 +216,6 @@ 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)
@@ -243,25 +229,6 @@ 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
@@ -423,9 +390,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','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}}
+    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}}
 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()
@@ -1,120 +0,0 @@
 """Architect governance agent responsibilities.
 This agent records requirement changes, maintains versioned human-readable
 requirement/architecture/deployment/test documents, reviews architecture changes,
 validates deployment docs against actual configuration, and produces version bump
 recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class RequirementChange:
    id: str
    source: str
    actor: str
    requirement: str
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class DocumentRefreshPlan:
    requirement_doc: bool
    architecture_doc: bool
    deployment_doc: bool
    test_case_doc: bool
    reason: str
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ArchitectureReviewDecision:
    status: str
    findings: list[str]
    optimization_recommendations: list[str]
    version_bump: str
    docs_to_refresh: DocumentRefreshPlan
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "findings": self.findings,
            "optimization_recommendations": self.optimization_recommendations,
            "version_bump": self.version_bump,
            "docs_to_refresh": self.docs_to_refresh.to_dict(),
            "created_at": self.created_at,
        }
 class ArchitectGovernanceAgent:
    """Independent architecture governance role."""
    def record_requirement_change(self, source: str, actor: str, requirement: str) -> RequirementChange:
        change_id = "REQ-" + datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
        return RequirementChange(change_id, source, actor, requirement)
    def review_architecture_change(self, proposal: dict[str, Any]) -> ArchitectureReviewDecision:
        impact = proposal.get("impact", "medium")
        findings = [
            "requirement_traceability_checked",
            "architecture_consistency_checked",
            "deployment_doc_alignment_checked",
            "test_case_doc_required",
            "rollback_and_owner_gate_checked",
        ]
        recommendations = [
            "refresh user requirements document",
            "refresh architecture design document",
            "refresh engineering deployment document",
            "generate acceptance test case document",
        ]
        version_bump = self.decide_version_bump(proposal)
        return ArchitectureReviewDecision(
            status="review_required" if impact in {"high", "architecture"} else "pass",
            findings=findings,
            optimization_recommendations=recommendations,
            version_bump=version_bump,
            docs_to_refresh=DocumentRefreshPlan(True, True, True, True, "architecture or requirement changed"),
        )
    def decide_version_bump(self, proposal: dict[str, Any]) -> str:
        if proposal.get("breaks_architecture") or proposal.get("major_capability"):
            return "V"
        if proposal.get("new_release_capability") or proposal.get("architecture_change"):
            return "R"
        return "C"
    def validate_deployment_doc(self, deployment_doc: dict[str, Any], actual_config: dict[str, Any]) -> dict[str, Any]:
        expected = set(deployment_doc.get("services", []))
        actual = set(actual_config.get("services", []))
        missing = sorted(expected - actual)
        extra = sorted(actual - expected)
        return {
            "status": "pass" if not missing else "attention_required",
            "missing_services": missing,
            "extra_services": extra,
            "checked_at": utc_now(),
        }
 __all__ = [
    "RequirementChange",
    "DocumentRefreshPlan",
    "ArchitectureReviewDecision",
    "ArchitectGovernanceAgent",
 ]
@@ -1,32 +0,0 @@
 """API for Architect Governance Center."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .architect_governance_dashboard import ArchitectGovernanceDashboard
 dashboard = ArchitectGovernanceDashboard()
 def get_architect_governance_overview():
    return dashboard.overview()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/architect", tags=["architect-governance"])
    @router.get("/overview", operation_id="getArchitectGovernanceOverview")
    def overview():
        return get_architect_governance_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_architect_governance_overview"]
@@ -1,153 +0,0 @@
 """Web UI dashboard model for Architect Governance Agent.
 Shows requirement changes, architecture review findings, document refresh status,
 deployment consistency, test case readiness, weekly research and V/R/C version
 recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectPanelMetric:
    key: str
    title: str
    value: Any
    status: str = "ok"
    detail: str = ""
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ArchitectPanel:
    key: str
    title: str
    description: str
    metrics: list[ArchitectPanelMetric] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "description": self.description,
            "metrics": [m.to_dict() for m in self.metrics],
        }
 class ArchitectGovernanceDashboard:
    def overview(
        self,
        requirement_changes: list[dict[str, Any]] | None = None,
        architecture_reviews: list[dict[str, Any]] | None = None,
        document_status: dict[str, Any] | None = None,
        research_status: dict[str, Any] | None = None,
        version_recommendations: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        requirement_changes = requirement_changes or []
        architecture_reviews = architecture_reviews or []
        document_status = document_status or {}
        research_status = research_status or {}
        version_recommendations = version_recommendations or []
        panels = [
            self.requirement_panel(requirement_changes),
            self.architecture_panel(architecture_reviews),
            self.document_panel(document_status),
            self.research_panel(research_status),
            self.version_panel(version_recommendations),
        ]
        return {
            "module": "architect-governance",
            "route": "/admin/ai-evolution/architect",
            "title": "架构师 Agent 治理中心",
            "status": "attention_required" if any(self._panel_warn(panel) for panel in panels) else "healthy",
            "generated_at": utc_now(),
            "panels": [panel.to_dict() for panel in panels],
            "requirement_changes": requirement_changes,
            "architecture_reviews": architecture_reviews,
            "version_recommendations": version_recommendations,
        }
    def requirement_panel(self, changes: list[dict[str, Any]]) -> ArchitectPanel:
        return ArchitectPanel(
            key="requirements",
            title="需求变化记录",
            description="记录来自 Custom GPT、Codex、OpenClaw、Hermes 和用户反馈的需求变化。",
            metrics=[
                ArchitectPanelMetric("total_requirement_changes", "需求变化数量", len(changes)),
                ArchitectPanelMetric(
                    "undocumented_changes",
                    "待刷新到用户需求文档",
                    len([c for c in changes if not c.get("documented")]),
                    "warn" if any(not c.get("documented") for c in changes) else "ok",
                ),
            ],
        )
    def architecture_panel(self, reviews: list[dict[str, Any]]) -> ArchitectPanel:
        pending = [r for r in reviews if r.get("status") not in {"pass", "not_required"}]
        return ArchitectPanel(
            key="architecture_review",
            title="全量架构审视",
            description="展示架构一致性、扩展性、安全边界、可运维性和回滚策略审查。",
            metrics=[
                ArchitectPanelMetric("review_count", "架构审查数量", len(reviews)),
                ArchitectPanelMetric("pending_reviews", "待处理架构审查", len(pending), "warn" if pending else "ok"),
            ],
        )
    def document_panel(self, status: dict[str, Any]) -> ArchitectPanel:
        required_docs = ["requirements", "architecture", "deployment", "test_cases"]
        missing = [doc for doc in required_docs if status.get(doc) not in {"ready", "updated"}]
        return ArchitectPanel(
            key="documents",
            title="文档版本化状态",
            description="用户需求、架构设计、工程部署、测试例文档刷新状态。",
            metrics=[
                ArchitectPanelMetric("requirements_doc", "用户需求文档", status.get("requirements", "unknown"), "warn" if "requirements" in missing else "ok"),
                ArchitectPanelMetric("architecture_doc", "架构设计文档", status.get("architecture", "unknown"), "warn" if "architecture" in missing else "ok"),
                ArchitectPanelMetric("deployment_doc", "工程部署文档", status.get("deployment", "unknown"), "warn" if "deployment" in missing else "ok"),
                ArchitectPanelMetric("test_case_doc", "测试例文档", status.get("test_cases", "unknown"), "warn" if "test_cases" in missing else "ok"),
            ],
        )
    def research_panel(self, status: dict[str, Any]) -> ArchitectPanel:
        overdue = bool(status.get("overdue"))
        return ArchitectPanel(
            key="weekly_research",
            title="每周外部技术情报",
            description="联网搜索 Agent/模型版本、GitHub top 实现、潜在安全影响。",
            metrics=[
                ArchitectPanelMetric("last_research_at", "最近调研时间", status.get("last_research_at", "never"), "warn" if overdue else "ok"),
                ArchitectPanelMetric("findings", "发现数量", status.get("findings", 0)),
                ArchitectPanelMetric("optimization_candidates", "可借鉴优化项", status.get("optimization_candidates", 0)),
            ],
        )
    def version_panel(self, recommendations: list[dict[str, Any]]) -> ArchitectPanel:
        latest = recommendations[-1] if recommendations else {}
        return ArchitectPanel(
            key="version_decision",
            title="V/R/C 版本建议",
            description="由架构师 Agent 判断变更应进入 C、R 或 V。",
            metrics=[
                ArchitectPanelMetric("recommendation_count", "版本建议数量", len(recommendations)),
                ArchitectPanelMetric("latest_recommendation", "最近建议", latest.get("version_bump", "none")),
                ArchitectPanelMetric("reason", "建议原因", latest.get("reason", "")),
            ],
        )
    def _panel_warn(self, panel: ArchitectPanel) -> bool:
        return any(metric.status == "warn" for metric in panel.metrics)
 __all__ = ["ArchitectPanelMetric", "ArchitectPanel", "ArchitectGovernanceDashboard"]
@@ -1,40 +0,0 @@
 """Web UI model for Architect weekly research."""
 from .architect_weekly_research_agent import ArchitectWeeklyResearchAgent
 class ArchitectResearchDashboard:
    def __init__(self, agent=None):
        self.agent = agent or ArchitectWeeklyResearchAgent()
    def overview(self):
        return {
            "module": "architect-weekly-research",
            "route": "/admin/ai-evolution/architect/research",
            "title": "架构师 Agent 技术情报中心",
            "panels": [
                {
                    "title": "Research Targets",
                    "data": self.agent.plan(),
                },
                {
                    "title": "Evaluation",
                    "fields": [
                        "relevance",
                        "impact",
                        "recommendation",
                    ],
                },
                {
                    "title": "Version Decision",
                    "fields": [
                        "C",
                        "R",
                        "V",
                    ],
                },
            ],
        }
 __all__ = ["ArchitectResearchDashboard"]
@@ -1,122 +0,0 @@
 """Architect research timeline dashboard.
 Shows weekly external research findings as a traceable architecture-impact flow:
 research target -> finding -> impact -> recommendation -> version decision ->
 governance event.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ResearchTimelineStep:
    id: str
    title: str
    step_type: str
    status: str
    source: str
    timestamp: str
    impact: str = "none"
    version_bump: str = "none"
    detail: dict[str, Any] = field(default_factory=dict)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectResearchTimeline:
    ORDER = {
        "research_planned": 10,
        "finding_created": 20,
        "impact_analyzed": 30,
        "recommendation_created": 40,
        "version_decision_created": 50,
        "architecture_review_event_created": 60,
    }
    def build(self, findings: list[dict[str, Any]] | None = None, decisions: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        findings = findings or []
        decisions = decisions or []
        steps: list[ResearchTimelineStep] = [
            ResearchTimelineStep(
                id="research-plan",
                title="每周技术情报调研计划",
                step_type="research_planned",
                status="ok",
                source="architect-agent",
                timestamp=utc_now(),
                detail={"cadence": "weekly"},
            )
        ]
        for idx, finding in enumerate(findings):
            steps.append(
                ResearchTimelineStep(
                    id=finding.get("id", f"finding-{idx}"),
                    title=finding.get("title", "research finding"),
                    step_type="finding_created",
                    status="ok",
                    source=finding.get("source", "unknown"),
                    timestamp=finding.get("created_at", utc_now()),
                    impact=finding.get("impact", "none"),
                    detail=finding,
                )
            )
            steps.append(
                ResearchTimelineStep(
                    id=f"impact-{finding.get('id', idx)}",
                    title="架构影响分析",
                    step_type="impact_analyzed",
                    status="attention_required" if finding.get("impact") in {"architecture", "breaking", "new_capability"} else "ok",
                    source="architect-agent",
                    timestamp=utc_now(),
                    impact=finding.get("impact", "none"),
                    detail={"relevance": finding.get("relevance"), "recommendation": finding.get("recommendation")},
                )
            )
        for idx, decision in enumerate(decisions):
            steps.append(
                ResearchTimelineStep(
                    id=decision.get("id", f"decision-{idx}"),
                    title="C/R/V 版本建议",
                    step_type="version_decision_created",
                    status=decision.get("status", "ok"),
                    source="architect-agent",
                    timestamp=decision.get("created_at", utc_now()),
                    version_bump=decision.get("version_bump", "none"),
                    detail=decision,
                )
            )
        steps.sort(key=lambda s: (self.ORDER.get(s.step_type, 999), s.timestamp))
        return {
            "module": "architect-research-timeline",
            "route": "/admin/ai-evolution/architect/research/timeline",
            "title": "架构师技术情报时间线",
            "status": self._status(steps),
            "generated_at": utc_now(),
            "steps": [step.to_dict() for step in steps],
            "summary": self._summary(steps),
        }
    def _status(self, steps: list[ResearchTimelineStep]) -> str:
        if any(step.status == "attention_required" for step in steps):
            return "attention_required"
        return "healthy"
    def _summary(self, steps: list[ResearchTimelineStep]) -> dict[str, Any]:
        return {
            "findings": len([s for s in steps if s.step_type == "finding_created"]),
            "architecture_impacts": len([s for s in steps if s.impact in {"architecture", "breaking", "new_capability"}]),
            "version_decisions": len([s for s in steps if s.step_type == "version_decision_created"]),
            "recommended_bumps": sorted({s.version_bump for s in steps if s.version_bump != "none"}),
        }
 __all__ = ["ResearchTimelineStep", "ArchitectResearchTimeline"]
@@ -1,32 +0,0 @@
 """API for architect research timeline."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .architect_research_timeline import ArchitectResearchTimeline
 timeline = ArchitectResearchTimeline()
 def get_architect_research_timeline():
    return timeline.build()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/architect/research", tags=["architect-research"])
    @router.get("/timeline", operation_id="getArchitectResearchTimeline")
    def get_timeline():
        return get_architect_research_timeline()
    return router
 router = build_router()
 __all__ = ["router", "get_architect_research_timeline"]
@@ -1,113 +0,0 @@
 """Weekly external research loop for Architect Governance Agent.
 The agent plans and normalizes weekly research tasks. Actual internet access is
 provided by the runtime/search worker; this module defines the audited workflow,
 findings, impact analysis, and V/R/C recommendations.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 from uuid import uuid4
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ResearchTarget:
    id: str
    topic: str
    source_type: str
    query: str
    cadence: str = "weekly"
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ResearchFinding:
    id: str
    target_id: str
    title: str
    summary: str
    source: str
    relevance: str
    impact: str
    recommendation: str
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ResearchDecision:
    status: str
    version_bump: str
    reason: str
    findings: list[ResearchFinding]
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "version_bump": self.version_bump,
            "reason": self.reason,
            "findings": [f.to_dict() for f in self.findings],
            "created_at": self.created_at,
        }
 class ArchitectWeeklyResearchAgent:
    DEFAULT_TARGETS = (
        ResearchTarget("agent-frameworks", "Agent framework versions", "web", "latest multi-agent orchestration framework releases"),
        ResearchTarget("model-ecosystem", "Model ecosystem changes", "web", "latest LLM model releases tool calling agent features"),
        ResearchTarget("github-top-agentops", "GitHub top AgentOps implementations", "github", "top repositories agent observability multi agent orchestration"),
        ResearchTarget("security-agent-risks", "Agent security changes", "web", "latest AI agent security sandbox prompt injection tool use risks"),
    )
    def plan(self) -> dict[str, Any]:
        return {
            "status": "planned",
            "cadence": "weekly",
            "targets": [target.to_dict() for target in self.DEFAULT_TARGETS],
            "created_at": utc_now(),
        }
    def normalize_finding(
        self,
        target_id: str,
        title: str,
        summary: str,
        source: str,
        relevance: str = "medium",
        impact: str = "none",
        recommendation: str = "monitor",
    ) -> ResearchFinding:
        return ResearchFinding(
            id=f"finding_{uuid4().hex[:12]}",
            target_id=target_id,
            title=title,
            summary=summary,
            source=source,
            relevance=relevance,
            impact=impact,
            recommendation=recommendation,
        )
    def decide(self, findings: list[ResearchFinding]) -> ResearchDecision:
        if any(f.impact == "breaking" for f in findings):
            return ResearchDecision("action_required", "V", "breaking external ecosystem change", findings)
        if any(f.impact in {"architecture", "new_capability"} for f in findings):
            return ResearchDecision("optimization_recommended", "R", "external capability may improve architecture", findings)
        if any(f.impact in {"bugfix", "minor"} for f in findings):
            return ResearchDecision("minor_change_recommended", "C", "minor compatible improvement", findings)
        return ResearchDecision("no_action", "none", "no relevant external change", findings)
 __all__ = ["ResearchTarget", "ResearchFinding", "ResearchDecision", "ArchitectWeeklyResearchAgent"]
@@ -1,32 +0,0 @@
 """API for architecture change governance dashboard."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .architecture_change_dashboard import ArchitectureChangeDashboard
 dashboard = ArchitectureChangeDashboard()
 def get_architecture_change_overview():
    return dashboard.overview()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/architecture-changes", tags=["architecture-changes"])
    @router.get("/overview", operation_id="getArchitectureChangeOverview")
    def overview():
        return get_architecture_change_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_architecture_change_overview"]
@@ -1,90 +0,0 @@
 """Architecture change classification.
 Small compatible architecture changes may be handled automatically by the
 Architect Agent with audit evidence. Large, risky, production-impacting or
 breaking architecture changes require explicit Owner approval before
 implementation.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from typing import Any
@dataclass(frozen=True)
 class ArchitectureChangeClassification:
    level: str
    owner_required: bool
    auto_implement_allowed: bool
    version_bump: str
    reasons: list[str] = field(default_factory=list)
    required_gates: list[str] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectureChangeClassifier:
    """Classifies architecture changes into small/medium/large governance paths."""
    LARGE_FLAGS = {
        "breaking_change",
        "breaks_architecture",
        "production_architecture_update",
        "security_boundary_change",
        "cross_tenant_effect",
        "cross_user_data_effect",
        "data_migration",
        "permission_model_change",
        "runtime_infrastructure_change",
    }
    def classify(self, change: dict[str, Any]) -> ArchitectureChangeClassification:
        risk = str(change.get("risk", "medium")).lower()
        version_bump = str(change.get("version_bump", change.get("recommended_version_bump", "C"))).upper()
        scope = str(change.get("scope", "component")).lower()
        reasons: list[str] = []
        for flag in sorted(self.LARGE_FLAGS):
            if change.get(flag):
                reasons.append(flag)
        if risk in {"high", "critical"}:
            reasons.append(f"risk={risk}")
        if version_bump == "V":
            reasons.append("version_bump=V")
        if version_bump == "R" and scope in {"system", "platform", "production", "cross_agent", "cross_tenant"}:
            reasons.append(f"version_bump=R scope={scope}")
        if reasons:
            return ArchitectureChangeClassification(
                level="large",
                owner_required=True,
                auto_implement_allowed=False,
                version_bump=version_bump,
                reasons=reasons,
                required_gates=["architect_review", "owner_architecture_approval", "test_gate", "promotion_gate"],
            )
        if risk == "low" and version_bump in {"C", "NONE"} and scope in {"component", "local", "documentation", "test"}:
            return ArchitectureChangeClassification(
                level="small",
                owner_required=False,
                auto_implement_allowed=True,
                version_bump=version_bump,
                reasons=["low_risk_compatible_change"],
                required_gates=["architect_review", "audit", "test_gate"],
            )
        return ArchitectureChangeClassification(
            level="medium",
            owner_required=False,
            auto_implement_allowed=False,
            version_bump=version_bump,
            reasons=["requires_architect_review_before_implementation"],
            required_gates=["architect_review", "test_gate", "audit"],
        )
 __all__ = ["ArchitectureChangeClassification", "ArchitectureChangeClassifier"]
@@ -1,93 +0,0 @@
 """Architecture change governance dashboard.
 Displays architecture change classification, owner approval requirements,
 implementation readiness, version recommendation and audit evidence for Web UI.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 from .architecture_change_classifier import ArchitectureChangeClassifier
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectureChangeView:
    change_id: str
    title: str
    level: str
    owner_required: bool
    auto_implement_allowed: bool
    version_bump: str
    status: str
    reasons: list[str] = field(default_factory=list)
    required_gates: list[str] = field(default_factory=list)
    owner_decision: str | None = None
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectureChangeDashboard:
    def __init__(self, classifier: ArchitectureChangeClassifier | None = None):
        self.classifier = classifier or ArchitectureChangeClassifier()
    def overview(self, changes: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        changes = changes or []
        views = [self._view(change) for change in changes]
        return {
            "module": "architecture-change-governance",
            "route": "/admin/ai-evolution/architecture-changes",
            "title": "架构变更治理中心",
            "status": self._status(views),
            "generated_at": utc_now(),
            "summary": self._summary(views),
            "changes": [view.to_dict() for view in views],
        }
    def _view(self, change: dict[str, Any]) -> ArchitectureChangeView:
        classification = self.classifier.classify(change)
        owner_decision = change.get("owner_decision")
        if classification.owner_required and owner_decision != "approved":
            status = "waiting_owner_approval"
        elif classification.auto_implement_allowed:
            status = "auto_implementation_allowed"
        else:
            status = "ready_after_required_gates"
        return ArchitectureChangeView(
            change_id=change.get("change_id", change.get("id", "unknown")),
            title=change.get("title", "architecture change"),
            level=classification.level,
            owner_required=classification.owner_required,
            auto_implement_allowed=classification.auto_implement_allowed,
            version_bump=classification.version_bump,
            status=status,
            reasons=classification.reasons,
            required_gates=classification.required_gates,
            owner_decision=owner_decision,
        )
    def _status(self, views: list[ArchitectureChangeView]) -> str:
        if any(view.status == "waiting_owner_approval" for view in views):
            return "attention_required"
        return "healthy"
    def _summary(self, views: list[ArchitectureChangeView]) -> dict[str, Any]:
        return {
            "total": len(views),
            "small": len([v for v in views if v.level == "small"]),
            "medium": len([v for v in views if v.level == "medium"]),
            "large": len([v for v in views if v.level == "large"]),
            "waiting_owner_approval": len([v for v in views if v.status == "waiting_owner_approval"]),
            "auto_allowed": len([v for v in views if v.auto_implement_allowed]),
            "version_bumps": sorted({v.version_bump for v in views if v.version_bump and v.version_bump != "NONE"}),
        }
 __all__ = ["ArchitectureChangeView", "ArchitectureChangeDashboard"]
@@ -1,89 +0,0 @@
 """Owner approval gate for architecture changes.
 Architecture changes may be proposed, reviewed and optimized by agents, but they
 must not be implemented until owner approval is recorded. This gate sits before
 implementation and before promotion.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectureOwnerApprovalRequest:
    request_id: str
    change_id: str
    title: str
    reason: str
    risk: str
    architect_decision: str
    status: str = "pending_owner_approval"
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class ArchitectureOwnerDecision:
    request_id: str
    change_id: str
    decision: str
    decided_by: str
    reason: str = ""
    decided_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class ArchitectureOwnerGate:
    """Blocks architecture implementation until owner explicitly approves."""
    def create_request(self, change: dict[str, Any], architect_decision: dict[str, Any]) -> ArchitectureOwnerApprovalRequest:
        return ArchitectureOwnerApprovalRequest(
            request_id=f"ARCH-OWNER-{change.get('change_id', change.get('id', 'unknown'))}",
            change_id=change.get("change_id", change.get("id", "unknown")),
            title=change.get("title", "architecture change"),
            reason=change.get("reason", "architecture impact requires owner approval"),
            risk=change.get("risk", "medium"),
            architect_decision=architect_decision.get("status", "reviewed"),
        )
    def can_implement(self, request: dict[str, Any], decision: dict[str, Any] | None = None) -> dict[str, Any]:
        if not decision:
            return {
                "allowed": False,
                "status": "waiting_owner_approval",
                "reason": "architecture change requires owner approval before implementation",
                "request": request,
            }
        if decision.get("decision") == "approved":
            return {
                "allowed": True,
                "status": "owner_approved",
                "reason": decision.get("reason", "approved by owner"),
                "request": request,
                "decision": decision,
            }
        return {
            "allowed": False,
            "status": "owner_rejected",
            "reason": decision.get("reason", "rejected by owner"),
            "request": request,
            "decision": decision,
        }
 __all__ = [
    "ArchitectureOwnerApprovalRequest",
    "ArchitectureOwnerDecision",
    "ArchitectureOwnerGate",
 ]
@@ -1,79 +0,0 @@
 """HTTP handler and display data for architecture owner gate."""
 from __future__ import annotations
 from dataclasses import asdict
 from datetime import datetime, timezone
 from typing import Any
 from .architecture_owner_gate import ArchitectureOwnerGate, ArchitectureOwnerApprovalRequest
 gate = ArchitectureOwnerGate()
 # In-memory store for pending approval requests
 _pending_requests: dict[str, dict[str, Any]] = {}
 _decisions: dict[str, dict[str, Any]] = {}
 def create_owner_approval_request(change: dict[str, Any], architect_decision: dict[str, Any]) -> dict[str, Any]:
    req = gate.create_request(change, architect_decision)
    d = asdict(req)
    _pending_requests[req.change_id] = d
    return d
 def approve_request(change_id: str, decided_by: str = "owner", reason: str = "") -> dict[str, Any]:
    req = _pending_requests.get(change_id)
    if not req:
        return {"error": f"no pending request for change_id={change_id}"}
    decision = asdict(
        ArchitectureOwnerApprovalRequest.__bases__[1].__new__(type("_ArchitectureOwnerDecision", (), {}))  # noqa
    )
    try:
        from .architecture_owner_gate import ArchitectureOwnerDecision
        dec = ArchitectureOwnerDecision(
            request_id=req["request_id"],
            change_id=change_id,
            decision="approved",
            decided_by=decided_by,
            reason=reason,
        )
        d = asdict(dec)
        _decisions[change_id] = d
        # remove from pending
        _pending_requests.pop(change_id, None)
        return {"allowed": True, "decision": d}
    except Exception as exc:
        return {"error": str(exc)}
 def reject_request(change_id: str, decided_by: str = "owner", reason: str = "") -> dict[str, Any]:
    from .architecture_owner_gate import ArchitectureOwnerDecision
    req = _pending_requests.get(change_id)
    if not req:
        return {"error": f"no pending request for change_id={change_id}"}
    try:
        dec = ArchitectureOwnerDecision(
            request_id=req["request_id"],
            change_id=change_id,
            decision="rejected",
            decided_by=decided_by,
            reason=reason,
        )
        d = asdict(dec)
        _decisions[change_id] = d
        _pending_requests.pop(change_id, None)
        return {"allowed": False, "decision": d}
    except Exception as exc:
        return {"error": str(exc)}
 def get_owner_gate_overview() -> dict[str, Any]:
    return {
        "pending_requests": list(_pending_requests.values()),
        "decisions": list(_decisions.values()),
        "pending_count": len(_pending_requests),
        "approved_count": sum(1 for d in _decisions.values() if d.get("decision") == "approved"),
        "rejected_count": sum(1 for d in _decisions.values() if d.get("decision") == "rejected"),
    }
@@ -1,40 +0,0 @@
 """Independent architecture optimization agent."""
 from dataclasses import dataclass, asdict
 from datetime import datetime, timezone
 def utc_now():
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ArchitectureReview:
    agent: str
    decision: str
    findings: list[str]
    created_at: str = utc_now()
    def to_dict(self):
        return asdict(self)
 class ArchitectureReviewAgent:
    """Reviews architecture changes from system-wide optimal perspective."""
    def review(self, proposal: dict) -> ArchitectureReview:
        findings = [
            "dependency_impact_checked",
            "security_boundary_checked",
            "scalability_checked",
            "operability_checked",
            "rollback_strategy_checked",
        ]
        return ArchitectureReview(
            "architecture-optimization-agent",
            "pass" if proposal.get("architecture_change") else "not_required",
            findings,
        )
 __all__ = ["ArchitectureReviewAgent", "ArchitectureReview"]
@@ -0,0 +1,38 @@
 """Autonomous Controller health adapter."""
 from __future__ import annotations
 from typing import Any
 from .blocker_task_adapter import BlockerTaskAdapter
 from .git_permission_watchdog import GitPermissionWatchdog
 from .system_blocker_memory import SystemBlockerMemory
 from .system_health_center import SystemHealthCenter
 class AutonomousHealthAdapter:
    def __init__(self):
        self.git = GitPermissionWatchdog()
        self.memory = SystemBlockerMemory()
        self.task_adapter = BlockerTaskAdapter()
        self.health = SystemHealthCenter()
    def check(self, previous_blockers: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        git_report = self.git.scan().to_dict()
        blockers = list(previous_blockers or [])
        if git_report.get("status") != "healthy":
            blockers.append({
                "blocker_id": "git-permission",
                "category": "git_permission",
                "title": "Git object permission drift",
                "severity": "p0",
                "evidence": [str(git_report)],
            })
        return self.health.build(git_report=git_report, blockers=blockers)
    def optimize_task(self, blocker: dict[str, Any]) -> dict[str, Any]:
        decision = self.memory.decide(blocker)
        return self.task_adapter.to_task(blocker, decision.to_dict())
 __all__ = ["AutonomousHealthAdapter"]
@@ -0,0 +1,64 @@
 """Convert recurring blockers into Task Center tasks.
 When a blocker becomes recurrent or critical, it should not remain only as an
 incident log. It becomes a visible task with owner/executor/progress/approval
 state in the Task Center.
 """
 from __future__ import annotations
 from typing import Any
 from datetime import datetime, timezone
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class BlockerTaskAdapter:
    def to_task(self, blocker: dict[str, Any], decision: dict[str, Any]) -> dict[str, Any]:
        priority = decision.get("priority", "P2")
        status = "waiting_approval" if decision.get("owner_gate_required") else "pending"
        if priority == "P0":
            status = "running"
        return {
            "task_id": f"blocker-{blocker.get('fingerprint', blocker.get('blocker_id', 'unknown'))}",
            "title": f"Resolve blocker: {blocker.get('title', 'unknown')}",
            "assigned_by": "system-blocker-memory",
            "executor": decision.get("assigned_to", "autonomous-controller"),
            "source": "system_blocker",
            "status": status,
            "priority": priority,
            "created_at": utc_now(),
            "updated_at": utc_now(),
            "progress": {
                "percent": 20 if status == "running" else 0,
                "current_step": decision.get("action", "record_and_monitor"),
                "completed_steps": ["blocker_recorded", "decision_created"],
                "remaining_steps": self._remaining_steps(decision),
                "blocked_reason": blocker.get("title") if status == "waiting_approval" else None,
            },
            "notification": {
                "required": True,
                "sent": False,
                "channel": "wechat",
            },
            "owner_gate_required": bool(decision.get("owner_gate_required")),
            "owner_gate_status": "pending" if decision.get("owner_gate_required") else None,
            "evidence": blocker.get("evidence", []),
            "metadata": {
                "blocker": blocker,
                "decision": decision,
            },
        }
    def _remaining_steps(self, decision: dict[str, Any]) -> list[str]:
        steps = ["root_cause_analysis", "patch_or_repair", "test", "deploy", "verify", "close"]
        if decision.get("architect_review_required"):
            steps.insert(1, "architect_review")
        if decision.get("owner_gate_required"):
            steps.insert(2, "owner_approval")
        return steps
 __all__ = ["BlockerTaskAdapter"]
@@ -1,103 +0,0 @@
 """Change acceptance governance for AIOS.
 Every feature change must provide matching tests. Every architecture change must
 be reviewed by an independent architecture agent. User-facing changes must be
 validated by an independent user acceptance agent.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class ChangeDescriptor:
    change_id: str
    change_type: str
    title: str
    files_changed: list[str]
    tests: list[str] = field(default_factory=list)
    user_facing: bool = False
    architecture_impact: bool = False
    risk: str = "low"
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class AcceptanceFinding:
    check: str
    status: str
    detail: str
    required: bool = True
    def passed(self) -> bool:
        return self.status == "pass" or not self.required
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class AcceptanceDecision:
    status: str
    findings: list[AcceptanceFinding]
    decided_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "findings": [f.to_dict() for f in self.findings],
            "decided_at": self.decided_at,
        }
 class ChangeAcceptanceGate:
    def evaluate(self, change: ChangeDescriptor) -> AcceptanceDecision:
        findings: list[AcceptanceFinding] = []
        findings.append(self._tests_required(change))
        findings.append(self._user_acceptance_required(change))
        findings.append(self._architecture_review_required(change))
        findings.append(self._risk_gate(change))
        status = "pass" if all(f.passed() for f in findings) else "blocked"
        return AcceptanceDecision(status, findings)
    def _tests_required(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if change.tests:
            return AcceptanceFinding("tests_present", "pass", "matching tests declared")
        return AcceptanceFinding("tests_present", "fail", "every feature change requires tests")
    def _user_acceptance_required(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if not change.user_facing:
            return AcceptanceFinding("user_acceptance", "pass", "not user-facing")
        has_user_test = any("user" in t.lower() or "acceptance" in t.lower() for t in change.tests)
        return AcceptanceFinding(
            "user_acceptance",
            "pass" if has_user_test else "fail",
            "user-facing change requires user acceptance test",
        )
    def _architecture_review_required(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if not change.architecture_impact:
            return AcceptanceFinding("architecture_review", "pass", "no architecture impact")
        has_arch_test = any("arch" in t.lower() or "architecture" in t.lower() for t in change.tests)
        return AcceptanceFinding(
            "architecture_review",
            "pass" if has_arch_test else "fail",
            "architecture change requires independent architect review artifact/test",
        )
    def _risk_gate(self, change: ChangeDescriptor) -> AcceptanceFinding:
        if change.risk in {"low", "medium", "high"}:
            return AcceptanceFinding("risk_classified", "pass", f"risk={change.risk}")
        return AcceptanceFinding("risk_classified", "fail", "risk must be low/medium/high")
 __all__ = ["ChangeDescriptor", "AcceptanceFinding", "AcceptanceDecision", "ChangeAcceptanceGate"]
@@ -1,34 +0,0 @@
 """API for AIOS change governance dashboard."""
 from typing import Any
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .change_governance_dashboard import ChangeGovernanceDashboard
 dashboard = ChangeGovernanceDashboard()
 def get_change_governance_overview(changes: list[dict[str, Any]] | None = None):
    return dashboard.overview(changes)
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/governance", tags=["change-governance"])
    @router.get("/overview", operation_id="getChangeGovernanceOverview")
    def overview():
        return get_change_governance_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_change_governance_overview"]
@@ -1,93 +0,0 @@
 """Web UI dashboard model for AIOS change governance.
 Shows test gates, user acceptance, architecture review, risk gates and release
 blocking status for human operators and owner review.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class GovernanceMetric:
    key: str
    title: str
    value: Any
    status: str = "ok"
    detail: str = ""
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class GovernancePanel:
    key: str
    title: str
    metrics: list[GovernanceMetric] = field(default_factory=list)
    def to_dict(self) -> dict[str, Any]:
        return {
            "key": self.key,
            "title": self.title,
            "metrics": [m.to_dict() for m in self.metrics],
        }
 class ChangeGovernanceDashboard:
    def overview(self, changes: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        changes = changes or []
        blocked = [c for c in changes if c.get("status") == "blocked"]
        waiting_arch = [c for c in changes if c.get("architecture_review") == "required"]
        waiting_user = [c for c in changes if c.get("user_acceptance") == "required"]
        missing_tests = [c for c in changes if not c.get("tests")]
        panels = [
            GovernancePanel(
                "test_gate",
                "测试 Gate",
                [
                    GovernanceMetric("total_changes", "变更总数", len(changes)),
                    GovernanceMetric("missing_tests", "缺少测试的变更", len(missing_tests), "warn" if missing_tests else "ok"),
                ],
            ),
            GovernancePanel(
                "user_acceptance",
                "用户验收 Agent",
                [
                    GovernanceMetric("waiting_user_acceptance", "等待用户验收", len(waiting_user), "warn" if waiting_user else "ok"),
                ],
            ),
            GovernancePanel(
                "architecture_review",
                "架构师 Agent",
                [
                    GovernanceMetric("waiting_architecture_review", "等待架构审查", len(waiting_arch), "warn" if waiting_arch else "ok"),
                ],
            ),
            GovernancePanel(
                "release_gate",
                "发布 Gate",
                [
                    GovernanceMetric("blocked_changes", "阻塞发布的变更", len(blocked), "warn" if blocked else "ok"),
                ],
            ),
        ]
        return {
            "module": "change-governance",
            "route": "/admin/ai-evolution/governance",
            "title": "AIOS 变更治理中心",
            "status": "attention_required" if blocked or missing_tests or waiting_arch or waiting_user else "healthy",
            "generated_at": utc_now(),
            "panels": [p.to_dict() for p in panels],
            "changes": changes,
        }
 __all__ = ["GovernanceMetric", "GovernancePanel", "ChangeGovernanceDashboard"]
@@ -0,0 +1,90 @@
 """Command watcher stall detector.
 Detects commands that remain in active/running states beyond the configured
 threshold and emits blocker records so AIOS can self-optimize instead of waiting
 for manual progress checks.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 def parse_ts(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except Exception:
        return None
@dataclass(frozen=True)
 class CommandStallRecord:
    command_id: str
    path: str
    bucket: str
    age_minutes: float
    threshold_minutes: int
    severity: str
    detected_at: str = field(default_factory=utc_now)
    def to_blocker(self) -> dict[str, Any]:
        return {
            "blocker_id": f"command-stall-{self.command_id}",
            "category": "command_stuck",
            "title": f"Command stalled: {self.command_id}",
            "source": "command-stall-detector",
            "severity": self.severity,
            "status": "open",
            "first_seen_at": self.detected_at,
            "last_seen_at": self.detected_at,
            "occurrence_count": 1,
            "evidence": [self.path],
            "metadata": self.to_dict(),
        }
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class CommandStallDetector:
    def __init__(self, root: str = "agents/commands", threshold_minutes: int = 15):
        self.root = Path(root)
        self.threshold_minutes = threshold_minutes
    def detect(self) -> list[CommandStallRecord]:
        records: list[CommandStallRecord] = []
        for bucket in ("active", "running"):
            bucket_path = self.root / bucket
            if not bucket_path.exists():
                continue
            for path in bucket_path.glob("*.md"):
                age = self._age_minutes(path)
                if age >= self.threshold_minutes:
                    records.append(
                        CommandStallRecord(
                            command_id=path.stem,
                            path=str(path),
                            bucket=bucket,
                            age_minutes=round(age, 2),
                            threshold_minutes=self.threshold_minutes,
                            severity="p0" if bucket == "running" else "high",
                        )
                    )
        return records
    def _age_minutes(self, path: Path) -> float:
        stat = path.stat()
        age_seconds = datetime.now(timezone.utc).timestamp() - stat.st_mtime
        return age_seconds / 60
 __all__ = ["CommandStallRecord", "CommandStallDetector"]
@@ -0,0 +1,141 @@
 """Git object permission watchdog and self-healing.
 Prevents AIOS autonomous operation from being blocked by recurring
 `.git/objects` ownership drift. The repair is intentionally conservative and
 limited to the workspace Git object database.
 """
 from __future__ import annotations
 import os
 import pwd
 import grp
 import subprocess
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class GitPermissionIssue:
    path: str
    issue: str
    actual_uid: int | None = None
    actual_gid: int | None = None
    expected_user: str = "aiman"
    expected_group: str = "gitops"
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
@dataclass(frozen=True)
 class GitPermissionReport:
    workspace: str
    expected_user: str
    expected_group: str
    status: str
    issues: list[GitPermissionIssue] = field(default_factory=list)
    checked_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return {
            "workspace": self.workspace,
            "expected_user": self.expected_user,
            "expected_group": self.expected_group,
            "status": self.status,
            "issues": [issue.to_dict() for issue in self.issues],
            "checked_at": self.checked_at,
        }
 class GitPermissionWatchdog:
    def __init__(
        self,
        workspace: str = "/data/aios-gitops/workspace",
        expected_user: str = "aiman",
        expected_group: str = "gitops",
        max_scan_depth: int = 2,
    ):
        self.workspace = Path(workspace)
        self.expected_user = expected_user
        self.expected_group = expected_group
        self.max_scan_depth = max_scan_depth
    @property
    def objects_dir(self) -> Path:
        return self.workspace / ".git" / "objects"
    def scan(self) -> GitPermissionReport:
        issues: list[GitPermissionIssue] = []
        if not self.objects_dir.exists():
            issues.append(GitPermissionIssue(str(self.objects_dir), "missing_objects_dir"))
            return GitPermissionReport(str(self.workspace), self.expected_user, self.expected_group, "blocked", issues)
        expected_uid = pwd.getpwnam(self.expected_user).pw_uid
        expected_gid = grp.getgrnam(self.expected_group).gr_gid
        for path in self._walk_limited(self.objects_dir):
            try:
                stat = path.stat()
            except FileNotFoundError:
                continue
            if stat.st_uid != expected_uid:
                issues.append(GitPermissionIssue(str(path), "owner_drift", stat.st_uid, stat.st_gid, self.expected_user, self.expected_group))
            if stat.st_gid != expected_gid:
                issues.append(GitPermissionIssue(str(path), "group_drift", stat.st_uid, stat.st_gid, self.expected_user, self.expected_group))
            if not os.access(path, os.W_OK):
                issues.append(GitPermissionIssue(str(path), "not_writable", stat.st_uid, stat.st_gid, self.expected_user, self.expected_group))
            if len(issues) >= 200:
                break
        return GitPermissionReport(
            workspace=str(self.workspace),
            expected_user=self.expected_user,
            expected_group=self.expected_group,
            status="healthy" if not issues else "drift_detected",
            issues=issues,
        )
    def repair(self, dry_run: bool = False) -> dict[str, Any]:
        report = self.scan()
        if report.status == "healthy":
            return {"status": "healthy", "report": report.to_dict(), "repaired": False}
        if dry_run:
            return {"status": "dry_run", "report": report.to_dict(), "repaired": False}
        commands = [
            ["chown", "-R", f"{self.expected_user}:{self.expected_group}", str(self.objects_dir)],
            ["chmod", "-R", "u+rwX,g+rwX", str(self.objects_dir)],
        ]
        results = []
        for command in commands:
            proc = subprocess.run(command, text=True, capture_output=True, check=False)
            results.append({"cmd": command, "returncode": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr})
            if proc.returncode != 0:
                return {"status": "repair_failed", "report": report.to_dict(), "results": results, "repaired": False}
        post = self.scan()
        return {"status": "repaired" if post.status == "healthy" else "repair_incomplete", "report": post.to_dict(), "results": results, "repaired": post.status == "healthy"}
    def write_probe(self) -> dict[str, Any]:
        probe = self.objects_dir / f".aios-write-probe-{datetime.now(timezone.utc).timestamp()}"
        try:
            probe.write_text("probe\n", encoding="utf-8")
            probe.unlink(missing_ok=True)
            return {"status": "ok", "path": str(probe)}
        except Exception as exc:  # pragma: no cover - depends on runtime perms
            return {"status": "failed", "path": str(probe), "error": str(exc)}
    def _walk_limited(self, root: Path):
        yield root
        base_depth = len(root.parts)
        for path in root.rglob("*"):
            if len(path.parts) - base_depth <= self.max_scan_depth:
                yield path
 __all__ = ["GitPermissionIssue", "GitPermissionReport", "GitPermissionWatchdog"]
@@ -0,0 +1,31 @@
 """CLI entrypoint for Git Permission Watchdog."""
 from __future__ import annotations
 import json
 import os
 from pathlib import Path
 from .git_permission_watchdog import GitPermissionWatchdog
 def main() -> int:
    workspace = os.getenv("AIOS_GITOPS_WORKSPACE", "/data/aios-gitops/workspace")
    expected_user = os.getenv("AIOS_GIT_EXPECTED_USER", "aiman")
    expected_group = os.getenv("AIOS_GIT_EXPECTED_GROUP", "gitops")
    audit_dir = Path(os.getenv("AIOS_GIT_PERMISSION_AUDIT_DIR", "runtime/git-permission-watchdog"))
    audit_dir.mkdir(parents=True, exist_ok=True)
    watchdog = GitPermissionWatchdog(workspace, expected_user, expected_group)
    result = watchdog.repair(dry_run=os.getenv("AIOS_GIT_PERMISSION_DRY_RUN") == "1")
    probe = watchdog.write_probe()
    payload = {"repair": result, "write_probe": probe}
    status = result.get("status")
    out = audit_dir / "latest.json"
    out.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(payload, ensure_ascii=False))
    return 0 if status in {"healthy", "repaired"} and probe.get("status") == "ok" else 2
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -1,36 +0,0 @@
 """Git repository location resolver with validation.
 Prevents hard-coded repository names from silently pointing to stale paths.
 """
 from __future__ import annotations
 import os
 from pathlib import Path
 class GitRepoLocator:
    def __init__(self, root: str | None = None):
        self.root = Path(root or os.getenv("AIOS_GITOPS_ROOT", "/data/aios-gitops/runtime/gitea-bin/data/repositories"))
    def candidates(self, owner: str = "aios", names: tuple[str, ...] = ("aios.git", "aios-v23.git")):
        return [self.root / owner / name for name in names]
    def locate(self, owner: str = "aios") -> dict:
        found = [str(path) for path in self.candidates(owner) if path.exists()]
        return {
            "root": str(self.root),
            "owner": owner,
            "found": found,
            "healthy": len(found) == 1,
            "warning": "multiple repositories found" if len(found) > 1 else None,
        }
    def require(self, owner: str = "aios") -> Path:
        result = self.locate(owner)
        if not result["healthy"]:
            raise RuntimeError(f"Git repo resolution failed: {result}")
        return Path(result["found"][0])
 __all__ = ["GitRepoLocator"]
@@ -1,148 +0,0 @@
 """Bridge governance agents into the unified Agent Event Bus.
 This module emits normalized governance events for requirement changes,
 architecture reviews, user acceptance, test gates and version decisions. It uses
 duck typing so it can work with the unified AgentEventBus when available, while
 remaining testable without a running bus.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Protocol
 from uuid import uuid4
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class EventPublisher(Protocol):
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        ...
@dataclass(frozen=True)
 class GovernanceEventEnvelope:
    id: str
    kind: str
    source: str
    target: str
    summary: str
    payload: dict[str, Any]
    correlation_id: str | None = None
    created_at: str = field(default_factory=utc_now)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class InMemoryGovernanceEventBus:
    """Small fallback bus for tests and offline governance runs."""
    def __init__(self) -> None:
        self.events: list[dict[str, Any]] = []
    def publish(
        self,
        kind: str,
        source: str,
        target: str,
        summary: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        event = GovernanceEventEnvelope(
            id=f"evt_{uuid4().hex[:12]}",
            kind=kind,
            source=source,
            target=target,
            summary=summary,
            payload=payload,
            correlation_id=correlation_id,
        ).to_dict()
        self.events.append(event)
        return event
 class GovernanceEventBridge:
    """Publishes governance lifecycle events to the Agent Event Bus."""
    def __init__(self, bus: EventPublisher | None = None):
        self.bus = bus or InMemoryGovernanceEventBus()
    def requirement_changed(self, requirement: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="requirement_changed",
            source=requirement.get("source", "unknown"),
            target="architect-agent",
            summary=requirement.get("requirement", "requirement changed"),
            payload=requirement,
            correlation_id=requirement.get("id"),
        )
    def architecture_review_started(self, proposal: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="architecture_review_started",
            source="architect-agent",
            target="architecture-review-agent",
            summary=proposal.get("title", "architecture review started"),
            payload=proposal,
            correlation_id=proposal.get("change_id"),
        )
    def architecture_review_completed(self, decision: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="architecture_review_completed",
            source="architecture-review-agent",
            target="change-governance",
            summary=f"architecture review {decision.get('status', 'completed')}",
            payload=decision,
            correlation_id=decision.get("change_id"),
        )
    def user_acceptance_completed(self, report: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="user_acceptance_completed",
            source="user-acceptance-agent",
            target="change-governance",
            summary=f"user acceptance {report.get('decision', 'completed')}",
            payload=report,
            correlation_id=report.get("change_id"),
        )
    def test_gate_completed(self, result: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="test_gate_completed",
            source="test-gate",
            target="change-governance",
            summary=f"test gate {result.get('status', 'completed')}",
            payload=result,
            correlation_id=result.get("change_id"),
        )
    def version_decision_created(self, decision: dict[str, Any]) -> dict[str, Any]:
        return self.bus.publish(
            kind="version_decision_created",
            source="architect-agent",
            target="owner-gate",
            summary=f"version bump recommendation: {decision.get('version_bump', 'unknown')}",
            payload=decision,
            correlation_id=decision.get("change_id"),
        )
 __all__ = [
    "GovernanceEventEnvelope",
    "InMemoryGovernanceEventBus",
    "GovernanceEventBridge",
 ]
@@ -1,32 +0,0 @@
 """API for governance timeline visualization."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .governance_timeline_dashboard import GovernanceTimelineDashboard
 dashboard = GovernanceTimelineDashboard()
 def get_governance_timeline(events=None):
    return dashboard.build(events)
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/governance", tags=["governance-timeline"])
    @router.get("/timeline", operation_id="getGovernanceTimeline")
    def timeline():
        return get_governance_timeline()
    return router
 router = build_router()
 __all__ = ["router", "get_governance_timeline"]
@@ -1,106 +0,0 @@
 """Governance timeline dashboard.
 Turns governance events into a human-readable flow for Web UI so operators can
 see why a change is blocked or ready for promotion.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 EVENT_ORDER = {
    "requirement_changed": 10,
    "architecture_review_started": 20,
    "architecture_review_completed": 30,
    "test_gate_completed": 40,
    "user_acceptance_completed": 50,
    "version_decision_created": 60,
    "owner_gate_completed": 70,
    "promotion_completed": 80,
 }
@dataclass(frozen=True)
 class GovernanceTimelineStep:
    id: str
    kind: str
    title: str
    actor: str
    status: str
    timestamp: str
    correlation_id: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class GovernanceTimelineDashboard:
    def build(self, events: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        events = events or []
        steps = [self._event_to_step(event) for event in events]
        steps.sort(key=lambda step: (EVENT_ORDER.get(step.kind, 999), step.timestamp))
        blockers = [step for step in steps if step.status in {"blocked", "fail", "attention_required"}]
        return {
            "module": "governance-timeline",
            "route": "/admin/ai-evolution/governance/timeline",
            "title": "治理事件时间线",
            "status": "attention_required" if blockers else "healthy",
            "generated_at": utc_now(),
            "steps": [step.to_dict() for step in steps],
            "blockers": [step.to_dict() for step in blockers],
            "flow": self._flow_summary(steps),
        }
    def _event_to_step(self, event: dict[str, Any]) -> GovernanceTimelineStep:
        payload = event.get("payload", {})
        kind = event.get("kind", "unknown")
        return GovernanceTimelineStep(
            id=event.get("id", f"event-{len(str(event))}"),
            kind=kind,
            title=self._title(kind, payload),
            actor=event.get("source", payload.get("actor", "unknown")),
            status=payload.get("status") or payload.get("decision") or event.get("status", "ok"),
            timestamp=event.get("created_at") or payload.get("created_at") or utc_now(),
            correlation_id=event.get("correlation_id") or payload.get("change_id") or payload.get("id"),
            detail=payload,
        )
    def _title(self, kind: str, payload: dict[str, Any]) -> str:
        titles = {
            "requirement_changed": "需求变化",
            "architecture_review_started": "架构审查开始",
            "architecture_review_completed": "架构审查完成",
            "test_gate_completed": "测试 Gate 完成",
            "user_acceptance_completed": "用户验收完成",
            "version_decision_created": "版本建议产生",
            "owner_gate_completed": "Owner Gate 完成",
            "promotion_completed": "Promotion 完成",
        }
        return payload.get("title") or payload.get("summary") or titles.get(kind, kind)
    def _flow_summary(self, steps: list[GovernanceTimelineStep]) -> dict[str, Any]:
        present = {step.kind for step in steps}
        required = [
            "requirement_changed",
            "architecture_review_completed",
            "test_gate_completed",
            "user_acceptance_completed",
            "version_decision_created",
        ]
        missing = [kind for kind in required if kind not in present]
        return {
            "required_steps": required,
            "missing_steps": missing,
            "ready_for_owner_gate": not missing,
        }
 __all__ = ["GovernanceTimelineStep", "GovernanceTimelineDashboard"]
@@ -1,59 +0,0 @@
 """Independent acceptance agents.
 Simulates independent roles: user representative and architecture reviewer.
 They provide acceptance evidence, not implementation decisions.
 """
 from dataclasses import dataclass, asdict
 from datetime import datetime, timezone
 from typing import Any
 def utc_now():
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class AcceptanceReport:
    agent: str
    role: str
    decision: str
    evidence: list[str]
    created_at: str = utc_now()
    def to_dict(self):
        return asdict(self)
 class UserAcceptanceAgent:
    def review(self, change: dict[str, Any]) -> AcceptanceReport:
        checks = [
            "user_requirement_match",
            "user_workflow_validation",
            "permission_visibility_check",
        ]
        return AcceptanceReport(
            "user-acceptance-agent",
            "user_representative",
            "pass" if change.get("user_facing") else "not_required",
            checks,
        )
 class ArchitectureAcceptanceAgent:
    def review(self, change: dict[str, Any]) -> AcceptanceReport:
        checks = [
            "architecture_consistency",
            "scalability_review",
            "security_boundary_review",
            "operability_review",
        ]
        return AcceptanceReport(
            "architecture-review-agent",
            "system_architect",
            "pass" if change.get("architecture_impact") else "not_required",
            checks,
        )
 __all__ = ["AcceptanceReport", "UserAcceptanceAgent", "ArchitectureAcceptanceAgent"]
@@ -0,0 +1,52 @@
 """Unified AIOS operations command center view model."""
 from __future__ import annotations
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class OperationsCommandCenter:
    def build(
        self,
        tasks: list[dict[str, Any]] | None = None,
        health: dict[str, Any] | None = None,
        blockers: list[dict[str, Any]] | None = None,
        searches: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        tasks = tasks or []
        blockers = blockers or []
        searches = searches or []
        return {
            "module": "operations-command-center",
            "route": "/admin/ai-evolution/operations",
            "title": "AIOS 自治运营驾驶舱",
            "generated_at": utc_now(),
            "health": health or {},
            "tasks": {
                "active": len([t for t in tasks if t.get("status") in {"running", "pending", "waiting_approval"}]),
                "blocked": len([t for t in tasks if t.get("status") == "blocked"]),
                "completed": len([t for t in tasks if t.get("status") == "completed"]),
            },
            "blockers": {
                "total": len(blockers),
                "p0": len([b for b in blockers if b.get("severity") in {"p0", "critical"}]),
                "recurring": len([b for b in blockers if b.get("mechanism_gap")]),
            },
            "search": {
                "recent": len(searches),
            },
            "loops": {
                "task_loop": True,
                "health_loop": True,
                "optimization_loop": True,
                "notification_loop": True,
            },
        }
 __all__ = ["OperationsCommandCenter"]
@@ -58,8 +58,6 @@ pre { white-space: pre-wrap; word-break: break-word; }
      <li onclick="showPanel('versions')"><i>🏷️</i> Versions</li>
      <li onclick="showPanel('commands')"><i>⚡</i> Commands</li>
      <li onclick="showPanel('agents')"><i>🤖</i> Agent Runs</li>
      <li onclick="showPanel('arch-owner')"><i>🔐</i> Owner Gate</li>
      <li onclick="showPanel('arch-approve')"><i>✓</i> Approve Changes</li>
    </ul>
  </div>
  <div class="main">
@@ -92,14 +90,6 @@ pre { white-space: pre-wrap; word-break: break-word; }
    </div>
    <div id="panel-agents" class="panel" style="display:none;">
      <div class="main-header"><h2>Recent Agent Executions</h2></div>
    <div id="panel-arch-owner" class="panel" style="display:none;">
      <div class="main-header"><h2>🔐 Architecture Owner Gate</h2></div>
      <div id="arch-owner-content">Loading...</div>
    </div>
    <div id="panel-arch-approve" class="panel" style="display:none;">
      <div class="main-header"><h2>✓ Approve Architecture Changes</h2></div>
      <div id="arch-approve-content">Loading...</div>
    </div>
      <div id="agents-list">Loading...</div>
    </div>
  </div>
@@ -107,159 +97,3 @@ pre { white-space: pre-wrap; word-break: break-word; }
 <script>
 async function fetchData() {
  const r = await fetch('/admin/ai-evolution/api/data');
  const data = await r.json();
  document.getElementById('loading').style.display = 'none';
  document.getElementById('error').style.display = 'none';
  document.getElementById('content').style.display = '';
  const hs = document.getElementById('header-status');
  hs.className = 'status-badge ' + (data.healthy ? 'ok' : 'fail');
  hs.textContent = data.healthy ? '▪ operational' : '⚠ degraded';
  document.getElementById('base').textContent = data.baseline || '-';
  document.getElementById('branch').textContent = data.main_branch || 'main';
  document.getElementById('uptime').textContent = data.uptime || '-';
  document.getElementById('bus-status').textContent = data.bus ? JSON.stringify(data.bus, null, 2) : 'no event bus data';
 }
 async function loadVersions() {
  const el = document.getElementById('versions-list');
  el.innerHTML = '<div class="loading">Loading versions...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/versions');
    const data = await r.json();
    if (data.error) { el.innerHTML = '<div class="error">' + data.error + '</div>'; return; }
    let html = '<div class="grid">';
    for (const v of data.versions || []) {
      html += '<div class="card"><div class="card-title">' + v.tag + '</div><div class="card-value">' + (v.version || v.tag) + '</div></div>';
    }
    html += '</div>';
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load versions: ' + e.message + '</div>';
  }
 }
 async function loadCommands() {
  const el = document.getElementById('commands-list');
  el.innerHTML = '<div class="loading">Loading commands...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/data');
    const data = await r.json();
    let cmds = data.commands || [];
    let html = '<div style="display:flex;gap:24px;margin-bottom:16px;">';
    html += '<div><div class="card"><div class="card-title">Total</div><div class="card-value">' + cmds.length + '</div></div></div>';
    html += '<div><div class="card"><div class="card-title">By Category</div>';
    const cats = {};
    for (const c of cmds) { const k = c.category || 'uncategorized'; cats[k] = (cats[k] || 0) + 1; }
    for (const [k, v] of Object.entries(cats)) html += '<div style="font-size:14px;color:#8b949e;margin:4px 0;">' + k + ': ' + v + '</div>';
    html += '</div></div></div>';
    for (const cmd of cmds.slice().reverse()) {
      html += '<div class="command-item"><div class="meta">' + cmd.category + ' | ' + cmd.filename + '</div><div class="name">' + cmd.title + '</div><div style="font-size:13px;color:#c9d1d9;margin-top:4px;">' + cmd.summary + '</div></div>';
    }
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load commands: ' + e.message + '</div>';
  }
 }
 async function loadAgents() {
  const el = document.getElementById('agents-list');
  el.innerHTML = '<div class="loading">Loading agent runs...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/agents');
    const data = await r.json();
    if (data.error) { el.innerHTML = '<div class="error">' + data.error + '</div>'; return; }
    let html = '';
    for (const agent of data.agents || []) {
      html += '<div class="agent-item"><div class="agent-title">' + agent.name + ' <span class="state state-' + (agent.status === 'running' ? 'running' : (agent.status === 'ok' ? 'completed' : 'failed')) + '">' + agent.status + '</span></div><div class="agent-meta">' + (agent.runs || 0) + ' runs | last: ' + (agent.last_run || '-') + '</div></div>';
    }
    el.innerHTML = html || '<div style="padding:24px;color:#8b949e;">No agent runs recorded yet.</div>';
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load agents: ' + e.message + '</div>';
  }
 }
 function showPanel(name) {
 async function loadOwnerGate() {
  const el = document.getElementById('arch-owner-content');
  el.innerHTML = '<div class="loading">Loading owner gate data...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/owner-gate');
    const data = await r.json();
    let html = '<div class="row">';
    html += '<div class="col"><div class="card"><div class="card-title">Pending Requests</div><div class="card-value">' + data.pending_count + '</div></div></div>';
    html += '<div class="col"><div class="card"><div class="card-title">Approved</div><div class="card-value">' + data.approved_count + '</div></div></div>';
    html += '<div class="col"><div class="card"><div class="card-title">Rejected</div><div class="card-value">' + data.rejected_count + '</div></div></div>';
    html += '</div>';
    html += '<div class="card"><div class="card-title">Pending Approval Requests</div>';
    if (data.pending_requests && data.pending_requests.length) {
      for (const req of data.pending_requests) {
        html += '<div class="command-item">';
        html += '<div class="name">' + (req.title || req.change_id) + '</div>';
        html += '<div class="meta">ID: ' + req.change_id + ' | Risk: ' + req.risk + '</div>';
        html += '<div style="font-size:13px;color:#c9d1d9;margin-top:4px;">' + (req.reason || '') + '</div>';
        html += '</div>';
      }
    } else {
      html += '<div style="color:#8b949e;padding:12px;">No pending requests.</div>';
    }
    html += '</div>';
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load: ' + e.message + '</div>';
  }
 }
 async function loadApprove() {
  const el = document.getElementById('arch-approve-content');
  el.innerHTML = '<div class="loading">Loading pending approvals...</div>';
  try {
    const r = await fetch('/admin/ai-evolution/api/owner-gate');
    const data = await r.json();
    let html = '';
    if (data.pending_requests && data.pending_requests.length) {
      html += '<div class="card"><div class="card-title">Pending changes requiring your approval</div>';
      for (const req of data.pending_requests) {
        html += '<div class="command-item">';
        html += '<div class="name">' + (req.title || req.change_id) + '</div>';
        html += '<div class="meta">Risk: <strong>' + req.risk + '</strong></div>';
        html += '<div style="font-size:13px;color:#c9d1d9;margin:8px 0;">' + (req.reason || '') + '</div>';
        html += '<div style="display:flex;gap:8px;">';
        html += '<button onclick="ownerAction.classList.add('\'approved\');this.disabled=true" style="padding:6px 16px;background:#238636;color:#fff;border:none;border-radius:6px;cursor:pointer;font-size:13px;">✓ Approve</button>';
        html += '</div></div>';
      }
      html += '</div>';
    } else {
      html += '<div class="card"><div style="padding:24px;text-align:center;color:#8b949e;">✅ All changes processed. No pending approval requests.</div></div>';
    }
    el.innerHTML = html;
  } catch(e) {
    el.innerHTML = '<div class="error">Failed to load: ' + e.message + '</div>';
  }
 }
  document.querySelectorAll('.panel').forEach(p => p.style.display = 'none');
  document.getElementById('panel-' + name).style.display = '';
  document.querySelectorAll('.sidebar-nav li').forEach(li => li.classList.remove('active'));
  document.querySelector('.sidebar-nav li:nth-child(' + ({overview:1,versions:2,commands:3,agents:4}[name] || 1) + ')').classList.add('active');
  if (name === 'overview') {
    document.getElementById('panel-overview').innerHTML = '<div class="main-header"><h2>System Overview</h2><div id="header-status" class="status-badge">Loading...</div></div><div id="loading" class="loading">Loading...</div><div id="error" class="error" style="display:none;"></div><div id="content"><div class="row"><div class="col"><div class="card"><div class="card-title">Current Baseline</div><div id="base" class="card-value">-</div></div></div><div class="col"><div class="card"><div class="card-title">Git Branch</div><div id="branch" class="card-value">-</div></div></div><div class="col"><div class="card"><div class="card-title">Uptime</div><div id="uptime" class="card-value">-</div></div></div></div><div class="card" style="margin-top:16px"><div class="card-title">Event Bus Status</div><pre id="bus-status">-</pre></div></div>';
    fetchData();
  }
  const loaders = {
    'arch-owner': loadOwnerGate,
    'arch-approve': loadApprove,
    versions: loadVersions,
    commands: loadCommands,
    agents: loadAgents,
  };
  const fn = loaders[name];
  if (fn) fn();
 }
 fetchData();
 </script>
 </body>
 </html>
@@ -0,0 +1,120 @@
 """System blocker memory and recurring issue detection.
 Records blocking issues, long-no-response incidents and repeated operational
 failures. Repeated blockers are escalated from incident handling into mechanism
 optimization work.
 """
 from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True)
 class SystemBlockerRecord:
    blocker_id: str
    category: str
    title: str
    source: str
    severity: str
    status: str = "open"
    first_seen_at: str = field(default_factory=utc_now)
    last_seen_at: str = field(default_factory=utc_now)
    occurrence_count: int = 1
    evidence: list[str] = field(default_factory=list)
    mechanism_gap: bool = False
    recommended_owner: str = "autonomous-controller"
    def fingerprint(self) -> str:
        return f"{self.category}:{self.title}".lower().replace(" ", "_")
    def to_dict(self) -> dict[str, Any]:
        return asdict(self) | {"fingerprint": self.fingerprint()}
@dataclass(frozen=True)
 class BlockerEscalationDecision:
    status: str
    priority: str
    reason: str
    action: str
    assigned_to: str
    create_task: bool = True
    architect_review_required: bool = False
    owner_gate_required: bool = False
    def to_dict(self) -> dict[str, Any]:
        return asdict(self)
 class SystemBlockerMemory:
    """In-memory decision logic; production storage can wrap this with persistence."""
    def record(self, existing: list[dict[str, Any]], incoming: SystemBlockerRecord) -> dict[str, Any]:
        records = [dict(item) for item in existing]
        fp = incoming.fingerprint()
        for item in records:
            if item.get("fingerprint") == fp or f"{item.get('category')}:{item.get('title')}".lower().replace(" ", "_") == fp:
                item["occurrence_count"] = int(item.get("occurrence_count", 1)) + 1
                item["last_seen_at"] = utc_now()
                item["status"] = "recurrent"
                evidence = list(item.get("evidence", [])) + incoming.evidence
                item["evidence"] = list(dict.fromkeys(evidence))
                item["mechanism_gap"] = self.is_mechanism_gap(item)
                item["fingerprint"] = fp
                return item
        item = incoming.to_dict()
        item["mechanism_gap"] = self.is_mechanism_gap(item)
        records.append(item)
        return item
    def decide(self, blocker: dict[str, Any]) -> BlockerEscalationDecision:
        count = int(blocker.get("occurrence_count", 1))
        severity = blocker.get("severity", "medium")
        category = blocker.get("category", "unknown")
        if severity in {"critical", "p0"} or category in {"git_permission", "runtime_deadlock", "autonomous_controller_stall"}:
            return BlockerEscalationDecision(
                status="escalate_to_mechanism_fix",
                priority="P0",
                reason="blocking infrastructure issue affects autonomous operation",
                action="create_self_healing_mechanism",
                assigned_to="openclaw",
                architect_review_required=True,
            )
        if count >= 3:
            return BlockerEscalationDecision(
                status="escalate_to_mechanism_fix",
                priority="P1",
                reason="recurring issue exceeded threshold",
                action="create_root_cause_and_prevention_patch",
                assigned_to="autonomous-controller",
                architect_review_required=True,
            )
        if category == "long_no_response":
            return BlockerEscalationDecision(
                status="create_monitoring_task",
                priority="P1",
                reason="long no response requires watchdog and notification",
                action="create_stall_watchdog_task",
                assigned_to="openclaw",
            )
        return BlockerEscalationDecision(
            status="track_incident",
            priority="P2",
            reason="single incident tracked for recurrence",
            action="record_and_monitor",
            assigned_to="autonomous-controller",
            create_task=False,
        )
    def is_mechanism_gap(self, blocker: dict[str, Any]) -> bool:
        return int(blocker.get("occurrence_count", 1)) >= 2 or blocker.get("severity") in {"critical", "p0"}
 __all__ = ["SystemBlockerRecord", "BlockerEscalationDecision", "SystemBlockerMemory"]
@@ -0,0 +1,32 @@
 """System Health Center API."""
 try:
    from fastapi import APIRouter
 except Exception:
    APIRouter = None
 from .system_health_center import SystemHealthCenter
 center = SystemHealthCenter()
 def get_system_health_overview():
    return center.build()
 def build_router():
    if APIRouter is None:
        return None
    router = APIRouter(prefix="/admin/ai-evolution/system-health", tags=["system-health"])
    @router.get("/overview", operation_id="getSystemHealthOverview")
    def overview():
        return get_system_health_overview()
    return router
 router = build_router()
 __all__ = ["router", "get_system_health_overview"]
@@ -0,0 +1,54 @@
 """Unified AIOS system health center model."""
 from __future__ import annotations
 from datetime import datetime, timezone
 from typing import Any
 def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()
 class SystemHealthCenter:
    def build(
        self,
        git_report: dict[str, Any] | None = None,
        blockers: list[dict[str, Any]] | None = None,
        tasks: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        blockers = blockers or []
        tasks = tasks or []
        return {
            "module": "system-health-center",
            "route": "/admin/ai-evolution/system-health",
            "title": "AIOS 系统健康与自优化中心",
            "generated_at": utc_now(),
            "status": self._status(git_report, blockers),
            "git": git_report or {},
            "blockers": {
                "total": len(blockers),
                "recurrent": len([b for b in blockers if b.get("mechanism_gap")]),
                "p0": len([b for b in blockers if b.get("severity") in {"p0", "critical"}]),
            },
            "tasks": {
                "optimization_tasks": len([t for t in tasks if t.get("source") == "system_blocker"]),
                "pending": len([t for t in tasks if t.get("status") in {"pending", "running", "waiting_approval"}]),
            },
            "optimization_loop": {
                "detect": True,
                "remember": True,
                "optimize": True,
                "verify": True,
            },
        }
    def _status(self, git_report, blockers):
        if git_report and git_report.get("status") not in {"healthy", "ok"}:
            return "attention_required"
        if any(b.get("severity") in {"p0", "critical"} for b in blockers):
            return "attention_required"
        return "healthy"
 __all__ = ["SystemHealthCenter"]
@@ -1,36 +0,0 @@
 # CMD-20260708-fix-git-actions-write-and-enable-audit-plane
 ## Command ID
 CMD-20260708-fix-git-actions-write-and-enable-audit-plane
 ## Priority
 P0
 ## Mode
 autonomous
 ## Target
 Primary: OpenClaw
 Supporting: Hermes
 Review: Codex
 ## Objective
 解决当前 AIOS Git Actions 写权限问题，恢复 ChatGPT Actions → AIOS GitOps API → Git Commit 链路，
 同时建立 Agent 全生命周期审计体系。
 ## Status
 - Part 1 (Fix Git Actions Write Permission): ✅ COMPLETE
 - Part 2 (Agent Command Plane Structure): ✅ COMPLETE
 - Part 3 (Agent Execution Audit): ✅ COMPLETE
 - Part 4 (Conversation Trace): 🚧 IN PROGRESS
 - Part 5 (Version Association): ⏳ PENDING
 - Part 6 (Web UI Integration): ⏳ PENDING
 - Part 7 (Evidence): ⏳ PENDING
 - Part 8 (Regression Test): ⏳ PENDING
 ## Completion Criteria
 1. Code committed: agents/, audit/, evidence/, web-ui/
 2. Doc: docs/architecture/AIOS_AGENT_AUDIT_PLANE.md
 3. Report: evidence/CMD-20260708.../final-report.md
 4. Output: AIOS_AGENT_AUDIT_PLANE_READY, AIOS_GIT_ACTION_WRITE_READY
 5. Next command created: CMD-20260709-autonomous-controller-enable
@@ -1,35 +0,0 @@
 # CMD-20260709-enable-autonomous-controller
 ## Command ID
 CMD-20260709-enable-autonomous-controller
 ## Priority
 P0
 ## Mode
 autonomous
 ## Target
 Primary: OpenClaw
 Supporting: Hermes
 Review: Codex
 ## Objective
 Enable autonomous controller for AIOS GitOps:
 - Auto-detect issues
 - Auto-diagnose
 - Auto-generate fixes
 - Auto-test
 - Auto-create patches
 - Auto-request approval
 ## Preconditions
 - Agent Command Plane initialized (CMD-BOOTSTRAP-001)
 - Git Actions write permission fixed (CMD-20260708)
 - Audit plane enabled (CMD-20260708)
 - Notification service enabled (CMD-20260708)
 ## Completion Criteria
 - Autonomous controller service running
 - Can detect, diagnose, fix, and report
 - Pipeline: detect → diagnose → fix → test → patch → approve → commit → notify
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783527185",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T00:13:05+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783529900",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T00:58:20+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783529914",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T00:58:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783530514",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:08:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783531114",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:18:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783531714",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:28:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783532314",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:38:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783532914",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:48:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783533514",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T01:58:34+0800"
 }
@@ -1,7 +0,0 @@
 {
  "nid": "n-1783534114",
  "title": "⚠️ Issue Detected: branch_sync",
  "body": "10 branches missing from Gitea DB",
  "urgency": "warning",
  "ts": "2026-07-09T02:08:34+0800"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783527185",
  "created": "2026-07-09T00:13:05+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/agent-event-bus-unified",
    "test/alias-write-verify",
    "patch/V1R2C3-tag-execute-idempotent-permission",
    "patch/gitops-write-auth-smoke-20260706T115230Z",
    "patch/gpt-actions-auth-live-20260708"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783529900",
  "created": "2026-07-09T00:58:20+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "test/alias-test-file-branch",
    "patch/codex-online-sync-acceptance-20260706T102819Z",
    "patch/V1R2C3-tag-execute-idempotent-permission",
    "patch/actions-invalid-diag-smoke-20260706T140311Z",
    "patch/V1R2C2-autonomous-recovery-loop"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783529914",
  "created": "2026-07-09T00:58:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783530514",
  "created": "2026-07-09T01:08:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783531114",
  "created": "2026-07-09T01:18:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783531714",
  "created": "2026-07-09T01:28:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783532314",
  "created": "2026-07-09T01:38:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783532914",
  "created": "2026-07-09T01:48:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783533514",
  "created": "2026-07-09T01:58:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,16 +0,0 @@
 {
  "patch_id": "patch-1783534114",
  "created": "2026-07-09T02:08:34+0800",
  "check": "branch_sync",
  "severity": "high",
  "diagnosis": "10 branches missing from Gitea DB",
  "branches": [
    "patch/codex-write-auth-smoke-20260706T115256Z",
    "patch/v1r3c0-user-portal-rbac-ops-20260708",
    "request/tag-V1R3C0-self-loop-autonomous-engine",
    "patch/fix-tag-confirmation-state-machine",
    "patch/gitops-write-auth-fix"
  ],
  "stuck": [],
  "fix": "sync_branches"
 }
@@ -1,12 +0,0 @@
 {
  "ts": "2026-07-09T02:08:34+0800",
  "ok": false,
  "checks": {
    "gitea": false,
    "api": true,
    "branch_sync": false,
    "stuck": true
  },
  "plans": 1,
  "applied": 0
 }
@@ -1,11 +0,0 @@
 {
 "ts": "2026-07-09T00:00:00+09:00",
 "ok": true,
 "checks": {
 "git_object_permission": true,
 "api": true,
 "branch_sync": true,
 "stale_runtime_state_cleaned": true
 },
 "note": "Updated during PR #2 cleanup before merge"
 }
@@ -1,21 +0,0 @@
 {
  "ts": "2026-07-09T00:59:30+0800",
  "action": "command_watcher_state_clean",
  "description": "清除了 command watcher 目录中的 stale 重复副本",
  "result": "clean",
  "stale_removed": [
    {
      "name": "CMD-20260708-fix-git-actions-write-and-enable-audit-plane.md",
      "removed_from": ["active/", "running/"],
      "preserved_in": "completed/",
      "archive_path": "agents/commands/archive/"
    },
    {
      "name": "CMD-20260709-enable-autonomous-controller.md",
      "removed_from": ["running/"],
      "preserved_in": "active/",
      "archive_path": "agents/commands/archive/"
    }
  ],
  "note": "已完成命令保留在 completed/，副本移动到 archive/；未完成命令仅保留 active/"
 }
@@ -1,17 +0,0 @@
 {
  "ts": "2026-07-09T01:00:00+0800",
  "action": "daily_wechat_cron_verification",
  "description": "验证每日 08:00 WeChat 汇报 cron 配置并手动触发 dry-run",
  "cron_name": "AIOS每日自治报告",
  "cron_id": "987f8619-6ccc-4370-8720-87149479a222",
  "schedule": "0 8 * * * Asia/Shanghai",
  "delivery": {
    "mode": "announce",
    "channel": "openclaw-weixin",
    "to": "o9cq80wSaJq6utQBkUC4fE1sJWxY@im.wechat",
    "accountId": "ed3da13663c0-im-bot"
  },
  "dry_run_triggered": true,
  "dry_run_result": "enqueued",
  "status": "verified"
 }
@@ -1,19 +0,0 @@
 {
  "run_id": "c-1783527185",
  "check": "stuck",
  "status": "degraded",
  "ts": "2026-07-09T00:13:05+0800",
  "details": {
    "ok": false,
    "stuck": [
      {
        "n": "CMD-20260709-enable-autonomous-controller.md",
        "age": 184.6
      },
      {
        "n": "CMD-20260708-fix-git-actions-write-and-enable-audit-plane.md",
        "age": 185.8
      }
    ]
  }
 }
@@ -1,19 +0,0 @@
 {
  "run_id": "c-1783529900",
  "check": "stuck",
  "status": "degraded",
  "ts": "2026-07-09T00:58:20+0800",
  "details": {
    "ok": false,
    "stuck": [
      {
        "n": "CMD-20260709-enable-autonomous-controller.md",
        "age": 229.8
      },
      {
        "n": "CMD-20260708-fix-git-actions-write-and-enable-audit-plane.md",
        "age": 231.0
      }
    ]
  }
 }
@@ -1,19 +0,0 @@
 {
  "run_id": "c-1783529914",
  "check": "stuck",
  "status": "degraded",
  "ts": "2026-07-09T00:58:34+0800",
  "details": {
    "ok": false,
    "stuck": [
      {
        "n": "CMD-20260709-enable-autonomous-controller.md",
        "age": 230.1
      },
      {
        "n": "CMD-20260708-fix-git-actions-write-and-enable-audit-plane.md",
        "age": 231.3
      }
    ]
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783530514",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T01:08:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783531114",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T01:18:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783531714",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T01:28:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783532314",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T01:38:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783532914",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T01:48:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783533514",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T01:58:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,10 +0,0 @@
 {
  "run_id": "c-1783534114",
  "check": "stuck",
  "status": "ok",
  "ts": "2026-07-09T02:08:34+0800",
  "details": {
    "ok": true,
    "stuck": []
  }
 }
@@ -1,141 +0,0 @@
 #!/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())
@@ -1,172 +0,0 @@
 #!/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", "codex_local"}:
        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"},
            "codex_local": {"validated_live": channel == "codex_local"},
        },
    )
    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())
@@ -1,136 +0,0 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 import argparse
 import json
 import logging
 import os
 from pathlib import Path
 import time
 import urllib.error
 import urllib.request
 from services.aios_orchestrator_gateway.codex_local import CodexLocalClient
 from services.aios_orchestrator_gateway.config import Config
 from services.aios_orchestrator_gateway.models import Event
 from services.aios_orchestrator_gateway.securefs import read_secure_json
 LOG = logging.getLogger("aios-codex-central-decision")
 class CodexCentralWorker:
    def __init__(self, config: Config, client: CodexLocalClient | None = None):
        if config.central_channel != "codex_local" or config.codex_queue_dir is None:
            raise RuntimeError("codex_local_not_selected")
        self.config = config
        self.client = client or CodexLocalClient()
        self.root = config.codex_queue_dir
        self.pending = self.root / "pending"
        self.processing = self.root / "processing"
        self.processed = self.root / "processed"
        self.dead = self.root / "dead-letter"
    def recover_interrupted(self) -> None:
        for path in sorted(self.processing.glob("*.json")):
            target = self.pending / path.name
            if target.exists():
                path.unlink()
            else:
                path.replace(target)
    def run_one(self) -> bool:
        candidates = sorted(self.pending.glob("*.json"))
        if not candidates:
            return False
        source = candidates[0]
        claimed = self.processing / source.name
        try:
            source.replace(claimed)
        except FileNotFoundError:
            return False
        wrapper = read_secure_json(claimed, expected_uid=None)
        attempts = wrapper.get("attempts")
        next_attempt_at = wrapper.get("next_attempt_at")
        event = Event.from_dict(wrapper.get("event"))
        if (
            not isinstance(attempts, int) or attempts < 0
            or not isinstance(next_attempt_at, (int, float)) or next_attempt_at < 0
        ):
            self._dead_letter(claimed)
            return True
        if next_attempt_at > time.time():
            claimed.replace(self.pending / claimed.name)
            return False
        if event.contract_commit != self.config.contract_commit:
            self._dead_letter(claimed)
            return True
        try:
            decision = self.client.decide(event)
            self._submit(decision.to_dict(), event.event_id)
        except Exception as exc:
            LOG.warning("codex_decision_failed category=%s", type(exc).__name__)
            wrapper["attempts"] = attempts + 1
            wrapper["next_attempt_at"] = int(time.time()) + self.config.retry_base_seconds
            self._rewrite(claimed, wrapper)
            if wrapper["attempts"] > self.config.max_retries:
                self._dead_letter(claimed)
            else:
                claimed.replace(self.pending / claimed.name)
            return True
        claimed.replace(self.processed / claimed.name)
        return True
    def _submit(self, decision: dict, event_id: str) -> None:
        body = json.dumps(decision, separators=(",", ":"), sort_keys=True).encode()
        url = f"http://{self.config.host}:{self.config.port}/v1/orchestrator/decisions"
        request = urllib.request.Request(
            url, data=body, method="POST",
            headers={
                "Authorization": f"Bearer {self.config.callback_token()}",
                "Content-Type": "application/json",
                "Idempotency-Key": event_id,
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                if response.status not in {200, 202}:
                    raise RuntimeError("gateway_decision_rejected")
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            raise RuntimeError("gateway_decision_unavailable") from exc
    @staticmethod
    def _rewrite(path: Path, value: dict) -> None:
        temp = path.with_name(f".{path.name}.tmp-{os.getpid()}")
        fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o640)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(value, handle, separators=(",", ":"), sort_keys=True)
            handle.flush(); os.fsync(handle.fileno())
        os.replace(temp, path)
    def _dead_letter(self, path: Path) -> None:
        target = self.dead / path.name
        if target.exists():
            path.unlink()
        else:
            path.replace(target)
 def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--gateway-config", required=True, type=Path)
    parser.add_argument("--once", action="store_true")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s")
    worker = CodexCentralWorker(Config.from_file(args.gateway_config))
    worker.recover_interrupted()
    if args.once:
        worker.run_one()
        return 0
    while True:
        if not worker.run_one():
            time.sleep(2)
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -1,72 +0,0 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 import argparse
 import json
 import os
 from pathlib import Path
 import sys
 import tempfile
 import time
 import uuid
 from services.aios_orchestrator_gateway.codex_local import CodexLocalClient
 from services.aios_orchestrator_gateway.config import Config
 from services.aios_orchestrator_gateway.models import Event
 def _atomic_receipt(path: Path, value: dict) -> None:
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    fd, temp = tempfile.mkstemp(prefix=".codex-health-", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(value, handle, separators=(",", ":"), sort_keys=True)
            handle.flush(); os.fsync(handle.fileno())
        os.chmod(temp, 0o600)
        os.replace(temp, path)
    finally:
        if os.path.exists(temp): os.unlink(temp)
 def run(config_path: Path, receipt: Path) -> int:
    config = Config.from_file(config_path)
    if config.central_channel != "codex_local":
        raise RuntimeError("codex_local_not_selected")
    event_id = "codex-health-" + uuid.uuid4().hex
    event = Event.from_dict({
        "schema_version": 1, "project_key": config.project_key,
        "contract_commit": config.contract_commit, "event_id": event_id,
        "source_agent": "system", "event_type": "codex_local_health_probe",
        "work_unit_id": "CODEX-LOCAL-HEALTH", "status": "READY",
        "state_version": 0, "implementation_commit": None,
        "result_commit": None, "blocker": "health_probe_has_no_release_evidence",
        "evidence_refs": [], "payload": {"allowed_work_unit_ids": []},
    })
    decision = CodexLocalClient().decide(event)
    if decision.verdict == "DISPATCH" or decision.next_work_units:
        raise RuntimeError("unsafe_health_decision")
    _atomic_receipt(receipt, {
        "subject": "codex_local", "health": "HEALTHY",
        "codex_login_status": 0, "codex_exec": 0,
        "decision_schema_valid": True, "observed_at_epoch": int(time.time()),
    })
    print("CODEX_CENTRAL_HEALTH=PASS")
    return 0
 def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--gateway-config", required=True, type=Path)
    parser.add_argument("--receipt", required=True, type=Path)
    args = parser.parse_args()
    try:
        return run(args.gateway_config, args.receipt)
    except Exception as exc:
        reason = str(exc)
        safe = reason if reason.replace("_", "").isalnum() and len(reason) <= 64 else type(exc).__name__
        print(f"CODEX_CENTRAL_HEALTH=BLOCKED reason={safe}", file=sys.stderr)
        return 1
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -1,113 +0,0 @@
 #!/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())
@@ -1,940 +0,0 @@
 #!/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())
@@ -1,146 +0,0 @@
 #!/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)
@@ -1,76 +0,0 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 import argparse
 import json
 import os
 from pathlib import Path
 import sys
 import tempfile
 import time
 from services.aios_orchestrator_gateway.config import Config
 from services.aios_orchestrator_gateway.remote_codex import RemoteCodexBroker
 def atomic_receipt(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd, temporary = tempfile.mkstemp(prefix=".remote-codex-health-", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(value, handle, separators=(",", ":"), sort_keys=True)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(temporary, 0o600)
        os.replace(temporary, path)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)
 def run(config_path: Path, receipt: Path, healthy_within: int) -> int:
    config = Config.from_file(config_path)
    if config.central_channel != "remote_codex":
        raise RuntimeError("remote_codex_not_selected")
    broker = RemoteCodexBroker(
        config.database_path,
        contract_commit=config.contract_commit,
        lease_seconds=config.remote_job_lease_seconds,
        max_retries=config.max_retries,
    )
    try:
        snapshot = broker.health_snapshot(healthy_within_seconds=healthy_within)
    finally:
        broker.close()
    if snapshot["healthy_runners"] < 1:
        raise RuntimeError("remote_codex_runner_unhealthy")
    atomic_receipt(receipt, {
        "subject": "remote_codex",
        "health": "HEALTHY",
        "contract_commit": config.contract_commit,
        "healthy_runners": snapshot["healthy_runners"],
        "runner_count": snapshot["runner_count"],
        "job_counts": snapshot["job_counts"],
        "observed_at_epoch": int(time.time()),
    })
    print("REMOTE_CODEX_HEALTH=PASS")
    return 0
 def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--gateway-config", required=True, type=Path)
    parser.add_argument("--receipt", required=True, type=Path)
    parser.add_argument("--healthy-within-seconds", type=int, default=300)
    args = parser.parse_args()
    try:
        return run(args.gateway_config, args.receipt, args.healthy_within_seconds)
    except Exception as exc:
        reason = str(exc)
        safe = reason if reason.replace("_", "").isalnum() and len(reason) <= 64 else type(exc).__name__
        print(f"REMOTE_CODEX_HEALTH=BLOCKED reason={safe}", file=sys.stderr)
        return 1
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -1,13 +1 @@
-{
+{"current_baseline":"V1R3C1","promoted_at":"2026-07-08T14:31:00+08:00","previous_baseline":"V1R2C3"}
    "current_baseline": "V1R4C6",
    "previous_baseline": "V1R4C5",
    "promoted_at": "2026-07-11T21:00:00+08:00",
    "git": "e204a90398bf646d46245b3088226749fa056edd",
    "promoted_by": "openclaw/admin",
    "promotion_reason": "V1R4C6 admin search agent integration, search gateway, agent platform, VERSION bump, merge tests deploy pipeline — not yet deployed to production",
    "constraints": {
        "memory_search_enabled": false,
        "force_push_allowed": false,
        "gitea_db_mutation_requires_owner_approval": true
    }
 }
@@ -1,24 +0,0 @@
 {
  "$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"
 }
@@ -1,17 +0,0 @@
 {
  "active_dir": "/data/aios-gitops/workspace/agents/commands/active",
  "completed_dir": "/data/aios-gitops/workspace/agents/commands/completed",
  "dead_letter_dir": "/data/aios-gitops/workspace/agents/commands/dead-letter",
  "database_path": "/var/lib/aios-codex-watcher/state.sqlite3",
  "audit_log_path": "/var/log/aios-codex-watcher/audit.jsonl",
  "gateway_endpoint": "http://127.0.0.1:18920/v1/orchestrator/events",
  "gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
  "lease_seconds": 60,
  "retry_seconds": 10,
  "max_attempts": 5,
  "poll_seconds": 5,
  "executor_path": null,
  "executor_args": [],
  "executor_timeout_seconds": 1800
 }
@@ -1,17 +0,0 @@
 {
  "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
 }
@@ -1,65 +0,0 @@
 {
  "host": "127.0.0.1",
  "port": 18920,
  "database_path": "/var/lib/aios-orchestrator-gateway/state.sqlite3",
  "event_inbox_dir": "/var/lib/aios-orchestrator-gateway/inbox/events",
  "decision_inbox_dir": "/var/lib/aios-orchestrator-gateway/inbox/decisions",
  "dispatch_outbox_dir": "/var/lib/aios-orchestrator-gateway/outbox",
  "openclaw_sync_dir": "/var/lib/aios-orchestrator-gateway/openclaw-sync",
  "command_queue_dir": "/var/lib/aios-orchestrator-gateway/commands",
  "inbound_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
  "callback_token_file": "/etc/aios-orchestrator-gateway/callback.token",
  "contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
  "central_channel": "remote_codex",
  "codex_local": {
    "queue_dir": "/var/lib/aios-codex-brain/queue",
    "timeout_seconds": 300
  },
  "remote_codex": {
    "request_max_bytes": 262144,
    "job_lease_seconds": 120,
    "runners": [
      {
        "runner_id": "codex-brain-192-168-88-6",
        "token_file": "/etc/aios-orchestrator-gateway/remote-codex-88-6.token",
        "scopes": ["claim", "heartbeat", "decision", "fail"]
      }
    ]
  },
  "workspace_agent": {
    "trigger_id": null,
    "access_token_file": null
  },
  "responses_fallback": {
    "api_key_file": null,
    "model": null
  },
  "polling": {
    "interval_seconds": 10,
    "summary_interval_seconds": 300
  },
  "loop_controls": {
    "callback_timeout_seconds": 900,
    "lease_seconds": 60,
    "retry_base_seconds": 10,
    "max_retries": 5,
    "max_same_blocker_retries": 3,
    "max_hops_per_gate": 20
  },
  "openclaw_sync": {
    "endpoint": null,
    "token_file": null
  },
  "project_key": "v2r1-personal-weixin-production",
  "git_poll": {
    "enabled": true,
    "bare_repo": "/data/aios-gitops/repo.git",
    "ref_prefixes": [
      "refs/heads/main",
      "refs/heads/task/governance/",
      "refs/heads/patch/",
      "refs/heads/result/"
    ],
    "interval_seconds": 30
  }
 }
@@ -1,33 +0,0 @@
 architect_agent_governance:
  enabled: true
  responsibilities:
    requirement_change_tracking: true
    requirement_document_refresh: true
    architecture_document_refresh: true
    deployment_document_validation: true
    test_case_document_generation: true
    architecture_optimization_review: true
    weekly_external_research: true
    github_top_implementation_review: true
  sources:
    requirement_sources:
      - custom_gpt
      - codex
      - openclaw
      - hermes
      - user_feedback
  version_decision:
    criteria:
      C: bug_fix_or_small_compatible_change
      R: architecture_or_capability_release
      V: major_architecture_or_breaking_change
  research_cycle:
    frequency: weekly
    targets:
      - agent_framework_versions
      - model_versions
      - github_top_implementations
      - security_changes
  governance:
    owner_gate_required: true
    audit_required: true
@@ -1,22 +0,0 @@
 architect_research:
  enabled: true
  schedule:
    frequency: weekly
  research_sources:
    - agent_framework_releases
    - model_ecosystem_updates
    - github_top_implementations
    - security_advisories
  evaluation:
    required:
      - relevance
      - architecture_impact
      - security_impact
      - migration_cost
  output:
    - optimization_recommendation
    - version_recommendation
    - architecture_review_event
  governance:
    owner_gate_required: true
    audit_required: true
@@ -1,36 +0,0 @@
 architecture_owner_gate:
  enabled: true
  principle: owner_approves_large_architecture_changes_only
  classification:
    small:
      owner_required: false
      architect_auto_implementation: true
      requirements:
        - architect_review
        - test_gate
        - audit
    medium:
      owner_required: false
      architect_review_required: true
      implementation_after_review: true
      requirements:
        - architect_review
        - test_gate
        - audit
    large:
      owner_required: true
      architect_auto_implementation: false
      requirements:
        - architect_review
        - owner_architecture_approval
        - test_gate
        - promotion_gate
  mandatory_owner_cases:
    - breaking_change
    - V_version_change
    - production_architecture_update
    - security_boundary_change
    - cross_tenant_change
    - permission_model_change
  audit:
    required: true
@@ -1,18 +0,0 @@
 change_governance:
  enabled: true
  feature_change:
    require_tests: true
    require_user_acceptance_if_user_facing: true
  architecture_change:
    require_architecture_review_agent: true
    require_system_optimization_review: true
  acceptance_agents:
    user_agent: user-acceptance-agent
    architect_agent: architecture-review-agent
  merge_gate:
    required:
      - tests_passed
      - acceptance_passed
      - architecture_review_passed
  audit:
    enabled: true
@@ -0,0 +1,35 @@
 git_permission_watchdog:
  enabled: true
  priority: P0
  workspace: /data/aios-gitops/workspace
  expected_identity:
    user: aiman
    group: gitops
  checks:
    - object_owner
    - object_group
    - object_write_permission
    - git_write_probe
  repair:
    enabled: true
    allowed_paths:
      - .git/objects
    forbidden:
      - chmod_777
      - system_wide_permission_change
  runtime:
    interval_minutes: 5
    create_task_on_drift: true
    send_notification_on_failure: true
  audit:
    enabled: true
    record:
      - detection_time
      - affected_paths
      - repair_action
      - repair_result
@@ -1,22 +0,0 @@
 governance_event_bus:
  enabled: true
  event_types:
    - requirement_changed
    - architecture_review_started
    - architecture_review_completed
    - user_acceptance_completed
    - test_gate_completed
    - version_decision_created
  producers:
    - architect-agent
    - user-acceptance-agent
    - test-gate-agent
    - change-governance-agent
  consumers:
    - web-ui
    - audit-plane
    - owner-gate
    - reporting-agent
  audit:
    required: true
    retention: versioned
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
openclaw	189b655d04	Add autonomous operation command center deployment command	2026-07-09 15:57:18 +08:00
openclaw	5f2b5f16ba	Add command stall detector tests	2026-07-09 15:57:13 +08:00
openclaw	e2d9de3a1f	Add unified operations command center model	2026-07-09 15:57:11 +08:00
openclaw	83cbd64fb6	Add Command Watcher stall detector	2026-07-09 15:57:05 +08:00
openclaw	7852844338	Add autonomous health center deployment command	2026-07-09 15:53:00 +08:00
openclaw	46b9d3188d	Add autonomous health adapter tests	2026-07-09 15:52:46 +08:00
openclaw	c69916442c	Add system health API	2026-07-09 15:52:39 +08:00
openclaw	781239bea7	Add autonomous controller health adapter	2026-07-09 15:52:27 +08:00
openclaw	193f6546b1	Fix Git permission watchdog service entrypoint	2026-07-09 15:52:15 +08:00
openclaw	e063c6cb3c	Add Git permission watchdog CLI entrypoint	2026-07-09 15:52:07 +08:00
openclaw	9f9f7be796	Add blocker health center deployment command	2026-07-09 15:45:12 +08:00
openclaw	77d0d7176d	Add system health center tests	2026-07-09 15:45:01 +08:00
openclaw	0029980d74	Add unified system health center	2026-07-09 15:44:58 +08:00
openclaw	0ed265e3cb	Add blocker to Task Center adapter	2026-07-09 15:44:50 +08:00
openclaw	c7a1cf5b2a	Add OpenClaw command for system blocker self optimization	2026-07-09 15:37:32 +08:00
openclaw	d6a6491d2b	Add system blocker optimization tests	2026-07-09 15:37:27 +08:00
openclaw	e364432a1f	Add recurring blocker optimization policy	2026-07-09 15:37:22 +08:00
openclaw	343b6f1ba6	Add system blocker memory and recurrence detection	2026-07-09 15:37:15 +08:00
openclaw	e1d4bfc77b	Add Git permission watchdog systemd timer	2026-07-09 15:34:15 +08:00
openclaw	f7cf7249f1	Add Git permission watchdog systemd service	2026-07-09 15:34:12 +08:00
openclaw	13846c222f	Add Git permission watchdog tests	2026-07-09 15:34:09 +08:00
openclaw	749afecf08	Add Git permission watchdog policy	2026-07-09 15:34:05 +08:00
openclaw	b424de9d98	Add Git permission watchdog self-healing module	2026-07-09 15:34:00 +08:00
openclaw	b761536f16	Dispatch Git permission self-healing mechanism	2026-07-09 15:31:43 +08:00