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: self.workspace.trigger(event) except TriggerError: if self.responses is None: raise 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, ) else: self.store.update_event_trigger( event.event_id, status="AWAITING_DECISION", channel="workspace_agent", remote_id=None, ) elif self.responses is not None: response_id = self.responses.start(event) self.store.update_event_trigger( event.event_id, status="POLLING_RESPONSE", channel="responses_api", remote_id=response_id, retry_after=self.config.poll_interval_seconds, ) else: raise TriggerError("no_model_channel") except TriggerError as exc: self._retry_or_dead_letter(event, attempts, str(exc)) def process_workspace_timeouts(self) -> None: cutoff = int(time.time()) - self.config.callback_timeout_seconds for row in self.store.claim_workspace_timeouts( owner=self.worker_id, older_than=cutoff, limit=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) -> None: next_attempt = attempts + 1 if next_attempt > self.config.max_retries: self.store.update_event_trigger( event.event_id, status="DEAD_LETTER", channel=None, remote_id=None, increment_attempt=True, ) record = lifecycle_record( contract_commit=self.config.contract_commit, project_key=event.project_key, kind="TASK_ENDED", work_unit_id=event.work_unit_id, payload={ "status": "FAIL", "reason": reason, "event_id": event.event_id, "attempts": next_attempt, }, ) self.store.enqueue_lifecycle( sync_id=record["sync_id"], project_key=event.project_key, work_unit_id=event.work_unit_id, kind=record["kind"], payload=record, ) return self.store.update_event_trigger( event.event_id, status="TRIGGER_FAILED", channel=None, remote_id=None, retry_after=self._backoff(attempts), increment_attempt=True, ) def _backoff(self, attempts: int) -> int: return min(3600, self.config.retry_base_seconds * (2 ** min(attempts, 10))) class Handler(BaseHTTPRequestHandler): runtime: Runtime def log_message(self, fmt: str, *args: object) -> None: LOG.info("%s %s", self.client_address[0], fmt % args) def _reply(self, status: int, payload: dict[str, Any]) -> None: body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _body(self) -> dict[str, Any]: try: length = int(self.headers.get("Content-Length", "0")) except ValueError as exc: raise ValidationError("invalid_content_length") from exc if length <= 0 or length > 1_048_576: raise ValidationError("invalid_body_size") try: value = json.loads(self.rfile.read(length)) except json.JSONDecodeError as exc: raise ValidationError("invalid_json") from exc if not isinstance(value, dict): raise ValidationError("json_root_not_object") return value def _authenticate(self, expected: str) -> None: header = self.headers.get("Authorization", "") if not header.startswith("Bearer "): raise AuthenticationError("missing_bearer") if not hmac.compare_digest(header[7:], expected): raise AuthenticationError("invalid_bearer") def do_GET(self) -> None: path = urlparse(self.path).path.rstrip("/") or "/" try: if path == "/healthz": self._reply( 200, { "service": "aios-orchestrator-gateway", "status": "active", "contract_commit": self.runtime.config.contract_commit, "polling": True, }, ) return if path.startswith("/v1/orchestrator/state/"): self._authenticate(self.runtime.config.inbound_token()) key = path.rsplit("/", 1)[-1] state = self.runtime.store.project_summary(key) self._reply(200 if state else 404, state or {"error": "project_not_found"}) return 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()