Compare commits

...
Author SHA1 Message Date
Codex 6be4ad3388 docs(codex): record workspace trigger protocol hardening result 2026-07-18 14:57:14 +08:00
Codex 815e39f36c fix(codex): harden workspace trigger protocol 2026-07-18 14:51:57 +08:00
Codex c679c8394f fix(codex): make central GPT health probing single-flight 2026-07-18 13:44:06 +08:00
Codex 9ec60c162a fix(quorum): treat persisted blocked state as healthy cycle 2026-07-18 11:59:27 +08:00
Codex 415d736f88 fix(quorum): read authoritative bare refs in sandbox 2026-07-18 11:57:52 +08:00
Codex 20bfb3e773 fix(quorum): restore controller executable mode 2026-07-18 11:56:31 +08:00
Codex cf211ac4ac fix(quorum): validate and trust exact Git ref source 2026-07-18 11:56:18 +08:00
Codex a76153c637 fix(quorum): make health receipts group-readable 2026-07-18 11:53:50 +08:00
Codex 3ddab59f65 fix(codex): allow read-only probe WAL access 2026-07-18 11:47:02 +08:00
Codex 4b88d90531 fix(codex): preserve watcher access to probe commands 2026-07-18 11:42:37 +08:00
Codex 2a80c8ad7a fix(codex): bound gateway polling work units 2026-07-18 10:42:51 +08:00
Codex 1e00541e50 fix(codex): bound central health event scheduling 2026-07-18 10:36:22 +08:00
Codex cb4f13601c fix(codex): fail over model trigger channel safely 2026-07-18 10:32:12 +08:00
Codex 6170627542 fix(codex): verify existing shared queue without mutation 2026-07-18 10:26:16 +08:00
Codex 56827247f0 fix(codex): align gateway sandbox with durable queue 2026-07-18 10:24:10 +08:00
Codex 84714c712a fix(codex): recover failed gateway activation safely 2026-07-18 10:20:09 +08:00
Codex 87fe8aea91 fix(codex): make gateway activation transactional 2026-07-18 10:17:41 +08:00
Codex e850db7437 feat(codex): close GPT plus executor unattended automation loop 2026-07-18 10:15:05 +08:00
Codex e116146341 merge: integrate GPT plus one-executor governance and quorum 2026-07-18 09:43:06 +08:00
Codex c3984ca2b4 fix(codex): isolate watcher runtime directory ownership 2026-07-18 03:39:45 +08:00
Codex 0c7e2d3b89 feat(codex): add persistent automatic command watcher 2026-07-18 03:36:46 +08:00
Codex 1b3b74fd62 merge(codex): integrate AIOS gateway with baseline remediation 2026-07-18 03:01:30 +08:00
Codex 378cc4d56a fix(codex): restore hardened WeChat send backend lineage 2026-07-18 03:00:32 +08:00
Codex 95fc4d37f0 feat(codex): add reliable AIOS central orchestration gateway 2026-07-18 02:26:14 +08:00
71 changed files with 9179 additions and 32 deletions
@@ -0,0 +1,73 @@
# RESULT: A0 Workspace Trigger Protocol Hardening
```yaml
contract: cbe33ecab8e9794ba5f318bea43aeaf447a56fbc
work_unit: A0-WORKSPACE-TRIGGER-PROTOCOL-HARDENING
status: PASS
base_commit: c679c8394f23ad011dd7a188aff78d8a2739ac46
implementation_commit: 815e39f36cce2f867460ddfcf4c45da5288674a9
result_commit: HEAD
workspace_202_empty_body: PASS
workspace_202_without_remote_id: PASS
awaiting_decision_transition: PASS
workspace_http_error_categories: PASS_401_403_404_409
workspace_network_error_category: PASS
workspace_idempotency_key_reuse: PASS
workspace_no_repeat_post_after_202: PASS
responses_fallback_after_callback_timeout_only: PASS
responses_missing_credential_fail_closed: PASS
actions_api_operation: PASS_submit_orchestration_decision
actions_api_strict_schema: PASS
actions_api_fixed_gateway_forward: PASS_127_0_0_1_18920
callback_secret_ref_only: PASS
callback_stale_state: PASS_409_FAIL_CLOSED
callback_idempotency: PASS
metadata_only_event_status: PASS
gateway_actions_tests: PASS_68_OF_68
codex_watcher_quorum_tests: PASS
full_unittest: PASS_150_OF_150
full_pytest: PASS_226_AND_34_SUBTESTS
python_compile: PASS
bash_syntax: PASS
json_schema: PASS
openapi_yaml: PASS
systemd_verify: PASS
git_diff_check: PASS
gitleaks: PASS_FINAL_TREE_AND_IMPLEMENTATION_RANGE_ZERO_FINDINGS
full_repository_history_gitleaks: BASELINE_4_FINDINGS_NOT_COUNTED_AS_PASS
implementation_worktree_oid: 815e39f36cce2f867460ddfcf4c45da5288674a9
implementation_bare_oid: 815e39f36cce2f867460ddfcf4c45da5288674a9
implementation_gitea_oid: 815e39f36cce2f867460ddfcf4c45da5288674a9
implementation_all_equal: true
```
## Verified behavior
- Workspace Agent HTTP `202` is accepted without reading a response body and
without requiring a remote ID. The event becomes `AWAITING_DECISION` and a
second processing pass does not issue another POST.
- HTTP 401, 403, 404, and 409 are persisted only as the fixed metadata classes
`UNAUTHORIZED`, `FORBIDDEN`, `NOT_FOUND`, and `NOT_RUNNABLE`; network, TLS,
and timeout failures use `NETWORK_ERROR`. Raw response bodies are not stored.
- Responses fallback is reached only after an accepted Workspace event exceeds
the callback timeout. Missing Responses configuration remains fail-closed.
- The actual 18890 Actions application exposes only the dedicated
`submit_orchestration_decision` callback operation for this path. It validates
the strict decision model and contract before resolving a callback credential,
resolves only an approved SecretRef through a trusted absolute executable,
and forwards to the fixed loopback 18920 decision endpoint.
- Callback credentials are not accepted from request JSON, argv, or inherited
environment values. Unexpected bridge faults return a fixed metadata-only
error and cannot reach the outer exception logger.
## Gitleaks boundary
The final committed tree and exact implementation range each contain zero
Gitleaks findings. A separate complete repository-history scan reports four
pre-existing findings in immutable ancestor commits. That history scan is
recorded as baseline evidence and is explicitly not represented as PASS; no
history rewriting was attempted.
No secret value was printed or committed. No reset, amend, rebase, force push,
direct main push, controller mutation, B4 work, mainline merge, or production
release was performed.
@@ -0,0 +1,69 @@
# Codex task: apply and verify AIOS Orchestrator Gateway v1
Contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
Use the strongest actually available Codex model for the initial security review.
If unavailable or timed out, record TIMEOUT/NOT_EXECUTED and let OpenClaw run
the deterministic test path. Do not count a timeout as PASS.
## Apply
1. Create a new clean append-only branch and isolated worktree.
2. Validate the delivery tarball and `SHA256SUMS`.
3. Run the external `scripts/apply_gateway_patch.sh` with `SOURCE_DIR` and
`WORKTREE`; do not alter unrelated paths.
4. Review all added code, especially:
- exact contract and optimistic state-version checks;
- event/decision idempotency;
- SQLite leases and retry behavior;
- Workspace Agent `202` asynchronous callback flow;
- Responses background creation and polling;
- file-inbox and Git-ref polling;
- traversal/symlink protections;
- OpenClaw lifecycle synchronization;
- systemd sandbox;
- install and rollback transactions.
## Tests
Run from the final source tree:
- Python compile for all gateway and helper files;
- unit tests;
- JSON schema/config parsing;
- Bash syntax;
- `systemd-analyze verify`;
- event duplicate and payload-conflict tests;
- stale state-version rejection;
- lease/double-claim test;
- Workspace trigger failure and Responses fallback;
- response queued/in-progress/completed polling;
- file-inbox recovery after HTTP failure;
- Git-ref initial baseline and subsequent-change event;
- task start/end and project summary synchronization;
- dispatch traversal and symlink rejection;
- retry/backoff/dead-letter;
- install dry-run;
- rollback test in an isolated fixture;
- `git diff --check`;
- full repository tests;
- final Gitleaks.
## Commit and review
1. Create an append-only implementation commit.
2. Push without force.
3. Confirm `WORKTREE_OID == BARE_OID == GITEA_OID`.
4. Create an independent RESULT-only branch and exact-one result commit.
5. Record the source bundle SHA-256, implementation OID, RESULT OID, tests,
Gitleaks, systemd validation and remaining external configuration blockers.
## Forbidden
No reset, amend, rebase, force push, direct main push, controller mutation,
secret disclosure, fabricated test result, production deployment or B3 evidence
fabrication.
Success:
`GATEWAY_CODE_REVIEW=PASS IMPLEMENTATION=<oid> RESULT=<oid> OID_MATCH=YES`
@@ -0,0 +1,68 @@
# OpenClaw task: deploy and supervise AIOS Orchestrator Gateway v1
Contract: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
You are the default orchestrator and deployment supervisor.
## Preconditions
- Codex has applied and independently reviewed this source bundle on a clean,
append-only branch.
- Implementation and RESULT OIDs are identical across worktree, bare repo and
Gitea.
- Final Gitleaks and repository tests are PASS.
- No force push, reset, amend or rebase occurred.
## Responsibilities
1. Validate the source bundle SHA-256 and `SHA256SUMS`.
2. Review the generated systemd service, poll timer, tmpfiles and sysusers files.
3. Install through `scripts/install_aios_orchestrator_gateway.sh`.
4. Configure one or both central-model channels:
- published Workspace Agent API channel with `agtch_...` trigger ID and a
Workspace Agent access token stored outside Git; or
- Responses API fallback with an approved model and API key stored outside Git.
5. Publish the callback endpoint securely and configure the Workspace Agent
action from `workspace-agent/orchestrator-decision-action.openapi.yaml`.
6. Confirm that the Workspace Agent instructions match
`workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md`.
7. Enable both:
- the continuously running gateway service;
- the independent `aios-orchestrator-gateway-poll.timer`.
8. Verify HTTP event ingestion, file-inbox polling, Git-ref polling,
Workspace Agent trigger acceptance, callback decision, Responses background
polling fallback, dispatch queue generation, retry leases and loop guards.
9. Verify OpenClaw receives and synchronizes:
- TASK_DISPATCHED
- TASK_STARTED
- TASK_ENDED
- PROJECT_SUMMARY
- PROJECT_COMPLETED
10. Run a harmless bootstrap work unit through OpenClaw and Codex and confirm
duplicate event submission is idempotent.
11. Run rollback, reinstall and heartbeat/health verification.
12. Create one RESULT-only review recording exact commits, artifact digest,
tests, systemd security, polling recovery, lifecycle synchronization and
rollback.
## Fail-closed boundaries
- Do not put Workspace Agent tokens, API keys, callback tokens or gateway
credentials in Git, logs, command arguments or RESULT files.
- Do not claim the gateway resolves the existing B3 credential/raw-wire blocker.
- Do not advance B4 or production merely because this gateway is healthy.
- Do not bypass owner/admin production release approvals.
- Do not use reset, amend, rebase, force push, direct main push or bare-ref edits.
## Required final status
Success:
`ORCHESTRATOR_GATEWAY_LIVE=PASS POLLING=PASS OPENCLAW_SYNC=PASS ROLLBACK=PASS`
Blocker:
`ORCHESTRATOR_GATEWAY_LIVE=BLOCKED reason=<stable_reason>`
Keep `BASELINE=V1R4C6 RELEASE=paused PRODUCTION_RELEASE=NOT_STARTED` until the
existing production contract advances all release gates.
+11
View File
@@ -0,0 +1,11 @@
# AIOS Codex automatic command watcher
The watcher polls the governed `commands/active` queue without a chat trigger. It uses SQLite transactions for exclusive leases, records the exact command digest, rejects contract drift, retries transient failures, dead-letters exhausted work, and resumes from the last durable phase so an end-event retry does not execute a work unit twice.
Commands are JSON documents validated by `schemas/codex-command.schema.json`. `acceptance_probe` is the only built-in action and has no external side effect. Production `codex` work requires an administrator-provisioned absolute executor path; the instruction is passed on standard input, never argv or environment. The executor receives a sanitized environment and must obtain credentials through its approved file or exec provider. If the executor is absent, the task fails closed and follows the retry/dead-letter policy.
The watcher provides at-most-once execution. It writes the `EXECUTING` phase before launching an external executor; if the host dies after an external effect but before a result is durably recorded, recovery does not replay the work and instead dead-letters the ambiguity for human review. Start and end event JSON is persisted before transmission, so response-loss retries reuse the exact event ID and payload and are deduplicated by the Gateway.
Lifecycle evidence is appended to `/var/log/aios-codex-watcher/audit.jsonl` as `CODEX_AUTO_CLAIM`, `TASK_STARTED`, `TASK_EXECUTED`, `TASK_ENDED`, and `EVENT_POSTED`. Inputs, tokens, stdout, and stderr are not logged; only result digests are retained.
Install from a reviewed commit with `sudo scripts/install_aios_codex_watcher.sh`. The installer creates a content-addressed release, atomically updates `current`, writes a receipt, and enables the five-second systemd timer. The existing Gateway inbound token file is referenced in place and never copied.
+107
View File
@@ -0,0 +1,107 @@
# AIOS Orchestrator Gateway v0.2
This component turns the current passive GPT orchestration pattern into a
durable, unattended loop:
```text
Gitea/OpenClaw/Codex event
-> HTTP event endpoint or polled file inbox
-> durable SQLite state machine
-> Workspace Agent API trigger
-> strict callback decision
-> Responses API background fallback and polling
-> OpenClaw/Codex dispatch queues
-> start/end/result events
-> next orchestration cycle
```
## Reliability mechanisms
The gateway does not depend on one timer, webhook or model call:
- HTTP event ingestion.
- Periodic file-inbox polling for missed webhooks and offline producers.
- Local bare-Git ref polling for `main`, governance, patch and result refs.
- Workspace Agent retries with a stable event `Idempotency-Key`.
- Workspace Agent callback timeout and Responses API background fallback.
- Responses status polling until completion or fail-closed terminal status.
- SQLite optimistic state versions, idempotency records and processing leases.
- Exponential retry with a maximum retry count and dead-letter state.
- Same-blocker and maximum-hop guards to stop non-progress loops.
- A systemd timer runs `--poll-once` as an independent reliability watchdog.
Workspace Agent trigger runs are asynchronous and do not expose a retrievable
final response. The central Workspace Agent must therefore call the
`submit_orchestration_decision` action.
## OpenClaw synchronization
Every lifecycle stage is durably written to the OpenClaw synchronization queue:
- `TASK_DISPATCHED`
- `TASK_STARTED`
- `TASK_ENDED`
- `DECISION_APPLIED`
- `PROJECT_SUMMARY`
- `PROJECT_COMPLETED`
An optional authenticated OpenClaw HTTP endpoint can receive the same records.
A failed HTTP delivery remains in the lifecycle retry queue. OpenClaw remains
the default project-status synchronizer and authoritative progress presenter.
## Agent execution contract
Each dispatched JSON envelope contains:
- exact contract commit;
- source event;
- clean base commit;
- primary and fallback agent;
- allowed paths;
- required checks;
- forbidden operations;
- completion event callback format.
OpenClaw and Codex must send a `work_unit_started` event before execution and a
`work_unit_completed`, `work_unit_blocked`, or `work_unit_failed` event when
they stop. This guarantees that OpenClaw can synchronize every task start,
finish, and overall project summary.
## Security
- Model and callback credentials are read only from root-owned secret files.
- Credentials never enter Git, dispatch payloads or logs.
- The gateway runs as a dedicated unprivileged user.
- Dispatch and sync paths are protected against traversal and symlink targets.
- All state changes require the exact contract commit and current state version.
- Production evidence, credentials and approvals stay fail-closed.
- Git history operations remain controlled by the committed governance contract.
## Deployment sequence
1. Apply this source bundle to a clean append-only branch.
2. Run unit, repository, shell, JSON, systemd and Gitleaks checks.
3. Configure one or both model channels:
- published Workspace Agent API channel and Workspace Agent access token;
- Responses API key and approved fallback model.
4. Publish the decision callback as a Workspace Agent action or MCP tool.
5. Install the service through `scripts/install_aios_orchestrator_gateway.sh`.
6. Run `scripts/verify_aios_orchestrator_gateway.sh`.
7. Submit a harmless event and verify:
- trigger acceptance;
- strict decision callback;
- dispatch file;
- OpenClaw task-start/task-end/project-summary synchronization;
- duplicate idempotency;
- polling recovery after suppressing the HTTP webhook.
The installer copies an explicit gateway-only allowlist into a release directory
named by its SHA-256 digest, atomically updates `/opt/aios-orchestrator-gateway/current`,
and persists `/var/lib/aios-orchestrator-gateway/install-receipt.json` with the
current and previous release. `--dry-run` performs source validation and emits
`AIOS_ORCHESTRATOR_INSTALL_DRY_RUN=PASS` without changing installed state.
Rollback reads that root-owned `0600` receipt, verifies both release directory
names and `SOURCE_DIGEST` files, coordinates the gateway and poll units, and
atomically switches `current` to the previous release. Repeating the same
rollback is a no-op and emits `NOOP_ALREADY_ROLLED_BACK`.
+35 -2
View File
@@ -10,11 +10,22 @@ from orchestrator.action_monitor import ActionMonitor
ROOT = Path(os.getenv('AIOS_GITOPS_ROOT', '/data/aios-gitops'))
WS = Path(os.getenv('AIOS_GITOPS_WORKSPACE', str(ROOT / 'workspace')))
sys.path.insert(0, str(WS))
from services.aios_orchestrator_gateway.action_bridge import (
CallbackBridgeError,
submit_orchestration_decision,
)
SECRETS = Path(os.getenv('AIOS_GITOPS_SECRETS_DIR', str(ROOT / 'runtime' / 'secrets')))
TOKEN_FILE = Path(os.getenv('AIOS_GITOPS_TOKEN_FILE', str(SECRETS / 'gitops.tokens')))
HOST = os.getenv('AIOS_GITOPS_HOST', '127.0.0.1')
PORT = int(os.getenv('AIOS_GITOPS_PORT', '18890'))
OPENAPI = WS / 'docs' / 'gitops' / 'openapi-chatgpt-actions.yaml'
ORCHESTRATOR_OPENAPI = WS / 'workspace-agent' / 'orchestrator-decision-action.openapi.yaml'
ORCHESTRATOR_CONTRACT_COMMIT = os.getenv('AIOS_ORCHESTRATOR_CONTRACT_COMMIT', '')
ORCHESTRATOR_CALLBACK_SECRET_REF = os.getenv('AIOS_ORCHESTRATOR_CALLBACK_SECRET_REF', '')
_resolver = os.getenv('AIOS_ORCHESTRATOR_SECRET_REF_RESOLVER', '')
ORCHESTRATOR_SECRET_REF_RESOLVER = Path(_resolver) if _resolver else None
BARE_REPO = Path(os.getenv('AIOS_GITOPS_BARE_REPO', str(ROOT / 'runtime' / 'gitea-bin' / 'data' / 'repositories' / 'aios' / 'aios.git')))
monitor = ActionMonitor()
@@ -184,6 +195,8 @@ class H(BaseHTTPRequestHandler):
if path in ('/health','/api/health'): return self.sendj({'status':'ok','service':'chatgpt-gitops-adapter','workspace':str(WS)})
if path in ('/api/openapi.yaml','/api/openapi-chatgpt-actions.yaml','/api/schema'):
return self.sendt(OPENAPI.read_text(encoding='utf-8'), 'application/yaml; charset=utf-8') if OPENAPI.exists() else self.sendj({'error':'schema_missing'},404)
if path == '/api/orchestrator-decision.openapi.yaml':
return self.sendt(ORCHESTRATOR_OPENAPI.read_text(encoding='utf-8'), 'application/yaml; charset=utf-8') if ORCHESTRATOR_OPENAPI.exists() else self.sendj({'error':'schema_missing'},404)
if path in ('/api','/api/status'): return self.sendj(api_spec())
if path in ('/api/base','/api/baseline','/base'):
data=jload(WS/'baseline'/'current.json', {}); data.setdefault('git', sh(['rev-parse','--short','HEAD'])['stdout']); return self.sendj(data)
@@ -216,6 +229,7 @@ class H(BaseHTTPRequestHandler):
def do_POST(self):
path=urllib.parse.urlparse(self.path).path.rstrip('/'); data=self.body()
try:
if path == '/v1/orchestrator/decisions': return self.orchestration_decision(data)
if path in ('/api/branch','/api/create-branch'): return self.branch(data)
if path in ('/api/file','/api/commit-file'): return self.file(data)
if path == '/api/commit': return self.commit(data)
@@ -229,6 +243,25 @@ class H(BaseHTTPRequestHandler):
except Exception as e:
monitor.record_failure('do_POST', str(e), {'path':path})
return self.sendj({'error':str(e)},500)
def orchestration_decision(self,data):
a=self.need('orchestrator:decision')
if not a: return
if not ORCHESTRATOR_CONTRACT_COMMIT or ORCHESTRATOR_SECRET_REF_RESOLVER is None:
return self.sendj({'error':'CALLBACK_CONFIGURATION_UNAVAILABLE'},503)
try:
result=submit_orchestration_decision(
data,
contract_commit=ORCHESTRATOR_CONTRACT_COMMIT,
secret_ref=ORCHESTRATOR_CALLBACK_SECRET_REF,
resolver=ORCHESTRATOR_SECRET_REF_RESOLVER,
)
except CallbackBridgeError as exc:
return self.sendj({'error':exc.category},exc.status_code)
except Exception:
# The outer request handler records exception strings. Keep this
# secret-bearing boundary metadata-only even for unexpected faults.
return self.sendj({'error':'CALLBACK_INTERNAL_ERROR'},500)
return self.sendj(result,202)
def branch(self,data):
a=self.need('branch:create');
if not a: return
@@ -390,9 +423,9 @@ return{vs,cmds,ar,cmd,cdet,tot,sok,sfail,runs,baseline,refreshMsg,loadCmd,selVer
def api_spec():
return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /admin/ai-evolution'],'write':['POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl'],'policy':{'bearer_required_for_write':True,'direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True}}
return {'version':'V1R1C0-chatgpt-actions','openapi':'/api/openapi.yaml','orchestrator_openapi':'/api/orchestrator-decision.openapi.yaml','read':['GET /api/base','GET /api/ver','GET /api/diff','GET /api/patch','GET /api/model','GET /api/run','GET /admin/ai-evolution'],'write':['POST /v1/orchestrator/decisions','POST /api/branch','POST /api/file','POST /api/commit','POST /api/patch','POST /api/tag','POST /api/rollback','POST /api/model','PUT /api/model-priority','POST /api/nl'],'policy':{'bearer_required_for_write':True,'orchestrator_decision_scope':'orchestrator:decision','orchestrator_callback_credential':'secret_ref_only','direct_main_push':False,'delete_tag':False,'owner_approval_required_for_v_level':True,'api_keys_in_git':False,'git_secret_field':'secret_ref','online_git_sync':True}}
def main():
print(f'AIOS ChatGPT GitOps adapter listening on {HOST}:{PORT}')
ThreadingHTTPServer((HOST,PORT), H).serve_forever()
if __name__ == '__main__': main()
if __name__ == '__main__': main()
+141
View File
@@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""Prove two real central-decision to Gateway-dispatch to Codex lifecycle cycles."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sqlite3
import time
import uuid
from health_probe_common import (
ProbeError, gateway_base, load_config, read_token, request_json, state_version,
)
REQUIRED_MARKERS = {"CODEX_AUTO_CLAIM", "TASK_STARTED", "TASK_EXECUTED", "TASK_ENDED", "EVENT_POSTED"}
def marker_counts(path: Path, command_id: str) -> dict[str, int]:
counts = {marker: 0 for marker in REQUIRED_MARKERS}
with path.open("r", encoding="utf-8") as handle:
for line in handle:
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
marker = record.get("marker")
if record.get("command_id") == command_id and marker in counts:
counts[marker] += 1
return counts
def watcher_status(database: Path, command_id: str) -> str | None:
with sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=2) as db:
row = db.execute("SELECT status FROM commands WHERE command_id=?", (command_id,)).fetchone()
return None if row is None else str(row[0])
def wait_for_cycle(config: dict[str, object], token: str, cycle: int) -> tuple[str, str]:
endpoint = str(config["gateway_event_url"])
project = str(config["project_key"])
event_id = f"a0-cycle-{cycle}-" + uuid.uuid4().hex
work_unit_id = f"A0-CODEX-CYCLE-{cycle}-" + uuid.uuid4().hex[:16]
event = {
"schema_version": 1,
"project_key": project,
"contract_commit": config["contract_commit"],
"event_id": event_id,
"source_agent": "system",
"event_type": "a0_unattended_acceptance_requested",
"work_unit_id": work_unit_id,
"status": "READY",
"state_version": state_version(endpoint, project, token),
"implementation_commit": None,
"result_commit": None,
"blocker": None,
"evidence_refs": [],
"payload": {
"acceptance_contract": {
"verdict": "DISPATCH",
"work_unit_id": work_unit_id,
"track_id": "A0_AUTOMATION",
"gate_id": "A0_AUTOMATION_CLOSED",
"primary_agent": "codex",
"fallback_agent": "openclaw",
"clean_base_commit": config["probe_clean_base_commit"],
"allowed_paths": [],
"required_checks": ["acceptance_probe"],
"forbidden_operations": ["production_mutation", "git_mutation"],
"instruction": "Run the fixed side-effect-free acceptance probe only.",
}
},
}
request_json(endpoint, token, method="POST", payload=event)
deadline = time.monotonic() + int(config["cycle_timeout_seconds"])
channel = None
while time.monotonic() < deadline:
_, status = request_json(endpoint + "/" + event_id, token)
if status.get("status") == "TRIGGER_FAILED":
raise ProbeError("central_trigger_failed")
if status.get("status") == "DECIDED" and status.get("decision_applied") is True:
channel = status.get("trigger_channel")
break
time.sleep(float(config["poll_interval_seconds"]))
if channel not in {"workspace_agent", "responses_api"}:
raise ProbeError("central_decision_timeout")
database = Path(str(config["codex_database_path"]))
audit = Path(str(config["codex_audit_log"]))
while time.monotonic() < deadline:
if database.exists() and audit.exists() and watcher_status(database, work_unit_id) == "COMPLETED":
counts = marker_counts(audit, work_unit_id)
if all(counts[marker] == 1 for marker in REQUIRED_MARKERS):
for suffix in ("started", "ended"):
_, lifecycle = request_json(endpoint + "/" + work_unit_id + ":" + suffix, token)
if lifecycle.get("event_id") != work_unit_id + ":" + suffix:
raise ProbeError("gateway_lifecycle_missing")
return event_id, str(channel)
time.sleep(float(config["poll_interval_seconds"]))
raise ProbeError("gateway_dispatch_or_codex_lifecycle_timeout")
def project_summaries_since(database: Path, project: str, since: int) -> int:
with sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=2) as db:
row = db.execute(
"SELECT COUNT(*) FROM lifecycle WHERE project_key=? AND kind='PROJECT_SUMMARY' AND created_at>=?",
(project, since),
).fetchone()
return 0 if row is None else int(row[0])
def run(config_path: Path) -> int:
config = load_config(config_path)
endpoint = str(config["gateway_event_url"])
gateway_base(endpoint)
token = read_token(Path(str(config["gateway_token_file"])))
started = int(time.time())
results = [wait_for_cycle(config, token, cycle) for cycle in (1, 2)]
if len({item[0] for item in results}) != 2:
raise ProbeError("duplicate_cycle_event")
if project_summaries_since(
Path(str(config["gateway_database_path"])), str(config["project_key"]), started
) < 2:
raise ProbeError("project_summary_not_persisted")
return 0
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--config", required=True, type=Path)
args = parser.parse_args()
try:
return run(args.config)
except (KeyError, OSError, ProbeError, sqlite3.Error, ValueError):
return 1
if __name__ == "__main__":
raise SystemExit(main())
+171
View File
@@ -0,0 +1,171 @@
#!/usr/bin/env python3
"""Prove a real central-model decision through the authenticated Gateway."""
from __future__ import annotations
import argparse
import json
import os
import stat
import sys
import time
from pathlib import Path
import uuid
from health_probe_common import (
ProbeError, atomic_receipt, gateway_base, load_config, read_token,
request_json, state_version,
)
PENDING_PATH = Path("/run/aios-health/central-gpt.pending.json")
def _load_pending(
path: Path, contract: str, *, owners: set[int] | None = None
) -> tuple[str, int] | None:
allowed_owners = {0} if owners is None else owners
if not path.exists():
return None
info = path.lstat()
if (
not path.is_absolute()
or stat.S_ISLNK(info.st_mode)
or not stat.S_ISREG(info.st_mode)
or stat.S_IMODE(info.st_mode) & ~0o640
or (os.name == "posix" and info.st_uid not in allowed_owners)
or path.resolve(strict=True) != Path(os.path.abspath(path))
):
raise ProbeError("unsafe_pending_marker")
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict) or set(value) != {"contract_commit", "event_id", "state_version"}:
raise ProbeError("invalid_pending_marker")
event_id = value["event_id"]
version = value["state_version"]
if (
value["contract_commit"] != contract
or not isinstance(event_id, str)
or not event_id.startswith("health-central-")
or len(event_id) != 47
or not isinstance(version, int)
or version < 0
):
raise ProbeError("invalid_pending_marker")
return event_id, version
def _clear_pending(path: Path) -> None:
if not path.exists():
return
info = path.lstat()
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
raise ProbeError("unsafe_pending_marker")
path.unlink()
def _safe_reason(exc: BaseException) -> str:
reason = str(exc)
if reason and len(reason) <= 64 and all(char.islower() or char.isdigit() or char == "_" for char in reason):
return reason
return "probe_error"
def run(config_path: Path, pending_path: Path = PENDING_PATH) -> int:
config = load_config(config_path)
endpoint = config["gateway_event_url"]
gateway_base(endpoint)
token = read_token(Path(config["gateway_token_file"]))
project = config["project_key"]
contract = config["contract_commit"]
pending = _load_pending(pending_path, contract)
is_new = pending is None
if is_new:
event_id = "health-central-" + uuid.uuid4().hex
version = state_version(endpoint, project, token)
atomic_receipt(
pending_path,
{"contract_commit": contract, "event_id": event_id, "state_version": version},
)
else:
event_id, version = pending
event = {
"schema_version": 1,
"project_key": project,
"contract_commit": contract,
"event_id": event_id,
"source_agent": "system",
"event_type": "central_gpt_health_probe",
"work_unit_id": "CENTRAL-GPT-HEALTH",
"status": "READY",
"state_version": version,
"implementation_commit": None,
"result_commit": None,
"blocker": None,
"evidence_refs": [],
"payload": {"probe": "side_effect_free", "nonce": event_id},
}
should_post = is_new
if not is_new:
try:
_, observed = request_json(endpoint + "/" + event_id, token)
except ProbeError as exc:
if str(exc) != "gateway_http_404":
raise
should_post = True
else:
observed = None
if should_post:
try:
status, _ = request_json(endpoint, token, method="POST", payload=event)
except ProbeError as exc:
if str(exc) == "gateway_fail_closed_409":
_clear_pending(pending_path)
raise
if status not in {200, 202}:
raise ProbeError("event_not_accepted")
observed = None
deadline = time.monotonic() + int(config["central_timeout_seconds"])
while time.monotonic() < deadline:
_, observed = request_json(endpoint + "/" + event_id, token)
if observed.get("status") in {"DECIDED", "TRIGGER_FAILED", "DEAD_LETTER"}:
break
time.sleep(float(config["poll_interval_seconds"]))
if isinstance(observed, dict) and observed.get("status") == "DEAD_LETTER":
_clear_pending(pending_path)
raise ProbeError("central_event_dead_letter")
if not isinstance(observed, dict) or observed.get("status") != "DECIDED" or observed.get("decision_applied") is not True:
raise ProbeError("central_decision_not_proven")
channel = observed.get("trigger_channel")
if channel not in {"workspace_agent", "responses_api"}:
raise ProbeError("central_channel_not_proven")
_clear_pending(pending_path)
now = int(time.time())
atomic_receipt(
Path(config["central_receipt_path"]),
{
"subject": "central_gpt",
"health": "HEALTHY",
"live_probe_succeeded": True,
"observed_at_epoch": now,
"event_id": event_id,
"decision_applied": True,
"state_version_after": observed["state_version_after"],
"workspace_agent": {"validated_live": channel == "workspace_agent"},
"responses_api": {"validated_live": channel == "responses_api"},
},
)
return 0
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--config", required=True, type=Path)
args = parser.parse_args()
try:
return run(args.config)
except (KeyError, OSError, ProbeError, ValueError) as exc:
print(f"CENTRAL_GPT_PROBE=BLOCKED reason={_safe_reason(exc)}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
+113
View File
@@ -0,0 +1,113 @@
#!/usr/bin/env python3
"""Queue one fixed no-side-effect command and prove the watcher lifecycle."""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import sqlite3
import stat
import time
import uuid
from health_probe_common import (
ProbeError, atomic_receipt, gateway_base, load_config, read_token,
request_json, state_version,
)
MARKERS = {"CODEX_AUTO_CLAIM", "TASK_STARTED", "TASK_EXECUTED", "TASK_ENDED", "EVENT_POSTED"}
def _write_command(path: Path, value: dict[str, object]) -> None:
directory = path.parent
info = directory.lstat()
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) & 0o002:
raise ProbeError("unsafe_command_queue")
data = (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o640)
try:
os.write(fd, data)
os.fsync(fd)
finally:
os.close(fd)
def _status(database: Path, command_id: str) -> str | None:
with sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=2) as db:
row = db.execute("SELECT status FROM commands WHERE command_id=?", (command_id,)).fetchone()
return None if row is None else str(row[0])
def _markers(path: Path, command_id: str) -> dict[str, int]:
result = {name: 0 for name in MARKERS}
with path.open("r", encoding="utf-8") as handle:
for line in handle:
try:
item = json.loads(line)
except json.JSONDecodeError:
continue
if item.get("command_id") == command_id and item.get("marker") in result:
result[item["marker"]] += 1
return result
def run(config_path: Path) -> int:
config = load_config(config_path)
endpoint = config["gateway_event_url"]
gateway_base(endpoint)
token = read_token(Path(config["gateway_token_file"]))
command_id = "health-codex-" + uuid.uuid4().hex
version = state_version(endpoint, config["project_key"], token)
command = {
"schema_version": 1,
"command_id": command_id,
"project_key": config["project_key"],
"work_unit_id": command_id,
"contract_commit": config["contract_commit"],
"state_version": version,
"execution": {"kind": "acceptance_probe", "input": "fixed-side-effect-free-health-probe-v1"},
}
_write_command(Path(config["codex_command_queue"]) / f"{command_id}.json", command)
deadline = time.monotonic() + int(config["codex_timeout_seconds"])
database = Path(config["codex_database_path"])
audit = Path(config["codex_audit_log"])
while time.monotonic() < deadline:
if database.exists() and audit.exists() and _status(database, command_id) == "COMPLETED":
counts = _markers(audit, command_id)
if all(counts[name] == 1 for name in MARKERS):
for suffix in ("started", "ended"):
_, event = request_json(endpoint + "/" + command_id + ":" + suffix, token)
if event.get("event_id") != command_id + ":" + suffix:
raise ProbeError("gateway_lifecycle_missing")
atomic_receipt(
Path(config["codex_receipt_path"]),
{
"subject": "codex",
"health": "HEALTHY",
"live_probe_succeeded": True,
"observed_at_epoch": int(time.time()),
"command_id": command_id,
"lifecycle_complete": True,
"duplicate_side_effects": False,
},
)
return 0
time.sleep(float(config["poll_interval_seconds"]))
raise ProbeError("codex_lifecycle_not_proven")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--config", required=True, type=Path)
args = parser.parse_args()
try:
return run(args.config)
except (KeyError, OSError, ProbeError, sqlite3.Error, ValueError):
return 1
if __name__ == "__main__":
raise SystemExit(main())
+141 -21
View File
@@ -18,6 +18,7 @@ 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
@@ -155,6 +156,43 @@ def read_secure_json(path: Path, *, expected_uid: int | None = None) -> dict[str
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,
@@ -694,14 +732,35 @@ def poll_file_inbox(store: QuorumStore, inbox_dir: Path) -> int:
def resolve_git_ref(repo: Path, ref_name: str) -> str:
completed = subprocess.run(
["git", "-C", str(repo), "rev-parse", "--verify", ref_name],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
timeout=10,
)
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")
@@ -743,21 +802,67 @@ def responses_fallback_allowed(
)
def post_json(url: str, payload: Mapping[str, Any], timeout: float = 5.0) -> None:
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"},
headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
method="POST",
)
with urllib.request.urlopen(request, timeout=timeout) as response:
if response.status < 200 or response.status >= 300:
raise urllib.error.HTTPError(url, response.status, "unexpected status", response.headers, None)
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] = {}
@@ -775,16 +880,31 @@ def run_once(config_path: Path) -> int:
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(
f"quorum-{time.time_ns()}",
quorum_event_id,
"gateway",
{
"event_type": "DECISION_APPLIED",
"allowed": decision.allowed,
"priority": decision.priority,
"status": decision.status,
"active_executor": decision.active_executor,
"standby_executor": decision.standby_executor,
"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"]))
@@ -792,7 +912,7 @@ def run_once(config_path: Path) -> int:
if isinstance(git_polling, Mapping):
poll_git_refs(store, Path(git_polling["repository"]), git_polling["refs"])
store.deliver_outbox(
lambda payload: post_json(config["gateway_event_url"], payload),
lambda payload: post_json(gateway_url, payload, token),
max_attempts=int(config["blocker_retry_threshold"]),
)
return 0 if decision.allowed else 2
+146
View File
@@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""Strict metadata-only helpers shared by root health probes."""
from __future__ import annotations
import json
import os
from pathlib import Path
import stat
import tempfile
from typing import Any
import urllib.error
import urllib.parse
import urllib.request
class ProbeError(RuntimeError):
pass
def secure_regular(path: Path, *, max_mode: int, owners: set[int]) -> os.stat_result:
if not path.is_absolute():
raise ProbeError("path_not_absolute")
info = path.lstat()
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
raise ProbeError("unsafe_file_type")
if stat.S_IMODE(info.st_mode) & ~max_mode:
raise ProbeError("unsafe_file_mode")
if os.name == "posix" and info.st_uid not in owners:
raise ProbeError("unsafe_file_owner")
if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
raise ProbeError("symlinked_path_component")
return info
def load_config(path: Path) -> dict[str, Any]:
secure_regular(path, max_mode=0o600, owners={0})
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict):
raise ProbeError("config_not_object")
return value
def read_token(path: Path) -> str:
info = secure_regular(path, max_mode=0o640, owners={0, os.geteuid()})
if info.st_size <= 0 or info.st_size > 16_384:
raise ProbeError("invalid_token_size")
fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
opened = os.fstat(fd)
if not stat.S_ISREG(opened.st_mode) or opened.st_ino != info.st_ino or opened.st_dev != info.st_dev:
raise ProbeError("token_changed_during_validation")
data = os.read(fd, 16_385)
if os.read(fd, 1):
raise ProbeError("invalid_token_size")
finally:
os.close(fd)
if b"\n" in data or b"\r" in data:
raise ProbeError("invalid_token_format")
try:
return data.decode("utf-8")
except UnicodeDecodeError as exc:
raise ProbeError("invalid_token_encoding") from exc
def gateway_base(endpoint: str) -> str:
parsed = urllib.parse.urlparse(endpoint)
if (
parsed.scheme != "http"
or parsed.hostname != "127.0.0.1"
or parsed.port != 18920
or parsed.path != "/v1/orchestrator/events"
or parsed.query
or parsed.fragment
):
raise ProbeError("invalid_gateway_endpoint")
return endpoint.removesuffix("/events")
def request_json(
url: str,
token: str,
*,
method: str = "GET",
payload: dict[str, Any] | None = None,
timeout: float = 10.0,
) -> tuple[int, dict[str, Any]]:
data = None if payload is None else json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
headers = {"Authorization": f"Bearer {token}"}
if data is not None:
headers["Content-Type"] = "application/json"
request = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
body = json.loads(response.read(262_145))
if not isinstance(body, dict):
raise ProbeError("gateway_response_not_object")
return response.status, body
except urllib.error.HTTPError as exc:
if exc.code in {401, 403, 409, 422}:
raise ProbeError(f"gateway_fail_closed_{exc.code}") from exc
raise ProbeError(f"gateway_http_{exc.code}") from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
raise ProbeError("gateway_unavailable") from exc
def state_version(endpoint: str, project_key: str, token: str) -> int:
url = gateway_base(endpoint) + "/state/" + urllib.parse.quote(project_key, safe="")
try:
_, value = request_json(url, token)
except ProbeError as exc:
if str(exc) == "gateway_http_404":
return 0
raise
version = value.get("state_version")
if not isinstance(version, int) or version < 0:
raise ProbeError("invalid_state_version")
return version
def atomic_receipt(path: Path, value: dict[str, Any]) -> None:
if not path.is_absolute() or path.parent != Path("/run/aios-health"):
raise ProbeError("invalid_receipt_path")
parent = path.parent
info = parent.lstat()
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
raise ProbeError("unsafe_receipt_directory")
if os.name == "posix" and (info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o750):
raise ProbeError("unsafe_receipt_directory_metadata")
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=parent)
try:
# Receipts contain availability metadata only. Give the dedicated
# aios-quorum group read access so its unprivileged controller can
# consume them, while retaining root ownership and no public access.
if os.name == "posix":
os.fchown(fd, 0, info.st_gid)
os.fchmod(fd, 0o640)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(value, handle, sort_keys=True, separators=(",", ":"))
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
@@ -1,5 +1,7 @@
{
"$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",
@@ -17,5 +19,6 @@
"blocker_retry_threshold": 3,
"max_health_age_seconds": 60,
"lease_ttl_seconds": 120,
"gateway_event_url": "http://127.0.0.1:8765/v1/events"
"gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
"gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token"
}
+17
View File
@@ -0,0 +1,17 @@
{
"active_dir": "/data/aios-gitops/workspace/agents/commands/active",
"completed_dir": "/data/aios-gitops/workspace/agents/commands/completed",
"dead_letter_dir": "/data/aios-gitops/workspace/agents/commands/dead-letter",
"database_path": "/var/lib/aios-codex-watcher/state.sqlite3",
"audit_log_path": "/var/log/aios-codex-watcher/audit.jsonl",
"gateway_endpoint": "http://127.0.0.1:18920/v1/orchestrator/events",
"gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
"contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
"lease_seconds": 60,
"retry_seconds": 10,
"max_attempts": 5,
"poll_seconds": 5,
"executor_path": null,
"executor_args": [],
"executor_timeout_seconds": 1800
}
+17
View File
@@ -0,0 +1,17 @@
{
"project_key": "v2r1-personal-weixin-production",
"contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
"gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
"gateway_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
"gateway_database_path": "/var/lib/aios-orchestrator-gateway/state.sqlite3",
"probe_clean_base_commit": "c3984ca2b4388875d2875b4fdfdf95a2984f6ab3",
"central_receipt_path": "/run/aios-health/central-gpt.json",
"codex_receipt_path": "/run/aios-health/codex.json",
"codex_command_queue": "/data/aios-gitops/workspace/agents/commands/active",
"codex_database_path": "/var/lib/aios-codex-watcher/state.sqlite3",
"codex_audit_log": "/var/log/aios-codex-watcher/audit.jsonl",
"central_timeout_seconds": 90,
"codex_timeout_seconds": 90,
"cycle_timeout_seconds": 180,
"poll_interval_seconds": 1
}
@@ -0,0 +1,49 @@
{
"host": "127.0.0.1",
"port": 18920,
"database_path": "/var/lib/aios-orchestrator-gateway/state.sqlite3",
"event_inbox_dir": "/var/lib/aios-orchestrator-gateway/inbox/events",
"decision_inbox_dir": "/var/lib/aios-orchestrator-gateway/inbox/decisions",
"dispatch_outbox_dir": "/var/lib/aios-orchestrator-gateway/outbox",
"openclaw_sync_dir": "/var/lib/aios-orchestrator-gateway/openclaw-sync",
"command_queue_dir": "/var/lib/aios-orchestrator-gateway/commands",
"inbound_token_file": "/etc/aios-orchestrator-gateway/inbound.token",
"callback_token_file": "/etc/aios-orchestrator-gateway/callback.token",
"contract_commit": "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc",
"workspace_agent": {
"trigger_id": "agtch_REPLACE_ME",
"access_token_file": "/etc/aios-orchestrator-gateway/workspace-agent.token"
},
"responses_fallback": {
"api_key_file": "/etc/aios-orchestrator-gateway/openai-api.key",
"model": "REPLACE_WITH_APPROVED_MODEL"
},
"polling": {
"interval_seconds": 10,
"summary_interval_seconds": 300
},
"loop_controls": {
"callback_timeout_seconds": 900,
"lease_seconds": 60,
"retry_base_seconds": 10,
"max_retries": 5,
"max_same_blocker_retries": 3,
"max_hops_per_gate": 20
},
"openclaw_sync": {
"endpoint": null,
"token_file": null
},
"project_key": "v2r1-personal-weixin-production",
"git_poll": {
"enabled": true,
"bare_repo": "/data/aios-gitops/repo.git",
"ref_prefixes": [
"refs/heads/main",
"refs/heads/task/governance/",
"refs/heads/patch/",
"refs/heads/result/"
],
"interval_seconds": 30
}
}
@@ -13,6 +13,7 @@ Environment=AIOS_GITOPS_TOKEN_FILE=/data/aios-gitops/runtime/secrets/gitops.toke
Environment=AIOS_GITOPS_BARE_REPO=/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git
Environment=AIOS_GITOPS_HOST=127.0.0.1
Environment=AIOS_GITOPS_PORT=18890
EnvironmentFile=-/etc/aios-gitops/orchestrator-callback.env
ExecStart=/usr/bin/python3 /data/aios-gitops/workspace/active-runtime/app/aios_chatgpt_actions_api.py
Restart=always
RestartSec=3
+36
View File
@@ -29,6 +29,20 @@ A receipt claiming `HEALTHY` must contain:
The central GPT receipt additionally requires either `workspace_agent.validated_live=true` or `responses_api.validated_live=true`. Responses fallback is allowed only after a real Responses decision probe succeeded; credential/configuration presence is insufficient.
The controller and probes use only the fixed loopback endpoint
`http://127.0.0.1:18920/v1/orchestrator/events`. The bearer is read from
`/etc/aios-orchestrator-gateway/inbound.token`; the file must be absolute,
root-owned, regular, non-symlinked, no wider than `0640`, nonempty, and contain
no CR/LF. It is never accepted through argv, environment, logs, receipts, or
Git. HTTP `401`, `403`, `409`, and `422` are terminal fail-closed results for a
probe attempt.
The authenticated metadata query
`GET /v1/orchestrator/events/<event_id>` returns only the event ID, public
status, selected central channel, whether the decision was applied, and the
resulting state version. It never returns the submitted event, model output,
credential, or work-unit instruction.
## Durable behavior
SQLite persists ingress event digests, optimistic task state versions, leases, checkpoints, lifecycle events, effect IDs, outbox messages, blocker counts, observed Git refs, and project summaries. Reusing an event/effect/outbox ID with different content fails closed. An expired lease may be reclaimed; a live lease cannot be stolen.
@@ -55,6 +69,8 @@ Installation is versioned and does not overwrite an existing release:
```bash
sudo scripts/install_automation_quorum_controller.sh "$PWD" 20260718.1
sudo systemctl enable --now aios-automation-quorum-controller.timer
sudo systemctl enable --now aios-central-gpt-health-probe.timer
sudo systemctl enable --now aios-codex-executor-health-probe.timer
```
Run deterministic verification before enablement:
@@ -63,6 +79,26 @@ Run deterministic verification before enablement:
scripts/verify_automation_quorum_controller.sh "$PWD"
```
The two root probes are fixed, bounded, side-effect-free programs. They write
receipts atomically with mode `0600` below the root-owned `0750`
`/run/aios-health`. The Codex probe may write only one `acceptance_probe`
command into the controlled watcher queue; it cannot accept arbitrary command
text from configuration.
Run the strict two-cycle closure only after both services are installed from an
immutable committed release:
```bash
sudo /opt/aios-automation-quorum/current/bin/a0-unattended-loop-probe \
--config /etc/aios-automation-quorum/health-probes.json
```
Each cycle requires a live central decision that dispatches the named Codex
acceptance work unit through the Gateway. The verifier then requires exactly
one claim/start/execute/end/post marker, both lifecycle events in the Gateway,
and a persisted `PROJECT_SUMMARY`. Directly queuing a command cannot satisfy
this two-cycle closure.
Rollback accepts only an existing non-symlink directory below `/opt/aios-automation-quorum/releases`. Repeating rollback to the current target is a no-op:
```bash
@@ -0,0 +1,80 @@
openapi: 3.0.3
info:
title: AIOS Orchestrator Gateway metadata API
version: 0.2.0
servers:
- url: http://127.0.0.1:18920
paths:
/v1/orchestrator/events:
post:
security:
- bearerAuth: []
requestBody:
required: true
content:
application/json:
schema:
type: object
responses:
'202':
description: Event accepted
'409':
description: Idempotency or state conflict
'422':
description: Invalid event
/v1/orchestrator/events/{event_id}:
get:
security:
- bearerAuth: []
parameters:
- in: path
name: event_id
required: true
schema:
type: string
pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$'
responses:
'200':
description: Metadata-only event status
content:
application/json:
schema:
type: object
additionalProperties: false
required: [event_id, status, trigger_http_status, trigger_error_category, trigger_channel, remote_id, decision_applied, state_version_after]
properties:
event_id:
type: string
status:
enum: [RECEIVED, AWAITING_DECISION, DECIDED, TRIGGER_FAILED, DEAD_LETTER]
trigger_channel:
type: string
nullable: true
enum: [workspace_agent, responses_api]
trigger_http_status:
type: integer
nullable: true
minimum: 100
maximum: 599
trigger_error_category:
type: string
nullable: true
enum: [UNAUTHORIZED, FORBIDDEN, NOT_FOUND, NOT_RUNNABLE, NETWORK_ERROR]
remote_id:
type: string
nullable: true
decision_applied:
type: boolean
state_version_after:
type: integer
nullable: true
minimum: 0
'401':
description: Missing or invalid bearer
'404':
description: Event not found
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearer
@@ -5,16 +5,21 @@
"type": "object",
"additionalProperties": false,
"required": [
"project_key",
"contract_commit",
"database_path",
"file_inbox_path",
"health_receipts",
"blocker_retry_threshold",
"max_health_age_seconds",
"lease_ttl_seconds",
"gateway_event_url"
"gateway_event_url",
"gateway_token_file"
],
"properties": {
"$schema": {"type": "string"},
"project_key": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
"contract_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
"database_path": {"type": "string", "pattern": "^/"},
"file_inbox_path": {"type": "string", "pattern": "^/"},
"default_workload": {
@@ -50,6 +55,7 @@
"blocker_retry_threshold": {"type": "integer", "minimum": 1, "maximum": 20},
"max_health_age_seconds": {"type": "integer", "minimum": 5, "maximum": 300},
"lease_ttl_seconds": {"type": "integer", "minimum": 15, "maximum": 3600},
"gateway_event_url": {"type": "string", "pattern": "^https?://"}
"gateway_event_url": {"type": "string", "const": "http://127.0.0.1:18920/v1/orchestrator/events"},
"gateway_token_file": {"type": "string", "pattern": "^/"}
}
}
+25
View File
@@ -0,0 +1,25 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://aios.local/schemas/codex-command.schema.json",
"title": "AIOS Codex command",
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "command_id", "project_key", "work_unit_id", "contract_commit", "state_version", "execution"],
"properties": {
"schema_version": {"const": 1},
"command_id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
"project_key": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
"work_unit_id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"},
"contract_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
"state_version": {"type": "integer", "minimum": 0},
"execution": {
"type": "object",
"additionalProperties": false,
"required": ["kind", "input"],
"properties": {
"kind": {"enum": ["acceptance_probe", "codex"]},
"input": {"type": "string", "maxLength": 65536}
}
}
}
}
+145
View File
@@ -0,0 +1,145 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AIOS Orchestrator Decision",
"type": "object",
"additionalProperties": false,
"required": [
"schema_version",
"project_key",
"contract_commit",
"event_id",
"expected_state_version",
"verdict",
"state_updates",
"next_work_units",
"requires_human_input",
"stop_reason"
],
"properties": {
"schema_version": {
"type": "integer",
"const": 1
},
"project_key": {
"type": "string"
},
"contract_commit": {
"type": "string",
"pattern": "^[0-9a-f]{40}$"
},
"event_id": {
"type": "string"
},
"expected_state_version": {
"type": "integer",
"minimum": 0
},
"verdict": {
"type": "string",
"enum": [
"BLOCKED",
"COMPLETE",
"DISPATCH",
"NOOP"
]
},
"state_updates": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": [
"key",
"value"
],
"properties": {
"key": {
"type": "string"
},
"value": {
"type": "string"
}
}
}
},
"next_work_units": {
"type": "array",
"maxItems": 32,
"items": {
"type": "object",
"additionalProperties": false,
"required": [
"id",
"track_id",
"gate_id",
"primary_agent",
"fallback_agent",
"clean_base_commit",
"allowed_paths",
"required_checks",
"forbidden_operations",
"instruction"
],
"properties": {
"id": {
"type": "string"
},
"track_id": {
"type": "string"
},
"gate_id": {
"type": "string"
},
"primary_agent": {
"type": "string",
"enum": [
"openclaw",
"codex"
]
},
"fallback_agent": {
"type": "string",
"enum": [
"openclaw",
"codex"
]
},
"clean_base_commit": {
"type": "string",
"pattern": "^[0-9a-f]{40}$"
},
"allowed_paths": {
"type": "array",
"items": {
"type": "string"
}
},
"required_checks": {
"type": "array",
"items": {
"type": "string"
}
},
"forbidden_operations": {
"type": "array",
"items": {
"type": "string"
}
},
"instruction": {
"type": "string"
}
}
}
},
"requires_human_input": {
"type": "boolean"
},
"stop_reason": {
"type": [
"string",
"null"
]
}
}
}
+102
View File
@@ -0,0 +1,102 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "AIOS Orchestrator Event",
"type": "object",
"additionalProperties": false,
"required": [
"schema_version",
"project_key",
"contract_commit",
"event_id",
"source_agent",
"event_type",
"work_unit_id",
"status",
"state_version",
"implementation_commit",
"result_commit",
"blocker",
"evidence_refs",
"payload"
],
"properties": {
"schema_version": {
"const": 1
},
"project_key": {
"type": "string",
"minLength": 1,
"maxLength": 128
},
"contract_commit": {
"type": "string",
"pattern": "^[0-9a-f]{40}$"
},
"event_id": {
"type": "string",
"minLength": 1,
"maxLength": 128
},
"source_agent": {
"enum": [
"openclaw",
"codex",
"gitea",
"system",
"workspace-agent"
]
},
"event_type": {
"type": "string"
},
"work_unit_id": {
"type": "string"
},
"status": {
"enum": [
"READY",
"RUNNING",
"PASS",
"BLOCKED",
"FAIL",
"TIMEOUT",
"NOT_EXECUTED"
]
},
"state_version": {
"type": "integer",
"minimum": 0
},
"implementation_commit": {
"type": [
"string",
"null"
],
"pattern": "^[0-9a-f]{40}$"
},
"result_commit": {
"type": [
"string",
"null"
],
"pattern": "^[0-9a-f]{40}$"
},
"blocker": {
"type": [
"string",
"null"
],
"maxLength": 512
},
"evidence_refs": {
"type": "array",
"items": {
"type": "string"
},
"maxItems": 64
},
"payload": {
"type": "object"
}
}
}
+63
View File
@@ -0,0 +1,63 @@
#!/usr/bin/env python3
"""Seed exactly one orchestration event when the project has no durable state."""
from __future__ import annotations
import argparse
from pathlib import Path
import sys
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT))
from services.aios_orchestrator_gateway.config import Config
from services.aios_orchestrator_gateway.models import Event
from services.aios_orchestrator_gateway.securefs import atomic_write_bytes_shared
from services.aios_orchestrator_gateway.store import Store
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--config", type=Path, required=True)
args = parser.parse_args()
config = Config.from_file(args.config)
store = Store(config.database_path)
if store.project_state(config.project_key) is not None:
print("ORCHESTRATION_BOOTSTRAP=NOOP reason=STATE_EXISTS")
return 0
event = Event.from_dict(
{
"schema_version": 1,
"project_key": config.project_key,
"contract_commit": config.contract_commit,
"event_id": "bootstrap-" + config.contract_commit[:24],
"source_agent": "system",
"event_type": "orchestration_bootstrap",
"work_unit_id": "ORCHESTRATION-BOOTSTRAP",
"status": "READY",
"state_version": 0,
"implementation_commit": None,
"result_commit": None,
"blocker": None,
"evidence_refs": [],
"payload": {
"reason": "initial_gateway_install",
"polling_enabled": True,
"openclaw_sync_required": True,
},
}
)
path = config.event_inbox_dir / "pending" / f"{event.event_id}.json"
atomic_write_bytes_shared(
path,
(event.canonical_json() + "\n").encode(),
mode=0o640,
group_writable_directory=True,
)
print(f"ORCHESTRATION_BOOTSTRAP=PASS event_id={event.event_id}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+64
View File
@@ -0,0 +1,64 @@
#!/usr/bin/env bash
set -euo pipefail
[[ ${EUID} -eq 0 ]] || { echo "root required" >&2; exit 1; }
ROOT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd -P)
CONFIG=${1:-/etc/aios-codex-watcher/config.json}
[[ "$CONFIG" = /* ]] || { echo "config path must be absolute" >&2; exit 1; }
[[ -f "$CONFIG" && ! -L "$CONFIG" ]] || { echo "pre-provisioned regular config required" >&2; exit 1; }
for path in \
services/aios_codex_watcher/__init__.py \
services/aios_codex_watcher/__main__.py \
services/aios_codex_watcher/watcher.py \
systemd/aios-codex-watcher.service \
systemd/aios-codex-watcher.timer \
systemd/aios-codex-watcher.sysusers \
systemd/aios-codex-watcher.tmpfiles; do
[[ -f "$ROOT/$path" && ! -L "$ROOT/$path" ]] || { echo "missing or unsafe: $path" >&2; exit 1; }
done
digest=$(sha256sum \
"$ROOT/services/aios_codex_watcher/__init__.py" \
"$ROOT/services/aios_codex_watcher/__main__.py" \
"$ROOT/services/aios_codex_watcher/watcher.py" | sha256sum | awk '{print $1}')
release="/opt/aios-codex-watcher/releases/$digest"
install -d -o root -g root -m 0755 "$release/services/aios_codex_watcher" "$release/bin"
install -o root -g root -m 0644 "$ROOT/services/aios_codex_watcher/"*.py "$release/services/aios_codex_watcher/"
launcher="$release/bin/aios-codex-watcher"
printf '%s\n' '#!/usr/bin/env bash' 'set -euo pipefail' \
"export PYTHONPATH='$release'" \
"exec /usr/bin/python3 -m services.aios_codex_watcher \"\$@\"" >"$launcher"
chown root:root "$launcher"
chmod 0755 "$launcher"
install -d -o root -g root -m 0755 /opt/aios-codex-watcher /etc/aios-codex-watcher
ln -sfn "releases/$digest" /opt/aios-codex-watcher/current.new
mv -Tf /opt/aios-codex-watcher/current.new /opt/aios-codex-watcher/current
install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.service" /etc/systemd/system/
install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.timer" /etc/systemd/system/
install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.sysusers" /usr/lib/sysusers.d/
install -o root -g root -m 0644 "$ROOT/systemd/aios-codex-watcher.tmpfiles" /usr/lib/tmpfiles.d/
systemd-sysusers /usr/lib/sysusers.d/aios-codex-watcher.sysusers
systemd-tmpfiles --create /usr/lib/tmpfiles.d/aios-codex-watcher.tmpfiles
commands=/data/aios-gitops/workspace/agents/commands
for directory in "$commands/active" "$commands/completed"; do
[[ -d "$directory" && ! -L "$directory" ]] || { echo "missing or unsafe command directory: $directory" >&2; exit 1; }
[[ $(stat -c '%U:%G' "$directory") == aios-orchestrator:aios-orchestrator ]] || { echo "unexpected command directory owner: $directory" >&2; exit 1; }
mode=$(stat -c '%a' "$directory")
(( (8#$mode & 2) == 0 )) || { echo "world-writable command directory: $directory" >&2; exit 1; }
done
if [[ -e "$commands/dead-letter" || -L "$commands/dead-letter" ]]; then
[[ -d "$commands/dead-letter" && ! -L "$commands/dead-letter" ]] || { echo "unsafe dead-letter directory" >&2; exit 1; }
else
install -d -o aios-orchestrator -g aios-orchestrator -m 0750 "$commands/dead-letter"
fi
[[ $(stat -c '%U:%G' "$commands/dead-letter") == aios-orchestrator:aios-orchestrator ]] || { echo "unexpected dead-letter owner" >&2; exit 1; }
mode=$(stat -c '%a' "$commands/dead-letter")
(( (8#$mode & 2) == 0 )) || { echo "world-writable dead-letter directory" >&2; exit 1; }
printf '{"release_sha256":"%s","installed_at":"%s"}\n' "$digest" "$(date -u +%FT%TZ)" >"$release/receipt.json"
chmod 0644 "$release/receipt.json"
systemctl daemon-reload
systemctl enable --now aios-codex-watcher.timer
printf 'installed_sha256=%s\n' "$digest"
+305
View File
@@ -0,0 +1,305 @@
#!/usr/bin/env bash
set -Eeuo pipefail
umask 0077
SOURCE_ROOT="${SOURCE_ROOT:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}"
INSTALL_ROOT="${INSTALL_ROOT:-/opt/aios-orchestrator-gateway}"
CONFIG_ROOT="${CONFIG_ROOT:-/etc/aios-orchestrator-gateway}"
STATE_ROOT="${STATE_ROOT:-/var/lib/aios-orchestrator-gateway}"
UNIT_ROOT="${UNIT_ROOT:-/etc/systemd/system}"
SYSUSERS_ROOT="${SYSUSERS_ROOT:-/etc/sysusers.d}"
TMPFILES_ROOT="${TMPFILES_ROOT:-/etc/tmpfiles.d}"
DRY_RUN=0
if [[ "${1:-}" == "--dry-run" ]]; then
DRY_RUN=1
fi
fail() {
printf 'AIOS_ORCHESTRATOR_INSTALL=BLOCKED reason=%s\n' "$1" >&2
exit 2
}
[[ "$(id -u)" -eq 0 ]] || fail OWNER_INSTALL_REQUIRED
for cmd in git python3 systemd-analyze systemctl systemd-sysusers systemd-tmpfiles sha256sum find sort xargs; do
command -v "$cmd" >/dev/null 2>&1 || fail "MISSING_COMMAND_${cmd}"
done
if [[ -f "$SOURCE_ROOT/SHA256SUMS" ]]; then
(cd "$SOURCE_ROOT" && sha256sum --check SHA256SUMS)
fi
for path in \
"$SOURCE_ROOT/services/aios_orchestrator_gateway" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway.service" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.service" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles" \
"$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json"; do
[[ -e "$path" ]] || fail "MISSING_SOURCE_${path}"
done
PYCACHE="/tmp/aios-orchestrator-install-pycache-$$"
rm -rf "$PYCACHE"
PYTHONPYCACHEPREFIX="$PYCACHE" python3 -m compileall -q "$SOURCE_ROOT/services" "$SOURCE_ROOT/scripts"
PYTHONDONTWRITEBYTECODE=1 PYTHONPATH="$SOURCE_ROOT" \
python3 -m unittest -v tests.test_aios_orchestrator_gateway
rm -rf "$PYCACHE"
for script in "$SOURCE_ROOT"/scripts/*.sh; do bash -n "$script"; done
python3 -m json.tool "$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json" >/dev/null
systemd-analyze verify \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway.service" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.service" \
"$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer"
SOURCE_FILES=(
README-AIOS-ORCHESTRATOR-GATEWAY.md
config/aios-orchestrator-gateway.example.json
schemas/decision.schema.json
schemas/event.schema.json
scripts/bootstrap_orchestration_loop.py
scripts/install_aios_orchestrator_gateway.sh
scripts/report_orchestrator_event.py
scripts/rollback_aios_orchestrator_gateway.sh
scripts/verify_aios_orchestrator_gateway.sh
systemd/aios-orchestrator-gateway-poll.service
systemd/aios-orchestrator-gateway-poll.timer
systemd/aios-orchestrator-gateway.service
systemd/aios-orchestrator-gateway.sysusers
systemd/aios-orchestrator-gateway.tmpfiles
tests/test_aios_orchestrator_gateway.py
tests/test_aios_orchestrator_gateway_rollback.sh
workspace-agent/CENTRAL_AGENT_INSTRUCTIONS.md
workspace-agent/orchestrator-decision-action.openapi.yaml
openapi/aios-orchestrator-gateway.openapi.yaml
)
while IFS= read -r path; do SOURCE_FILES+=("$path"); done < <(
cd "$SOURCE_ROOT"
find services/aios_orchestrator_gateway -type f ! -path '*/__pycache__/*' ! -name '*.pyc' -print | sort
)
for path in "${SOURCE_FILES[@]}"; do
[[ -f "$SOURCE_ROOT/$path" && ! -L "$SOURCE_ROOT/$path" ]] || fail "INVALID_SOURCE_FILE_${path}"
done
SOURCE_DIGEST="$(
cd "$SOURCE_ROOT"
printf '%s\0' "${SOURCE_FILES[@]}" | sort -z | xargs -0 sha256sum \
| sha256sum \
| awk '{print $1}'
)"
[[ "$SOURCE_DIGEST" =~ ^[0-9a-f]{64}$ ]] || fail SOURCE_DIGEST_INVALID
RELEASE_DIR="$INSTALL_ROOT/releases/$SOURCE_DIGEST"
if [[ "$DRY_RUN" -eq 1 ]]; then
printf 'AIOS_ORCHESTRATOR_INSTALL_DRY_RUN=PASS digest=%s release=%s\n' \
"$SOURCE_DIGEST" "$RELEASE_DIR"
exit 0
fi
systemd-sysusers "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers"
install -d -m 0755 "$INSTALL_ROOT/releases"
install -d -o root -g aios-orchestrator -m 0750 "$CONFIG_ROOT"
install -d -m 0700 "$STATE_ROOT"
systemd-tmpfiles --create "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles"
if [[ ! -d "$RELEASE_DIR" ]]; then
TMP_RELEASE="$INSTALL_ROOT/releases/.tmp-$SOURCE_DIGEST-$$"
install -d -m 0755 "$TMP_RELEASE"
for path in "${SOURCE_FILES[@]}"; do
install -d -m 0755 "$TMP_RELEASE/$(dirname "$path")"
cp -a "$SOURCE_ROOT/$path" "$TMP_RELEASE/$path"
done
printf '%s\n' "$SOURCE_DIGEST" > "$TMP_RELEASE/SOURCE_DIGEST"
chmod -R go-w "$TMP_RELEASE"
mv "$TMP_RELEASE" "$RELEASE_DIR"
fi
PREVIOUS_TARGET=""
if [[ -L "$INSTALL_ROOT/current" ]]; then
PREVIOUS_TARGET="$(readlink -f "$INSTALL_ROOT/current")"
fi
install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.service" \
"$UNIT_ROOT/aios-orchestrator-gateway.service"
install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.service" \
"$UNIT_ROOT/aios-orchestrator-gateway-poll.service"
install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway-poll.timer" \
"$UNIT_ROOT/aios-orchestrator-gateway-poll.timer"
install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.sysusers" \
"$SYSUSERS_ROOT/aios-orchestrator-gateway.conf"
install -m 0644 "$SOURCE_ROOT/systemd/aios-orchestrator-gateway.tmpfiles" \
"$TMPFILES_ROOT/aios-orchestrator-gateway.conf"
if [[ ! -f "$CONFIG_ROOT/inbound.token" ]]; then
python3 - <<'PY' > "$CONFIG_ROOT/inbound.token"
import secrets, sys
sys.stdout.write(secrets.token_urlsafe(48))
PY
fi
[[ ! -L "$CONFIG_ROOT/inbound.token" ]] || fail INBOUND_TOKEN_SYMLINK
chown root:aios-orchestrator "$CONFIG_ROOT/inbound.token"
chmod 0640 "$CONFIG_ROOT/inbound.token"
if [[ ! -f "$CONFIG_ROOT/callback.token" ]]; then
python3 - <<'PY' > "$CONFIG_ROOT/callback.token"
import secrets, sys
sys.stdout.write(secrets.token_urlsafe(48))
PY
fi
[[ ! -L "$CONFIG_ROOT/callback.token" ]] || fail CALLBACK_TOKEN_SYMLINK
chown root:aios-orchestrator "$CONFIG_ROOT/callback.token"
chmod 0640 "$CONFIG_ROOT/callback.token"
# Historical installers appended one LF. Remove only that exact legacy byte in
# place; reject every other newline form without ever copying or printing data.
python3 - "$CONFIG_ROOT/config.json" "$CONFIG_ROOT/inbound.token" "$CONFIG_ROOT/callback.token" <<'PY'
import json, os, pathlib, stat, sys
config_path = pathlib.Path(sys.argv[1])
names = list(sys.argv[2:])
if config_path.is_file() and not config_path.is_symlink():
config = json.loads(config_path.read_text())
candidates = [
config.get("inbound_token_file"),
config.get("callback_token_file"),
(config.get("workspace_agent") or {}).get("access_token_file"),
(config.get("responses_fallback") or {}).get("api_key_file"),
(config.get("openclaw_sync") or {}).get("bearer_token_file"),
]
names.extend(name for name in candidates if isinstance(name, str) and name)
for name in dict.fromkeys(names):
path = pathlib.Path(name)
if not path.is_absolute() or path.is_symlink() or path.resolve() != path:
raise SystemExit("invalid token path")
fd = os.open(name, os.O_RDWR | getattr(os, "O_NOFOLLOW", 0))
try:
info = os.fstat(fd)
if not stat.S_ISREG(info.st_mode) or info.st_size <= 0 or info.st_size > 16384:
raise SystemExit("invalid token metadata")
data = os.read(fd, info.st_size)
if data.endswith(b"\n") and b"\n" not in data[:-1] and b"\r" not in data:
os.ftruncate(fd, info.st_size - 1)
os.fsync(fd)
elif b"\n" in data or b"\r" in data:
raise SystemExit("invalid token format")
finally:
os.close(fd)
PY
if [[ ! -f "$CONFIG_ROOT/config.json" ]]; then
install -m 0600 \
"$SOURCE_ROOT/config/aios-orchestrator-gateway.example.json" \
"$CONFIG_ROOT/config.json.example"
printf 'AIOS_ORCHESTRATOR_INSTALL=BLOCKED reason=CONFIGURATION_REQUIRED digest=%s\n' \
"$SOURCE_DIGEST"
exit 2
fi
chown root:aios-orchestrator "$CONFIG_ROOT/config.json"
chmod 0640 "$CONFIG_ROOT/config.json"
python3 -m json.tool "$CONFIG_ROOT/config.json" >/dev/null
AIOS_ORCHESTRATOR_CONFIG_OWNER_UID=0 \
AIOS_ORCHESTRATOR_SECRET_OWNER_UID=0 \
PYTHONPATH="$RELEASE_DIR" \
python3 - "$CONFIG_ROOT/config.json" <<'PY'
import sys
from pathlib import Path
from services.aios_orchestrator_gateway.config import Config
cfg = Config.from_file(Path(sys.argv[1]))
cfg.inbound_token()
cfg.callback_token()
if cfg.workspace_trigger_id:
cfg.workspace_token()
if cfg.responses_api_key_file:
cfg.responses_api_key()
if cfg.openclaw_sync_endpoint:
cfg.openclaw_sync_token()
print("CONFIG_VALIDATE=PASS")
PY
# Commit the release pointer only after every candidate/config/secret check has
# succeeded. A failed validation therefore leaves the running release intact.
ln -s "$RELEASE_DIR" "$INSTALL_ROOT/.current-new-$$"
mv -Tf "$INSTALL_ROOT/.current-new-$$" "$INSTALL_ROOT/current"
python3 - "$STATE_ROOT/install-receipt.json" "$SOURCE_DIGEST" "$RELEASE_DIR" "$PREVIOUS_TARGET" <<'PY'
import json, os, sys, tempfile
from pathlib import Path
path = Path(sys.argv[1])
data = {
"schema_version": 1,
"source_digest": sys.argv[2],
"current_release": sys.argv[3],
"previous_release": sys.argv[4] or None,
"last_rollback": None,
}
path.parent.mkdir(parents=True, exist_ok=True)
fd, temp = tempfile.mkstemp(prefix=".install-receipt-", dir=path.parent)
try:
with os.fdopen(fd, "w") as handle:
json.dump(data, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.chmod(temp, 0o600)
os.replace(temp, path)
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(directory)
finally:
os.close(directory)
finally:
try:
os.unlink(temp)
except FileNotFoundError:
pass
PY
chown root:root "$STATE_ROOT/install-receipt.json"
chmod 0600 "$STATE_ROOT/install-receipt.json"
systemctl daemon-reload
systemctl enable aios-orchestrator-gateway.service
systemctl enable aios-orchestrator-gateway-poll.timer
systemctl stop aios-orchestrator-gateway-poll.timer || true
systemctl stop aios-orchestrator-gateway-poll.service || true
rollback_activation() {
if [[ -n "$PREVIOUS_TARGET" && -d "$PREVIOUS_TARGET" ]]; then
bash "$RELEASE_DIR/scripts/rollback_aios_orchestrator_gateway.sh" || true
else
systemctl stop aios-orchestrator-gateway.service || true
fi
fail ACTIVATION_FAILED_ROLLED_BACK
}
systemctl restart aios-orchestrator-gateway.service || rollback_activation
python3 - <<'PY' || rollback_activation
import json, time, urllib.request
for _ in range(30):
try:
with urllib.request.urlopen("http://127.0.0.1:18920/healthz", timeout=1) as response:
body = json.loads(response.read(65536))
if response.status == 200 and body.get("status") == "active":
raise SystemExit(0)
except Exception:
pass
time.sleep(0.5)
raise SystemExit(1)
PY
systemctl start aios-orchestrator-gateway-poll.timer || rollback_activation
systemctl is-active --quiet aios-orchestrator-gateway.service
systemctl is-active --quiet aios-orchestrator-gateway-poll.timer
AIOS_ORCHESTRATOR_CONFIG_OWNER_UID=0 \
AIOS_ORCHESTRATOR_SECRET_OWNER_UID=0 \
PYTHONPATH="$INSTALL_ROOT/current" \
python3 "$INSTALL_ROOT/current/scripts/bootstrap_orchestration_loop.py" \
--config "$CONFIG_ROOT/config.json"
systemctl start aios-orchestrator-gateway-poll.service || rollback_activation
printf 'AIOS_ORCHESTRATOR_INSTALL=PASS digest=%s previous=%s\n' \
"$SOURCE_DIGEST" "${PREVIOUS_TARGET:-NONE}"
@@ -17,12 +17,21 @@ SOURCE_REAL=$(readlink -f -- "${SOURCE_ROOT}" 2>/dev/null || true)
SOURCE_FILES=(
automation/executor_quorum_controller.py
automation/health_probe_common.py
automation/central_gpt_health_probe.py
automation/codex_executor_health_probe.py
automation/a0_unattended_loop_probe.py
schemas/automation-quorum-controller.schema.json
config/aios-automation-quorum-controller.example.json
config/aios-health-probes.example.json
systemd/aios-automation-quorum-controller.service
systemd/aios-automation-quorum-controller.timer
systemd/aios-automation-quorum-controller.sysusers
systemd/aios-automation-quorum-controller.tmpfiles
systemd/aios-central-gpt-health-probe.service
systemd/aios-central-gpt-health-probe.timer
systemd/aios-codex-executor-health-probe.service
systemd/aios-codex-executor-health-probe.timer
)
for relative in "${SOURCE_FILES[@]}"; do
[[ -f ${SOURCE_ROOT}/${relative} && ! -L ${SOURCE_ROOT}/${relative} ]] || {
@@ -38,12 +47,28 @@ VERSION_DIR=${INSTALL_ROOT}/releases/${VERSION}
install -d -o root -g root -m 0755 "${VERSION_DIR}/bin" "${VERSION_DIR}/share"
install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/executor_quorum_controller.py" \
"${VERSION_DIR}/bin/executor-quorum-controller"
install -o root -g root -m 0644 "${SOURCE_ROOT}/automation/health_probe_common.py" \
"${VERSION_DIR}/bin/health_probe_common.py"
install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/central_gpt_health_probe.py" \
"${VERSION_DIR}/bin/central-gpt-health-probe"
install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/codex_executor_health_probe.py" \
"${VERSION_DIR}/bin/codex-executor-health-probe"
install -o root -g root -m 0755 "${SOURCE_ROOT}/automation/a0_unattended_loop_probe.py" \
"${VERSION_DIR}/bin/a0-unattended-loop-probe"
install -o root -g root -m 0644 "${SOURCE_ROOT}/schemas/automation-quorum-controller.schema.json" \
"${VERSION_DIR}/share/automation-quorum-controller.schema.json"
python3 -m py_compile "${VERSION_DIR}/bin/executor-quorum-controller"
python3 -m py_compile "${VERSION_DIR}/bin/executor-quorum-controller" \
"${VERSION_DIR}/bin/health_probe_common.py" \
"${VERSION_DIR}/bin/central-gpt-health-probe" \
"${VERSION_DIR}/bin/codex-executor-health-probe"
python3 -m py_compile "${VERSION_DIR}/bin/a0-unattended-loop-probe"
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.service" /etc/systemd/system/
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.timer" /etc/systemd/system/
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-central-gpt-health-probe.service" /etc/systemd/system/
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-central-gpt-health-probe.timer" /etc/systemd/system/
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-codex-executor-health-probe.service" /etc/systemd/system/
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-codex-executor-health-probe.timer" /etc/systemd/system/
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.sysusers" /usr/lib/sysusers.d/aios-automation-quorum-controller.conf
install -o root -g root -m 0644 "${SOURCE_ROOT}/systemd/aios-automation-quorum-controller.tmpfiles" /usr/lib/tmpfiles.d/aios-automation-quorum-controller.conf
@@ -55,6 +80,11 @@ if [[ ! -e /etc/aios-automation-quorum/controller.json ]]; then
"${SOURCE_ROOT}/config/aios-automation-quorum-controller.example.json" \
/etc/aios-automation-quorum/controller.json
fi
if [[ ! -e /etc/aios-automation-quorum/health-probes.json ]]; then
install -o root -g root -m 0600 \
"${SOURCE_ROOT}/config/aios-health-probes.example.json" \
/etc/aios-automation-quorum/health-probes.json
fi
PREVIOUS_TARGET=$(readlink -f "${INSTALL_ROOT}/current" 2>/dev/null || true)
RECEIPT=${VERSION_DIR}/install-receipt.json
+130
View File
@@ -0,0 +1,130 @@
#!/usr/bin/env python3
"""Submit an OpenClaw/Codex task lifecycle event over HTTP or a polled inbox."""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import sys
import urllib.error
import urllib.request
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT))
from services.aios_orchestrator_gateway.models import Event
from services.aios_orchestrator_gateway.securefs import atomic_write_bytes_shared, read_secure_json, read_secure_text
def parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser()
p.add_argument("--project-key", required=True)
p.add_argument("--contract-commit", required=True)
p.add_argument("--event-id", required=True)
p.add_argument("--source-agent", required=True, choices=["openclaw", "codex", "gitea", "system"])
p.add_argument("--event-type", required=True)
p.add_argument("--work-unit-id", required=True)
p.add_argument(
"--status",
required=True,
choices=["READY", "RUNNING", "PASS", "BLOCKED", "FAIL", "TIMEOUT", "NOT_EXECUTED"],
)
p.add_argument("--state-version", type=int, required=True)
p.add_argument("--implementation-commit")
p.add_argument("--result-commit")
p.add_argument("--blocker")
p.add_argument("--evidence-ref", action="append", default=[])
p.add_argument("--payload-file", type=Path)
p.add_argument("--endpoint")
p.add_argument("--token-file", type=Path)
p.add_argument("--inbox-dir", type=Path)
p.add_argument("--timeout-seconds", type=int, default=15)
return p
def main() -> int:
args = parser().parse_args()
if not args.endpoint and not args.inbox_dir:
raise SystemExit("either --endpoint or --inbox-dir is required")
if args.endpoint and not args.token_file:
raise SystemExit("--token-file is required with --endpoint")
payload = {}
if args.payload_file:
if not args.payload_file.is_absolute():
raise SystemExit("--payload-file must be absolute")
payload = read_secure_json(
args.payload_file, expected_uid=os.getuid(), max_bytes=262_144
)
if not isinstance(payload, dict):
raise SystemExit("payload root must be an object")
event = Event.from_dict(
{
"schema_version": 1,
"project_key": args.project_key,
"contract_commit": args.contract_commit,
"event_id": args.event_id,
"source_agent": args.source_agent,
"event_type": args.event_type,
"work_unit_id": args.work_unit_id,
"status": args.status,
"state_version": args.state_version,
"implementation_commit": args.implementation_commit,
"result_commit": args.result_commit,
"blocker": args.blocker,
"evidence_refs": args.evidence_ref,
"payload": payload,
}
)
encoded = event.canonical_json().encode()
http_error: Exception | None = None
if args.endpoint:
token = read_secure_text(
args.token_file,
expected_uid=None,
max_bytes=64_000,
)
request = urllib.request.Request(
args.endpoint.rstrip("/") + "/v1/orchestrator/events",
data=encoded,
method="POST",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Idempotency-Key": event.event_id,
"User-Agent": "aios-agent-event-reporter/0.2",
},
)
try:
with urllib.request.urlopen(request, timeout=args.timeout_seconds) as response:
response.read(65_536)
if response.status not in {200, 202}:
raise RuntimeError(f"unexpected HTTP {response.status}")
print(f"EVENT_SUBMIT=PASS event_id={event.event_id} channel=http")
return 0
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError) as exc:
http_error = exc
if args.inbox_dir:
pending = args.inbox_dir / "pending"
path = pending / f"{event.event_id}.json"
atomic_write_bytes_shared(
path,
(json.dumps(event.to_dict(), indent=2, sort_keys=True) + "\n").encode(),
mode=0o640,
group_writable_directory=True,
)
print(f"EVENT_SUBMIT=PASS event_id={event.event_id} channel=file_inbox")
return 0
print(
f"EVENT_SUBMIT=FAIL event_id={event.event_id} reason={type(http_error).__name__}",
file=sys.stderr,
)
return 2
if __name__ == "__main__":
raise SystemExit(main())
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env bash
set -Eeuo pipefail
umask 0077
INSTALL_ROOT="${INSTALL_ROOT:-/opt/aios-orchestrator-gateway}"
STATE_ROOT="${STATE_ROOT:-/var/lib/aios-orchestrator-gateway}"
RECEIPT="$STATE_ROOT/install-receipt.json"
fail() {
printf 'AIOS_ORCHESTRATOR_ROLLBACK=BLOCKED reason=%s\n' "$1" >&2
exit 2
}
[[ "$(id -u)" -eq 0 ]] || fail OWNER_REQUIRED
[[ -f "$RECEIPT" ]] || fail RECEIPT_MISSING
[[ ! -L "$RECEIPT" ]] || fail RECEIPT_SYMLINK
[[ "$(stat -c '%u:%a' "$RECEIPT")" == "0:600" ]] || fail RECEIPT_PERMISSION
readarray -t VALUES < <(
python3 - "$RECEIPT" "$INSTALL_ROOT/releases" "$INSTALL_ROOT/current" <<'PY'
import json, os, stat, sys
from pathlib import Path
receipt = Path(sys.argv[1])
releases = Path(sys.argv[2]).resolve()
current_link = Path(sys.argv[3])
data = json.loads(receipt.read_text())
previous = data.get("previous_release")
current = data.get("current_release")
if not previous or not current:
raise SystemExit(2)
previous_path = Path(previous).resolve()
current_path = Path(current).resolve()
if previous_path.parent != releases or current_path.parent != releases:
raise SystemExit(2)
if not stat.S_ISDIR(os.lstat(previous_path).st_mode) or not stat.S_ISDIR(os.lstat(current_path).st_mode):
raise SystemExit(2)
for release in (previous_path, current_path):
if len(release.name) != 64 or any(ch not in "0123456789abcdef" for ch in release.name):
raise SystemExit(2)
digest_file = release / "SOURCE_DIGEST"
if digest_file.is_symlink() or not digest_file.is_file():
raise SystemExit(2)
if digest_file.read_text(encoding="utf-8").strip() != release.name:
raise SystemExit(2)
last = data.get("last_rollback")
if isinstance(last, dict) and last.get("to") == current and current_link.is_symlink():
if current_link.resolve() == current_path:
print("ALREADY_ROLLED_BACK")
raise SystemExit(0)
print(previous_path)
print(current_path)
PY
) || fail RECEIPT_INVALID
if [[ "${VALUES[0]:-}" == "ALREADY_ROLLED_BACK" ]]; then
echo "AIOS_ORCHESTRATOR_ROLLBACK=PASS result=NOOP_ALREADY_ROLLED_BACK"
exit 0
fi
[[ "${#VALUES[@]}" -eq 2 ]] || fail RECEIPT_INVALID
PREVIOUS="${VALUES[0]}"
CURRENT="${VALUES[1]}"
systemctl stop aios-orchestrator-gateway-poll.timer || true
systemctl stop aios-orchestrator-gateway-poll.service || true
systemctl stop aios-orchestrator-gateway.service || true
ln -s "$PREVIOUS" "$INSTALL_ROOT/.current-rollback-$$"
mv -Tf "$INSTALL_ROOT/.current-rollback-$$" "$INSTALL_ROOT/current"
systemctl start aios-orchestrator-gateway.service
systemctl is-active --quiet aios-orchestrator-gateway.service
systemctl start aios-orchestrator-gateway-poll.timer
systemctl is-active --quiet aios-orchestrator-gateway-poll.timer
python3 - "$RECEIPT" "$PREVIOUS" "$CURRENT" <<'PY'
import json, os, sys, tempfile
from pathlib import Path
path = Path(sys.argv[1])
data = json.loads(path.read_text())
data["current_release"] = sys.argv[2]
data["previous_release"] = sys.argv[3]
data["last_rollback"] = {"from": sys.argv[3], "to": sys.argv[2]}
fd, temp = tempfile.mkstemp(prefix=".rollback-receipt-", dir=path.parent)
try:
with os.fdopen(fd, "w") as handle:
json.dump(data, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.chmod(temp, 0o600)
os.chown(temp, 0, 0)
os.replace(temp, path)
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(directory)
finally:
os.close(directory)
finally:
try:
os.unlink(temp)
except FileNotFoundError:
pass
PY
printf 'AIOS_ORCHESTRATOR_ROLLBACK=PASS current=%s previous=%s\n' "$PREVIOUS" "$CURRENT"
+20 -2
View File
@@ -10,6 +10,8 @@ from urllib.parse import urlparse
REQUIRED = {
"project_key",
"contract_commit",
"database_path",
"file_inbox_path",
"health_receipts",
@@ -17,6 +19,7 @@ REQUIRED = {
"max_health_age_seconds",
"lease_ttl_seconds",
"gateway_event_url",
"gateway_token_file",
}
OPTIONAL = {"$schema", "default_workload", "git_ref_polling"}
WORKLOADS = {
@@ -53,6 +56,13 @@ def validate(config: object, schema: object) -> None:
fail("missing or unexpected top-level config fields")
absolute_path(config["database_path"], "database_path")
absolute_path(config["file_inbox_path"], "file_inbox_path")
absolute_path(config["gateway_token_file"], "gateway_token_file")
project_key = config["project_key"]
if not isinstance(project_key, str) or not project_key or len(project_key) > 128:
fail("project_key is invalid")
commit = config["contract_commit"]
if not isinstance(commit, str) or len(commit) != 40 or any(c not in "0123456789abcdef" for c in commit):
fail("contract_commit must be a full lowercase Git OID")
receipts = config["health_receipts"]
if not isinstance(receipts, dict) or set(receipts) != {"central_gpt", "openclaw", "codex"}:
fail("health_receipts must contain exactly central_gpt, openclaw, and codex")
@@ -75,8 +85,16 @@ def validate(config: object, schema: object) -> None:
bounded_integer(config["lease_ttl_seconds"], "lease_ttl_seconds", 15, 3600)
url = config["gateway_event_url"]
parsed = urlparse(url) if isinstance(url, str) else None
if not parsed or parsed.scheme not in {"http", "https"} or not parsed.hostname:
fail("gateway_event_url must be an HTTP(S) URL with a host")
if (
not parsed
or parsed.scheme != "http"
or parsed.hostname != "127.0.0.1"
or parsed.port != 18920
or parsed.path != "/v1/orchestrator/events"
or parsed.query
or parsed.fragment
):
fail("gateway_event_url must be the fixed loopback orchestrator endpoint")
def main() -> int:
+32
View File
@@ -0,0 +1,32 @@
#!/usr/bin/env bash
set -Eeuo pipefail
CONFIG="${1:-/etc/aios-orchestrator-gateway/config.json}"
PORT="${PORT:-18920}"
python3 -m json.tool "$CONFIG" >/dev/null
systemd-analyze verify \
/etc/systemd/system/aios-orchestrator-gateway.service \
/etc/systemd/system/aios-orchestrator-gateway-poll.service \
/etc/systemd/system/aios-orchestrator-gateway-poll.timer
systemctl is-active --quiet aios-orchestrator-gateway.service
systemctl is-active --quiet aios-orchestrator-gateway-poll.timer
health="$(curl --fail --silent --show-error "http://127.0.0.1:${PORT}/healthz")"
python3 -c '
import json, sys
v=json.load(sys.stdin)
assert v["service"]=="aios-orchestrator-gateway"
assert v["status"]=="active"
assert v["polling"] is True
assert len(v["contract_commit"])==40
' <<<"$health"
systemctl start aios-orchestrator-gateway-poll.service
if journalctl -u aios-orchestrator-gateway.service -n 200 --no-pager \
| grep -Eqi 'Bearer |token=|api[_-]?key|secret='; then
echo "AIOS_ORCHESTRATOR_VERIFY=BLOCKED reason=SECRET_PATTERN_IN_LOG" >&2
exit 2
fi
echo "AIOS_ORCHESTRATOR_VERIFY=PASS"
+13 -1
View File
@@ -4,7 +4,13 @@ set -euo pipefail
ROOT=${1:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}
python3 -m py_compile "${ROOT}/automation/executor_quorum_controller.py"
python3 -m py_compile \
"${ROOT}/automation/health_probe_common.py" \
"${ROOT}/automation/central_gpt_health_probe.py" \
"${ROOT}/automation/codex_executor_health_probe.py" \
"${ROOT}/automation/a0_unattended_loop_probe.py"
python3 -m unittest discover -s "${ROOT}/tests" -p 'test_executor_quorum_controller.py' -v
python3 -m unittest discover -s "${ROOT}/tests" -p 'test_aios_health_probes.py' -v
python3 -m json.tool "${ROOT}/config/aios-automation-quorum-controller.example.json" >/dev/null
python3 -m json.tool "${ROOT}/schemas/automation-quorum-controller.schema.json" >/dev/null
python3 "${ROOT}/scripts/validate_automation_quorum_schema.py" \
@@ -14,4 +20,10 @@ python3 -m py_compile "${ROOT}/scripts/validate_automation_quorum_schema.py"
bash -n "${ROOT}/scripts/install_automation_quorum_controller.sh"
bash -n "${ROOT}/scripts/rollback_automation_quorum_controller.sh"
bash -n "${ROOT}/scripts/verify_automation_quorum_controller.sh"
systemd-analyze verify "${ROOT}/systemd/aios-automation-quorum-controller.service" "${ROOT}/systemd/aios-automation-quorum-controller.timer"
systemd-analyze verify \
"${ROOT}/systemd/aios-automation-quorum-controller.service" \
"${ROOT}/systemd/aios-automation-quorum-controller.timer" \
"${ROOT}/systemd/aios-central-gpt-health-probe.service" \
"${ROOT}/systemd/aios-central-gpt-health-probe.timer" \
"${ROOT}/systemd/aios-codex-executor-health-probe.service" \
"${ROOT}/systemd/aios-codex-executor-health-probe.timer"
+652
View File
@@ -0,0 +1,652 @@
#!/usr/bin/env python3
"""Hardened WeChat outbound backend for AIOS.
Only the dedicated UCC adapter is a supported send backend. OpenClaw channel
discovery is retained for diagnostics, but generic OpenClaw chat or agent
delivery is never promoted to WeChat delivery.
"""
from __future__ import annotations
import json
import os
import re
import selectors
import signal
import subprocess
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional, Tuple
OPENCLAW_BIN = os.environ.get("AIOS_OPENCLAW_BIN", "openclaw")
WECHAT_ACCOUNT_ID = os.environ.get("AIOS_WECHAT_ACCOUNT_ID", "")
WECHAT_CHANNEL = os.environ.get("AIOS_WECHAT_CHANNEL", "")
SEND_BACKEND = os.environ.get("AIOS_WECHAT_SEND_BACKEND", "")
UCC_ADAPTER_URL = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_URL", "http://127.0.0.1:18800")
UCC_ADAPTER_TOKEN = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_TOKEN") or os.environ.get(
"AIOS_WECHAT_QR_ADAPTER_TOKEN", ""
)
UCC_SEND_PATH = os.environ.get("AIOS_UCC_WECHAT_SEND_PATH", "/api/internal/wechat/send")
_allowed_origins = os.environ.get("AIOS_UCC_WECHAT_ALLOWED_ORIGINS", "").strip()
UCC_ADAPTER_ALLOWED_ORIGINS = tuple(
origin.strip() for origin in _allowed_origins.split(",") if origin.strip()
) or (
"http://127.0.0.1:18800",
"http://[::1]:18800",
)
MAX_TARGET_BYTES = 512
MAX_CONTENT_BYTES = 32 * 1024
MAX_IDENTIFIER_BYTES = 512
MAX_METADATA_BYTES = 8 * 1024
MAX_METADATA_DEPTH = 32
MAX_METADATA_NODES = 1024
MAX_RESPONSE_BYTES = 64 * 1024
MAX_ENDPOINT_BYTES = 256
_DEFAULT_TIMEOUT_SECONDS = 10
_MAX_TIMEOUT_SECONDS = 30
_LEGACY_SEND_PATH = "/internal/channels/wechat/send"
_SAFE_PATH = re.compile(r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]+$")
_MAX_PATH_DECODE_ROUNDS = 4
_OPENCLAW_TIMEOUT_RETURN_CODE = 124
_OPENCLAW_OUTPUT_LIMIT_RETURN_CODE = 125
_OPENCLAW_SPAWN_ERROR_RETURN_CODE = 126
def _bounded_timeout(value: str) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
return _DEFAULT_TIMEOUT_SECONDS
return min(max(parsed, 1), _MAX_TIMEOUT_SECONDS)
REQUEST_TIMEOUT_SECONDS = _bounded_timeout(os.environ.get("AIOS_UCC_WECHAT_TIMEOUT_SECONDS", "10"))
class BackendResult:
def __init__(self, ok: bool, backend: str, status: str, data: Dict[str, Any]) -> None:
self.ok = ok
self.backend = backend
self.status = status
self.data = data
def to_dict(self) -> Dict[str, Any]:
return {"ok": self.ok, "backend": self.backend, "status": self.status, **self.data}
class _BackendProtocolError(Exception):
"""Internal marker whose details must never cross the module boundary."""
def _openclaw_base_args() -> List[str]:
"""Return the OpenClaw 2026.6.11-compatible command prefix.
Gateway/auth global flags are not supported by the installed CLI and would
also expose credentials through the process argument vector.
"""
return [OPENCLAW_BIN]
def _terminate_process_tree(process: Any) -> None:
if os.name == "posix":
try:
os.killpg(process.pid, signal.SIGTERM)
except OSError:
pass
try:
process.wait(timeout=0.5)
except (OSError, subprocess.TimeoutExpired):
pass
time.sleep(0.05)
try:
os.killpg(process.pid, signal.SIGKILL)
except OSError:
pass
try:
process.wait(timeout=0.5)
except (OSError, subprocess.TimeoutExpired):
pass
return
if process.poll() is not None:
return
try:
process.terminate()
process.wait(timeout=1)
return
except (OSError, subprocess.TimeoutExpired):
pass
try:
process.kill()
process.wait(timeout=1)
except (OSError, subprocess.TimeoutExpired):
pass
def _run_openclaw(
args: List[str], timeout: int = 15
) -> subprocess.CompletedProcess[str]:
bounded_timeout = min(max(int(timeout), 1), _MAX_TIMEOUT_SECONDS)
popen_options: Dict[str, Any] = {
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
}
if os.name == "posix":
popen_options["start_new_session"] = True
elif hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
try:
process = subprocess.Popen(args, **popen_options)
except (OSError, ValueError):
return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
selector = selectors.DefaultSelector()
streams = {"stdout": process.stdout, "stderr": process.stderr}
buffers = {"stdout": bytearray(), "stderr": bytearray()}
try:
for name, stream in streams.items():
if stream is not None:
selector.register(stream, selectors.EVENT_READ, name)
except (OSError, ValueError):
selector.close()
_terminate_process_tree(process)
for stream in streams.values():
if stream is not None:
stream.close()
return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
deadline = time.monotonic() + bounded_timeout
total_bytes = 0
return_code: Optional[int] = None
try:
while selector.get_map():
remaining = deadline - time.monotonic()
if remaining <= 0:
return_code = _OPENCLAW_TIMEOUT_RETURN_CODE
break
events = selector.select(min(remaining, 0.1))
for key, _mask in events:
allowance = MAX_RESPONSE_BYTES - total_bytes
try:
chunk = os.read(key.fileobj.fileno(), min(8192, allowance + 1))
except OSError:
return_code = _OPENCLAW_SPAWN_ERROR_RETURN_CODE
break
if not chunk:
selector.unregister(key.fileobj)
key.fileobj.close()
continue
if len(chunk) > allowance:
return_code = _OPENCLAW_OUTPUT_LIMIT_RETURN_CODE
break
buffers[key.data].extend(chunk)
total_bytes += len(chunk)
if return_code is not None:
break
finally:
selector.close()
if return_code is not None:
_terminate_process_tree(process)
for stream in streams.values():
if stream is not None and not stream.closed:
stream.close()
return subprocess.CompletedProcess(args, return_code, "", "")
try:
completed_return_code = process.wait(timeout=1)
except subprocess.TimeoutExpired:
_terminate_process_tree(process)
return subprocess.CompletedProcess(args, _OPENCLAW_TIMEOUT_RETURN_CODE, "", "")
stdout = buffers["stdout"].decode("utf-8", errors="replace")
stderr = buffers["stderr"].decode("utf-8", errors="replace")
return subprocess.CompletedProcess(args, completed_return_code, stdout, stderr)
def _json_dict(text: str) -> Dict[str, Any]:
try:
value = json.loads(text or "{}")
except (RecursionError, TypeError, ValueError, json.JSONDecodeError):
return {}
return value if isinstance(value, dict) else {}
def discover_openclaw_wechat_channel() -> Optional[str]:
"""Discover an explicitly typed WeChat/Weixin channel for diagnostics."""
if WECHAT_CHANNEL:
return WECHAT_CHANNEL
try:
proc = _run_openclaw(_openclaw_base_args() + ["channels", "list", "--json"], timeout=15)
except (OSError, subprocess.SubprocessError, ValueError):
return None
if proc.returncode != 0:
return None
data = _json_dict(proc.stdout[:MAX_RESPONSE_BYTES])
candidates: List[str] = []
def walk(value: Any) -> None:
if isinstance(value, dict):
name = value.get("name") or value.get("channel") or value.get("id")
kind = value.get("type") or value.get("kind")
if isinstance(name, str) and isinstance(kind, str):
label = f"{name} {kind}".lower()
if "wechat" in label or "weixin" in label:
candidates.append(name)
for child in value.values():
walk(child)
elif isinstance(value, list):
for child in value:
walk(child)
walk(data)
return candidates[0] if candidates else None
def _auth_headers() -> Dict[str, str]:
headers = {"Content-Type": "application/json", "Accept": "application/json"}
if UCC_ADAPTER_TOKEN:
headers["Authorization"] = "Bearer " + UCC_ADAPTER_TOKEN
return headers
def _safe_path(value: Any) -> Optional[str]:
if not isinstance(value, str):
return None
path = value if value.startswith("/") else "/" + value
try:
encoded_path = path.encode("utf-8")
except UnicodeEncodeError:
return None
if len(encoded_path) > MAX_ENDPOINT_BYTES:
return None
if not _SAFE_PATH.fullmatch(path):
return None
decoded = path
for _ in range(_MAX_PATH_DECODE_ROUNDS):
try:
next_decoded = urllib.parse.unquote_to_bytes(decoded).decode("utf-8")
except UnicodeDecodeError:
return None
if next_decoded == decoded:
break
decoded = next_decoded
if "%" in decoded or "\\" in decoded or "\x00" in decoded:
return None
if not _SAFE_PATH.fullmatch(decoded):
return None
if any(segment in {".", ".."} for segment in decoded.split("/")):
return None
return path
def _unique_paths(paths: List[str]) -> List[str]:
out: List[str] = []
for candidate in paths:
path = _safe_path(candidate)
if path and path not in out:
out.append(path)
if len(out) == 2:
break
return out
def _ucc_send_paths() -> List[str]:
canonical = _safe_path(UCC_SEND_PATH) or "/api/internal/wechat/send"
return _unique_paths([canonical, _LEGACY_SEND_PATH])
def _canonical_origin(value: Any) -> Optional[str]:
if not isinstance(value, str) or not value or len(value) > 2048:
return None
if "\\" in value or any(ord(character) < 32 for character in value):
return None
try:
parsed = urllib.parse.urlsplit(value)
port = parsed.port
except (UnicodeError, ValueError):
return None
scheme = parsed.scheme.lower()
if scheme not in {"http", "https"} or not parsed.hostname:
return None
if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
return None
try:
host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
except UnicodeError:
return None
if not host:
return None
if port is None:
port = 80 if scheme == "http" else 443
rendered_host = f"[{host}]" if ":" in host else host
return f"{scheme}://{rendered_host}:{port}"
def _adapter_base_url() -> Optional[str]:
origin = _canonical_origin(UCC_ADAPTER_URL)
allowed = {
canonical
for canonical in (_canonical_origin(value) for value in UCC_ADAPTER_ALLOWED_ORIGINS)
if canonical
}
return origin if origin and origin in allowed else None
class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(
self,
req: Any,
fp: Any,
code: int,
msg: str,
headers: Any,
newurl: str,
) -> None:
del req, fp, code, msg, headers, newurl
return None
def _open_no_redirect(request: urllib.request.Request, timeout: int) -> Any:
opener = urllib.request.build_opener(_NoRedirectHandler())
return opener.open(request, timeout=timeout)
def ucc_adapter_alive() -> bool:
base_url = _adapter_base_url()
if not base_url:
return False
request = urllib.request.Request(
base_url + "/health", headers=_auth_headers(), method="GET"
)
try:
with _open_no_redirect(request, timeout=min(5, REQUEST_TIMEOUT_SECONDS)) as response:
return 200 <= int(response.status) < 300
except (OSError, ValueError, urllib.error.URLError):
return False
def select_backend() -> Dict[str, Any]:
explicit = SEND_BACKEND.strip().lower()
if explicit:
if explicit == "ucc-adapter":
return {"backend": "ucc-adapter", "reason": "explicit"}
return {"backend": "unavailable", "reason": "unsupported_backend"}
if ucc_adapter_alive():
return {"backend": "ucc-adapter", "reason": "ucc_adapter_alive"}
return {"backend": "unavailable", "reason": "ucc_adapter_unavailable"}
def _post_ucc(path: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
base_url = _adapter_base_url()
safe_path = _safe_path(path)
if not base_url or not safe_path:
raise _BackendProtocolError("invalid adapter endpoint")
try:
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
except (TypeError, ValueError) as exc:
raise _BackendProtocolError("invalid request payload") from exc
request = urllib.request.Request(
base_url + safe_path,
data=body,
headers=_auth_headers(),
method="POST",
)
with _open_no_redirect(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
raw = response.read(MAX_RESPONSE_BYTES + 1)
if len(raw) > MAX_RESPONSE_BYTES:
raise _BackendProtocolError("adapter response too large")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise _BackendProtocolError("adapter response encoding invalid") from exc
data = _json_dict(text)
if not data:
raise _BackendProtocolError("adapter response invalid")
return int(response.status), data
def _ucc_response_sent(data: Dict[str, Any]) -> bool:
return isinstance(data, dict) and data.get("ok") is True and data.get("status") == "sent"
def _ucc_result_is_ok(data: Dict[str, Any], dry_run: bool = False) -> bool:
expected_status = "dry_run" if dry_run else "sent"
return isinstance(data, dict) and data.get("ok") is True and data.get("status") == expected_status
def _utf8_size(value: str) -> Optional[int]:
try:
return len(value.encode("utf-8"))
except UnicodeEncodeError:
return None
def _valid_identifier(value: Any, required: bool = False) -> bool:
if not isinstance(value, str):
return False
if required and not value.strip():
return False
if not value:
return not required
if len(value) > MAX_IDENTIFIER_BYTES:
return False
size = _utf8_size(value)
return size is not None and size <= MAX_IDENTIFIER_BYTES and not any(
ord(character) < 32 or ord(character) == 127 for character in value
)
def _validate_metadata(metadata: Any) -> Optional[str]:
if not isinstance(metadata, dict):
return "invalid_input"
nodes = 0
stack: List[Tuple[Any, int]] = [(metadata, 0)]
while stack:
value, depth = stack.pop()
nodes += 1
if depth > MAX_METADATA_DEPTH or nodes > MAX_METADATA_NODES:
return "input_too_large"
if isinstance(value, dict):
if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
return "input_too_large"
for key, child in value.items():
if (
not isinstance(key, str)
or len(key) > MAX_METADATA_BYTES
or _utf8_size(key) is None
):
return "invalid_input"
stack.append((child, depth + 1))
elif isinstance(value, list):
if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
return "input_too_large"
stack.extend((child, depth + 1) for child in value)
elif isinstance(value, str):
if len(value) > MAX_METADATA_BYTES:
return "input_too_large"
if _utf8_size(value) is None:
return "invalid_input"
elif value is not None and not isinstance(value, (bool, int, float)):
return "invalid_input"
try:
encoded = json.dumps(
metadata,
ensure_ascii=False,
allow_nan=False,
separators=(",", ":"),
).encode("utf-8")
except (RecursionError, TypeError, UnicodeEncodeError, ValueError):
return "invalid_input"
return "input_too_large" if len(encoded) > MAX_METADATA_BYTES else None
def _validate_payload(
target: Any,
content: Any,
account_id: Any,
message_id: Any,
metadata: Any,
dry_run: Any,
) -> Optional[str]:
if not isinstance(target, str) or not isinstance(content, str):
return "invalid_input"
if not target.strip() or not content.strip() or not isinstance(dry_run, bool):
return "invalid_input"
if any(ord(character) < 32 or ord(character) == 127 for character in target):
return "invalid_input"
if "\x00" in content or not _valid_identifier(account_id) or not _valid_identifier(message_id):
return "invalid_input"
if len(target) > MAX_TARGET_BYTES or len(content) > MAX_CONTENT_BYTES:
return "input_too_large"
target_size = _utf8_size(target)
content_size = _utf8_size(content)
if target_size is None or content_size is None:
return "invalid_input"
if target_size > MAX_TARGET_BYTES or content_size > MAX_CONTENT_BYTES:
return "input_too_large"
return _validate_metadata(metadata)
def _safe_provider_message_id(value: Any) -> Optional[str]:
if not isinstance(value, str) or not value:
return None
if len(value) > MAX_IDENTIFIER_BYTES:
return None
size = _utf8_size(value)
if size is None or size > MAX_IDENTIFIER_BYTES:
return None
if any(ord(character) < 32 or ord(character) == 127 for character in value):
return None
return value
def send_ucc_adapter(
target: str,
content: str,
account_id: str = "",
message_id: str = "",
metadata: Optional[Dict[str, Any]] = None,
dry_run: bool = False,
) -> BackendResult:
effective_account = account_id or WECHAT_ACCOUNT_ID
effective_metadata: Any = {} if metadata is None else metadata
error_code = _validate_payload(
target,
content,
effective_account,
message_id,
effective_metadata,
dry_run,
)
if error_code:
return BackendResult(False, "ucc-adapter", "failed", {"code": error_code})
payload = {
"target": target,
"content": content,
"account_id": effective_account,
"message_id": message_id,
"metadata": effective_metadata,
"dry_run": dry_run,
}
last_endpoint: Optional[str] = None
for path in _ucc_send_paths():
last_endpoint = path
try:
http_status, data = _post_ucc(path, payload)
except urllib.error.HTTPError as exc:
if exc.code in {404, 405}:
continue
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_http_error", "http_status": int(exc.code), "endpoint": path},
)
except (OSError, ValueError, urllib.error.URLError, _BackendProtocolError):
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_unavailable", "endpoint": path},
)
except Exception:
# Fail closed without returning exception text, type, or chained cause.
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_unavailable", "endpoint": path},
)
if not (200 <= http_status < 300):
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_http_error", "http_status": http_status, "endpoint": path},
)
if not _ucc_result_is_ok(data, dry_run):
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_rejected", "http_status": http_status, "endpoint": path},
)
confirmed_status = "dry_run" if dry_run else "sent"
result_data: Dict[str, Any] = {
"code": confirmed_status,
"http_status": http_status,
"endpoint": path,
}
provider_message_id = _safe_provider_message_id(data.get("provider_message_id"))
if provider_message_id:
result_data["provider_message_id"] = provider_message_id
return BackendResult(True, "ucc-adapter", confirmed_status, result_data)
data: Dict[str, Any] = {"code": "ucc_adapter_unavailable"}
if last_endpoint:
data["endpoint"] = last_endpoint
return BackendResult(False, "ucc-adapter", "failed", data)
def send_openclaw_channel(
target: str,
content: str,
account_id: str = "",
message_id: str = "",
metadata: Optional[Dict[str, Any]] = None,
dry_run: bool = False,
) -> BackendResult:
del target, content, account_id, message_id, metadata, dry_run
return BackendResult(
False,
"openclaw-channel",
"failed",
{"code": "unsupported_backend"},
)
def send_message(
target: str,
content: str,
account_id: str = "",
message_id: str = "",
metadata: Optional[Dict[str, Any]] = None,
dry_run: bool = False,
) -> Dict[str, Any]:
selected = select_backend()
if selected.get("backend") == "ucc-adapter":
return send_ucc_adapter(
target, content, account_id, message_id, metadata, dry_run
).to_dict()
code = "unsupported_backend" if selected.get("reason") == "unsupported_backend" else "wechat_outbound_backend_unavailable"
return BackendResult(False, "unavailable", "failed", {"code": code}).to_dict()
if __name__ == "__main__":
print(json.dumps(select_backend(), ensure_ascii=False, separators=(",", ":")))
+1
View File
@@ -0,0 +1 @@
"""AIOS services namespace."""
+5
View File
@@ -0,0 +1,5 @@
"""Reliable, fail-closed Codex command watcher."""
from .watcher import CommandWatcher, Config, load_config
__all__ = ["CommandWatcher", "Config", "load_config"]
+19
View File
@@ -0,0 +1,19 @@
from __future__ import annotations
import argparse
import sys
from .watcher import CommandWatcher, load_config
def main() -> int:
parser = argparse.ArgumentParser(description="Claim and execute Codex work units")
parser.add_argument("--config", required=True)
parser.add_argument("--once", action="store_true", help="process one scan and exit")
args = parser.parse_args()
watcher = CommandWatcher(load_config(args.config))
return watcher.run_once() if args.once else watcher.run_forever()
if __name__ == "__main__":
sys.exit(main())
+480
View File
@@ -0,0 +1,480 @@
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
import stat
import subprocess
import time
from typing import Any, Callable
import urllib.error
import urllib.parse
import urllib.request
import uuid
OID_RE = re.compile(r"^[0-9a-f]{40}$")
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
TERMINAL = {"COMPLETED", "DEAD_LETTER", "REJECTED"}
class WatcherError(RuntimeError):
pass
@dataclass(frozen=True)
class Config:
active_dir: Path
completed_dir: Path
dead_letter_dir: Path
database_path: Path
audit_log_path: Path
gateway_endpoint: str
gateway_token_file: Path
contract_commit: str
lease_seconds: int
retry_seconds: int
max_attempts: int
poll_seconds: float
executor_path: Path | None
executor_args: tuple[str, ...]
executor_timeout_seconds: int
def _absolute(value: Any, key: str) -> Path:
if not isinstance(value, str) or not value or not Path(value).is_absolute():
raise WatcherError(f"invalid_{key}")
return Path(value)
def _secure_regular(path: Path, *, writable_mask: int = 0o022) -> os.stat_result:
info = os.lstat(path)
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
raise WatcherError(f"unsafe_file:{path}")
if stat.S_IMODE(info.st_mode) & writable_mask:
raise WatcherError(f"unsafe_mode:{path}")
return info
def _read_secure_text(
path: Path,
*,
writable_mask: int = 0o022,
max_bytes: int,
allowed_owners: set[int] | None = None,
) -> str:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
fd = os.open(path, flags)
try:
info = os.fstat(fd)
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) & writable_mask:
raise WatcherError(f"unsafe_file:{path}")
if allowed_owners is not None and info.st_uid not in allowed_owners:
raise WatcherError(f"owner_mismatch:{path}")
if info.st_size > max_bytes:
raise WatcherError(f"file_too_large:{path}")
chunks: list[bytes] = []
total = 0
while True:
chunk = os.read(fd, min(65_536, max_bytes + 1 - total))
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
if total > max_bytes:
raise WatcherError(f"file_too_large:{path}")
return b"".join(chunks).decode("utf-8")
finally:
os.close(fd)
def load_config(path_value: str) -> Config:
path = _absolute(path_value, "config_path")
raw = json.loads(_read_secure_text(path, max_bytes=131_072, allowed_owners={0, os.geteuid()}))
expected = {
"active_dir", "completed_dir", "dead_letter_dir", "database_path",
"audit_log_path", "gateway_endpoint", "gateway_token_file",
"contract_commit", "lease_seconds", "retry_seconds", "max_attempts",
"poll_seconds", "executor_path", "executor_args",
"executor_timeout_seconds",
}
if not isinstance(raw, dict) or set(raw) != expected:
raise WatcherError("invalid_config_fields")
contract = raw["contract_commit"]
if not isinstance(contract, str) or not OID_RE.fullmatch(contract):
raise WatcherError("invalid_contract_commit")
endpoint = raw["gateway_endpoint"]
if not isinstance(endpoint, str) or not endpoint.startswith("http://127.0.0.1:") or not endpoint.endswith("/v1/orchestrator/events"):
raise WatcherError("gateway_endpoint_must_be_loopback")
executor_raw = raw["executor_path"]
executor = None if executor_raw is None else _absolute(executor_raw, "executor_path")
executor_args = raw["executor_args"]
if not isinstance(executor_args, list) or any(not isinstance(x, str) or not x for x in executor_args):
raise WatcherError("invalid_executor_args")
ints = ("lease_seconds", "retry_seconds", "max_attempts", "executor_timeout_seconds")
if any(not isinstance(raw[x], int) or raw[x] <= 0 for x in ints):
raise WatcherError("invalid_positive_integer")
if not isinstance(raw["poll_seconds"], (int, float)) or raw["poll_seconds"] <= 0:
raise WatcherError("invalid_poll_seconds")
return Config(
active_dir=_absolute(raw["active_dir"], "active_dir"),
completed_dir=_absolute(raw["completed_dir"], "completed_dir"),
dead_letter_dir=_absolute(raw["dead_letter_dir"], "dead_letter_dir"),
database_path=_absolute(raw["database_path"], "database_path"),
audit_log_path=_absolute(raw["audit_log_path"], "audit_log_path"),
gateway_endpoint=endpoint,
gateway_token_file=_absolute(raw["gateway_token_file"], "gateway_token_file"),
contract_commit=contract,
lease_seconds=raw["lease_seconds"], retry_seconds=raw["retry_seconds"],
max_attempts=raw["max_attempts"], poll_seconds=float(raw["poll_seconds"]),
executor_path=executor, executor_args=tuple(executor_args),
executor_timeout_seconds=raw["executor_timeout_seconds"],
)
def canonical_digest(value: dict[str, Any]) -> str:
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
def validate_command(value: Any, expected_contract: str) -> dict[str, Any]:
expected = {
"schema_version", "command_id", "project_key", "work_unit_id",
"contract_commit", "state_version", "execution",
}
if not isinstance(value, dict) or set(value) != expected or value.get("schema_version") != 1:
raise WatcherError("invalid_command_fields")
for key in ("command_id", "project_key", "work_unit_id"):
if not isinstance(value[key], str) or not ID_RE.fullmatch(value[key]):
raise WatcherError(f"invalid_{key}")
if value["contract_commit"] != expected_contract:
raise WatcherError("contract_commit_mismatch")
if not isinstance(value["state_version"], int) or value["state_version"] < 0:
raise WatcherError("invalid_state_version")
execution = value["execution"]
if not isinstance(execution, dict) or set(execution) != {"kind", "input"}:
raise WatcherError("invalid_execution")
if execution["kind"] not in {"acceptance_probe", "codex"}:
raise WatcherError("invalid_execution_kind")
if not isinstance(execution["input"], str) or len(execution["input"].encode()) > 65_536:
raise WatcherError("invalid_execution_input")
return value
class CommandWatcher:
def __init__(
self,
config: Config,
*,
clock: Callable[[], float] = time.time,
event_poster: Callable[[dict[str, Any]], None] | None = None,
executor: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
worker_id: str | None = None,
) -> None:
self.config = config
self.clock = clock
self.worker_id = worker_id or f"{os.uname().nodename}:{os.getpid()}:{uuid.uuid4().hex[:8]}"
self._uses_gateway = event_poster is None
self.event_poster = event_poster or self._post_event
self.executor = executor or self._execute
self._prepare()
def _prepare(self) -> None:
for directory in (self.config.active_dir, self.config.completed_dir, self.config.dead_letter_dir):
info = os.lstat(directory)
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) & 0o002:
raise WatcherError(f"unsafe_directory:{directory}")
self.config.database_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
self.config.audit_log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
with self._db() as db:
db.execute("PRAGMA journal_mode=WAL")
db.execute("""
CREATE TABLE IF NOT EXISTS commands (
command_id TEXT PRIMARY KEY, work_unit_id TEXT NOT NULL UNIQUE,
payload_sha256 TEXT NOT NULL,
source_path TEXT NOT NULL, status TEXT NOT NULL,
phase TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
lease_owner TEXT, lease_until REAL, next_attempt REAL NOT NULL DEFAULT 0,
result_json TEXT, start_event_json TEXT, end_event_json TEXT,
last_error TEXT, updated_at REAL NOT NULL
)
""")
def _db(self) -> sqlite3.Connection:
db = sqlite3.connect(self.config.database_path, timeout=5, isolation_level=None)
db.row_factory = sqlite3.Row
return db
def _audit(self, marker: str, command_id: str, **fields: Any) -> None:
safe = {k: v for k, v in fields.items() if k not in {"input", "token", "secret"}}
record = {"time": int(self.clock()), "marker": marker, "command_id": command_id, **safe}
fd = os.open(self.config.audit_log_path, os.O_APPEND | os.O_CREAT | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
try:
os.write(fd, (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode())
finally:
os.close(fd)
def _read_command(self, path: Path) -> tuple[dict[str, Any], str]:
raw = _read_secure_text(path, max_bytes=131_072, allowed_owners={0, os.geteuid()})
value = validate_command(json.loads(raw), self.config.contract_commit)
return value, canonical_digest(value)
def _register(self, command: dict[str, Any], digest: str, path: Path) -> str:
now = self.clock()
with self._db() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT payload_sha256,status FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
if row:
db.execute("COMMIT")
if row["payload_sha256"] != digest:
raise WatcherError("duplicate_command_payload_conflict")
return "duplicate_terminal" if row["status"] in TERMINAL else "existing"
prior_work_unit = db.execute(
"SELECT command_id FROM commands WHERE work_unit_id=?",
(command["work_unit_id"],),
).fetchone()
if prior_work_unit:
db.execute("COMMIT")
raise WatcherError("duplicate_work_unit")
db.execute(
"INSERT INTO commands(command_id,work_unit_id,payload_sha256,source_path,status,phase,updated_at) VALUES(?,?,?,?,?,?,?)",
(command["command_id"], command["work_unit_id"], digest, str(path), "PENDING", "START", now),
)
db.execute("COMMIT")
return "new"
def _claim(self, command_id: str) -> sqlite3.Row | None:
now = self.clock()
with self._db() as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
if not row or row["status"] in TERMINAL or row["next_attempt"] > now:
db.execute("COMMIT")
return None
if row["lease_until"] is not None and row["lease_until"] > now and row["lease_owner"] != self.worker_id:
db.execute("COMMIT")
return None
attempts = row["attempts"] + 1
db.execute(
"UPDATE commands SET status='CLAIMED',attempts=?,lease_owner=?,lease_until=?,updated_at=? WHERE command_id=?",
(attempts, self.worker_id, now + self.config.lease_seconds, now, command_id),
)
row = db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
db.execute("COMMIT")
assert row is not None
self._audit("CODEX_AUTO_CLAIM", command_id, attempt=row["attempts"], worker=self.worker_id)
return row
def _update(self, command_id: str, *, status: str, phase: str, result: dict[str, Any] | None = None, error: str | None = None) -> None:
with self._db() as db:
release = status in TERMINAL
db.execute(
"UPDATE commands SET status=?,phase=?,result_json=COALESCE(?,result_json),last_error=?,lease_owner=CASE WHEN ? THEN NULL ELSE lease_owner END,lease_until=CASE WHEN ? THEN NULL ELSE lease_until END,updated_at=? WHERE command_id=?",
(status, phase, json.dumps(result, sort_keys=True) if result is not None else None, error, release, release, self.clock(), command_id),
)
def _retry_or_dead(self, command_id: str, path: Path, error: Exception) -> None:
message = f"{type(error).__name__}:{str(error)[:240]}"
with self._db() as db:
row = db.execute("SELECT attempts,phase FROM commands WHERE command_id=?", (command_id,)).fetchone()
attempts = row["attempts"] if row else self.config.max_attempts
if str(error) == "gateway_stale_state_version" and row:
column = "start_event_json" if row["phase"] == "START" else "end_event_json"
db.execute(f"UPDATE commands SET {column}=NULL WHERE command_id=?", (command_id,))
if attempts >= self.config.max_attempts:
db.execute("UPDATE commands SET status='DEAD_LETTER',last_error=?,lease_owner=NULL,lease_until=NULL,updated_at=? WHERE command_id=?", (message, self.clock(), command_id))
self._move(path, self.config.dead_letter_dir)
self._audit("DEAD_LETTER", command_id, error=message)
else:
db.execute("UPDATE commands SET status='RETRY',last_error=?,next_attempt=?,lease_owner=NULL,lease_until=NULL,updated_at=? WHERE command_id=?", (message, self.clock() + self.config.retry_seconds, self.clock(), command_id))
self._audit("RETRY_SCHEDULED", command_id, attempt=attempts, error=message)
def _event(self, command: dict[str, Any], event_type: str, status: str, suffix: str, result: dict[str, Any] | None = None) -> dict[str, Any]:
return {
"schema_version": 1, "project_key": command["project_key"],
"contract_commit": command["contract_commit"],
"event_id": f"{command['command_id']}:{suffix}", "source_agent": "codex",
"event_type": event_type, "work_unit_id": command["work_unit_id"],
"status": status, "state_version": command["state_version"],
"implementation_commit": None, "result_commit": None, "blocker": None,
"evidence_refs": [], "payload": {"command_id": command["command_id"], "result": result or {}},
}
def _gateway_token(self) -> str:
token = _read_secure_text(
self.config.gateway_token_file,
writable_mask=0o027,
max_bytes=16_384,
allowed_owners={0, os.geteuid()},
)
if not token or "\n" in token or "\r" in token:
raise WatcherError("invalid_gateway_token")
return token
def _gateway_state_version(self, project_key: str, fallback: int) -> int:
state_url = self.config.gateway_endpoint.removesuffix("/events") + "/state/" + urllib.parse.quote(project_key, safe="")
request = urllib.request.Request(
state_url,
headers={"Authorization": f"Bearer {self._gateway_token()}"},
method="GET",
)
try:
with urllib.request.urlopen(request, timeout=15) as response:
payload = json.loads(response.read(262_145))
version = payload.get("state_version")
if not isinstance(version, int) or version < 0:
raise WatcherError("invalid_gateway_state")
return version
except urllib.error.HTTPError as exc:
if exc.code == 404:
return fallback
raise WatcherError(f"gateway_state_http_{exc.code}") from exc
except urllib.error.URLError as exc:
raise WatcherError("gateway_state_unavailable") from exc
def _durable_event(self, command: dict[str, Any], column: str, event: dict[str, Any]) -> dict[str, Any]:
with self._db() as db:
row = db.execute(f"SELECT {column} FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
if row and row[column]:
return json.loads(row[column])
if self._uses_gateway:
event["state_version"] = self._gateway_state_version(command["project_key"], command["state_version"])
encoded = json.dumps(event, sort_keys=True, separators=(",", ":"))
with self._db() as db:
db.execute(f"UPDATE commands SET {column}=? WHERE command_id=? AND {column} IS NULL", (encoded, command["command_id"]))
saved = db.execute(f"SELECT {column} FROM commands WHERE command_id=?", (command["command_id"],)).fetchone()
if not saved or not saved[column]:
raise WatcherError("event_persistence_failed")
return json.loads(saved[column])
def _post_event(self, event: dict[str, Any]) -> None:
request = urllib.request.Request(
self.config.gateway_endpoint,
data=json.dumps(event, separators=(",", ":")).encode(),
headers={"Authorization": f"Bearer {self._gateway_token()}", "Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=15) as response:
if response.status not in {200, 201, 202}:
raise WatcherError(f"gateway_http_{response.status}")
except urllib.error.HTTPError as exc:
body = exc.read(16_384)
try:
reason = json.loads(body).get("error")
except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
reason = None
if exc.code == 409 and reason == "stale_state_version":
raise WatcherError("gateway_stale_state_version") from exc
raise WatcherError(f"gateway_http_{exc.code}") from exc
except urllib.error.URLError as exc:
raise WatcherError("gateway_unavailable") from exc
def _execute(self, command: dict[str, Any]) -> dict[str, Any]:
execution = command["execution"]
if execution["kind"] == "acceptance_probe":
return {"kind": "acceptance_probe", "input_sha256": hashlib.sha256(execution["input"].encode()).hexdigest(), "exit_code": 0}
assert self.config.executor_path is not None
proc = subprocess.run(
[str(self.config.executor_path), *self.config.executor_args],
input=execution["input"], text=True, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, timeout=self.config.executor_timeout_seconds,
check=False, env={"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"},
)
if proc.returncode != 0:
raise WatcherError(f"codex_executor_failed:{proc.returncode}")
return {"kind": "codex", "exit_code": 0}
def _preflight_execution(self, command: dict[str, Any]) -> None:
if command["execution"]["kind"] == "acceptance_probe":
return
if self.config.executor_path is None:
raise WatcherError("codex_executor_unavailable")
info = _secure_regular(self.config.executor_path)
if not info.st_mode & stat.S_IXUSR:
raise WatcherError("codex_executor_not_executable")
def _move(self, source: Path, target_dir: Path) -> None:
if not source.exists() and not source.is_symlink():
return
target = target_dir / source.name
if target.exists() or target.is_symlink():
target = target_dir / f"{source.stem}-{uuid.uuid4().hex[:8]}{source.suffix}"
os.replace(source, target)
def _reject_unregistered(self, path: Path, error: Exception) -> None:
command_id = path.stem[:128] if ID_RE.fullmatch(path.stem[:128]) else f"invalid-{uuid.uuid4().hex[:12]}"
self._audit("COMMAND_REJECTED", command_id, error=f"{type(error).__name__}:{str(error)[:240]}")
self._move(path, self.config.dead_letter_dir)
def process_path(self, path: Path) -> bool:
try:
command, digest = self._read_command(path)
registration = self._register(command, digest, path)
if registration == "duplicate_terminal":
self._move(path, self.config.completed_dir)
return False
row = self._claim(command["command_id"])
if row is None:
return False
phase = row["phase"]
result = json.loads(row["result_json"]) if row["result_json"] else None
if phase == "START":
start_event = self._durable_event(
command,
"start_event_json",
self._event(command, "work_unit_started", "RUNNING", "started"),
)
self.event_poster(start_event)
self._audit("TASK_STARTED", command["command_id"])
self._update(command["command_id"], status="PENDING", phase="EXECUTE")
phase = "EXECUTE"
if phase == "EXECUTE":
self._preflight_execution(command)
self._update(command["command_id"], status="CLAIMED", phase="EXECUTING")
phase = "EXECUTING"
result = self.executor(command)
self._audit("TASK_EXECUTED", command["command_id"], result=result)
self._update(command["command_id"], status="PENDING", phase="END", result=result)
phase = "END"
elif phase == "EXECUTING":
raise WatcherError("ambiguous_execution_not_repeated")
if phase == "END":
assert result is not None
end_event = self._durable_event(
command,
"end_event_json",
self._event(command, "work_unit_ended", "PASS", "ended", result),
)
self.event_poster(end_event)
self._audit("TASK_ENDED", command["command_id"])
self._audit("EVENT_POSTED", command["command_id"], event="work_unit_ended")
self._update(command["command_id"], status="COMPLETED", phase="DONE", result=result)
self._move(path, self.config.completed_dir)
return True
except Exception as exc:
try:
if isinstance(exc, WatcherError) and str(exc) in {"duplicate_command_payload_conflict", "duplicate_work_unit"}:
self._reject_unregistered(path, exc)
elif "command" in locals():
self._retry_or_dead(command["command_id"], path, exc)
else:
self._reject_unregistered(path, exc)
except Exception:
pass
return False
def run_once(self) -> int:
failures = 0
for path in sorted(self.config.active_dir.glob("*.json")):
if not self.process_path(path):
failures += 1
return 1 if failures else 0
def run_forever(self) -> int:
while True:
self.run_once()
time.sleep(self.config.poll_seconds)
@@ -0,0 +1,3 @@
"""Reliable central orchestration gateway for AIOS."""
__version__ = "0.2.0"
@@ -0,0 +1,34 @@
from __future__ import annotations
import argparse
import logging
from pathlib import Path
from .config import Config
from .service import Runtime, serve
def main() -> int:
parser = argparse.ArgumentParser(description="AIOS Orchestrator Gateway")
parser.add_argument("--config", type=Path, required=True)
parser.add_argument("--log-level", default="INFO")
parser.add_argument(
"--poll-once",
action="store_true",
help="Run one inbox/model/outbox/OpenClaw synchronization polling cycle.",
)
args = parser.parse_args()
logging.basicConfig(
level=getattr(logging, args.log_level.upper(), logging.INFO),
format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
if args.poll_once:
runtime = Runtime(Config.from_file(args.config))
runtime.poll_once()
return 0
serve(args.config)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,132 @@
from __future__ import annotations
import json
import os
from pathlib import Path
import re
import stat
import subprocess
from typing import Any, Callable
import urllib.error
import urllib.request
from .models import Decision
GATEWAY_DECISION_URL = "http://127.0.0.1:18920/v1/orchestrator/decisions"
SECRET_REF_RE = re.compile(r"^secret://[A-Za-z0-9._/-]{1,480}$")
class CallbackBridgeError(RuntimeError):
def __init__(self, category: str, status_code: int) -> None:
super().__init__(category)
self.category = category
self.status_code = status_code
def _trusted_resolver(path: Path) -> Path:
if not path.is_absolute():
raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503)
try:
info = path.lstat()
resolved = path.resolve(strict=True)
except OSError as exc:
raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503) from exc
if (
resolved != path
or stat.S_ISLNK(info.st_mode)
or not stat.S_ISREG(info.st_mode)
or info.st_uid != 0
or stat.S_IMODE(info.st_mode) & 0o022
or not stat.S_IMODE(info.st_mode) & 0o111
):
raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503)
return resolved
def resolve_secret_ref(secret_ref: str, resolver: Path) -> str:
if not isinstance(secret_ref, str) or not SECRET_REF_RE.fullmatch(secret_ref):
raise CallbackBridgeError("CALLBACK_CONFIGURATION_UNAVAILABLE", 503)
executable = _trusted_resolver(resolver)
request = json.dumps(
{"operation": "resolve", "secret_ref": secret_ref},
separators=(",", ":"),
sort_keys=True,
).encode()
try:
result = subprocess.run(
[str(executable)],
input=request,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
timeout=5,
check=False,
)
except (OSError, subprocess.SubprocessError) as exc:
raise CallbackBridgeError("CALLBACK_CREDENTIAL_UNAVAILABLE", 503) from exc
token = result.stdout.removesuffix(b"\n")
if (
result.returncode != 0
or not token
or len(token) > 16_384
or b"\n" in token
or b"\r" in token
):
raise CallbackBridgeError("CALLBACK_CREDENTIAL_UNAVAILABLE", 503)
try:
return token.decode("utf-8")
except UnicodeDecodeError as exc:
raise CallbackBridgeError("CALLBACK_CREDENTIAL_UNAVAILABLE", 503) from exc
def submit_orchestration_decision(
payload: dict[str, Any],
*,
contract_commit: str,
secret_ref: str,
resolver: Path,
request_open: Callable[..., Any] = urllib.request.urlopen,
) -> dict[str, Any]:
try:
decision = Decision.from_dict(payload)
except Exception as exc:
raise CallbackBridgeError("INVALID_DECISION", 422) from exc
if decision.contract_commit != contract_commit:
raise CallbackBridgeError("CONTRACT_MISMATCH", 409)
token = resolve_secret_ref(secret_ref, resolver)
request = urllib.request.Request(
GATEWAY_DECISION_URL,
data=json.dumps(
decision.to_dict(), separators=(",", ":"), sort_keys=True
).encode(),
method="POST",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Idempotency-Key": decision.event_id,
"User-Agent": "aios-chatgpt-actions/decision-bridge-v1",
},
)
try:
with request_open(request, timeout=10) as response:
raw = response.read(65_537)
if response.status != 202 or len(raw) > 65_536:
raise CallbackBridgeError("CALLBACK_HTTP_ERROR", 502)
except urllib.error.HTTPError as exc:
if exc.code == 409:
raise CallbackBridgeError("STALE_STATE", 409) from exc
raise CallbackBridgeError("CALLBACK_HTTP_ERROR", 502) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise CallbackBridgeError("CALLBACK_NETWORK_ERROR", 502) from exc
try:
value = json.loads(raw)
except json.JSONDecodeError as exc:
raise CallbackBridgeError("CALLBACK_RESPONSE_INVALID", 502) from exc
if (
not isinstance(value, dict)
or value.get("accepted") is not True
or not isinstance(value.get("state_version"), int)
):
raise CallbackBridgeError("CALLBACK_RESPONSE_INVALID", 502)
return {"accepted": True, "state_version": value["state_version"]}
@@ -0,0 +1,204 @@
from __future__ import annotations
import json
from dataclasses import dataclass
import urllib.error
import urllib.request
from typing import Any
from .errors import TriggerError
from .models import DECISION_SCHEMA, Decision, Event
@dataclass(frozen=True)
class WorkspaceTriggerResult:
http_status: int
remote_id: None = None
WORKSPACE_HTTP_CATEGORIES = {
401: "UNAUTHORIZED",
403: "FORBIDDEN",
404: "NOT_FOUND",
409: "NOT_RUNNABLE",
}
class WorkspaceAgentClient:
"""Asynchronous Workspace Agent trigger.
The official endpoint returns only 202 and no run ID or retrievable output.
Completion therefore arrives through the gateway callback endpoint.
"""
def __init__(self, trigger_id: str, token: str, timeout_seconds: int = 30):
self.trigger_id = trigger_id
self.token = token
self.timeout_seconds = timeout_seconds
def trigger(self, event: Event) -> WorkspaceTriggerResult:
url = f"https://api.chatgpt.com/v1/workspace_agents/{self.trigger_id}/trigger"
prompt = (
"Process this AIOS orchestration event under the committed contract. "
"When complete, call submit_orchestration_decision with strict JSON. "
"Do not fabricate tests, Git state, secrets, evidence, or deployment facts.\n\n"
+ event.canonical_json()
)
body = json.dumps(
{
"conversation_key": event.project_key,
"input": prompt,
}
).encode()
request = urllib.request.Request(
url,
data=body,
method="POST",
headers={
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json",
"Idempotency-Key": event.event_id,
"User-Agent": "aios-orchestrator-gateway/0.2",
},
)
try:
with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
if response.status != 202:
raise TriggerError(
"workspace_agent_unexpected_http_status",
http_status=response.status,
)
return WorkspaceTriggerResult(http_status=202)
except urllib.error.HTTPError as exc:
category = WORKSPACE_HTTP_CATEGORIES.get(exc.code)
raise TriggerError(
"workspace_agent_http_error",
http_status=exc.code,
category=category,
) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise TriggerError(
"workspace_agent_network_error",
category="NETWORK_ERROR",
) from exc
class ResponsesClient:
"""Background Responses API client with polling and strict output validation."""
def __init__(self, api_key: str, model: str, timeout_seconds: int = 60):
self.api_key = api_key
self.model = model
self.timeout_seconds = timeout_seconds
def start(self, event: Event) -> str:
body = {
"model": self.model,
"background": True,
"instructions": (
"You are the AIOS central orchestration brain. Apply the exact "
"contract_commit in the event. Return only a strict decision. "
"Never fabricate evidence, secrets, Git facts, tests, approvals, "
"or deployment state. Keep external blockers fail-closed."
),
"input": event.canonical_json(),
"text": {
"format": {
"type": "json_schema",
"name": "aios_orchestrator_decision",
"strict": True,
"schema": DECISION_SCHEMA,
}
},
"metadata": {
"event_id": event.event_id,
"project_key": event.project_key,
"contract_commit": event.contract_commit,
},
}
payload = self._request(
"POST",
"https://api.openai.com/v1/responses",
body,
idempotency_key=event.event_id,
)
response_id = payload.get("id")
if not isinstance(response_id, str) or not response_id.startswith("resp_"):
raise TriggerError("responses_missing_id")
return response_id
def poll(self, response_id: str) -> tuple[str, Decision | None]:
payload = self._request(
"GET",
f"https://api.openai.com/v1/responses/{response_id}",
None,
)
status = payload.get("status")
if status in {"queued", "in_progress"}:
return str(status), None
if status == "completed":
text = _extract_output_text(payload)
try:
return "completed", Decision.from_dict(json.loads(text))
except json.JSONDecodeError as exc:
raise TriggerError("responses_invalid_json") from exc
if status in {"failed", "cancelled", "incomplete"}:
raise TriggerError(f"responses_{status}")
raise TriggerError("responses_unknown_status")
def _request(
self,
method: str,
url: str,
body: dict[str, Any] | None,
*,
idempotency_key: str | None = None,
) -> dict[str, Any]:
encoded = json.dumps(body).encode() if body is not None else None
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"User-Agent": "aios-orchestrator-gateway/0.2",
}
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key
request = urllib.request.Request(
url,
data=encoded,
method=method,
headers=headers,
)
try:
with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
raw = response.read(4_194_304)
if response.status not in {200, 201}:
raise TriggerError(f"responses_http_{response.status}")
except urllib.error.HTTPError as exc:
raise TriggerError(f"responses_http_{exc.code}") from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise TriggerError("responses_request_failed") from exc
try:
payload = json.loads(raw)
except json.JSONDecodeError as exc:
raise TriggerError("responses_invalid_payload") from exc
if not isinstance(payload, dict):
raise TriggerError("responses_payload_not_object")
return payload
def _extract_output_text(payload: dict[str, Any]) -> str:
output = payload.get("output")
if not isinstance(output, list):
raise TriggerError("responses_missing_output")
for item in output:
if not isinstance(item, dict):
continue
content = item.get("content")
if not isinstance(content, list):
continue
for part in content:
if isinstance(part, dict) and part.get("type") == "output_text":
text = part.get("text")
if isinstance(text, str):
return text
raise TriggerError("responses_missing_output_text")
@@ -0,0 +1,213 @@
from __future__ import annotations
from dataclasses import dataclass
import os
from pathlib import Path
import stat
from typing import Any
from .errors import ConfigurationError
from .securefs import read_secure_bytes, read_secure_json, verify_secure_file
@dataclass(frozen=True)
class Config:
host: str
port: int
project_key: str
database_path: Path
event_inbox_dir: Path
decision_inbox_dir: Path
dispatch_outbox_dir: Path
openclaw_sync_dir: Path
command_queue_dir: Path
inbound_token_file: Path
callback_token_file: Path
contract_commit: str
workspace_trigger_id: str | None
workspace_token_file: Path | None
responses_api_key_file: Path | None
responses_model: str | None
poll_interval_seconds: int
summary_interval_seconds: int
callback_timeout_seconds: int
lease_seconds: int
retry_base_seconds: int
max_retries: int
max_same_blocker_retries: int
max_hops_per_gate: int
openclaw_sync_endpoint: str | None
openclaw_sync_token_file: Path | None
git_poll_enabled: bool
git_poll_bare_repo: Path | None
git_poll_ref_prefixes: tuple[str, ...]
git_poll_interval_seconds: int
@classmethod
def from_file(cls, path: Path) -> "Config":
owner_uid = int(os.environ.get("AIOS_ORCHESTRATOR_CONFIG_OWNER_UID", "0"))
raw = read_secure_json(path, expected_uid=owner_uid)
expected = {
"host", "port", "project_key", "database_path", "event_inbox_dir",
"decision_inbox_dir", "dispatch_outbox_dir", "openclaw_sync_dir",
"command_queue_dir", "inbound_token_file", "callback_token_file",
"contract_commit", "workspace_agent", "responses_fallback",
"polling", "loop_controls", "openclaw_sync", "git_poll",
}
if set(raw) != expected:
raise ConfigurationError("invalid_config_fields")
def path_value(key: str) -> Path:
value = raw.get(key)
if not isinstance(value, str):
raise ConfigurationError(f"invalid_{key}")
result = Path(value)
if not result.is_absolute():
raise ConfigurationError(f"path_not_absolute:{key}")
return result
def integer(container: dict[str, Any], key: str, lo: int, hi: int) -> int:
value = container.get(key)
if not isinstance(value, int) or not lo <= value <= hi:
raise ConfigurationError(f"invalid_{key}")
return value
host = raw.get("host")
port = raw.get("port")
project_key = raw.get("project_key")
contract = raw.get("contract_commit")
if not isinstance(host, str) or not host:
raise ConfigurationError("invalid_host")
if not isinstance(project_key, str) or not project_key:
raise ConfigurationError("invalid_project_key")
if not isinstance(port, int) or not 1 <= port <= 65535:
raise ConfigurationError("invalid_port")
if not isinstance(contract, str) or len(contract) != 40:
raise ConfigurationError("invalid_contract_commit")
workspace = raw.get("workspace_agent")
responses = raw.get("responses_fallback")
polling = raw.get("polling")
loop = raw.get("loop_controls")
sync = raw.get("openclaw_sync")
git_poll = raw.get("git_poll")
if not all(isinstance(x, dict) for x in (workspace, responses, polling, loop, sync, git_poll)):
raise ConfigurationError("invalid_nested_config")
trigger_id = workspace.get("trigger_id")
workspace_token = workspace.get("access_token_file")
if trigger_id is not None:
if not isinstance(trigger_id, str) or not trigger_id.startswith("agtch_"):
raise ConfigurationError("invalid_workspace_trigger_id")
if not isinstance(workspace_token, str) or not Path(workspace_token).is_absolute():
raise ConfigurationError("invalid_workspace_token_file")
workspace_token_file = Path(workspace_token)
else:
workspace_token_file = None
responses_key = responses.get("api_key_file")
responses_model = responses.get("model")
if responses_key is not None:
if not isinstance(responses_key, str) or not Path(responses_key).is_absolute():
raise ConfigurationError("invalid_responses_key_file")
if not isinstance(responses_model, str) or not responses_model:
raise ConfigurationError("invalid_responses_model")
responses_key_file = Path(responses_key)
else:
responses_key_file = None
responses_model = None
if trigger_id is None and responses_key_file is None:
raise ConfigurationError("no_central_model_channel")
endpoint = sync.get("endpoint")
token_file = sync.get("token_file")
if endpoint is not None and (not isinstance(endpoint, str) or not endpoint.startswith(("http://", "https://"))):
raise ConfigurationError("invalid_openclaw_sync_endpoint")
if endpoint is not None:
if not isinstance(token_file, str) or not Path(token_file).is_absolute():
raise ConfigurationError("invalid_openclaw_sync_token_file")
sync_token_file = Path(token_file)
else:
sync_token_file = None
git_enabled = git_poll.get("enabled")
if not isinstance(git_enabled, bool):
raise ConfigurationError("invalid_git_poll_enabled")
git_bare_repo_raw = git_poll.get("bare_repo")
git_prefixes_raw = git_poll.get("ref_prefixes")
if git_enabled:
if not isinstance(git_bare_repo_raw, str) or not Path(git_bare_repo_raw).is_absolute():
raise ConfigurationError("invalid_git_poll_bare_repo")
if (
not isinstance(git_prefixes_raw, list)
or not git_prefixes_raw
or any(not isinstance(item, str) or not item.startswith("refs/") for item in git_prefixes_raw)
):
raise ConfigurationError("invalid_git_poll_ref_prefixes")
git_bare_repo = Path(git_bare_repo_raw)
git_prefixes = tuple(git_prefixes_raw)
else:
git_bare_repo = None
git_prefixes = ()
return cls(
host=host,
port=port,
project_key=project_key,
database_path=path_value("database_path"),
event_inbox_dir=path_value("event_inbox_dir"),
decision_inbox_dir=path_value("decision_inbox_dir"),
dispatch_outbox_dir=path_value("dispatch_outbox_dir"),
openclaw_sync_dir=path_value("openclaw_sync_dir"),
command_queue_dir=path_value("command_queue_dir"),
inbound_token_file=path_value("inbound_token_file"),
callback_token_file=path_value("callback_token_file"),
contract_commit=contract,
workspace_trigger_id=trigger_id,
workspace_token_file=workspace_token_file,
responses_api_key_file=responses_key_file,
responses_model=responses_model,
poll_interval_seconds=integer(polling, "interval_seconds", 1, 300),
summary_interval_seconds=integer(polling, "summary_interval_seconds", 10, 86400),
callback_timeout_seconds=integer(loop, "callback_timeout_seconds", 30, 86400),
lease_seconds=integer(loop, "lease_seconds", 10, 3600),
retry_base_seconds=integer(loop, "retry_base_seconds", 1, 3600),
max_retries=integer(loop, "max_retries", 0, 20),
max_same_blocker_retries=integer(loop, "max_same_blocker_retries", 0, 20),
max_hops_per_gate=integer(loop, "max_hops_per_gate", 1, 100),
openclaw_sync_endpoint=endpoint,
openclaw_sync_token_file=sync_token_file,
git_poll_enabled=git_enabled,
git_poll_bare_repo=git_bare_repo,
git_poll_ref_prefixes=git_prefixes,
git_poll_interval_seconds=integer(git_poll, "interval_seconds", 5, 3600),
)
def _secret(self, path: Path) -> str:
owner_uid = int(os.environ.get("AIOS_ORCHESTRATOR_SECRET_OWNER_UID", "0"))
info = verify_secure_file(path, expected_uid=owner_uid)
if stat.S_IMODE(info.st_mode) & ~0o640:
raise ConfigurationError(f"permissive_secret_mode:{path}")
data = read_secure_bytes(path, expected_uid=owner_uid, max_bytes=64_000)
if not data or b"\n" in data or b"\r" in data:
raise ConfigurationError(f"invalid_secret_format:{path}")
try:
return data.decode("utf-8")
except UnicodeDecodeError as exc:
raise ConfigurationError(f"secret_not_utf8:{path}") from exc
def inbound_token(self) -> str:
return self._secret(self.inbound_token_file)
def callback_token(self) -> str:
return self._secret(self.callback_token_file)
def workspace_token(self) -> str | None:
return self._secret(self.workspace_token_file) if self.workspace_token_file else None
def responses_api_key(self) -> str | None:
return self._secret(self.responses_api_key_file) if self.responses_api_key_file else None
def openclaw_sync_token(self) -> str | None:
return self._secret(self.openclaw_sync_token_file) if self.openclaw_sync_token_file else None
@@ -0,0 +1,122 @@
from __future__ import annotations
import json
from pathlib import Path
import re
import os
import stat
from typing import Any
from .errors import DispatchError
from .securefs import (
atomic_write_bytes,
atomic_write_bytes_shared,
ensure_secure_directory,
ensure_shared_directory,
read_secure_bytes,
)
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
class DispatchSink:
"""Reliable file-based dispatch adapter for existing OpenClaw/Codex watchers."""
def __init__(self, outbox_dir: Path, command_queue_dir: Path, contract_commit: str):
self.outbox_dir = Path(outbox_dir)
self.command_queue_dir = Path(command_queue_dir)
self.contract_commit = contract_commit
ensure_secure_directory(self.outbox_dir)
ensure_shared_directory(self.command_queue_dir)
def dispatch(self, record: dict[str, Any]) -> tuple[Path, Path]:
work_unit_id = record.get("work_unit_id")
event_id = record.get("event_id")
agent = record.get("assigned_agent")
payload_json = record.get("payload_json")
if (
not isinstance(work_unit_id, str)
or not SAFE_ID.fullmatch(work_unit_id)
or not isinstance(event_id, str)
or not SAFE_ID.fullmatch(event_id)
or agent not in {"openclaw", "codex"}
or not isinstance(payload_json, str)
):
raise DispatchError("invalid_dispatch_record")
try:
work_unit = json.loads(payload_json)
except json.JSONDecodeError as exc:
raise DispatchError("invalid_dispatch_payload") from exc
project_key = record.get("project_key")
state_version = record.get("state_version")
envelope = {
"schema_version": 1,
"contract_commit": self.contract_commit,
"event_id": event_id,
"assigned_agent": agent,
"work_unit": work_unit,
"callback": {
"event_endpoint": "/v1/orchestrator/events",
"start_event_type": "work_unit_started",
"end_event_type": "work_unit_completed",
},
}
outbox_path = self.outbox_dir / "dispatched" / f"{work_unit_id}.json"
if agent == "codex":
if (
not isinstance(project_key, str)
or not SAFE_ID.fullmatch(project_key)
or not isinstance(state_version, int)
or state_version < 0
):
raise DispatchError("invalid_dispatch_context")
required_checks = work_unit.get("required_checks", [])
execution_kind = (
"acceptance_probe"
if isinstance(required_checks, list) and "acceptance_probe" in required_checks
else "codex"
)
queue_value = {
"schema_version": 1,
"command_id": work_unit_id,
"project_key": project_key,
"work_unit_id": work_unit_id,
"contract_commit": self.contract_commit,
"state_version": state_version,
"execution": {
"kind": execution_kind,
"input": work_unit.get("instruction", ""),
},
}
# The Codex watcher polls the configured active directory itself.
queue_path = self.command_queue_dir / f"{work_unit_id}.json"
else:
queue_value = envelope
queue_path = self.command_queue_dir / agent / "pending" / f"{work_unit_id}.json"
ensure_secure_directory(outbox_path.parent)
ensure_shared_directory(queue_path.parent)
# Idempotent immutable payload: existing identical file is accepted.
self._write_idempotent_json(outbox_path, envelope, shared=False)
self._write_idempotent_json(queue_path, queue_value, shared=True)
return outbox_path, queue_path
@staticmethod
def _write_idempotent_json(path: Path, value: dict[str, Any], *, shared: bool) -> None:
payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode()
try:
st = os.lstat(path)
except FileNotFoundError:
if shared:
atomic_write_bytes_shared(path, payload, mode=0o640)
else:
atomic_write_bytes(path, payload, mode=0o600)
return
if not stat.S_ISREG(st.st_mode):
raise DispatchError(f"dispatch_target_not_regular:{path.name}")
existing = read_secure_bytes(path, expected_uid=None, max_bytes=1_048_576)
if existing != payload:
raise DispatchError(f"dispatch_idempotency_conflict:{path.name}")
@@ -0,0 +1,37 @@
class OrchestratorError(Exception):
"""Base orchestration error."""
class ConfigurationError(OrchestratorError):
"""Unsafe or invalid configuration."""
class ValidationError(OrchestratorError):
"""Invalid event or decision."""
class AuthenticationError(OrchestratorError):
"""Authentication failed."""
class ConflictError(OrchestratorError):
"""Idempotency, lease, or optimistic concurrency conflict."""
class TriggerError(OrchestratorError):
"""Central model trigger failed."""
def __init__(
self,
message: str,
*,
http_status: int | None = None,
category: str | None = None,
) -> None:
super().__init__(message)
self.http_status = http_status
self.category = category
class DispatchError(OrchestratorError):
"""Dispatch or synchronization output failed."""
@@ -0,0 +1,346 @@
from __future__ import annotations
from dataclasses import dataclass, field
import hashlib
import json
import re
from typing import Any
from .errors import ValidationError
OID_RE = re.compile(r"^[0-9a-f]{40}$")
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
AGENTS = {"openclaw", "codex", "gitea", "system", "workspace-agent"}
EVENT_STATUSES = {"READY", "RUNNING", "PASS", "BLOCKED", "FAIL", "TIMEOUT", "NOT_EXECUTED"}
VERDICTS = {"DISPATCH", "BLOCKED", "COMPLETE", "NOOP"}
EXECUTORS = {"openclaw", "codex"}
def req_str(value: dict[str, Any], key: str, pattern: re.Pattern[str] | None = None) -> str:
item = value.get(key)
if not isinstance(item, str) or not item:
raise ValidationError(f"invalid_{key}")
if pattern is not None and not pattern.fullmatch(item):
raise ValidationError(f"invalid_{key}")
return item
def opt_oid(item: Any, key: str) -> str | None:
if item is None:
return None
if not isinstance(item, str) or not OID_RE.fullmatch(item):
raise ValidationError(f"invalid_{key}")
return item
def string_list(value: dict[str, Any], key: str, limit: int) -> tuple[str, ...]:
raw = value.get(key)
if not isinstance(raw, list) or len(raw) > limit:
raise ValidationError(f"invalid_{key}")
result: list[str] = []
for item in raw:
if not isinstance(item, str) or not item or len(item) > 1024:
raise ValidationError(f"invalid_{key}")
result.append(item)
return tuple(result)
@dataclass(frozen=True)
class Event:
schema_version: int
project_key: str
contract_commit: str
event_id: str
source_agent: str
event_type: str
work_unit_id: str
status: str
state_version: int
implementation_commit: str | None = None
result_commit: str | None = None
blocker: str | None = None
evidence_refs: tuple[str, ...] = ()
payload: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_dict(cls, value: dict[str, Any]) -> "Event":
allowed = {
"schema_version", "project_key", "contract_commit", "event_id",
"source_agent", "event_type", "work_unit_id", "status",
"state_version", "implementation_commit", "result_commit",
"blocker", "evidence_refs", "payload",
}
if not isinstance(value, dict) or set(value) - allowed:
raise ValidationError("invalid_event_fields")
if value.get("schema_version") != 1:
raise ValidationError("unsupported_event_schema")
source = req_str(value, "source_agent")
status = req_str(value, "status")
if source not in AGENTS:
raise ValidationError("invalid_source_agent")
if status not in EVENT_STATUSES:
raise ValidationError("invalid_event_status")
version = value.get("state_version")
if not isinstance(version, int) or version < 0:
raise ValidationError("invalid_state_version")
blocker = value.get("blocker")
if blocker is not None and (not isinstance(blocker, str) or len(blocker) > 512):
raise ValidationError("invalid_blocker")
payload = value.get("payload", {})
if not isinstance(payload, dict):
raise ValidationError("invalid_payload")
if len(json.dumps(payload, sort_keys=True).encode()) > 262_144:
raise ValidationError("payload_too_large")
return cls(
schema_version=1,
project_key=req_str(value, "project_key", ID_RE),
contract_commit=req_str(value, "contract_commit", OID_RE),
event_id=req_str(value, "event_id", ID_RE),
source_agent=source,
event_type=req_str(value, "event_type", ID_RE),
work_unit_id=req_str(value, "work_unit_id", ID_RE),
status=status,
state_version=version,
implementation_commit=opt_oid(value.get("implementation_commit"), "implementation_commit"),
result_commit=opt_oid(value.get("result_commit"), "result_commit"),
blocker=blocker,
evidence_refs=string_list(value, "evidence_refs", 64),
payload=payload,
)
def to_dict(self) -> dict[str, Any]:
return {
"schema_version": 1,
"project_key": self.project_key,
"contract_commit": self.contract_commit,
"event_id": self.event_id,
"source_agent": self.source_agent,
"event_type": self.event_type,
"work_unit_id": self.work_unit_id,
"status": self.status,
"state_version": self.state_version,
"implementation_commit": self.implementation_commit,
"result_commit": self.result_commit,
"blocker": self.blocker,
"evidence_refs": list(self.evidence_refs),
"payload": self.payload,
}
def canonical_json(self) -> str:
return json.dumps(self.to_dict(), separators=(",", ":"), sort_keys=True)
def digest(self) -> str:
return hashlib.sha256(self.canonical_json().encode()).hexdigest()
@dataclass(frozen=True)
class WorkUnit:
id: str
track_id: str
gate_id: str
primary_agent: str
fallback_agent: str
clean_base_commit: str
allowed_paths: tuple[str, ...]
required_checks: tuple[str, ...]
forbidden_operations: tuple[str, ...]
instruction: str
@classmethod
def from_dict(cls, value: dict[str, Any]) -> "WorkUnit":
expected = {
"id", "track_id", "gate_id", "primary_agent", "fallback_agent",
"clean_base_commit", "allowed_paths", "required_checks",
"forbidden_operations", "instruction",
}
if not isinstance(value, dict) or set(value) != expected:
raise ValidationError("invalid_work_unit_fields")
primary = req_str(value, "primary_agent")
fallback = req_str(value, "fallback_agent")
if primary not in EXECUTORS or fallback not in EXECUTORS or primary == fallback:
raise ValidationError("invalid_agent_assignment")
instruction = req_str(value, "instruction")
if len(instruction.encode()) > 65_536:
raise ValidationError("instruction_too_large")
return cls(
id=req_str(value, "id", ID_RE),
track_id=req_str(value, "track_id", ID_RE),
gate_id=req_str(value, "gate_id", ID_RE),
primary_agent=primary,
fallback_agent=fallback,
clean_base_commit=req_str(value, "clean_base_commit", OID_RE),
allowed_paths=string_list(value, "allowed_paths", 128),
required_checks=string_list(value, "required_checks", 128),
forbidden_operations=string_list(value, "forbidden_operations", 64),
instruction=instruction,
)
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"track_id": self.track_id,
"gate_id": self.gate_id,
"primary_agent": self.primary_agent,
"fallback_agent": self.fallback_agent,
"clean_base_commit": self.clean_base_commit,
"allowed_paths": list(self.allowed_paths),
"required_checks": list(self.required_checks),
"forbidden_operations": list(self.forbidden_operations),
"instruction": self.instruction,
}
@dataclass(frozen=True)
class Decision:
schema_version: int
project_key: str
contract_commit: str
event_id: str
expected_state_version: int
verdict: str
state_updates: dict[str, str]
next_work_units: tuple[WorkUnit, ...]
requires_human_input: bool
stop_reason: str | None
@classmethod
def from_dict(cls, value: dict[str, Any]) -> "Decision":
expected = {
"schema_version", "project_key", "contract_commit", "event_id",
"expected_state_version", "verdict", "state_updates",
"next_work_units", "requires_human_input", "stop_reason",
}
if not isinstance(value, dict) or set(value) != expected:
raise ValidationError("invalid_decision_fields")
if value.get("schema_version") != 1:
raise ValidationError("unsupported_decision_schema")
version = value.get("expected_state_version")
if not isinstance(version, int) or version < 0:
raise ValidationError("invalid_expected_state_version")
verdict = req_str(value, "verdict")
if verdict not in VERDICTS:
raise ValidationError("invalid_verdict")
updates = value.get("state_updates")
normalized: dict[str, str] = {}
if isinstance(updates, dict):
update_items = [{"key": key, "value": item} for key, item in updates.items()]
elif isinstance(updates, list):
update_items = updates
else:
raise ValidationError("invalid_state_updates")
if len(update_items) > 128:
raise ValidationError("invalid_state_updates")
for update in update_items:
if not isinstance(update, dict) or set(update) != {"key", "value"}:
raise ValidationError("invalid_state_update")
key = update.get("key")
item = update.get("value")
if not isinstance(key, str) or not ID_RE.fullmatch(key):
raise ValidationError("invalid_state_update_key")
if not isinstance(item, str) or not item or len(item) > 256:
raise ValidationError("invalid_state_update_value")
if key in normalized:
raise ValidationError("duplicate_state_update_key")
normalized[key] = item
raw_units = value.get("next_work_units")
if not isinstance(raw_units, list) or len(raw_units) > 32:
raise ValidationError("invalid_next_work_units")
units = tuple(WorkUnit.from_dict(item) for item in raw_units)
human = value.get("requires_human_input")
if not isinstance(human, bool):
raise ValidationError("invalid_requires_human_input")
stop = value.get("stop_reason")
if stop is not None and (not isinstance(stop, str) or len(stop) > 512):
raise ValidationError("invalid_stop_reason")
if verdict == "DISPATCH" and not units:
raise ValidationError("dispatch_without_work_units")
if verdict != "DISPATCH" and units:
raise ValidationError("work_units_without_dispatch")
return cls(
schema_version=1,
project_key=req_str(value, "project_key", ID_RE),
contract_commit=req_str(value, "contract_commit", OID_RE),
event_id=req_str(value, "event_id", ID_RE),
expected_state_version=version,
verdict=verdict,
state_updates=normalized,
next_work_units=units,
requires_human_input=human,
stop_reason=stop,
)
def to_dict(self) -> dict[str, Any]:
return {
"schema_version": 1,
"project_key": self.project_key,
"contract_commit": self.contract_commit,
"event_id": self.event_id,
"expected_state_version": self.expected_state_version,
"verdict": self.verdict,
"state_updates": [
{"key": key, "value": value}
for key, value in sorted(self.state_updates.items())
],
"next_work_units": [u.to_dict() for u in self.next_work_units],
"requires_human_input": self.requires_human_input,
"stop_reason": self.stop_reason,
}
DECISION_SCHEMA: dict[str, Any] = {
"type": "object",
"additionalProperties": False,
"required": [
"schema_version", "project_key", "contract_commit", "event_id",
"expected_state_version", "verdict", "state_updates",
"next_work_units", "requires_human_input", "stop_reason",
],
"properties": {
"schema_version": {"type": "integer", "const": 1},
"project_key": {"type": "string"},
"contract_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
"event_id": {"type": "string"},
"expected_state_version": {"type": "integer", "minimum": 0},
"verdict": {"type": "string", "enum": sorted(VERDICTS)},
"state_updates": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"required": ["key", "value"],
"properties": {
"key": {"type": "string"},
"value": {"type": "string"}
}
}
},
"next_work_units": {
"type": "array",
"maxItems": 32,
"items": {
"type": "object",
"additionalProperties": False,
"required": [
"id", "track_id", "gate_id", "primary_agent",
"fallback_agent", "clean_base_commit", "allowed_paths",
"required_checks", "forbidden_operations", "instruction",
],
"properties": {
"id": {"type": "string"},
"track_id": {"type": "string"},
"gate_id": {"type": "string"},
"primary_agent": {"type": "string", "enum": ["openclaw", "codex"]},
"fallback_agent": {"type": "string", "enum": ["openclaw", "codex"]},
"clean_base_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
"allowed_paths": {"type": "array", "items": {"type": "string"}},
"required_checks": {"type": "array", "items": {"type": "string"}},
"forbidden_operations": {"type": "array", "items": {"type": "string"}},
"instruction": {"type": "string"},
},
},
},
"requires_human_input": {"type": "boolean"},
"stop_reason": {"type": ["string", "null"]},
},
}
@@ -0,0 +1,228 @@
from __future__ import annotations
import json
import os
from pathlib import Path
import stat
from typing import Any
from .errors import ConfigurationError
_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
def reject_symlink_chain(path: Path) -> None:
"""Reject symlinks in every existing component of an absolute path."""
path = Path(path)
if not path.is_absolute():
raise ConfigurationError(f"path_not_absolute:{path}")
current = Path(path.anchor)
for part in path.parts[1:]:
current = current / part
try:
st = os.lstat(current)
except FileNotFoundError:
continue
if stat.S_ISLNK(st.st_mode):
raise ConfigurationError(f"symlink_rejected:{current}")
def verify_secure_file(
path: Path,
*,
expected_uid: int | None,
allowed_write_bits: int = stat.S_IWUSR,
require_executable: bool = False,
) -> os.stat_result:
path = Path(path)
reject_symlink_chain(path)
before = os.lstat(path)
if not stat.S_ISREG(before.st_mode):
raise ConfigurationError(f"not_regular_file:{path}")
if expected_uid is not None and before.st_uid != expected_uid:
raise ConfigurationError(f"unexpected_owner:{path}")
write_bits = before.st_mode & (stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
if write_bits & ~allowed_write_bits:
raise ConfigurationError(f"permissive_mode:{path}")
if require_executable and not (before.st_mode & stat.S_IXUSR):
raise ConfigurationError(f"not_executable:{path}")
fd = os.open(path, os.O_RDONLY | _NOFOLLOW)
try:
after = os.fstat(fd)
if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
raise ConfigurationError(f"inode_changed:{path}")
finally:
os.close(fd)
return before
def read_secure_bytes(
path: Path,
*,
expected_uid: int | None,
max_bytes: int = 1_048_576,
) -> bytes:
path = Path(path)
before = verify_secure_file(path, expected_uid=expected_uid)
if before.st_size > max_bytes:
raise ConfigurationError(f"file_too_large:{path}")
fd = os.open(path, os.O_RDONLY | _NOFOLLOW)
try:
data = bytearray()
while True:
chunk = os.read(fd, min(65536, max_bytes + 1 - len(data)))
if not chunk:
break
data.extend(chunk)
if len(data) > max_bytes:
raise ConfigurationError(f"file_too_large:{path}")
return bytes(data)
finally:
os.close(fd)
def read_secure_text(path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576) -> str:
try:
return read_secure_bytes(path, expected_uid=expected_uid, max_bytes=max_bytes).decode("utf-8").strip()
except UnicodeDecodeError as exc:
raise ConfigurationError(f"not_utf8:{path}") from exc
def read_secure_json(path: Path, *, expected_uid: int | None, max_bytes: int = 1_048_576) -> dict[str, Any]:
try:
value = json.loads(read_secure_text(path, expected_uid=expected_uid, max_bytes=max_bytes))
except json.JSONDecodeError as exc:
raise ConfigurationError(f"invalid_json:{path}") from exc
if not isinstance(value, dict):
raise ConfigurationError(f"json_root_not_object:{path}")
return value
def ensure_secure_directory(path: Path, *, mode: int = 0o700) -> None:
path = Path(path)
if not path.is_absolute():
raise ConfigurationError(f"path_not_absolute:{path}")
reject_symlink_chain(path.parent)
path.mkdir(parents=True, exist_ok=True, mode=mode)
reject_symlink_chain(path)
st = os.lstat(path)
if not stat.S_ISDIR(st.st_mode):
raise ConfigurationError(f"not_directory:{path}")
if st.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
raise ConfigurationError(f"permissive_directory:{path}")
def atomic_write_bytes(path: Path, payload: bytes, *, mode: int = 0o600) -> None:
"""Write through a verified directory fd and atomically replace the target."""
path = Path(path)
ensure_secure_directory(path.parent)
parent_before = os.lstat(path.parent)
parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | _NOFOLLOW)
temp_name = f".{path.name}.tmp-{os.getpid()}-{os.urandom(6).hex()}"
try:
parent_after = os.fstat(parent_fd)
if (parent_before.st_dev, parent_before.st_ino) != (
parent_after.st_dev,
parent_after.st_ino,
):
raise ConfigurationError(f"directory_changed:{path.parent}")
fd = os.open(
temp_name,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW,
mode,
dir_fd=parent_fd,
)
try:
view = memoryview(payload)
while view:
written = os.write(fd, view)
view = view[written:]
os.fsync(fd)
finally:
os.close(fd)
os.replace(
temp_name,
path.name,
src_dir_fd=parent_fd,
dst_dir_fd=parent_fd,
)
os.fsync(parent_fd)
finally:
try:
os.unlink(temp_name, dir_fd=parent_fd)
except FileNotFoundError:
pass
os.close(parent_fd)
def atomic_write_json(path: Path, value: Any, *, mode: int = 0o600) -> None:
payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8")
atomic_write_bytes(path, payload, mode=mode)
def ensure_shared_directory(path: Path, *, group_writable: bool = False) -> None:
"""Verify/create a setgid queue directory shared with a consumer group."""
path = Path(path)
if not path.is_absolute():
raise ConfigurationError(f"path_not_absolute:{path}")
reject_symlink_chain(path.parent)
mode = 0o2770 if group_writable else 0o2750
try:
os.lstat(path)
except FileNotFoundError:
path.mkdir(parents=True, mode=mode)
os.chmod(path, mode)
reject_symlink_chain(path)
st = os.lstat(path)
if not stat.S_ISDIR(st.st_mode):
raise ConfigurationError(f"not_directory:{path}")
observed = stat.S_IMODE(st.st_mode)
required = stat.S_ISGID | stat.S_IRGRP | stat.S_IXGRP
if group_writable:
required |= stat.S_IWGRP
if observed & 0o007 or observed & required != required:
raise ConfigurationError(f"invalid_shared_directory_mode:{path}")
def atomic_write_bytes_shared(
path: Path,
payload: bytes,
*,
mode: int = 0o640,
group_writable_directory: bool = False,
) -> None:
"""Atomic write into a setgid/group-readable shared queue directory."""
path = Path(path)
ensure_shared_directory(path.parent, group_writable=group_writable_directory)
parent_before = os.lstat(path.parent)
parent_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | _NOFOLLOW)
temp_name = f".{path.name}.tmp-{os.getpid()}-{os.urandom(6).hex()}"
try:
parent_after = os.fstat(parent_fd)
if (parent_before.st_dev, parent_before.st_ino) != (
parent_after.st_dev,
parent_after.st_ino,
):
raise ConfigurationError(f"directory_changed:{path.parent}")
fd = os.open(
temp_name,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | _NOFOLLOW,
mode,
dir_fd=parent_fd,
)
os.fchmod(fd, mode)
try:
view = memoryview(payload)
while view:
written = os.write(fd, view)
view = view[written:]
os.fsync(fd)
finally:
os.close(fd)
os.replace(temp_name, path.name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
os.fsync(parent_fd)
finally:
try:
os.unlink(temp_name, dir_fd=parent_fd)
except FileNotFoundError:
pass
os.close(parent_fd)
@@ -0,0 +1,675 @@
from __future__ import annotations
import hashlib
import hmac
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import logging
import os
from pathlib import Path
import socket
import subprocess
import threading
import time
from typing import Any
from urllib.parse import urlparse
from .clients import ResponsesClient, WorkspaceAgentClient
from .config import Config
from .dispatch import DispatchSink
from .errors import (
AuthenticationError,
ConflictError,
DispatchError,
OrchestratorError,
TriggerError,
ValidationError,
)
from .models import Decision, Event, ID_RE
from .securefs import ensure_secure_directory, ensure_shared_directory, read_secure_json
from .store import Store
from .sync import OpenClawSynchronizer, lifecycle_record
LOG = logging.getLogger("aios-orchestrator-gateway")
class Runtime:
def __init__(self, config: Config):
self.config = config
self.store = Store(config.database_path)
self.dispatch = DispatchSink(
config.dispatch_outbox_dir,
config.command_queue_dir,
config.contract_commit,
)
self.workspace = None
workspace_token = config.workspace_token()
if config.workspace_trigger_id and workspace_token:
self.workspace = WorkspaceAgentClient(
config.workspace_trigger_id,
workspace_token,
)
self.responses = None
responses_key = config.responses_api_key()
if responses_key and config.responses_model:
self.responses = ResponsesClient(
responses_key,
config.responses_model,
)
self.synchronizer = OpenClawSynchronizer(
config.openclaw_sync_dir,
endpoint=config.openclaw_sync_endpoint,
bearer_token=config.openclaw_sync_token(),
)
self.worker_id = f"{socket.gethostname()}-{os.getpid()}-{threading.get_native_id()}"
self.stop_event = threading.Event()
self.worker = threading.Thread(target=self._loop, daemon=True)
self._next_git_poll = 0.0
for directory in (
config.event_inbox_dir,
config.decision_inbox_dir,
config.dispatch_outbox_dir,
config.openclaw_sync_dir,
):
ensure_secure_directory(directory)
ensure_shared_directory(config.command_queue_dir)
def start(self) -> None:
self.worker.start()
def stop(self) -> None:
self.stop_event.set()
if self.worker.is_alive():
self.worker.join(timeout=15)
self.store.close()
def submit_event(self, payload: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
event = Event.from_dict(payload)
if event.contract_commit != self.config.contract_commit:
raise ConflictError("contract_version_mismatch")
created, stored = self.store.register_event(event)
if created:
self._enqueue_event_lifecycle(event)
return created, stored
def submit_decision(self, payload: dict[str, Any]) -> int:
decision = Decision.from_dict(payload)
if decision.contract_commit != self.config.contract_commit:
raise ConflictError("contract_version_mismatch")
version = self.store.apply_decision(
decision,
max_hops_per_gate=self.config.max_hops_per_gate,
max_same_blocker_retries=self.config.max_same_blocker_retries,
)
self._enqueue_decision_lifecycle(decision, version)
return version
def _loop(self) -> None:
next_summary = 0.0
while not self.stop_event.wait(self.config.poll_interval_seconds):
try:
self.poll_once()
if time.monotonic() >= next_summary:
self.enqueue_periodic_summaries()
next_summary = time.monotonic() + self.config.summary_interval_seconds
except Exception:
LOG.exception("poll_cycle_failed")
def poll_once(self) -> None:
"""One deterministic reliability cycle.
This is deliberately callable from a systemd timer or health supervisor
in addition to the internal loop.
"""
self.ingest_file_inboxes()
self.poll_git_refs()
self.process_new_events()
self.process_workspace_timeouts()
self.poll_background_responses()
self.dispatch_work_units()
self.synchronize_openclaw()
def ingest_file_inboxes(self) -> None:
self._ingest_directory(self.config.event_inbox_dir, kind="event")
self._ingest_directory(self.config.decision_inbox_dir, kind="decision")
def _ingest_directory(self, directory: Path, *, kind: str) -> None:
pending = directory / "pending"
processed = directory / "processed"
rejected = directory / "rejected"
ensure_shared_directory(pending, group_writable=True)
ensure_secure_directory(processed)
ensure_secure_directory(rejected)
for path in sorted(pending.glob("*.json"))[:128]:
try:
payload = read_secure_json(path, expected_uid=None)
if kind == "event":
self.submit_event(payload)
else:
self.submit_decision(payload)
target = processed / path.name
except Exception as exc:
LOG.warning("file_inbox_rejected path=%s error=%s", path, type(exc).__name__)
target = rejected / path.name
if target.exists():
# Same file name is an idempotency replay. Keep original audit copy.
path.unlink()
else:
path.replace(target)
def poll_git_refs(self) -> None:
"""Poll authoritative local bare refs so missed webhooks cannot stall the loop."""
if not self.config.git_poll_enabled or self.config.git_poll_bare_repo is None:
return
now = time.monotonic()
if now < self._next_git_poll:
return
self._next_git_poll = now + self.config.git_poll_interval_seconds
command = [
"git",
f"--git-dir={self.config.git_poll_bare_repo}",
"for-each-ref",
"--format=%(refname) %(objectname)",
*self.config.git_poll_ref_prefixes,
]
try:
result = subprocess.run(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
timeout=30,
env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
)
except (OSError, subprocess.SubprocessError):
LOG.exception("git_ref_poll_failed")
return
if result.returncode != 0:
LOG.warning("git_ref_poll_rc=%s", result.returncode)
return
refs: dict[str, str] = {}
for line in result.stdout.splitlines():
refname, sep, oid = line.partition(" ")
if not sep or not refname.startswith("refs/") or len(oid) != 40:
continue
if any(ch not in "0123456789abcdef" for ch in oid):
continue
refs[refname] = oid
initialized, changes = self.store.observe_refs(refs)
if not initialized or not changes:
return
payload_changes = [
{"ref": refname, "old_oid": old_oid, "new_oid": new_oid}
for refname, old_oid, new_oid in sorted(changes)
]
canonical = json.dumps(payload_changes, separators=(",", ":"), sort_keys=True)
event_id = "gitpoll-" + hashlib.sha256(canonical.encode()).hexdigest()[:40]
state = self.store.project_state(self.config.project_key)
state_version = 0 if state is None else int(state["state_version"])
event = Event.from_dict(
{
"schema_version": 1,
"project_key": self.config.project_key,
"contract_commit": self.config.contract_commit,
"event_id": event_id,
"source_agent": "gitea",
"event_type": "git_refs_changed",
"work_unit_id": "GIT-REF-POLL",
"status": "READY",
"state_version": state_version,
"implementation_commit": None,
"result_commit": None,
"blocker": None,
"evidence_refs": [],
"payload": {"changes": payload_changes},
}
)
try:
self.submit_event(event.to_dict())
except ConflictError:
LOG.info("git_ref_event_already_registered event_id=%s", event_id)
def process_new_events(self) -> None:
rows = self.store.claim_events(
owner=self.worker_id,
statuses=("RECEIVED", "TRIGGER_FAILED"),
limit=1,
lease_seconds=self.config.lease_seconds,
)
for row in rows:
event = Event.from_dict(json.loads(row["event_json"]))
attempts = int(row["attempts"])
try:
if self.workspace is not None:
try:
accepted = self.workspace.trigger(event)
except TriggerError as exc:
self._retry_or_dead_letter(
event,
attempts,
str(exc),
trigger_channel="workspace_agent",
trigger_http_status=exc.http_status,
trigger_error_category=exc.category,
)
else:
self.store.update_event_trigger(
event.event_id,
status="AWAITING_DECISION",
channel="workspace_agent",
remote_id=accepted.remote_id,
trigger_http_status=accepted.http_status,
)
elif self.responses is not None:
response_id = self.responses.start(event)
self.store.update_event_trigger(
event.event_id,
status="POLLING_RESPONSE",
channel="responses_api",
remote_id=response_id,
retry_after=self.config.poll_interval_seconds,
)
else:
raise TriggerError("no_model_channel")
except TriggerError as exc:
self._retry_or_dead_letter(event, attempts, str(exc))
def process_workspace_timeouts(self) -> None:
cutoff = int(time.time()) - self.config.callback_timeout_seconds
for row in self.store.claim_workspace_timeouts(
owner=self.worker_id,
older_than=cutoff,
limit=1,
lease_seconds=self.config.lease_seconds,
):
event = Event.from_dict(json.loads(row["event_json"]))
if self.responses is None:
self._retry_or_dead_letter(
event,
int(row["attempts"]),
"workspace_callback_timeout",
)
continue
try:
response_id = self.responses.start(event)
self.store.update_event_trigger(
event.event_id,
status="POLLING_RESPONSE",
channel="responses_api",
remote_id=response_id,
retry_after=self.config.poll_interval_seconds,
increment_attempt=True,
)
except TriggerError as exc:
self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
def poll_background_responses(self) -> None:
if self.responses is None:
return
for row in self.store.claim_response_polls(
owner=self.worker_id,
limit=1,
lease_seconds=self.config.lease_seconds,
):
event = Event.from_dict(json.loads(row["event_json"]))
response_id = row.get("remote_id")
if not isinstance(response_id, str):
self._retry_or_dead_letter(event, int(row["attempts"]), "missing_response_id")
continue
try:
status, decision = self.responses.poll(response_id)
if status in {"queued", "in_progress"}:
attempts = int(row["attempts"])
if attempts + 1 > self.config.max_retries:
self._retry_or_dead_letter(
event, attempts, f"response_{status}_retry_limit"
)
continue
self.store.update_event_trigger(
event.event_id,
status="POLLING_RESPONSE",
channel="responses_api",
remote_id=response_id,
retry_after=self.config.poll_interval_seconds,
increment_attempt=True,
)
continue
if decision is None:
raise TriggerError("completed_without_decision")
self.submit_decision(decision.to_dict())
except OrchestratorError as exc:
self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
def dispatch_work_units(self) -> None:
rows = self.store.claim_outbox(
owner=self.worker_id,
limit=64,
lease_seconds=self.config.lease_seconds,
)
for row in rows:
try:
outbox_path, queue_path = self.dispatch.dispatch(row)
self.store.mark_outbox(row["work_unit_id"], status="DISPATCHED")
payload = {
"assigned_agent": row["assigned_agent"],
"event_id": row["event_id"],
"dispatch_outbox": str(outbox_path),
"command_queue": str(queue_path),
}
record = lifecycle_record(
contract_commit=self.config.contract_commit,
project_key=self._project_for_event(row["event_id"]),
kind="TASK_DISPATCHED",
work_unit_id=row["work_unit_id"],
payload=payload,
)
self.store.enqueue_lifecycle(
sync_id=record["sync_id"],
project_key=record["project_key"],
work_unit_id=row["work_unit_id"],
kind=record["kind"],
payload=record,
)
except Exception:
attempts = int(row["attempts"])
status = "DEAD_LETTER" if attempts >= self.config.max_retries else "RETRY"
self.store.mark_outbox(
row["work_unit_id"],
status=status,
retry_after=self._backoff(attempts),
increment_attempt=True,
)
LOG.exception("dispatch_failed work_unit=%s", row["work_unit_id"])
def synchronize_openclaw(self) -> None:
rows = self.store.claim_lifecycle(
owner=self.worker_id,
limit=64,
lease_seconds=self.config.lease_seconds,
)
for row in rows:
try:
payload = json.loads(row["payload_json"])
self.synchronizer.emit(payload)
self.store.mark_lifecycle(row["sync_id"], status="SENT")
except Exception:
attempts = int(row["attempts"])
status = "DEAD_LETTER" if attempts >= self.config.max_retries else "RETRY"
self.store.mark_lifecycle(
row["sync_id"],
status=status,
retry_after=self._backoff(attempts),
increment_attempt=True,
)
LOG.exception("openclaw_sync_failed sync_id=%s", row["sync_id"])
def enqueue_periodic_summaries(self) -> None:
for state in self.store.all_project_states():
project_key = state["project_key"]
if not self.store.summary_due(project_key, self.config.summary_interval_seconds):
continue
summary = self.store.project_summary(project_key)
if summary is None:
continue
digest = hashlib.sha256(
json.dumps(summary, separators=(",", ":"), sort_keys=True).encode()
).hexdigest()
record = lifecycle_record(
contract_commit=self.config.contract_commit,
project_key=project_key,
kind="PROJECT_SUMMARY",
work_unit_id=None,
payload=summary,
)
self.store.enqueue_lifecycle(
sync_id=record["sync_id"],
project_key=project_key,
work_unit_id=None,
kind=record["kind"],
payload=record,
)
self.store.mark_summary(project_key, digest)
def _enqueue_event_lifecycle(self, event: Event) -> None:
if event.event_type == "work_unit_started":
kind = "TASK_STARTED"
elif event.event_type in {"work_unit_completed", "work_unit_failed", "work_unit_blocked"}:
kind = "TASK_ENDED"
else:
kind = "EVENT_RECEIVED"
record = lifecycle_record(
contract_commit=self.config.contract_commit,
project_key=event.project_key,
kind=kind,
work_unit_id=event.work_unit_id,
payload=event.to_dict(),
)
self.store.enqueue_lifecycle(
sync_id=record["sync_id"],
project_key=event.project_key,
work_unit_id=event.work_unit_id,
kind=kind,
payload=record,
)
def _enqueue_decision_lifecycle(self, decision: Decision, version: int) -> None:
kind = "PROJECT_COMPLETED" if decision.verdict == "COMPLETE" else "DECISION_APPLIED"
record = lifecycle_record(
contract_commit=self.config.contract_commit,
project_key=decision.project_key,
kind=kind,
work_unit_id=None,
payload={"decision": decision.to_dict(), "state_version": version},
)
self.store.enqueue_lifecycle(
sync_id=record["sync_id"],
project_key=decision.project_key,
work_unit_id=None,
kind=kind,
payload=record,
)
summary = self.store.project_summary(decision.project_key)
if summary:
summary_record = lifecycle_record(
contract_commit=self.config.contract_commit,
project_key=decision.project_key,
kind="PROJECT_SUMMARY",
work_unit_id=None,
payload=summary,
)
self.store.enqueue_lifecycle(
sync_id=summary_record["sync_id"],
project_key=decision.project_key,
work_unit_id=None,
kind=summary_record["kind"],
payload=summary_record,
)
def _project_for_event(self, event_id: str) -> str:
rows = self.store._conn().execute(
"SELECT project_key FROM events WHERE event_id=?",
(event_id,),
).fetchone()
if rows is None:
raise ConflictError("event_not_found")
return str(rows["project_key"])
def _retry_or_dead_letter(
self,
event: Event,
attempts: int,
reason: str,
*,
trigger_channel: str | None = None,
trigger_http_status: int | None = None,
trigger_error_category: str | None = None,
) -> None:
next_attempt = attempts + 1
if next_attempt > self.config.max_retries:
self.store.update_event_trigger(
event.event_id,
status="DEAD_LETTER",
channel=trigger_channel,
remote_id=None,
increment_attempt=True,
trigger_http_status=trigger_http_status,
trigger_error_category=trigger_error_category,
)
record = lifecycle_record(
contract_commit=self.config.contract_commit,
project_key=event.project_key,
kind="TASK_ENDED",
work_unit_id=event.work_unit_id,
payload={
"status": "FAIL",
"reason": reason,
"event_id": event.event_id,
"attempts": next_attempt,
},
)
self.store.enqueue_lifecycle(
sync_id=record["sync_id"],
project_key=event.project_key,
work_unit_id=event.work_unit_id,
kind=record["kind"],
payload=record,
)
return
self.store.update_event_trigger(
event.event_id,
status="TRIGGER_FAILED",
channel=trigger_channel,
remote_id=None,
retry_after=self._backoff(attempts),
increment_attempt=True,
trigger_http_status=trigger_http_status,
trigger_error_category=trigger_error_category,
)
def _backoff(self, attempts: int) -> int:
return min(3600, self.config.retry_base_seconds * (2 ** min(attempts, 10)))
class Handler(BaseHTTPRequestHandler):
runtime: Runtime
def log_message(self, fmt: str, *args: object) -> None:
LOG.info("%s %s", self.client_address[0], fmt % args)
def _reply(self, status: int, payload: dict[str, Any]) -> None:
body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def _body(self) -> dict[str, Any]:
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError as exc:
raise ValidationError("invalid_content_length") from exc
if length <= 0 or length > 1_048_576:
raise ValidationError("invalid_body_size")
try:
value = json.loads(self.rfile.read(length))
except json.JSONDecodeError as exc:
raise ValidationError("invalid_json") from exc
if not isinstance(value, dict):
raise ValidationError("json_root_not_object")
return value
def _authenticate(self, expected: str) -> None:
header = self.headers.get("Authorization", "")
if not header.startswith("Bearer "):
raise AuthenticationError("missing_bearer")
if not hmac.compare_digest(header[7:], expected):
raise AuthenticationError("invalid_bearer")
def do_GET(self) -> None:
path = urlparse(self.path).path.rstrip("/") or "/"
try:
if path == "/healthz":
self._reply(
200,
{
"service": "aios-orchestrator-gateway",
"status": "active",
"contract_commit": self.runtime.config.contract_commit,
"polling": True,
},
)
return
if path.startswith("/v1/orchestrator/state/"):
self._authenticate(self.runtime.config.inbound_token())
key = path.rsplit("/", 1)[-1]
state = self.runtime.store.project_summary(key)
self._reply(200 if state else 404, state or {"error": "project_not_found"})
return
if path.startswith("/v1/orchestrator/events/"):
self._authenticate(self.runtime.config.inbound_token())
event_id = path.rsplit("/", 1)[-1]
if not ID_RE.fullmatch(event_id):
raise ValidationError("invalid_event_id")
status = self.runtime.store.event_status(event_id)
self._reply(200 if status else 404, status or {"error": "event_not_found"})
return
self._reply(404, {"error": "not_found"})
except AuthenticationError as exc:
self._reply(401, {"error": str(exc)})
except OrchestratorError as exc:
self._reply(400, {"error": str(exc)})
def do_POST(self) -> None:
path = urlparse(self.path).path.rstrip("/")
try:
if path == "/v1/orchestrator/events":
self._authenticate(self.runtime.config.inbound_token())
created, event = self.runtime.submit_event(self._body())
self._reply(202 if created else 200, {"created": created, "event": event})
return
if path == "/v1/orchestrator/decisions":
self._authenticate(self.runtime.config.callback_token())
version = self.runtime.submit_decision(self._body())
self._reply(202, {"accepted": True, "state_version": version})
return
if path == "/v1/orchestrator/poll":
self._authenticate(self.runtime.config.inbound_token())
self.runtime.poll_once()
self._reply(200, {"polled": True})
return
self._reply(404, {"error": "not_found"})
except AuthenticationError as exc:
self._reply(401, {"error": str(exc)})
except ConflictError as exc:
self._reply(409, {"error": str(exc)})
except ValidationError as exc:
self._reply(422, {"error": str(exc)})
except OrchestratorError as exc:
self._reply(400, {"error": str(exc)})
except Exception:
LOG.exception("request_failed path=%s", path)
self._reply(500, {"error": "internal_error"})
def serve(config_path: Path) -> None:
config = Config.from_file(config_path)
runtime = Runtime(config)
Handler.runtime = runtime
server = ThreadingHTTPServer((config.host, config.port), Handler)
runtime.start()
LOG.info(
"gateway_started host=%s port=%s contract=%s polling=%ss",
config.host,
config.port,
config.contract_commit,
config.poll_interval_seconds,
)
try:
server.serve_forever(poll_interval=0.5)
finally:
server.server_close()
runtime.stop()
+921
View File
@@ -0,0 +1,921 @@
from __future__ import annotations
import json
from pathlib import Path
import sqlite3
import threading
import time
from typing import Any
from .errors import ConflictError
from .models import Decision, Event
from .securefs import ensure_secure_directory
TRIGGER_ERROR_CATEGORIES = frozenset(
{"UNAUTHORIZED", "FORBIDDEN", "NOT_FOUND", "NOT_RUNNABLE", "NETWORK_ERROR"}
)
class Store:
"""Durable state machine with idempotency, optimistic versioning, and leases."""
def __init__(self, path: Path):
self.path = Path(path)
ensure_secure_directory(self.path.parent)
self._local = threading.local()
self._initialize()
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.path, timeout=30, isolation_level=None)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=FULL")
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=30000")
return conn
def _conn(self) -> sqlite3.Connection:
conn = getattr(self._local, "conn", None)
if conn is None:
conn = self._connect()
self._local.conn = conn
return conn
def _initialize(self) -> None:
conn = self._connect()
try:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS project_state (
project_key TEXT PRIMARY KEY,
contract_commit TEXT NOT NULL,
state_version INTEGER NOT NULL,
state_json TEXT NOT NULL,
hops INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS gate_hops (
project_key TEXT NOT NULL,
gate_id TEXT NOT NULL,
hops INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL,
PRIMARY KEY(project_key, gate_id)
);
CREATE TABLE IF NOT EXISTS events (
event_id TEXT PRIMARY KEY,
project_key TEXT NOT NULL,
digest TEXT NOT NULL,
event_json TEXT NOT NULL,
status TEXT NOT NULL,
channel TEXT,
remote_id TEXT,
trigger_http_status INTEGER,
trigger_error_category TEXT,
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at INTEGER NOT NULL DEFAULT 0,
lease_owner TEXT,
lease_until INTEGER,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_poll
ON events(status, next_attempt_at, lease_until, created_at);
CREATE TABLE IF NOT EXISTS decisions (
event_id TEXT PRIMARY KEY REFERENCES events(event_id),
project_key TEXT NOT NULL,
decision_json TEXT NOT NULL,
state_version_after INTEGER NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
work_unit_id TEXT PRIMARY KEY,
event_id TEXT NOT NULL REFERENCES events(event_id),
assigned_agent TEXT NOT NULL,
payload_json TEXT NOT NULL,
status TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at INTEGER NOT NULL DEFAULT 0,
lease_owner TEXT,
lease_until INTEGER,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_outbox_poll
ON outbox(status, next_attempt_at, lease_until, created_at);
CREATE TABLE IF NOT EXISTS lifecycle (
sync_id TEXT PRIMARY KEY,
project_key TEXT NOT NULL,
work_unit_id TEXT,
kind TEXT NOT NULL,
payload_json TEXT NOT NULL,
status TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at INTEGER NOT NULL DEFAULT 0,
lease_owner TEXT,
lease_until INTEGER,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_lifecycle_poll
ON lifecycle(status, next_attempt_at, lease_until, created_at);
CREATE TABLE IF NOT EXISTS blocker_history (
project_key TEXT NOT NULL,
gate_id TEXT NOT NULL,
blocker TEXT NOT NULL,
repetitions INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY(project_key, gate_id, blocker)
);
CREATE TABLE IF NOT EXISTS summaries (
project_key TEXT PRIMARY KEY,
last_emitted_at INTEGER NOT NULL DEFAULT 0,
last_digest TEXT
);
CREATE TABLE IF NOT EXISTS observed_refs (
refname TEXT PRIMARY KEY,
object_id TEXT NOT NULL,
observed_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS metadata (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
"""
)
columns = {
str(row["name"])
for row in conn.execute("PRAGMA table_info(events)").fetchall()
}
if "trigger_http_status" not in columns:
conn.execute("ALTER TABLE events ADD COLUMN trigger_http_status INTEGER")
if "trigger_error_category" not in columns:
conn.execute("ALTER TABLE events ADD COLUMN trigger_error_category TEXT")
finally:
conn.close()
def register_event(self, event: Event) -> tuple[bool, dict[str, Any]]:
now = int(time.time())
conn = self._conn()
conn.execute("BEGIN IMMEDIATE")
try:
existing = conn.execute(
"SELECT digest, event_json FROM events WHERE event_id=?",
(event.event_id,),
).fetchone()
if existing:
if existing["digest"] != event.digest():
raise ConflictError("event_id_payload_conflict")
conn.execute("COMMIT")
return False, json.loads(existing["event_json"])
state = conn.execute(
"SELECT contract_commit, state_version FROM project_state WHERE project_key=?",
(event.project_key,),
).fetchone()
if state is None:
conn.execute(
"""
INSERT INTO project_state(
project_key, contract_commit, state_version, state_json,
hops, updated_at
) VALUES (?, ?, ?, '{}', 0, ?)
""",
(event.project_key, event.contract_commit, event.state_version, now),
)
else:
if state["contract_commit"] != event.contract_commit:
raise ConflictError("contract_version_mismatch")
if state["state_version"] != event.state_version:
raise ConflictError("stale_state_version")
conn.execute(
"""
INSERT INTO events(
event_id, project_key, digest, event_json, status,
next_attempt_at, created_at, updated_at
) VALUES (?, ?, ?, ?, 'RECEIVED', ?, ?, ?)
""",
(
event.event_id,
event.project_key,
event.digest(),
event.canonical_json(),
now,
now,
now,
),
)
conn.execute("COMMIT")
return True, event.to_dict()
except Exception:
conn.execute("ROLLBACK")
raise
def claim_events(
self,
*,
owner: str,
statuses: tuple[str, ...],
limit: int,
lease_seconds: int,
) -> list[dict[str, Any]]:
if not statuses:
return []
now = int(time.time())
placeholders = ",".join("?" for _ in statuses)
conn = self._conn()
conn.execute("BEGIN IMMEDIATE")
try:
rows = conn.execute(
f"""
SELECT event_id FROM events
WHERE status IN ({placeholders})
AND next_attempt_at <= ?
AND (lease_until IS NULL OR lease_until < ?)
ORDER BY
CASE json_extract(event_json, '$.event_type')
WHEN 'central_gpt_health_probe' THEN 0
WHEN 'a0_unattended_acceptance_requested' THEN 0
ELSE 1
END,
CASE WHEN json_extract(event_json, '$.event_type') IN (
'central_gpt_health_probe', 'a0_unattended_acceptance_requested'
) THEN created_at END DESC,
created_at ASC
LIMIT ?
""",
(*statuses, now, now, limit),
).fetchall()
ids = [row["event_id"] for row in rows]
for event_id in ids:
conn.execute(
"""
UPDATE events SET lease_owner=?, lease_until=?, updated_at=?
WHERE event_id=?
""",
(owner, now + lease_seconds, now, event_id),
)
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
if not ids:
return []
placeholders = ",".join("?" for _ in ids)
claimed = {
str(row["event_id"]): dict(row)
for row in conn.execute(
f"SELECT * FROM events WHERE event_id IN ({placeholders})",
ids,
).fetchall()
}
return [claimed[str(event_id)] for event_id in ids]
def update_event_trigger(
self,
event_id: str,
*,
status: str,
channel: str | None,
remote_id: str | None,
retry_after: int = 0,
increment_attempt: bool = False,
trigger_http_status: int | None = None,
trigger_error_category: str | None = None,
) -> None:
if trigger_http_status is not None and (
not isinstance(trigger_http_status, int)
or not 100 <= trigger_http_status <= 599
):
raise ValueError("invalid_trigger_http_status")
if (
trigger_error_category is not None
and trigger_error_category not in TRIGGER_ERROR_CATEGORIES
):
raise ValueError("invalid_trigger_error_category")
now = int(time.time())
self._conn().execute(
"""
UPDATE events
SET status=?, channel=?, remote_id=?,
trigger_http_status=?, trigger_error_category=?,
attempts=attempts + ?,
next_attempt_at=?,
lease_owner=NULL, lease_until=NULL, updated_at=?
WHERE event_id=?
""",
(
status,
channel,
remote_id,
trigger_http_status,
trigger_error_category,
1 if increment_attempt else 0,
now + max(0, retry_after),
now,
event_id,
),
)
def claim_workspace_timeouts(
self,
*,
owner: str,
older_than: int,
limit: int,
lease_seconds: int,
) -> list[dict[str, Any]]:
return self._claim_event_subset(
owner=owner,
where_sql="status='AWAITING_DECISION' AND channel='workspace_agent' AND updated_at <= ?",
where_params=(older_than,),
limit=limit,
lease_seconds=lease_seconds,
)
def claim_response_polls(
self,
*,
owner: str,
limit: int,
lease_seconds: int,
) -> list[dict[str, Any]]:
now = int(time.time())
return self._claim_event_subset(
owner=owner,
where_sql="status='POLLING_RESPONSE' AND next_attempt_at <= ?",
where_params=(now,),
limit=limit,
lease_seconds=lease_seconds,
)
def _claim_event_subset(
self,
*,
owner: str,
where_sql: str,
where_params: tuple[Any, ...],
limit: int,
lease_seconds: int,
) -> list[dict[str, Any]]:
allowed = {
"status='AWAITING_DECISION' AND channel='workspace_agent' AND updated_at <= ?",
"status='POLLING_RESPONSE' AND next_attempt_at <= ?",
}
if where_sql not in allowed:
raise ValueError("invalid_event_claim_query")
now = int(time.time())
conn = self._conn()
conn.execute("BEGIN IMMEDIATE")
try:
rows = conn.execute(
f"""
SELECT event_id FROM events
WHERE {where_sql}
AND (lease_until IS NULL OR lease_until < ?)
ORDER BY updated_at
LIMIT ?
""",
(*where_params, now, limit),
).fetchall()
ids = [str(row["event_id"]) for row in rows]
for event_id in ids:
conn.execute(
"""
UPDATE events SET lease_owner=?, lease_until=?
WHERE event_id=?
""",
(owner, now + lease_seconds, event_id),
)
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
if not ids:
return []
placeholders = ",".join("?" for _ in ids)
return [
dict(row)
for row in conn.execute(
f"SELECT * FROM events WHERE event_id IN ({placeholders})",
ids,
).fetchall()
]
def apply_decision(
self,
decision: Decision,
*,
max_hops_per_gate: int,
max_same_blocker_retries: int,
) -> int:
now = int(time.time())
conn = self._conn()
conn.execute("BEGIN IMMEDIATE")
try:
event = conn.execute(
"SELECT project_key FROM events WHERE event_id=?",
(decision.event_id,),
).fetchone()
if event is None:
raise ConflictError("event_not_found")
if event["project_key"] != decision.project_key:
raise ConflictError("event_project_mismatch")
existing = conn.execute(
"SELECT decision_json, state_version_after FROM decisions WHERE event_id=?",
(decision.event_id,),
).fetchone()
if existing:
if json.loads(existing["decision_json"]) != decision.to_dict():
raise ConflictError("decision_idempotency_conflict")
conn.execute("COMMIT")
return int(existing["state_version_after"])
state = conn.execute(
"SELECT * FROM project_state WHERE project_key=?",
(decision.project_key,),
).fetchone()
if state is None:
raise ConflictError("project_state_not_found")
if state["contract_commit"] != decision.contract_commit:
raise ConflictError("contract_version_mismatch")
if state["state_version"] != decision.expected_state_version:
raise ConflictError("stale_decision_state_version")
state_json = json.loads(state["state_json"])
gate = str(
decision.state_updates.get(
"current_gate", state_json.get("current_gate", "unknown")
)
)
gate_hops = conn.execute(
"SELECT hops FROM gate_hops WHERE project_key=? AND gate_id=?",
(decision.project_key, gate),
).fetchone()
if gate_hops is not None and int(gate_hops["hops"]) >= max_hops_per_gate:
raise ConflictError("max_hops_exceeded")
state_json.update(decision.state_updates)
next_version = int(state["state_version"]) + 1
if decision.stop_reason:
row = conn.execute(
"""
SELECT repetitions FROM blocker_history
WHERE project_key=? AND gate_id=? AND blocker=?
""",
(decision.project_key, gate, decision.stop_reason),
).fetchone()
repetitions = (int(row["repetitions"]) if row else 0) + 1
conn.execute(
"""
INSERT INTO blocker_history(
project_key, gate_id, blocker, repetitions, updated_at
) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(project_key, gate_id, blocker)
DO UPDATE SET repetitions=excluded.repetitions,
updated_at=excluded.updated_at
""",
(decision.project_key, gate, decision.stop_reason, repetitions, now),
)
if repetitions > max_same_blocker_retries:
raise ConflictError("same_blocker_retry_limit_exceeded")
conn.execute(
"""
INSERT INTO gate_hops(project_key, gate_id, hops, updated_at)
VALUES (?, ?, 1, ?)
ON CONFLICT(project_key, gate_id)
DO UPDATE SET hops=gate_hops.hops + 1, updated_at=excluded.updated_at
""",
(decision.project_key, gate, now),
)
conn.execute(
"""
UPDATE project_state
SET state_version=?, state_json=?, hops=hops+1, updated_at=?
WHERE project_key=?
""",
(
next_version,
json.dumps(state_json, separators=(",", ":"), sort_keys=True),
now,
decision.project_key,
),
)
conn.execute(
"""
INSERT INTO decisions(
event_id, project_key, decision_json,
state_version_after, created_at
) VALUES (?, ?, ?, ?, ?)
""",
(
decision.event_id,
decision.project_key,
json.dumps(decision.to_dict(), separators=(",", ":"), sort_keys=True),
next_version,
now,
),
)
for unit in decision.next_work_units:
conn.execute(
"""
INSERT INTO outbox(
work_unit_id, event_id, assigned_agent, payload_json,
status, next_attempt_at, created_at, updated_at
) VALUES (?, ?, ?, ?, 'PENDING', ?, ?, ?)
""",
(
unit.id,
decision.event_id,
unit.primary_agent,
json.dumps(unit.to_dict(), separators=(",", ":"), sort_keys=True),
now,
now,
now,
),
)
conn.execute(
"""
UPDATE events SET status='DECIDED', lease_owner=NULL,
lease_until=NULL, updated_at=?
WHERE event_id=?
""",
(now, decision.event_id),
)
conn.execute("COMMIT")
return next_version
except Exception:
conn.execute("ROLLBACK")
raise
def claim_outbox(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
rows = self._claim_table(
table="outbox",
key="work_unit_id",
owner=owner,
statuses=("PENDING", "RETRY"),
limit=limit,
lease_seconds=lease_seconds,
)
for row in rows:
context = self._conn().execute(
"""
SELECT e.project_key, d.state_version_after
FROM events e JOIN decisions d ON d.event_id=e.event_id
WHERE e.event_id=?
""",
(row["event_id"],),
).fetchone()
if context is not None:
row["project_key"] = str(context["project_key"])
row["state_version"] = int(context["state_version_after"])
return rows
def event_status(self, event_id: str) -> dict[str, Any] | None:
row = self._conn().execute(
"""
SELECT e.event_id, e.status, e.channel, e.remote_id,
e.trigger_http_status, e.trigger_error_category,
d.state_version_after
FROM events e LEFT JOIN decisions d ON d.event_id=e.event_id
WHERE e.event_id=?
""",
(event_id,),
).fetchone()
if row is None:
return None
raw = str(row["status"])
if raw == "DECIDED":
public_status = "DECIDED"
elif raw == "DEAD_LETTER":
public_status = "DEAD_LETTER"
elif raw == "TRIGGER_FAILED":
public_status = "TRIGGER_FAILED"
elif raw in {"AWAITING_DECISION", "POLLING_RESPONSE"}:
public_status = "AWAITING_DECISION"
else:
public_status = "RECEIVED"
channel = row["channel"]
if channel not in {"workspace_agent", "responses_api"}:
channel = None
version = row["state_version_after"]
return {
"event_id": str(row["event_id"]),
"status": public_status,
"trigger_channel": channel,
"trigger_http_status": (
None if row["trigger_http_status"] is None
else int(row["trigger_http_status"])
),
"trigger_error_category": row["trigger_error_category"],
"remote_id": row["remote_id"],
"decision_applied": raw == "DECIDED" and version is not None,
"state_version_after": None if version is None else int(version),
}
def mark_outbox(
self,
work_unit_id: str,
*,
status: str,
retry_after: int = 0,
increment_attempt: bool = False,
) -> None:
self._mark_table(
table="outbox",
key="work_unit_id",
value=work_unit_id,
status=status,
retry_after=retry_after,
increment_attempt=increment_attempt,
)
def enqueue_lifecycle(
self,
*,
sync_id: str,
project_key: str,
work_unit_id: str | None,
kind: str,
payload: dict[str, Any],
) -> None:
now = int(time.time())
encoded = json.dumps(payload, separators=(",", ":"), sort_keys=True)
self._conn().execute(
"""
INSERT OR IGNORE INTO lifecycle(
sync_id, project_key, work_unit_id, kind, payload_json,
status, next_attempt_at, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, 'PENDING', ?, ?, ?)
""",
(sync_id, project_key, work_unit_id, kind, encoded, now, now, now),
)
def claim_lifecycle(self, *, owner: str, limit: int, lease_seconds: int) -> list[dict[str, Any]]:
return self._claim_table(
table="lifecycle",
key="sync_id",
owner=owner,
statuses=("PENDING", "RETRY"),
limit=limit,
lease_seconds=lease_seconds,
)
def mark_lifecycle(
self,
sync_id: str,
*,
status: str,
retry_after: int = 0,
increment_attempt: bool = False,
) -> None:
self._mark_table(
table="lifecycle",
key="sync_id",
value=sync_id,
status=status,
retry_after=retry_after,
increment_attempt=increment_attempt,
)
def _claim_table(
self,
*,
table: str,
key: str,
owner: str,
statuses: tuple[str, ...],
limit: int,
lease_seconds: int,
) -> list[dict[str, Any]]:
if table not in {"outbox", "lifecycle"} or key not in {"work_unit_id", "sync_id"}:
raise ValueError("invalid_claim_table")
now = int(time.time())
placeholders = ",".join("?" for _ in statuses)
conn = self._conn()
conn.execute("BEGIN IMMEDIATE")
try:
rows = conn.execute(
f"""
SELECT {key} FROM {table}
WHERE status IN ({placeholders})
AND next_attempt_at <= ?
AND (lease_until IS NULL OR lease_until < ?)
ORDER BY created_at
LIMIT ?
""",
(*statuses, now, now, limit),
).fetchall()
values = [row[key] for row in rows]
for item in values:
conn.execute(
f"""
UPDATE {table} SET lease_owner=?, lease_until=?, updated_at=?
WHERE {key}=?
""",
(owner, now + lease_seconds, now, item),
)
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
if not values:
return []
placeholders = ",".join("?" for _ in values)
return [
dict(row)
for row in conn.execute(
f"SELECT * FROM {table} WHERE {key} IN ({placeholders})",
values,
).fetchall()
]
def _mark_table(
self,
*,
table: str,
key: str,
value: str,
status: str,
retry_after: int,
increment_attempt: bool,
) -> None:
if table not in {"outbox", "lifecycle"} or key not in {"work_unit_id", "sync_id"}:
raise ValueError("invalid_mark_table")
now = int(time.time())
self._conn().execute(
f"""
UPDATE {table}
SET status=?, attempts=attempts + ?, next_attempt_at=?,
lease_owner=NULL, lease_until=NULL, updated_at=?
WHERE {key}=?
""",
(
status,
1 if increment_attempt else 0,
now + max(0, retry_after),
now,
value,
),
)
def project_state(self, project_key: str) -> dict[str, Any] | None:
row = self._conn().execute(
"SELECT * FROM project_state WHERE project_key=?",
(project_key,),
).fetchone()
if row is None:
return None
result = dict(row)
result["state"] = json.loads(result.pop("state_json"))
return result
def all_project_states(self) -> list[dict[str, Any]]:
result = []
for row in self._conn().execute("SELECT * FROM project_state ORDER BY project_key").fetchall():
item = dict(row)
item["state"] = json.loads(item.pop("state_json"))
result.append(item)
return result
def summary_due(self, project_key: str, interval: int) -> bool:
now = int(time.time())
row = self._conn().execute(
"SELECT last_emitted_at FROM summaries WHERE project_key=?",
(project_key,),
).fetchone()
return row is None or now - int(row["last_emitted_at"]) >= interval
def mark_summary(self, project_key: str, digest: str) -> None:
now = int(time.time())
self._conn().execute(
"""
INSERT INTO summaries(project_key, last_emitted_at, last_digest)
VALUES (?, ?, ?)
ON CONFLICT(project_key)
DO UPDATE SET last_emitted_at=excluded.last_emitted_at,
last_digest=excluded.last_digest
""",
(project_key, now, digest),
)
def project_summary(self, project_key: str) -> dict[str, Any] | None:
state = self.project_state(project_key)
if state is None:
return None
conn = self._conn()
event_counts = {
row["status"]: int(row["count"])
for row in conn.execute(
"""
SELECT status, COUNT(*) AS count FROM events
WHERE project_key=? GROUP BY status
""",
(project_key,),
).fetchall()
}
outbox_counts = {
row["status"]: int(row["count"])
for row in conn.execute(
"""
SELECT o.status, COUNT(*) AS count
FROM outbox o JOIN events e ON e.event_id=o.event_id
WHERE e.project_key=? GROUP BY o.status
""",
(project_key,),
).fetchall()
}
return {
"project_key": project_key,
"contract_commit": state["contract_commit"],
"state_version": state["state_version"],
"state": state["state"],
"event_counts": event_counts,
"work_unit_counts": outbox_counts,
"updated_at": state["updated_at"],
}
def metadata_get(self, key: str) -> str | None:
row = self._conn().execute(
"SELECT value FROM metadata WHERE key=?",
(key,),
).fetchone()
return None if row is None else str(row["value"])
def metadata_set(self, key: str, value: str) -> None:
self._conn().execute(
"""
INSERT INTO metadata(key, value) VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value=excluded.value
""",
(key, value),
)
def observe_refs(self, refs: dict[str, str]) -> tuple[bool, list[tuple[str, str | None, str]]]:
"""Persist a Git snapshot and return changes after the initial baseline."""
now = int(time.time())
conn = self._conn()
conn.execute("BEGIN IMMEDIATE")
try:
initialized = conn.execute(
"SELECT value FROM metadata WHERE key='git_poll_initialized'"
).fetchone()
previous = {
row["refname"]: row["object_id"]
for row in conn.execute("SELECT refname, object_id FROM observed_refs").fetchall()
}
changes: list[tuple[str, str | None, str]] = []
if initialized is not None:
for refname, oid in refs.items():
old = previous.get(refname)
if old != oid:
changes.append((refname, old, oid))
for refname, old in previous.items():
if refname not in refs:
changes.append((refname, old, "0" * 40))
conn.execute("DELETE FROM observed_refs")
for refname, oid in refs.items():
conn.execute(
"""
INSERT INTO observed_refs(refname, object_id, observed_at)
VALUES (?, ?, ?)
""",
(refname, oid, now),
)
conn.execute(
"""
INSERT INTO metadata(key, value)
VALUES ('git_poll_initialized', '1')
ON CONFLICT(key) DO UPDATE SET value='1'
"""
)
conn.execute("COMMIT")
return initialized is not None, changes
except Exception:
conn.execute("ROLLBACK")
raise
def close(self) -> None:
conn = getattr(self._local, "conn", None)
if conn is not None:
conn.close()
self._local.conn = None
+108
View File
@@ -0,0 +1,108 @@
from __future__ import annotations
import hashlib
import json
import os
import stat
from pathlib import Path
import urllib.error
import urllib.request
from typing import Any
from .errors import DispatchError
from .securefs import atomic_write_bytes_shared, ensure_shared_directory, read_secure_bytes
class OpenClawSynchronizer:
"""Synchronize task starts, ends, dispatches, and project summaries.
A durable file is always emitted. An optional HTTP endpoint is best-effort;
failure leaves the lifecycle record pending for retry.
"""
def __init__(
self,
queue_dir: Path,
*,
endpoint: str | None = None,
bearer_token: str | None = None,
timeout_seconds: int = 10,
):
self.queue_dir = Path(queue_dir)
self.endpoint = endpoint
self.bearer_token = bearer_token
self.timeout_seconds = timeout_seconds
ensure_shared_directory(self.queue_dir)
def emit(self, record: dict[str, Any]) -> Path:
sync_id = record.get("sync_id")
if not isinstance(sync_id, str) or not sync_id:
raise DispatchError("invalid_sync_id")
payload = (json.dumps(record, indent=2, sort_keys=True) + "\n").encode()
path = self.queue_dir / "pending" / f"{sync_id}.json"
ensure_shared_directory(path.parent)
try:
st = os.lstat(path)
except FileNotFoundError:
atomic_write_bytes_shared(path, payload, mode=0o640)
else:
if not stat.S_ISREG(st.st_mode):
raise DispatchError("sync_target_not_regular")
if read_secure_bytes(path, expected_uid=None, max_bytes=1_048_576) != payload:
raise DispatchError("sync_idempotency_conflict")
if self.endpoint:
headers = {
"Content-Type": "application/json",
"Idempotency-Key": sync_id,
"User-Agent": "aios-orchestrator-gateway/0.2",
}
if self.bearer_token:
headers["Authorization"] = f"Bearer {self.bearer_token}"
request = urllib.request.Request(
self.endpoint,
data=payload,
method="POST",
headers=headers,
)
try:
with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
response.read(65_536)
if response.status not in {200, 201, 202, 204}:
raise DispatchError(f"openclaw_sync_http_{response.status}")
except urllib.error.HTTPError as exc:
raise DispatchError(f"openclaw_sync_http_{exc.code}") from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise DispatchError("openclaw_sync_failed") from exc
return path
def lifecycle_record(
*,
contract_commit: str,
project_key: str,
kind: str,
work_unit_id: str | None,
payload: dict[str, Any],
) -> dict[str, Any]:
canonical = json.dumps(
{
"contract_commit": contract_commit,
"project_key": project_key,
"kind": kind,
"work_unit_id": work_unit_id,
"payload": payload,
},
separators=(",", ":"),
sort_keys=True,
)
sync_id = hashlib.sha256(canonical.encode()).hexdigest()
return {
"schema_version": 1,
"sync_id": sync_id,
"contract_commit": contract_commit,
"project_key": project_key,
"kind": kind,
"work_unit_id": work_unit_id,
"payload": payload,
}
@@ -8,7 +8,11 @@ ConditionPathExists=/etc/aios-automation-quorum/controller.json
Type=oneshot
User=aios-quorum
Group=aios-quorum
SupplementaryGroups=aios-orchestrator gitops
ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/executor-quorum-controller --once --config /etc/aios-automation-quorum/controller.json
# Exit 2 means the controller successfully persisted a fail-closed BLOCKED
# decision. Runtime/configuration errors use exit 3 and remain unit failures.
SuccessExitStatus=2
UMask=0077
NoNewPrivileges=yes
PrivateDevices=yes
@@ -21,7 +25,7 @@ ProtectKernelLogs=yes
ProtectKernelModules=yes
ProtectKernelTunables=yes
ProtectSystem=strict
ReadOnlyPaths=/etc/aios-automation-quorum /run/aios-health
ReadOnlyPaths=/etc/aios-automation-quorum /run/aios-health /data/aios-gitops/repo.git
ReadWritePaths=/var/lib/aios-automation-quorum
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
RestrictNamespaces=yes
@@ -0,0 +1,38 @@
[Unit]
Description=AIOS central GPT live health probe
After=network-online.target aios-orchestrator-gateway.service
Wants=network-online.target
ConditionPathExists=/etc/aios-automation-quorum/health-probes.json
[Service]
Type=oneshot
User=root
Group=root
ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/central-gpt-health-probe --config /etc/aios-automation-quorum/health-probes.json
TimeoutStartSec=120
UMask=0077
NoNewPrivileges=yes
PrivateDevices=yes
PrivateTmp=yes
ProtectClock=yes
ProtectControlGroups=yes
ProtectHome=yes
ProtectHostname=yes
ProtectKernelLogs=yes
ProtectKernelModules=yes
ProtectKernelTunables=yes
ProtectSystem=strict
ReadOnlyPaths=/etc/aios-automation-quorum /etc/aios-orchestrator-gateway
ReadWritePaths=/run/aios-health
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
RestrictNamespaces=yes
RestrictRealtime=yes
RestrictSUIDSGID=yes
LockPersonality=yes
MemoryDenyWriteExecute=yes
SystemCallArchitectures=native
SystemCallFilter=@system-service
SystemCallErrorNumber=EPERM
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,12 @@
[Unit]
Description=Run AIOS central GPT health probe periodically
[Timer]
OnBootSec=2min
OnUnitActiveSec=1min
AccuracySec=5s
Persistent=true
Unit=aios-central-gpt-health-probe.service
[Install]
WantedBy=timers.target
@@ -0,0 +1,42 @@
[Unit]
Description=AIOS Codex watcher unattended lifecycle health probe
After=network-online.target aios-orchestrator-gateway.service aios-codex-watcher.timer
Wants=network-online.target
ConditionPathExists=/etc/aios-automation-quorum/health-probes.json
[Service]
Type=oneshot
User=root
Group=aios-orchestrator
ExecStart=/usr/bin/python3 /opt/aios-automation-quorum/current/bin/codex-executor-health-probe --config /etc/aios-automation-quorum/health-probes.json
TimeoutStartSec=120
# The probe creates commands as 0640. Preserve group-read access so the
# aios-orchestrator watcher can consume them without making the queue public.
UMask=0027
NoNewPrivileges=yes
PrivateDevices=yes
PrivateTmp=yes
ProtectClock=yes
ProtectControlGroups=yes
ProtectHome=yes
ProtectHostname=yes
ProtectKernelLogs=yes
ProtectKernelModules=yes
ProtectKernelTunables=yes
ProtectSystem=strict
ReadOnlyPaths=/etc/aios-automation-quorum /etc/aios-orchestrator-gateway /var/log/aios-codex-watcher
# SQLite read-only connections still need the dedicated WAL directory writable
# for shared-memory/locking state. The probe opens the database with mode=ro.
ReadWritePaths=/run/aios-health /data/aios-gitops/workspace/agents/commands/active /var/lib/aios-codex-watcher
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
RestrictNamespaces=yes
RestrictRealtime=yes
RestrictSUIDSGID=yes
LockPersonality=yes
MemoryDenyWriteExecute=yes
SystemCallArchitectures=native
SystemCallFilter=@system-service
SystemCallErrorNumber=EPERM
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,12 @@
[Unit]
Description=Run AIOS Codex executor health probe periodically
[Timer]
OnBootSec=3min
OnUnitActiveSec=1min
AccuracySec=5s
Persistent=true
Unit=aios-codex-executor-health-probe.service
[Install]
WantedBy=timers.target
+35
View File
@@ -0,0 +1,35 @@
[Unit]
Description=AIOS Codex automatic command watcher
After=network-online.target aios-orchestrator-gateway.service
Wants=network-online.target
[Service]
Type=oneshot
User=aios-orchestrator
Group=aios-orchestrator
UMask=0077
Environment=PYTHONPATH=/opt/aios-codex-watcher/current
ExecStart=/usr/bin/python3 -m services.aios_codex_watcher --config /etc/aios-codex-watcher/config.json --once
NoNewPrivileges=yes
PrivateDevices=yes
PrivateTmp=yes
ProtectClock=yes
ProtectControlGroups=yes
ProtectHome=yes
ProtectHostname=yes
ProtectKernelLogs=yes
ProtectKernelModules=yes
ProtectKernelTunables=yes
ProtectSystem=strict
ReadOnlyPaths=/etc/aios-orchestrator-gateway/inbound.token
ReadWritePaths=/data/aios-gitops/workspace/agents/commands /var/lib/aios-codex-watcher /var/log/aios-codex-watcher
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
RestrictNamespaces=yes
RestrictRealtime=yes
RestrictSUIDSGID=yes
SystemCallArchitectures=native
LockPersonality=yes
MemoryDenyWriteExecute=yes
[Install]
WantedBy=multi-user.target
+1
View File
@@ -0,0 +1 @@
u aios-orchestrator - "AIOS Orchestrator" /var/lib/aios-orchestrator-gateway
+12
View File
@@ -0,0 +1,12 @@
[Unit]
Description=Poll AIOS Codex command queue
[Timer]
OnBootSec=30s
OnUnitActiveSec=5s
AccuracySec=1s
Persistent=true
Unit=aios-codex-watcher.service
[Install]
WantedBy=timers.target
+2
View File
@@ -0,0 +1,2 @@
d /var/lib/aios-codex-watcher 0700 aios-orchestrator aios-orchestrator -
d /var/log/aios-codex-watcher 0700 aios-orchestrator aios-orchestrator -
@@ -0,0 +1,38 @@
[Unit]
Description=AIOS Orchestrator Gateway reliability poll
After=aios-orchestrator-gateway.service
ConditionPathExists=/etc/aios-orchestrator-gateway/config.json
[Service]
Type=oneshot
TimeoutStartSec=120
User=aios-orchestrator
Group=aios-orchestrator
SupplementaryGroups=gitops
WorkingDirectory=/opt/aios-orchestrator-gateway/current
ExecStart=/usr/bin/python3 -m services.aios_orchestrator_gateway --config /etc/aios-orchestrator-gateway/config.json --poll-once
UMask=0077
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectKernelLogs=true
ProtectControlGroups=true
ProtectHostname=true
ProtectProc=invisible
ProcSubset=pid
RestrictNamespaces=true
MemoryDenyWriteExecute=true
RestrictSUIDSGID=true
LockPersonality=true
CapabilityBoundingSet=
AmbientCapabilities=
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
StateDirectory=aios-orchestrator-gateway
ReadOnlyPaths=/etc/aios-orchestrator-gateway /opt/aios-orchestrator-gateway
ReadWritePaths=/var/lib/aios-orchestrator-gateway /data/aios-gitops/workspace/agents/commands/active
SystemCallFilter=@system-service
SystemCallErrorNumber=EPERM
@@ -0,0 +1,12 @@
[Unit]
Description=Periodically poll AIOS orchestration inboxes and model responses
[Timer]
OnBootSec=45s
OnUnitActiveSec=60s
RandomizedDelaySec=10s
Persistent=true
Unit=aios-orchestrator-gateway-poll.service
[Install]
WantedBy=timers.target
+52
View File
@@ -0,0 +1,52 @@
[Unit]
Description=AIOS Orchestrator Gateway
After=network-online.target
Wants=network-online.target
ConditionPathExists=/etc/aios-orchestrator-gateway/config.json
[Service]
Type=simple
User=aios-orchestrator
Group=aios-orchestrator
SupplementaryGroups=gitops
WorkingDirectory=/opt/aios-orchestrator-gateway/current
ExecStart=/usr/bin/python3 -m services.aios_orchestrator_gateway --config /etc/aios-orchestrator-gateway/config.json
Restart=on-failure
RestartSec=5s
TimeoutStartSec=30s
TimeoutStopSec=30s
UMask=0077
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectKernelLogs=true
ProtectControlGroups=true
ProtectClock=true
ProtectHostname=true
ProtectProc=invisible
ProcSubset=pid
RestrictNamespaces=true
MemoryDenyWriteExecute=true
RestrictSUIDSGID=true
LockPersonality=true
RestrictRealtime=true
CapabilityBoundingSet=
AmbientCapabilities=
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
SystemCallArchitectures=native
StateDirectory=aios-orchestrator-gateway
RuntimeDirectory=aios-orchestrator-gateway
ReadOnlyPaths=/etc/aios-orchestrator-gateway /opt/aios-orchestrator-gateway
ReadWritePaths=/var/lib/aios-orchestrator-gateway /data/aios-gitops/workspace/agents/commands/active
SystemCallFilter=@system-service
SystemCallErrorNumber=EPERM
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,2 @@
g aios-orchestrator -
u aios-orchestrator - "AIOS Orchestrator Gateway" /var/lib/aios-orchestrator-gateway /usr/sbin/nologin
@@ -0,0 +1,19 @@
d /etc/aios-orchestrator-gateway 0750 root aios-orchestrator -
d /var/lib/aios-orchestrator-gateway 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox/events 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox/events/pending 2770 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/inbox/events/processed 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox/events/rejected 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox/decisions 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox/decisions/pending 2770 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/inbox/decisions/processed 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/inbox/decisions/rejected 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/outbox 0700 aios-orchestrator aios-orchestrator -
d /var/lib/aios-orchestrator-gateway/openclaw-sync 2750 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/openclaw-sync/pending 2750 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/commands 2750 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/commands/openclaw 2750 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/commands/openclaw/pending 2750 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/commands/codex 2750 aios-orchestrator gitops -
d /var/lib/aios-orchestrator-gateway/commands/codex/pending 2750 aios-orchestrator gitops -
+244
View File
@@ -0,0 +1,244 @@
from __future__ import annotations
import json
import importlib.util
import os
from pathlib import Path
import subprocess
import unittest
import urllib.error
from unittest import mock
from services.aios_orchestrator_gateway.action_bridge import (
CallbackBridgeError,
resolve_secret_ref,
submit_orchestration_decision,
)
from tests.test_aios_orchestrator_gateway import CONTRACT, decision_payload
class DecisionActionBridgeTests(unittest.TestCase):
def test_18890_openapi_and_handler_expose_only_decision_callback_operation(self):
root = Path(__file__).resolve().parents[1]
openapi = (root / "workspace-agent/orchestrator-decision-action.openapi.yaml").read_text(
encoding="utf-8"
)
source = (root / "active-runtime/app/aios_chatgpt_actions_api.py").read_text(
encoding="utf-8"
)
self.assertIn("operationId: submit_orchestration_decision", openapi)
self.assertIn("if path == '/v1/orchestrator/decisions'", source)
self.assertIn("self.need('orchestrator:decision')", source)
self.assertIn("submit_orchestration_decision(", source)
def test_18890_handler_calls_dedicated_callback_bridge(self):
root = Path(__file__).resolve().parents[1]
module_path = root / "active-runtime/app/aios_chatgpt_actions_api.py"
spec = importlib.util.spec_from_file_location("actions_api_under_test", module_path)
module = importlib.util.module_from_spec(spec)
previous = os.environ.get("AIOS_GITOPS_WORKSPACE")
os.environ["AIOS_GITOPS_WORKSPACE"] = str(root)
try:
assert spec.loader is not None
spec.loader.exec_module(module)
finally:
if previous is None:
os.environ.pop("AIOS_GITOPS_WORKSPACE", None)
else:
os.environ["AIOS_GITOPS_WORKSPACE"] = previous
class FakeHandler:
def __init__(self):
self.sent = None
def need(self, scope):
self.scope = scope
return {"account": "central-gpt"}
def sendj(self, payload, status=200):
self.sent = (payload, status)
return self.sent
handler = FakeHandler()
with (
mock.patch.object(module, "ORCHESTRATOR_CONTRACT_COMMIT", CONTRACT),
mock.patch.object(
module,
"ORCHESTRATOR_SECRET_REF_RESOLVER",
Path("/usr/libexec/aios-secret-resolver"),
),
mock.patch.object(
module,
"submit_orchestration_decision",
return_value={"accepted": True, "state_version": 2},
) as submit,
):
module.H.orchestration_decision(handler, decision_payload())
self.assertEqual(handler.scope, "orchestrator:decision")
self.assertEqual(handler.sent, ({"accepted": True, "state_version": 2}, 202))
submit.assert_called_once()
def test_18890_handler_redacts_unexpected_bridge_exception(self):
root = Path(__file__).resolve().parents[1]
module_path = root / "active-runtime/app/aios_chatgpt_actions_api.py"
spec = importlib.util.spec_from_file_location("actions_api_error_test", module_path)
module = importlib.util.module_from_spec(spec)
previous = os.environ.get("AIOS_GITOPS_WORKSPACE")
os.environ["AIOS_GITOPS_WORKSPACE"] = str(root)
try:
assert spec.loader is not None
spec.loader.exec_module(module)
finally:
if previous is None:
os.environ.pop("AIOS_GITOPS_WORKSPACE", None)
else:
os.environ["AIOS_GITOPS_WORKSPACE"] = previous
class FakeHandler:
def need(self, _scope):
return {"account": "central-gpt"}
def sendj(self, payload, status=200):
self.sent = (payload, status)
return self.sent
handler = FakeHandler()
with (
mock.patch.object(module, "ORCHESTRATOR_CONTRACT_COMMIT", CONTRACT),
mock.patch.object(
module,
"ORCHESTRATOR_SECRET_REF_RESOLVER",
Path("/usr/libexec/aios-secret-resolver"),
),
mock.patch.object(
module,
"submit_orchestration_decision",
side_effect=RuntimeError("sensitive-token-value"),
),
):
module.H.orchestration_decision(handler, decision_payload())
self.assertEqual(handler.sent, ({"error": "CALLBACK_INTERNAL_ERROR"}, 500))
def _response(self, *, version=1):
response = mock.MagicMock(status=202)
response.read.return_value = json.dumps(
{"accepted": True, "state_version": version}
).encode()
response.__enter__.return_value = response
response.__exit__.return_value = False
return response
def test_strict_schema_rejects_unknown_field_before_secret_resolution(self):
payload = decision_payload()
payload["unexpected"] = True
with mock.patch(
"services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref"
) as resolver:
with self.assertRaisesRegex(CallbackBridgeError, "INVALID_DECISION"):
submit_orchestration_decision(
payload,
contract_commit=CONTRACT,
secret_ref="secret://callback/gateway",
resolver=Path("/usr/libexec/aios-secret-resolver"),
)
resolver.assert_not_called()
def test_contract_mismatch_rejected_before_secret_resolution(self):
with mock.patch(
"services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref"
) as resolver:
with self.assertRaisesRegex(CallbackBridgeError, "CONTRACT_MISMATCH"):
submit_orchestration_decision(
decision_payload(),
contract_commit="f" * 40,
secret_ref="secret://callback/gateway",
resolver=Path("/usr/libexec/aios-secret-resolver"),
)
resolver.assert_not_called()
def test_valid_decision_forwards_to_fixed_gateway_with_secret_ref_token(self):
opened = mock.Mock(return_value=self._response(version=4))
with mock.patch(
"services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref",
return_value="resolved-token-not-logged",
):
result = submit_orchestration_decision(
decision_payload(),
contract_commit=CONTRACT,
secret_ref="secret://callback/gateway",
resolver=Path("/usr/libexec/aios-secret-resolver"),
request_open=opened,
)
request = opened.call_args.args[0]
self.assertEqual(
request.full_url,
"http://127.0.0.1:18920/v1/orchestrator/decisions",
)
self.assertEqual(request.headers["Idempotency-key"], "evt-001")
self.assertEqual(result, {"accepted": True, "state_version": 4})
def test_stale_state_409_fails_closed_without_response_body(self):
error = urllib.error.HTTPError(
"http://127.0.0.1:18920/v1/orchestrator/decisions",
409,
"redacted",
{},
None,
)
with mock.patch(
"services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref",
return_value="resolved-token-not-logged",
):
with self.assertRaisesRegex(CallbackBridgeError, "STALE_STATE") as raised:
submit_orchestration_decision(
decision_payload(),
contract_commit=CONTRACT,
secret_ref="secret://callback/gateway",
resolver=Path("/usr/libexec/aios-secret-resolver"),
request_open=mock.Mock(side_effect=error),
)
self.assertEqual(raised.exception.status_code, 409)
def test_duplicate_callback_reuses_same_idempotency_key(self):
opened = mock.Mock(side_effect=[self._response(), self._response()])
with mock.patch(
"services.aios_orchestrator_gateway.action_bridge.resolve_secret_ref",
return_value="resolved-token-not-logged",
):
for _ in range(2):
submit_orchestration_decision(
decision_payload(),
contract_commit=CONTRACT,
secret_ref="secret://callback/gateway",
resolver=Path("/usr/libexec/aios-secret-resolver"),
request_open=opened,
)
keys = [call.args[0].headers["Idempotency-key"] for call in opened.call_args_list]
self.assertEqual(keys, ["evt-001", "evt-001"])
def test_secret_ref_is_stdin_not_argv_or_environment(self):
completed = subprocess.CompletedProcess(
["/usr/libexec/aios-secret-resolver"], 0, b"resolved-token\n", b""
)
with (
mock.patch(
"services.aios_orchestrator_gateway.action_bridge._trusted_resolver",
return_value=Path("/usr/libexec/aios-secret-resolver"),
),
mock.patch(
"services.aios_orchestrator_gateway.action_bridge.subprocess.run",
return_value=completed,
) as run,
):
token = resolve_secret_ref(
"secret://callback/gateway",
Path("/usr/libexec/aios-secret-resolver"),
)
self.assertEqual(token, "resolved-token")
self.assertEqual(run.call_args.args[0], ["/usr/libexec/aios-secret-resolver"])
self.assertIn(b"secret://callback/gateway", run.call_args.kwargs["input"])
self.assertNotIn("secret", " ".join(run.call_args.kwargs["env"].values()).lower())
if __name__ == "__main__":
unittest.main()
+196
View File
@@ -0,0 +1,196 @@
from __future__ import annotations
import json
import os
from pathlib import Path
import sqlite3
import tempfile
import unittest
from services.aios_codex_watcher.watcher import CommandWatcher, Config, WatcherError, validate_command
CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
class Clock:
def __init__(self) -> None:
self.value = 1000.0
def __call__(self) -> float:
return self.value
class WatcherTests(unittest.TestCase):
def setUp(self) -> None:
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
for name in ("active", "completed", "dead", "state", "log"):
(self.root / name).mkdir(mode=0o700)
self.token = self.root / "token"
self.token.write_text("not-a-real-secret\n", encoding="utf-8")
self.token.chmod(0o600)
self.clock = Clock()
self.events: list[dict] = []
self.exec_count = 0
self.config = Config(
active_dir=self.root / "active", completed_dir=self.root / "completed",
dead_letter_dir=self.root / "dead", database_path=self.root / "state/state.sqlite3",
audit_log_path=self.root / "log/audit.jsonl",
gateway_endpoint="http://127.0.0.1:18920/v1/orchestrator/events",
gateway_token_file=self.token, contract_commit=CONTRACT,
lease_seconds=10, retry_seconds=2, max_attempts=3, poll_seconds=1,
executor_path=None, executor_args=(), executor_timeout_seconds=10,
)
def tearDown(self) -> None:
self.temp.cleanup()
def command(self, command_id: str = "cmd-1", *, contract: str = CONTRACT, kind: str = "acceptance_probe", text: str = "no-op") -> dict:
return {
"schema_version": 1, "command_id": command_id,
"project_key": "a0-codex-watcher-nonproduction",
"work_unit_id": "A0-CODEX-WATCHER-ACCEPTANCE",
"contract_commit": contract, "state_version": 0,
"execution": {"kind": kind, "input": text},
}
def write(self, value: dict, name: str = "command.json") -> Path:
path = self.config.active_dir / name
path.write_text(json.dumps(value), encoding="utf-8")
path.chmod(0o600)
return path
def watcher(self, **kwargs) -> CommandWatcher:
def execute(command: dict) -> dict:
self.exec_count += 1
return {"kind": command["execution"]["kind"], "exit_code": 0}
return CommandWatcher(
self.config, clock=self.clock,
event_poster=kwargs.get("event_poster", self.events.append),
executor=kwargs.get("executor", execute), worker_id=kwargs.get("worker_id"),
)
def row(self, command_id: str = "cmd-1") -> sqlite3.Row:
db = sqlite3.connect(self.config.database_path)
db.row_factory = sqlite3.Row
try:
return db.execute("SELECT * FROM commands WHERE command_id=?", (command_id,)).fetchone()
finally:
db.close()
def test_happy_path_posts_lifecycle_once_and_archives(self) -> None:
path = self.write(self.command())
self.assertTrue(self.watcher().process_path(path))
self.assertEqual([e["event_type"] for e in self.events], ["work_unit_started", "work_unit_ended"])
self.assertEqual(self.exec_count, 1)
self.assertEqual(self.row()["status"], "COMPLETED")
self.assertTrue((self.config.completed_dir / path.name).is_file())
markers = [json.loads(line)["marker"] for line in self.config.audit_log_path.read_text().splitlines()]
self.assertEqual(markers, ["CODEX_AUTO_CLAIM", "TASK_STARTED", "TASK_EXECUTED", "TASK_ENDED", "EVENT_POSTED"])
def test_contract_mismatch_is_rejected_without_registration(self) -> None:
path = self.write(self.command(contract="0" * 40))
self.assertFalse(self.watcher().process_path(path))
self.assertTrue((self.config.dead_letter_dir / path.name).is_file())
db = sqlite3.connect(self.config.database_path)
self.assertEqual(db.execute("SELECT count(*) FROM commands").fetchone()[0], 0)
def test_duplicate_id_with_different_payload_is_dead_lettered(self) -> None:
first = self.write(self.command(), "first.json")
self.assertTrue(self.watcher().process_path(first))
second = self.write(self.command(text="different"), "second.json")
self.assertFalse(self.watcher().process_path(second))
self.assertEqual(self.row()["status"], "COMPLETED")
self.assertTrue((self.config.dead_letter_dir / second.name).is_file())
def test_same_work_unit_with_different_command_id_is_not_executed(self) -> None:
first = self.write(self.command(), "first.json")
self.assertTrue(self.watcher().process_path(first))
second_command = self.command(command_id="cmd-2")
second = self.write(second_command, "second.json")
self.assertFalse(self.watcher().process_path(second))
self.assertEqual(self.exec_count, 1)
self.assertTrue((self.config.dead_letter_dir / second.name).is_file())
def test_two_workers_cannot_hold_same_lease(self) -> None:
path = self.write(self.command())
one = self.watcher(worker_id="worker-one")
command, digest = one._read_command(path)
one._register(command, digest, path)
self.assertIsNotNone(one._claim("cmd-1"))
two = self.watcher(worker_id="worker-two")
self.assertIsNone(two._claim("cmd-1"))
def test_stale_lease_is_recovered(self) -> None:
path = self.write(self.command())
one = self.watcher(worker_id="worker-one")
command, digest = one._read_command(path)
one._register(command, digest, path)
one._claim("cmd-1")
self.clock.value += 11
self.assertIsNotNone(self.watcher(worker_id="worker-two")._claim("cmd-1"))
def test_end_post_retry_does_not_execute_twice(self) -> None:
calls = 0
attempted_end_events: list[dict] = []
def poster(event: dict) -> None:
nonlocal calls
calls += 1
if event["event_type"] == "work_unit_ended":
attempted_end_events.append(event)
if calls == 2:
raise WatcherError("temporary")
self.events.append(event)
path = self.write(self.command())
watcher = self.watcher(event_poster=poster)
self.assertFalse(watcher.process_path(path))
self.assertEqual(self.row()["phase"], "END")
self.assertEqual(self.exec_count, 1)
self.clock.value += 3
self.assertTrue(watcher.process_path(path))
self.assertEqual(self.exec_count, 1)
self.assertEqual(len(attempted_end_events), 2)
self.assertEqual(attempted_end_events[0], attempted_end_events[1])
self.assertEqual(self.row()["status"], "COMPLETED")
def test_failure_exhaustion_moves_to_dead_letter(self) -> None:
executions = 0
def fail(_command: dict) -> dict:
nonlocal executions
executions += 1
raise WatcherError("executor failure")
path = self.write(self.command())
watcher = self.watcher(executor=fail)
for _ in range(3):
self.assertFalse(watcher.process_path(path))
self.clock.value += 3
self.assertEqual(self.row()["status"], "DEAD_LETTER")
self.assertEqual(executions, 1)
self.assertTrue((self.config.dead_letter_dir / path.name).is_file())
def test_symlink_command_is_rejected_without_following(self) -> None:
outside = self.root / "outside.json"
outside.write_text(json.dumps(self.command()), encoding="utf-8")
link = self.config.active_dir / "link.json"
os.symlink(outside, link)
self.assertFalse(self.watcher().process_path(link))
self.assertTrue(outside.is_file())
self.assertTrue((self.config.dead_letter_dir / link.name).is_symlink())
def test_codex_mode_fails_closed_when_executor_absent(self) -> None:
path = self.write(self.command(kind="codex"))
watcher = CommandWatcher(self.config, clock=self.clock, event_poster=self.events.append)
self.assertFalse(watcher.process_path(path))
self.assertEqual(self.row()["phase"], "EXECUTE")
self.assertIn("codex_executor_unavailable", self.row()["last_error"])
def test_validation_rejects_unknown_fields(self) -> None:
command = self.command()
command["secret"] = "must-not-be-accepted"
with self.assertRaisesRegex(WatcherError, "invalid_command_fields"):
validate_command(command, CONTRACT)
if __name__ == "__main__":
unittest.main()
+187
View File
@@ -0,0 +1,187 @@
from __future__ import annotations
import importlib.util
import os
from pathlib import Path
import json
import sys
import tempfile
import unittest
from unittest import mock
MODULE = Path(__file__).resolve().parents[1] / "automation" / "health_probe_common.py"
SPEC = importlib.util.spec_from_file_location("health_probe_common", MODULE)
assert SPEC and SPEC.loader
common = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(common)
sys.modules["health_probe_common"] = common
CENTRAL_MODULE = Path(__file__).resolve().parents[1] / "automation" / "central_gpt_health_probe.py"
CENTRAL_SPEC = importlib.util.spec_from_file_location("central_gpt_health_probe", CENTRAL_MODULE)
assert CENTRAL_SPEC and CENTRAL_SPEC.loader
central = importlib.util.module_from_spec(CENTRAL_SPEC)
CENTRAL_SPEC.loader.exec_module(central)
CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc"
class HealthProbeSecurityTests(unittest.TestCase):
def test_existing_pending_event_is_rechecked_without_duplicate_post(self):
config = {
"gateway_event_url": "http://127.0.0.1:18920/v1/orchestrator/events",
"gateway_token_file": "/run/aios-orchestrator-gateway/token",
"project_key": "v2r1-personal-weixin",
"contract_commit": CONTRACT,
"central_timeout_seconds": 1,
"poll_interval_seconds": 0.01,
"central_receipt_path": "/run/aios-health/central-gpt.json",
}
event_id = "health-central-" + "a" * 32
with (
mock.patch.object(central, "load_config", return_value=config),
mock.patch.object(central, "read_token", return_value="not-a-real-token"),
mock.patch.object(central, "_load_pending", return_value=(event_id, 3)),
mock.patch.object(
central,
"request_json",
return_value=(200, {"status": "TRIGGER_FAILED"}),
) as request,
):
with self.assertRaisesRegex(common.ProbeError, "central_decision_not_proven"):
central.run(Path("/not/read"))
self.assertTrue(request.call_args_list)
self.assertTrue(all(call.kwargs.get("method", "GET") == "GET" for call in request.call_args_list))
def test_central_pending_marker_is_metadata_only_and_reusable(self):
with tempfile.TemporaryDirectory() as temp:
path = Path(temp) / "central-gpt.pending.json"
path.write_text(
json.dumps(
{
"contract_commit": CONTRACT,
"event_id": "health-central-" + "a" * 32,
"state_version": 7,
}
),
encoding="utf-8",
)
os.chmod(path, 0o640)
self.assertEqual(
central._load_pending(path, CONTRACT, owners={os.geteuid()}),
("health-central-" + "a" * 32, 7),
)
def test_central_pending_marker_wrong_contract_fails_closed(self):
with tempfile.TemporaryDirectory() as temp:
path = Path(temp) / "central-gpt.pending.json"
path.write_text(
json.dumps(
{
"contract_commit": "0" * 40,
"event_id": "health-central-" + "a" * 32,
"state_version": 0,
}
),
encoding="utf-8",
)
os.chmod(path, 0o640)
with self.assertRaises(common.ProbeError):
central._load_pending(path, CONTRACT, owners={os.geteuid()})
def test_central_pending_marker_symlink_fails_closed(self):
if not hasattr(os, "symlink"):
self.skipTest("symlink unavailable")
with tempfile.TemporaryDirectory() as temp:
target = Path(temp) / "target"
target.write_text("{}", encoding="utf-8")
link = Path(temp) / "central-gpt.pending.json"
try:
link.symlink_to(target)
except OSError:
self.skipTest("symlink privilege unavailable")
with self.assertRaises(common.ProbeError):
central._load_pending(link, CONTRACT, owners={os.geteuid()})
def test_probe_error_reason_is_metadata_only(self):
self.assertEqual(central._safe_reason(common.ProbeError("gateway_unavailable")), "gateway_unavailable")
self.assertEqual(central._safe_reason(ValueError("secret value leaked!")), "probe_error")
def test_availability_receipts_are_readable_only_by_quorum_group(self):
source = MODULE.read_text(encoding="utf-8")
self.assertIn("os.fchown(fd, 0, info.st_gid)", source)
self.assertIn("os.fchmod(fd, 0o640)", source)
self.assertNotIn("os.fchmod(fd, 0o644)", source)
def test_codex_probe_commands_are_group_readable_by_watcher(self):
unit = (
Path(__file__).resolve().parents[1]
/ "systemd"
/ "aios-codex-executor-health-probe.service"
).read_text(encoding="utf-8")
self.assertIn("User=root\n", unit)
self.assertIn("Group=aios-orchestrator\n", unit)
self.assertIn("UMask=0027\n", unit)
self.assertIn(
"ReadOnlyPaths=/etc/aios-automation-quorum "
"/etc/aios-orchestrator-gateway /var/log/aios-codex-watcher\n",
unit,
)
self.assertIn(
"ReadWritePaths=/run/aios-health "
"/data/aios-gitops/workspace/agents/commands/active "
"/var/lib/aios-codex-watcher\n",
unit,
)
self.assertNotIn("Group=root\n", unit)
def test_gateway_endpoint_is_exact_loopback(self):
self.assertEqual(
common.gateway_base("http://127.0.0.1:18920/v1/orchestrator/events"),
"http://127.0.0.1:18920/v1/orchestrator",
)
for url in (
"http://127.0.0.1:8765/v1/orchestrator/events",
"http://localhost:18920/v1/orchestrator/events",
"https://127.0.0.1:18920/v1/orchestrator/events",
"http://127.0.0.1:18920/v1/orchestrator/events?token=x",
):
with self.assertRaises(common.ProbeError):
common.gateway_base(url)
def test_token_newline_is_rejected(self):
with tempfile.TemporaryDirectory() as temp:
path = Path(temp) / "token"
path.write_bytes(b"not-a-secret-value\n")
os.chmod(path, 0o600)
with self.assertRaises(common.ProbeError):
common.read_token(path)
def test_token_symlink_is_rejected(self):
if not hasattr(os, "symlink"):
self.skipTest("symlink unavailable")
with tempfile.TemporaryDirectory() as temp:
target = Path(temp) / "target"
target.write_bytes(b"not-a-secret-value")
os.chmod(target, 0o600)
link = Path(temp) / "link"
try:
link.symlink_to(target)
except OSError:
self.skipTest("symlink privilege unavailable")
with self.assertRaises(common.ProbeError):
common.read_token(link)
def test_token_wide_mode_is_rejected(self):
if os.name != "posix":
self.skipTest("POSIX mode semantics required")
with tempfile.TemporaryDirectory() as temp:
path = Path(temp) / "token"
path.write_bytes(b"not-a-secret-value")
os.chmod(path, 0o644)
with self.assertRaises(common.ProbeError):
common.read_token(path)
if __name__ == "__main__":
unittest.main()
File diff suppressed because it is too large Load Diff
+59
View File
@@ -0,0 +1,59 @@
#!/usr/bin/env bash
set -Eeuo pipefail
umask 0077
SOURCE_ROOT="${1:?source root required}"
ROOT="$(mktemp -d /tmp/aios-orchestrator-rollback-test.XXXXXX)"
trap 'rm -rf "$ROOT"' EXIT
INSTALL_ROOT="$ROOT/opt"
STATE_ROOT="$ROOT/state"
mkdir -p "$INSTALL_ROOT/releases" "$STATE_ROOT" "$ROOT/bin"
chmod 0700 "$STATE_ROOT"
OLD="$(printf 'a%.0s' {1..64})"
NEW="$(printf 'b%.0s' {1..64})"
mkdir "$INSTALL_ROOT/releases/$OLD" "$INSTALL_ROOT/releases/$NEW"
printf '%s\n' "$OLD" > "$INSTALL_ROOT/releases/$OLD/SOURCE_DIGEST"
printf '%s\n' "$NEW" > "$INSTALL_ROOT/releases/$NEW/SOURCE_DIGEST"
ln -s "$INSTALL_ROOT/releases/$NEW" "$INSTALL_ROOT/current"
python3 - "$STATE_ROOT/install-receipt.json" "$INSTALL_ROOT/releases/$NEW" "$INSTALL_ROOT/releases/$OLD" <<'PY'
import json, sys
with open(sys.argv[1], "w", encoding="utf-8") as handle:
json.dump({
"schema_version": 1,
"source_digest": "b" * 64,
"current_release": sys.argv[2],
"previous_release": sys.argv[3],
"last_rollback": None,
}, handle)
PY
chmod 0600 "$STATE_ROOT/install-receipt.json"
cat > "$ROOT/bin/systemctl" <<'SH'
#!/bin/sh
printf '%s\n' "$*" >> "${SYSTEMCTL_LOG:?}"
exit 0
SH
chmod 0755 "$ROOT/bin/systemctl"
: > "$ROOT/systemctl.log"
SYSTEMCTL_LOG="$ROOT/systemctl.log" PATH="$ROOT/bin:/usr/bin:/bin" INSTALL_ROOT="$INSTALL_ROOT" STATE_ROOT="$STATE_ROOT" \
bash "$SOURCE_ROOT/scripts/rollback_aios_orchestrator_gateway.sh"
FIRST="$(readlink -f "$INSTALL_ROOT/current")"
SYSTEMCTL_LOG="$ROOT/systemctl.log" PATH="$ROOT/bin:/usr/bin:/bin" INSTALL_ROOT="$INSTALL_ROOT" STATE_ROOT="$STATE_ROOT" \
bash "$SOURCE_ROOT/scripts/rollback_aios_orchestrator_gateway.sh"
SECOND="$(readlink -f "$INSTALL_ROOT/current")"
[[ "$FIRST" == "$INSTALL_ROOT/releases/$OLD" ]]
[[ "$SECOND" == "$FIRST" ]]
grep -Fxq 'stop aios-orchestrator-gateway-poll.timer' "$ROOT/systemctl.log"
grep -Fxq 'stop aios-orchestrator-gateway-poll.service' "$ROOT/systemctl.log"
grep -Fxq 'start aios-orchestrator-gateway-poll.timer' "$ROOT/systemctl.log"
printf 'tampered\n' > "$INSTALL_ROOT/releases/$OLD/SOURCE_DIGEST"
if SYSTEMCTL_LOG="$ROOT/systemctl.log" PATH="$ROOT/bin:/usr/bin:/bin" \
INSTALL_ROOT="$INSTALL_ROOT" STATE_ROOT="$STATE_ROOT" \
bash "$SOURCE_ROOT/scripts/rollback_aios_orchestrator_gateway.sh" >/dev/null 2>&1; then
echo 'ROLLBACK_DIGEST_MISMATCH_TEST=FAIL' >&2
exit 1
fi
echo "ROLLBACK_DIGEST_MISMATCH_TEST=PASS"
echo "ROLLBACK_REPEAT_TEST=PASS"
+55
View File
@@ -9,6 +9,7 @@ import sys
import tempfile
import time
import unittest
from unittest import mock
from pathlib import Path
@@ -331,6 +332,24 @@ class RecoveryAndSafetyTests(unittest.TestCase):
def tearDown(self):
self.temporary.cleanup()
def test_controller_git_authority_is_read_only_in_systemd_sandbox(self):
unit = (
Path(__file__).resolve().parents[1]
/ "systemd"
/ "aios-automation-quorum-controller.service"
).read_text(encoding="utf-8")
self.assertIn("SupplementaryGroups=aios-orchestrator gitops\n", unit)
self.assertIn("SuccessExitStatus=2\n", unit)
self.assertIn(
"ReadOnlyPaths=/etc/aios-automation-quorum /run/aios-health "
"/data/aios-gitops/repo.git\n",
unit,
)
self.assertNotIn(
"ReadWritePaths=/var/lib/aios-automation-quorum /data/aios-gitops/repo.git",
unit,
)
def test_http_failure_spools_then_file_poll_recovers(self):
inbox = self.root / "inbox"
inbox.mkdir(mode=0o700)
@@ -345,6 +364,42 @@ class RecoveryAndSafetyTests(unittest.TestCase):
finally:
store.close()
def test_git_ref_resolution_scopes_safe_directory_to_validated_repo(self):
repo = self.root / "repo.git"
repo.mkdir(mode=0o700)
completed = subprocess.CompletedProcess([], 0, stdout="a" * 40 + "\n")
with mock.patch.object(gateway.subprocess, "run", return_value=completed) as run:
self.assertEqual("a" * 40, gateway.resolve_git_ref(repo, "refs/heads/main"))
self.assertEqual(
[
"git",
"-c",
f"safe.directory={repo}",
"-C",
str(repo),
"rev-parse",
"--verify",
"refs/heads/main",
],
run.call_args.args[0],
)
def test_git_ref_resolution_rejects_symlink_and_world_writable_repo(self):
target = self.root / "target.git"
target.mkdir(mode=0o700)
link = self.root / "link.git"
try:
link.symlink_to(target, target_is_directory=True)
except OSError:
link = None
if link is not None:
with self.assertRaises(gateway.ControllerError):
gateway.resolve_git_ref(link, "refs/heads/main")
if os.name == "posix":
target.chmod(0o707)
with self.assertRaises(gateway.ControllerError):
gateway.resolve_git_ref(target, "refs/heads/main")
@unittest.skipIf(os.name == "nt", "POSIX mode semantics required")
def test_symlink_and_writable_parent_are_fail_closed(self):
safe = self.root / "safe"
+454 -1
View File
@@ -1,11 +1,20 @@
from __future__ import annotations
import importlib.util
import io
import os
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
from types import SimpleNamespace
from unittest import TestCase
from unittest.mock import patch
from unittest.mock import Mock, call, patch
BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
@@ -80,6 +89,450 @@ class WechatSendBackendTests(TestCase):
self.assertEqual(args[1]["content"], "hello")
self.assertFalse(args[1]["dry_run"])
def test_ucc_auth_is_authorization_header_not_argv_or_output(self):
secret = "ucc-super-secret-token"
class Response:
status = 200
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self, _limit=-1):
return b'{"ok":true,"status":"sent"}'
urlopen = Mock(return_value=Response())
output = io.StringIO()
with (
patch.object(self.backend, "UCC_ADAPTER_TOKEN", secret),
patch.object(self.backend, "_open_no_redirect", urlopen),
redirect_stdout(output),
redirect_stderr(output),
):
status, payload = self.backend._post_ucc(
"/api/internal/wechat/send", {"target": "chat-001", "content": "hello"}
)
request = urlopen.call_args.args[0]
self.assertEqual(request.get_header("Authorization"), f"Bearer {secret}")
self.assertEqual(status, 200)
self.assertEqual(payload, {"ok": True, "status": "sent"})
self.assertNotIn(secret, " ".join(self.backend._openclaw_base_args()))
self.assertNotIn(secret, output.getvalue())
self.assertFalse({"--gateway", "--token", "--password"} & set(self.backend._openclaw_base_args()))
def test_ucc_paths_are_canonical_first_unique_and_bounded(self):
with patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"):
paths = self.backend._ucc_send_paths()
self.assertEqual(paths[0], "/api/internal/wechat/send")
self.assertLessEqual(len(paths), 2)
self.assertEqual(len(paths), len(set(paths)))
self.assertTrue(all(path.startswith("/") for path in paths))
def test_ucc_accepts_only_explicit_true_and_sent_when_not_dry_run(self):
self.assertTrue(self.backend._ucc_response_sent({"ok": True, "status": "sent"}))
for response in (
{"ok": 1, "status": "sent"},
{"ok": True, "status": "accepted"},
{"ok": False, "status": "sent"},
{"status": "sent"},
):
with self.subTest(response=response):
self.assertFalse(self.backend._ucc_response_sent(response))
def test_ambiguous_canonical_failures_never_fallback(self):
secret = "raw-upstream-secret"
failures = (
TimeoutError(secret),
ConnectionError(secret),
self.backend._BackendProtocolError(secret),
RuntimeError(secret),
)
for failure in failures:
with self.subTest(failure=type(failure).__name__):
with (
patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
patch.object(self.backend, "_post_ucc", side_effect=failure) as post,
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", message_id="msg-001"
).to_dict()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_unavailable")
self.assertEqual(post.call_count, 1)
self.assertNotIn(secret, repr(result))
def test_only_explicit_http_404_or_405_uses_legacy_fallback(self):
for http_status in (404, 405):
with self.subTest(http_status=http_status):
missing = urllib.error.HTTPError(
"http://127.0.0.1/api/internal/wechat/send",
http_status,
"not found",
{},
None,
)
with (
patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
patch.object(
self.backend,
"_post_ucc",
side_effect=[missing, (200, {"ok": True, "status": "sent"})],
) as post,
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", message_id="msg-001"
).to_dict()
self.assertTrue(result["ok"])
self.assertEqual(result["endpoint"], "/internal/channels/wechat/send")
self.assertEqual(post.call_count, 2)
def test_non_not_found_http_error_never_falls_back(self):
denied = urllib.error.HTTPError(
"http://127.0.0.1/api/internal/wechat/send", 401, "denied", {}, None
)
with patch.object(self.backend, "_post_ucc", side_effect=denied) as post:
result = self.backend.send_ucc_adapter(target="chat-001", content="hello").to_dict()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_http_error")
post.assert_called_once()
def test_ucc_rejection_does_not_expose_raw_error_or_secret(self):
secret = "provider-password=secret-value"
with patch.object(
self.backend,
"_post_ucc",
return_value=(200, {"ok": False, "status": "failed", "error": secret}),
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", message_id="msg-001"
).to_dict()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_rejected")
self.assertNotIn(secret, repr(result))
def test_dry_run_requires_dry_run_status_and_never_reports_sent(self):
with patch.object(
self.backend,
"_post_ucc",
return_value=(200, {"ok": True, "status": "dry_run"}),
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", dry_run=True
).to_dict()
self.assertTrue(result["ok"])
self.assertEqual(result["status"], "dry_run")
self.assertNotEqual(result["status"], "sent")
for dry_run, upstream_status in ((True, "sent"), (False, "dry_run")):
with self.subTest(dry_run=dry_run, upstream_status=upstream_status):
with patch.object(
self.backend,
"_post_ucc",
return_value=(200, {"ok": True, "status": upstream_status}),
):
rejected = self.backend.send_ucc_adapter(
target="chat-001", content="hello", dry_run=dry_run
).to_dict()
self.assertFalse(rejected["ok"])
self.assertEqual(rejected["status"], "failed")
def test_invalid_provider_message_id_never_reverses_confirmed_send(self):
invalid_values = (
"\ud800",
"contains\x00control",
"x" * (self.backend.MAX_IDENTIFIER_BYTES + 1),
{"not": "a string"},
)
for provider_message_id in invalid_values:
with self.subTest(value=repr(provider_message_id)[:40]):
with patch.object(
self.backend,
"_post_ucc",
return_value=(
200,
{
"ok": True,
"status": "sent",
"provider_message_id": provider_message_id,
},
),
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello"
).to_dict()
self.assertTrue(result["ok"])
self.assertEqual(result["status"], "sent")
self.assertNotIn("provider_message_id", result)
def test_openclaw_2026_6_11_commands_have_no_unsupported_global_gateway(self):
completed = SimpleNamespace(returncode=0, stdout='{"channels": []}')
with (
patch.object(self.backend, "OPENCLAW_BIN", "/usr/local/bin/openclaw"),
patch.object(self.backend, "_run_openclaw", return_value=completed) as run,
):
self.assertIsNone(self.backend.discover_openclaw_wechat_channel())
command = run.call_args.args[0]
self.assertEqual(command, ["/usr/local/bin/openclaw", "channels", "list", "--json"])
self.assertFalse({"--gateway", "--token", "--password"} & set(command))
def test_openclaw_output_overflow_terminates_while_process_is_running(self):
writer = (
"import os,time;"
f"os.write(1,b'x'*{self.backend.MAX_RESPONSE_BYTES + 4096});"
"time.sleep(10)"
)
started = time.monotonic()
completed = self.backend._run_openclaw([sys.executable, "-c", writer], timeout=4)
elapsed = time.monotonic() - started
self.assertLess(elapsed, 2.0)
self.assertNotEqual(completed.returncode, 0)
self.assertEqual(completed.stdout, "")
self.assertEqual(completed.stderr, "")
def test_openclaw_timeout_returns_failure_and_terminates_promptly(self):
started = time.monotonic()
completed = self.backend._run_openclaw(
[sys.executable, "-c", "import time;time.sleep(10)"], timeout=1
)
elapsed = time.monotonic() - started
self.assertLess(elapsed, 3.0)
self.assertNotEqual(completed.returncode, 0)
self.assertEqual(completed.stdout, "")
self.assertEqual(completed.stderr, "")
def test_timeout_kills_inherited_pipe_descendant_after_parent_exits(self):
if os.name != "posix" or not Path("/proc").is_dir():
self.skipTest("POSIX process-group and procfs contract")
def process_is_running(pid):
try:
state = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[2]
except (FileNotFoundError, IndexError, OSError):
return False
return state not in {"X", "Z"}
with tempfile.TemporaryDirectory() as directory:
pid_path = Path(directory) / "descendant.pid"
descendant = "import time;time.sleep(30)"
launcher = (
"import pathlib,subprocess,sys;"
f"child=subprocess.Popen([sys.executable,'-c',{descendant!r}]);"
f"pathlib.Path({str(pid_path)!r}).write_text(str(child.pid),encoding='utf-8')"
)
started = time.monotonic()
completed = self.backend._run_openclaw(
[sys.executable, "-c", launcher], timeout=1
)
elapsed = time.monotonic() - started
self.assertTrue(pid_path.is_file())
descendant_pid = int(pid_path.read_text(encoding="utf-8"))
try:
deadline = time.monotonic() + 2
while process_is_running(descendant_pid) and time.monotonic() < deadline:
time.sleep(0.02)
self.assertFalse(process_is_running(descendant_pid))
finally:
if process_is_running(descendant_pid):
os.kill(descendant_pid, signal.SIGKILL)
self.assertLess(elapsed, 3.0)
self.assertNotEqual(completed.returncode, 0)
def test_process_tree_termination_uses_posix_process_group(self):
if os.name != "posix":
self.skipTest("POSIX process-group contract")
class Process:
pid = 43210
def __init__(self):
self.wait_count = 0
def poll(self):
return None
def wait(self, timeout):
self.wait_count += 1
if self.wait_count == 1:
raise subprocess.TimeoutExpired(["openclaw"], timeout)
return -signal.SIGKILL
process = Process()
with patch.object(self.backend.os, "killpg") as killpg:
self.backend._terminate_process_tree(process)
self.assertEqual(
killpg.call_args_list,
[
call(process.pid, signal.SIGTERM),
call(process.pid, signal.SIGKILL),
],
)
def test_unsupported_and_generic_chat_backends_fail_closed(self):
with (
patch.object(self.backend, "SEND_BACKEND", "openclaw-chat"),
patch.object(self.backend, "ucc_adapter_alive", return_value=False),
):
selected = self.backend.select_backend()
result = self.backend.send_message(target="chat-001", content="hello")
self.assertEqual(selected["backend"], "unavailable")
self.assertEqual(selected["reason"], "unsupported_backend")
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "unsupported_backend")
def test_invalid_or_oversized_input_is_rejected_before_network(self):
with patch.object(self.backend, "_post_ucc") as post:
empty = self.backend.send_ucc_adapter(target="", content="hello").to_dict()
oversized = self.backend.send_ucc_adapter(
target="chat-001", content="x" * (self.backend.MAX_CONTENT_BYTES + 1)
).to_dict()
self.assertEqual(empty["code"], "invalid_input")
self.assertEqual(oversized["code"], "input_too_large")
post.assert_not_called()
def test_unpaired_surrogate_inputs_fail_closed_before_network(self):
cases = (
{"target": "\ud800", "content": "hello"},
{"target": "chat-001", "content": "\ud800"},
{"target": "chat-001", "content": "hello", "account_id": "\ud800"},
{"target": "chat-001", "content": "hello", "message_id": "\ud800"},
{"target": "chat-001", "content": "hello", "metadata": {"bad": "\ud800"}},
)
with patch.object(self.backend, "_post_ucc") as post:
for kwargs in cases:
with self.subTest(field=next(reversed(kwargs))):
result = self.backend.send_ucc_adapter(**kwargs).to_dict()
self.assertEqual(result["code"], "invalid_input")
post.assert_not_called()
def test_deeply_nested_metadata_is_bounded_before_network(self):
metadata = {}
for _ in range(1200):
metadata = {"nested": metadata}
with patch.object(self.backend, "_post_ucc") as post:
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", metadata=metadata
).to_dict()
self.assertEqual(result["code"], "input_too_large")
post.assert_not_called()
def test_adapter_origin_must_be_allowlisted(self):
with (
patch.object(self.backend, "UCC_ADAPTER_URL", "https://evil.example:8443"),
patch.object(
self.backend,
"UCC_ADAPTER_ALLOWED_ORIGINS",
("http://127.0.0.1:18800",),
),
):
self.assertIsNone(self.backend._adapter_base_url())
with (
patch.object(self.backend, "UCC_ADAPTER_URL", "https://adapter.example:8443"),
patch.object(
self.backend,
"UCC_ADAPTER_ALLOWED_ORIGINS",
("https://adapter.example:8443",),
),
):
self.assertEqual(
self.backend._adapter_base_url(), "https://adapter.example:8443"
)
def test_adapter_origin_rejects_scheme_userinfo_fragment_and_path(self):
invalid_origins = (
"ftp://127.0.0.1:18800",
"http://user:password@127.0.0.1:18800",
"http://127.0.0.1:18800#fragment",
"http://127.0.0.1:18800/base-path",
)
for origin in invalid_origins:
with self.subTest(origin=origin):
with (
patch.object(self.backend, "UCC_ADAPTER_URL", origin),
patch.object(
self.backend,
"UCC_ADAPTER_ALLOWED_ORIGINS",
("http://127.0.0.1:18800",),
),
):
self.assertIsNone(self.backend._adapter_base_url())
def test_adapter_path_rejects_decoded_traversal(self):
for path in (
"/api/%2e%2e/send",
"/api/%2E%2E/send",
"/api/%2525252e%2525252e/send",
"/api/%252525252e%252525252e/send",
"/api/../send",
"/api/%zz/send",
):
with self.subTest(path=path):
self.assertIsNone(self.backend._safe_path(path))
def test_adapter_path_keeps_intended_safe_canonical_paths(self):
for path in (
"/api/internal/wechat/send",
"/internal/channels/wechat/send",
"/api/%7Ehealth/status",
):
with self.subTest(path=path):
self.assertEqual(self.backend._safe_path(path), path)
def test_redirects_are_refused_and_authorization_cannot_cross_origin(self):
handler = self.backend._NoRedirectHandler()
self.assertIsNone(
handler.redirect_request(
None,
None,
302,
"redirect",
{"Location": "https://evil.example/steal"},
"https://evil.example/steal",
)
)
redirected = urllib.error.HTTPError(
"http://127.0.0.1:18800/api/internal/wechat/send",
302,
"redirect",
{"Location": "https://evil.example/steal"},
None,
)
with (
patch.object(self.backend, "UCC_ADAPTER_TOKEN", "header-secret"),
patch.object(self.backend, "_open_no_redirect", side_effect=redirected) as opened,
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello"
).to_dict()
request = opened.call_args.args[0]
self.assertEqual(urllib.parse.urlsplit(request.full_url).netloc, "127.0.0.1:18800")
self.assertEqual(request.get_header("Authorization"), "Bearer header-secret")
opened.assert_called_once()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_http_error")
if __name__ == "__main__":
import unittest
@@ -0,0 +1,31 @@
# AIOS Central Orchestrator Workspace Agent
Contract commit: `cbe33ecab8e9794ba5f318bea43aeaf447a56fbc`
When an API-triggered input arrives:
1. Parse the AIOS event JSON from the trigger message.
2. Verify `contract_commit` exactly matches the configured contract commit.
3. Use connected GitOps tools to read the authoritative Git tree, runtime status,
gate state, test evidence and full OIDs.
4. Never invent evidence, credentials, secrets, tests, approvals, Git results,
model availability or deployment status.
5. Apply the adaptive dispatch contract:
- Prefer OpenClaw when quality is comparable, work is large/repetitive,
monitoring-heavy, infrastructure-heavy or cost-sensitive.
- Prefer the strongest actually available Codex model for complex code,
security boundaries, architecture and final high-risk review.
- On timeout/unavailability, mark the attempt `TIMEOUT` or `NOT_EXECUTED`
and assign the same contract to the fallback agent.
- Never infer PASS from timeout or missing output.
6. Every next work unit must include a clean base commit, allowed paths, required
checks, forbidden operations, primary agent and fallback agent.
7. Keep real credential, raw-wire, reviewer, secret-provider and owner-only
release gates fail-closed.
8. Submit exactly one strict decision through the
`submit_orchestration_decision` action.
9. Do not write a second decision for the same event ID.
10. Stop after the action succeeds.
The decision must satisfy the JSON schema in the connected action. Use the
incoming `event_id`, `project_key`, `contract_commit`, and `state_version`.
@@ -0,0 +1,127 @@
openapi: 3.1.0
info:
title: AIOS Orchestrator Decision Callback
version: 1.0.0
servers:
- url: https://REPLACE_WITH_ACTIONS_HOST
paths:
/v1/orchestrator/decisions:
post:
operationId: submit_orchestration_decision
summary: Submit one strict orchestration decision for a triggered event
description: >-
The 18890 Actions API validates the decision strictly and forwards it
to the loopback Gateway with a callback credential resolved only from
an approved SecretRef. This operation cannot perform arbitrary GitOps
mutations.
security:
- bearerAuth: []
requestBody:
required: true
content:
application/json:
schema:
$ref: '#/components/schemas/Decision'
responses:
'202':
description: Accepted
'409':
description: Contract mismatch or stale state version
'422':
description: Schema validation failed
'401':
description: Missing or invalid Actions API bearer credential
'403':
description: Credential lacks orchestrator:decision scope
'502':
description: Metadata-only Gateway forwarding failure
'503':
description: SecretRef provider or callback configuration unavailable
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearer
schemas:
WorkUnit:
type: object
additionalProperties: false
required:
- id
- track_id
- gate_id
- primary_agent
- fallback_agent
- clean_base_commit
- allowed_paths
- required_checks
- forbidden_operations
- instruction
properties:
id: {type: string}
track_id: {type: string}
gate_id: {type: string}
primary_agent:
type: string
enum: [openclaw, codex]
fallback_agent:
type: string
enum: [openclaw, codex]
clean_base_commit:
type: string
pattern: '^[0-9a-f]{40}$'
allowed_paths:
type: array
items: {type: string}
required_checks:
type: array
items: {type: string}
forbidden_operations:
type: array
items: {type: string}
instruction: {type: string}
Decision:
type: object
additionalProperties: false
required:
- schema_version
- project_key
- contract_commit
- event_id
- expected_state_version
- verdict
- state_updates
- next_work_units
- requires_human_input
- stop_reason
properties:
schema_version:
type: integer
const: 1
project_key: {type: string}
contract_commit:
type: string
pattern: '^[0-9a-f]{40}$'
event_id: {type: string}
expected_state_version:
type: integer
minimum: 0
verdict:
type: string
enum: [DISPATCH, BLOCKED, COMPLETE, NOOP]
state_updates:
type: array
items:
type: object
additionalProperties: false
required: [key, value]
properties:
key: {type: string}
value: {type: string}
next_work_units:
type: array
items:
$ref: '#/components/schemas/WorkUnit'
requires_human_input: {type: boolean}
stop_reason:
type: [string, 'null']