941 lines
37 KiB
Python
Executable File
941 lines
37 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Durable GPT + one-executor quorum and failover controller.
|
|
|
|
The controller persists decisions before dispatch and never stores credentials.
|
|
Health is accepted only from metadata-only, live-probe receipts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import urllib.parse
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterable, Mapping
|
|
|
|
|
|
ALLOWED_LIFECYCLE_EVENTS = frozenset(
|
|
{
|
|
"TASK_DISPATCHED",
|
|
"TASK_STARTED",
|
|
"TASK_ENDED",
|
|
"DECISION_APPLIED",
|
|
"PROJECT_SUMMARY",
|
|
"PROJECT_COMPLETED",
|
|
}
|
|
)
|
|
EXECUTORS = ("openclaw", "codex")
|
|
SECRET_KEYS = frozenset(
|
|
{
|
|
"secret",
|
|
"token",
|
|
"password",
|
|
"api_key",
|
|
"authorization",
|
|
"credential",
|
|
"private_key",
|
|
"environment",
|
|
"env",
|
|
"argv",
|
|
"stdout",
|
|
"stderr",
|
|
}
|
|
)
|
|
|
|
|
|
class Health(str, Enum):
|
|
HEALTHY = "HEALTHY"
|
|
UNHEALTHY = "UNHEALTHY"
|
|
UNKNOWN = "UNKNOWN"
|
|
|
|
|
|
class ControllerError(RuntimeError):
|
|
"""Fail-closed controller error."""
|
|
|
|
|
|
class EventConflict(ControllerError):
|
|
"""An event id was reused with a different payload."""
|
|
|
|
|
|
class StaleStateVersion(ControllerError):
|
|
"""A dispatch attempted to overwrite a newer task state."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class QuorumDecision:
|
|
allowed: bool
|
|
priority: str
|
|
status: str
|
|
active_executor: str | None
|
|
standby_executor: str | None
|
|
repair_required: bool
|
|
|
|
|
|
def canonical_json(value: Any) -> str:
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
|
|
|
|
|
def digest_json(value: Any) -> str:
|
|
return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _contains_forbidden_key(value: Any) -> bool:
|
|
if isinstance(value, Mapping):
|
|
for key, child in value.items():
|
|
normalized = str(key).strip().lower().replace("-", "_")
|
|
if normalized in SECRET_KEYS or _contains_forbidden_key(child):
|
|
return True
|
|
elif isinstance(value, list):
|
|
return any(_contains_forbidden_key(item) for item in value)
|
|
return False
|
|
|
|
|
|
def validate_metadata_only(value: Mapping[str, Any]) -> None:
|
|
if _contains_forbidden_key(value):
|
|
raise ControllerError("secret-bearing or output-bearing fields are forbidden")
|
|
|
|
|
|
def atomic_json_write(path: Path, value: Mapping[str, Any], mode: int = 0o600) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
validate_secure_directory(path.parent)
|
|
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
|
|
try:
|
|
os.fchmod(fd, mode)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
handle.write(canonical_json(value))
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
|
|
def validate_secure_directory(path: Path, *, expected_uid: int | None = None) -> None:
|
|
info = path.lstat()
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
|
|
raise ControllerError(f"unsafe directory type rejected: {path}")
|
|
if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
|
|
raise ControllerError(f"symlinked directory component rejected: {path}")
|
|
if os.name == "posix" and info.st_mode & 0o022:
|
|
raise ControllerError(f"group/world-writable directory rejected: {path}")
|
|
if expected_uid is not None and info.st_uid != expected_uid:
|
|
raise ControllerError(f"unexpected owner for directory {path}")
|
|
|
|
|
|
def read_secure_json(path: Path, *, expected_uid: int | None = None) -> dict[str, Any]:
|
|
info = path.lstat()
|
|
if stat.S_ISLNK(info.st_mode):
|
|
raise ControllerError(f"symlink rejected: {path}")
|
|
if not stat.S_ISREG(info.st_mode):
|
|
raise ControllerError(f"not a regular file: {path}")
|
|
if os.name == "posix" and info.st_mode & 0o022:
|
|
raise ControllerError(f"group/world-writable file rejected: {path}")
|
|
validate_secure_directory(path.parent, expected_uid=expected_uid)
|
|
if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
|
|
raise ControllerError(f"symlinked file component rejected: {path}")
|
|
if expected_uid is not None and info.st_uid != expected_uid:
|
|
raise ControllerError(f"unexpected owner for {path}")
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
value = json.load(handle)
|
|
if not isinstance(value, dict):
|
|
raise ControllerError(f"JSON object required: {path}")
|
|
validate_metadata_only(value)
|
|
return value
|
|
|
|
|
|
def read_gateway_token(path: Path) -> str:
|
|
if not path.is_absolute():
|
|
raise ControllerError("gateway token path must be absolute")
|
|
try:
|
|
info = path.lstat()
|
|
except FileNotFoundError as exc:
|
|
raise ControllerError("gateway token unavailable") from exc
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
|
|
raise ControllerError("gateway token must be a regular non-symlink file")
|
|
if stat.S_IMODE(info.st_mode) & ~0o640:
|
|
raise ControllerError("gateway token mode exceeds 0640")
|
|
if os.name == "posix" and info.st_uid not in {0, os.geteuid()}:
|
|
raise ControllerError("gateway token owner is not approved")
|
|
if os.name == "posix" and path.resolve(strict=True) != Path(os.path.abspath(path)):
|
|
raise ControllerError("gateway token has a symlinked path component")
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
fd = os.open(path, flags)
|
|
try:
|
|
opened = os.fstat(fd)
|
|
if not stat.S_ISREG(opened.st_mode) or opened.st_ino != info.st_ino or opened.st_dev != info.st_dev:
|
|
raise ControllerError("gateway token changed during validation")
|
|
data = os.read(fd, 16_385)
|
|
if os.read(fd, 1):
|
|
raise ControllerError("gateway token too large")
|
|
finally:
|
|
os.close(fd)
|
|
if not data or len(data) > 16_384 or b"\n" in data or b"\r" in data:
|
|
raise ControllerError("gateway token format rejected")
|
|
try:
|
|
token = data.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise ControllerError("gateway token encoding rejected") from exc
|
|
if not token:
|
|
raise ControllerError("gateway token unavailable")
|
|
return token
|
|
|
|
|
|
def validate_health_receipt(
|
|
receipt: Mapping[str, Any],
|
|
subject: str,
|
|
*,
|
|
now: float | None = None,
|
|
max_age_seconds: int = 60,
|
|
) -> Health:
|
|
validate_metadata_only(receipt)
|
|
if receipt.get("subject") != subject:
|
|
return Health.UNKNOWN
|
|
try:
|
|
health = Health(str(receipt.get("health", "UNKNOWN")))
|
|
except ValueError:
|
|
return Health.UNKNOWN
|
|
if health is not Health.HEALTHY:
|
|
return health
|
|
if receipt.get("live_probe_succeeded") is not True:
|
|
return Health.UNKNOWN
|
|
observed_at = receipt.get("observed_at_epoch")
|
|
if not isinstance(observed_at, (int, float)):
|
|
return Health.UNKNOWN
|
|
now = time.time() if now is None else now
|
|
if observed_at > now + 5 or now - observed_at > max_age_seconds:
|
|
return Health.UNKNOWN
|
|
if subject == "central_gpt":
|
|
workspace = receipt.get("workspace_agent")
|
|
responses = receipt.get("responses_api")
|
|
workspace_ok = isinstance(workspace, Mapping) and workspace.get("validated_live") is True
|
|
responses_ok = isinstance(responses, Mapping) and responses.get("validated_live") is True
|
|
if not (workspace_ok or responses_ok):
|
|
return Health.UNKNOWN
|
|
return Health.HEALTHY
|
|
|
|
|
|
def choose_executor(workload: str) -> str:
|
|
if workload in {"complex_code", "security_boundary", "architecture", "complex_code_security_architecture"}:
|
|
return "codex"
|
|
return "openclaw"
|
|
|
|
|
|
def evaluate_quorum(
|
|
central_gpt: Health,
|
|
openclaw: Health,
|
|
codex: Health,
|
|
workload: str,
|
|
) -> QuorumDecision:
|
|
if central_gpt is not Health.HEALTHY:
|
|
status = "BLOCKED_CENTRAL_GPT_UNKNOWN" if central_gpt is Health.UNKNOWN else "BLOCKED_CENTRAL_GPT_UNHEALTHY"
|
|
return QuorumDecision(False, "P0", status, None, None, True)
|
|
|
|
healthy = [name for name, health in (("openclaw", openclaw), ("codex", codex)) if health is Health.HEALTHY]
|
|
if not healthy:
|
|
has_unknown = Health.UNKNOWN in {openclaw, codex}
|
|
status = "BLOCKED_EXECUTOR_HEALTH_UNKNOWN" if has_unknown else "BLOCKED_NO_HEALTHY_EXECUTOR"
|
|
return QuorumDecision(False, "P1", status, None, None, True)
|
|
|
|
if len(healthy) == 1:
|
|
active = healthy[0]
|
|
standby = "codex" if active == "openclaw" else "openclaw"
|
|
return QuorumDecision(True, "P3", "EXECUTOR_QUORUM_PASS", active, standby, False)
|
|
|
|
active = choose_executor(workload)
|
|
standby = "codex" if active == "openclaw" else "openclaw"
|
|
return QuorumDecision(True, "NONE", "EXECUTOR_QUORUM_PASS", active, standby, False)
|
|
|
|
|
|
class QuorumStore:
|
|
def __init__(self, path: Path):
|
|
self.path = path
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
expected_uid = os.geteuid() if os.name == "posix" else None
|
|
validate_secure_directory(path.parent, expected_uid=expected_uid)
|
|
if path.exists() or path.is_symlink():
|
|
info = path.lstat()
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
|
|
raise ControllerError(f"unsafe database path rejected: {path}")
|
|
if expected_uid is not None and info.st_uid != expected_uid:
|
|
raise ControllerError(f"unexpected database owner: {path}")
|
|
self.db = sqlite3.connect(str(path), isolation_level=None, timeout=30)
|
|
if os.name == "posix":
|
|
path.chmod(0o600)
|
|
self.db.row_factory = sqlite3.Row
|
|
self.db.execute("PRAGMA journal_mode=WAL")
|
|
self.db.execute("PRAGMA synchronous=FULL")
|
|
self._initialize()
|
|
|
|
def close(self) -> None:
|
|
self.db.close()
|
|
|
|
def _initialize(self) -> None:
|
|
self.db.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS events (
|
|
event_id TEXT PRIMARY KEY,
|
|
payload_digest TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
created_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
work_unit TEXT PRIMARY KEY,
|
|
contract_commit TEXT NOT NULL,
|
|
event_id TEXT NOT NULL,
|
|
state_version INTEGER NOT NULL,
|
|
workload TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
active_executor TEXT,
|
|
standby_executor TEXT,
|
|
checkpoint TEXT,
|
|
lease_owner TEXT,
|
|
lease_until REAL,
|
|
interrupted_executor TEXT,
|
|
recovery_priority TEXT NOT NULL,
|
|
updated_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS lifecycle (
|
|
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
work_unit TEXT NOT NULL,
|
|
event_type TEXT NOT NULL,
|
|
executor TEXT,
|
|
state_version INTEGER NOT NULL,
|
|
metadata_json TEXT NOT NULL,
|
|
created_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS effects (
|
|
effect_id TEXT PRIMARY KEY,
|
|
work_unit TEXT NOT NULL,
|
|
payload_digest TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
created_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS outbox (
|
|
outbox_id TEXT PRIMARY KEY,
|
|
destination TEXT NOT NULL,
|
|
payload_digest TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
next_attempt REAL NOT NULL,
|
|
created_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS blockers (
|
|
blocker_key TEXT PRIMARY KEY,
|
|
occurrences INTEGER NOT NULL,
|
|
status TEXT NOT NULL,
|
|
updated_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS observed_refs (
|
|
ref_name TEXT PRIMARY KEY,
|
|
oid TEXT NOT NULL,
|
|
updated_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS summaries (
|
|
work_unit TEXT PRIMARY KEY,
|
|
summary_digest TEXT NOT NULL,
|
|
summary_json TEXT NOT NULL,
|
|
created_at REAL NOT NULL
|
|
);
|
|
CREATE UNIQUE INDEX IF NOT EXISTS lifecycle_once
|
|
ON lifecycle(work_unit, event_type, COALESCE(executor, ''), state_version);
|
|
"""
|
|
)
|
|
|
|
def ingest_event(self, event_id: str, payload: Mapping[str, Any], source: str) -> bool:
|
|
validate_metadata_only(payload)
|
|
encoded = canonical_json(payload)
|
|
digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
existing = self.db.execute("SELECT payload_digest FROM events WHERE event_id=?", (event_id,)).fetchone()
|
|
if existing:
|
|
if existing["payload_digest"] != digest:
|
|
self.db.execute("UPDATE events SET status='DEAD_LETTER_CONFLICT' WHERE event_id=?", (event_id,))
|
|
raise EventConflict(f"event id payload conflict: {event_id}")
|
|
return False
|
|
self.db.execute(
|
|
"INSERT INTO events VALUES(?,?,?,?,?,?)",
|
|
(event_id, digest, encoded, source, "ACCEPTED", time.time()),
|
|
)
|
|
return True
|
|
|
|
def upsert_task(
|
|
self,
|
|
*,
|
|
work_unit: str,
|
|
contract_commit: str,
|
|
event_id: str,
|
|
state_version: int,
|
|
workload: str,
|
|
decision: QuorumDecision,
|
|
checkpoint: str | None = None,
|
|
) -> str:
|
|
now = time.time()
|
|
self.db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
current = self.db.execute("SELECT * FROM tasks WHERE work_unit=?", (work_unit,)).fetchone()
|
|
if current and state_version < current["state_version"]:
|
|
raise StaleStateVersion(f"stale state version for {work_unit}")
|
|
interrupted: str | None = None
|
|
recovery_priority = decision.priority
|
|
status = "BLOCKED"
|
|
if decision.allowed:
|
|
status = "READY"
|
|
if current and current["active_executor"] and current["active_executor"] != decision.active_executor:
|
|
interrupted = current["active_executor"]
|
|
recovery_priority = "P2"
|
|
status = "INTERRUPTED_AGENT_FAILOVER"
|
|
if checkpoint is None:
|
|
checkpoint = current["checkpoint"]
|
|
if current:
|
|
if contract_commit != current["contract_commit"] or event_id != current["event_id"]:
|
|
raise ControllerError("failover must preserve contract_commit and event_id")
|
|
if (
|
|
current["state_version"] == state_version
|
|
and current["active_executor"] == decision.active_executor
|
|
and current["status"] in {"RUNNING", "COMPLETED"}
|
|
):
|
|
self.db.execute("COMMIT")
|
|
return current["status"]
|
|
self.db.execute(
|
|
"""UPDATE tasks SET state_version=?, workload=?, status=?, active_executor=?,
|
|
standby_executor=?, checkpoint=?, interrupted_executor=?, recovery_priority=?, updated_at=? WHERE work_unit=?""",
|
|
(
|
|
state_version,
|
|
workload,
|
|
status,
|
|
decision.active_executor,
|
|
decision.standby_executor,
|
|
checkpoint,
|
|
interrupted,
|
|
recovery_priority,
|
|
now,
|
|
work_unit,
|
|
),
|
|
)
|
|
else:
|
|
self.db.execute(
|
|
"INSERT INTO tasks VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
work_unit,
|
|
contract_commit,
|
|
event_id,
|
|
state_version,
|
|
workload,
|
|
status,
|
|
decision.active_executor,
|
|
decision.standby_executor,
|
|
checkpoint,
|
|
None,
|
|
None,
|
|
interrupted,
|
|
recovery_priority,
|
|
now,
|
|
),
|
|
)
|
|
self.db.execute("COMMIT")
|
|
return status
|
|
except Exception:
|
|
self.db.execute("ROLLBACK")
|
|
raise
|
|
|
|
def claim_lease(self, work_unit: str, worker: str, ttl_seconds: int, now: float | None = None) -> bool:
|
|
now = time.time() if now is None else now
|
|
self.db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
task = self.db.execute(
|
|
"SELECT active_executor, status, lease_owner, lease_until FROM tasks WHERE work_unit=?",
|
|
(work_unit,),
|
|
).fetchone()
|
|
if not task:
|
|
raise ControllerError(f"unknown work unit: {work_unit}")
|
|
if worker != task["active_executor"]:
|
|
self.db.execute("ROLLBACK")
|
|
return False
|
|
if task["status"] in {"BLOCKED", "COMPLETED"}:
|
|
self.db.execute("ROLLBACK")
|
|
return False
|
|
lease_live = task["lease_owner"] and task["lease_until"] and task["lease_until"] > now
|
|
if lease_live and task["lease_owner"] != worker:
|
|
self.db.execute("ROLLBACK")
|
|
return False
|
|
self.db.execute(
|
|
"UPDATE tasks SET lease_owner=?, lease_until=?, status='RUNNING', updated_at=? WHERE work_unit=?",
|
|
(worker, now + ttl_seconds, now, work_unit),
|
|
)
|
|
self.db.execute("COMMIT")
|
|
return True
|
|
except Exception:
|
|
if self.db.in_transaction:
|
|
self.db.execute("ROLLBACK")
|
|
raise
|
|
|
|
def dispatch_task(
|
|
self,
|
|
*,
|
|
work_unit: str,
|
|
contract_commit: str,
|
|
event_id: str,
|
|
state_version: int,
|
|
workload: str,
|
|
decision: QuorumDecision,
|
|
checkpoint: str | None = None,
|
|
) -> str:
|
|
status = self.upsert_task(
|
|
work_unit=work_unit,
|
|
contract_commit=contract_commit,
|
|
event_id=event_id,
|
|
state_version=state_version,
|
|
workload=workload,
|
|
decision=decision,
|
|
checkpoint=checkpoint,
|
|
)
|
|
if decision.allowed and status != "COMPLETED":
|
|
metadata = {
|
|
"contract_commit": contract_commit,
|
|
"event_id": event_id,
|
|
"status": status,
|
|
}
|
|
self.append_lifecycle(work_unit, "TASK_DISPATCHED", decision.active_executor, state_version, metadata)
|
|
self.enqueue(
|
|
f"dispatch:{work_unit}:{state_version}:{decision.active_executor}",
|
|
"gateway",
|
|
{"event_type": "TASK_DISPATCHED", "work_unit": work_unit, **metadata},
|
|
)
|
|
return status
|
|
|
|
def start_task(self, work_unit: str, executor: str, state_version: int, lease_ttl_seconds: int) -> bool:
|
|
if not self.claim_lease(work_unit, executor, lease_ttl_seconds):
|
|
return False
|
|
self.append_lifecycle(work_unit, "TASK_STARTED", executor, state_version, {"status": "RUNNING"})
|
|
self.enqueue(
|
|
f"started:{work_unit}:{state_version}:{executor}",
|
|
"gateway",
|
|
{"event_type": "TASK_STARTED", "work_unit": work_unit, "executor": executor, "state_version": state_version},
|
|
)
|
|
return True
|
|
|
|
def end_task(
|
|
self,
|
|
work_unit: str,
|
|
executor: str,
|
|
state_version: int,
|
|
result: str,
|
|
summary: Mapping[str, Any],
|
|
*,
|
|
project_completed: bool = False,
|
|
) -> None:
|
|
validate_metadata_only(summary)
|
|
self.db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
task = self.db.execute("SELECT active_executor,state_version FROM tasks WHERE work_unit=?", (work_unit,)).fetchone()
|
|
if not task or task["active_executor"] != executor:
|
|
raise ControllerError("only the active executor may end a task")
|
|
if task["state_version"] != state_version:
|
|
raise StaleStateVersion(f"stale state version for {work_unit}")
|
|
self.db.execute(
|
|
"UPDATE tasks SET status='COMPLETED', lease_owner=NULL, lease_until=NULL, updated_at=? WHERE work_unit=?",
|
|
(time.time(), work_unit),
|
|
)
|
|
self.append_lifecycle(work_unit, "TASK_ENDED", executor, state_version, {"result": result})
|
|
self.append_lifecycle(work_unit, "DECISION_APPLIED", executor, state_version, {"result": result})
|
|
self.append_lifecycle(work_unit, "PROJECT_SUMMARY", executor, state_version, summary)
|
|
if project_completed:
|
|
self.append_lifecycle(work_unit, "PROJECT_COMPLETED", executor, state_version, {"result": result})
|
|
self.persist_summary(work_unit, summary)
|
|
outbound = (
|
|
("TASK_ENDED", {"result": result}),
|
|
("DECISION_APPLIED", {"result": result}),
|
|
("PROJECT_SUMMARY", dict(summary)),
|
|
)
|
|
for event_type, metadata in outbound:
|
|
self.enqueue(
|
|
f"{event_type.lower()}:{work_unit}:{state_version}:{executor}",
|
|
"gateway",
|
|
{
|
|
"event_type": event_type,
|
|
"work_unit": work_unit,
|
|
"executor": executor,
|
|
"state_version": state_version,
|
|
**metadata,
|
|
},
|
|
)
|
|
if project_completed:
|
|
self.enqueue(
|
|
f"project_completed:{work_unit}:{state_version}:{executor}",
|
|
"gateway",
|
|
{
|
|
"event_type": "PROJECT_COMPLETED",
|
|
"work_unit": work_unit,
|
|
"executor": executor,
|
|
"state_version": state_version,
|
|
"result": result,
|
|
},
|
|
)
|
|
self.db.execute("COMMIT")
|
|
except Exception:
|
|
if self.db.in_transaction:
|
|
self.db.execute("ROLLBACK")
|
|
raise
|
|
|
|
def record_effect(self, effect_id: str, work_unit: str, payload: Mapping[str, Any]) -> bool:
|
|
validate_metadata_only(payload)
|
|
digest = digest_json(payload)
|
|
existing = self.db.execute("SELECT payload_digest FROM effects WHERE effect_id=?", (effect_id,)).fetchone()
|
|
if existing:
|
|
if existing["payload_digest"] != digest:
|
|
raise EventConflict(f"effect id payload conflict: {effect_id}")
|
|
return False
|
|
self.db.execute(
|
|
"INSERT INTO effects VALUES(?,?,?,?,?)",
|
|
(effect_id, work_unit, digest, "COMPLETED", time.time()),
|
|
)
|
|
return True
|
|
|
|
def append_lifecycle(
|
|
self,
|
|
work_unit: str,
|
|
event_type: str,
|
|
executor: str | None,
|
|
state_version: int,
|
|
metadata: Mapping[str, Any],
|
|
) -> None:
|
|
if event_type not in ALLOWED_LIFECYCLE_EVENTS:
|
|
raise ControllerError(f"unsupported lifecycle event: {event_type}")
|
|
validate_metadata_only(metadata)
|
|
self.db.execute(
|
|
"INSERT OR IGNORE INTO lifecycle(work_unit,event_type,executor,state_version,metadata_json,created_at) VALUES(?,?,?,?,?,?)",
|
|
(work_unit, event_type, executor, state_version, canonical_json(metadata), time.time()),
|
|
)
|
|
|
|
def persist_summary(self, work_unit: str, summary: Mapping[str, Any]) -> None:
|
|
validate_metadata_only(summary)
|
|
encoded = canonical_json(summary)
|
|
digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
self.db.execute(
|
|
"INSERT OR REPLACE INTO summaries VALUES(?,?,?,?)",
|
|
(work_unit, digest, encoded, time.time()),
|
|
)
|
|
|
|
def enqueue(self, outbox_id: str, destination: str, payload: Mapping[str, Any]) -> bool:
|
|
validate_metadata_only(payload)
|
|
encoded = canonical_json(payload)
|
|
digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
existing = self.db.execute("SELECT payload_digest FROM outbox WHERE outbox_id=?", (outbox_id,)).fetchone()
|
|
if existing:
|
|
if existing["payload_digest"] != digest:
|
|
raise EventConflict(f"outbox id payload conflict: {outbox_id}")
|
|
return False
|
|
self.db.execute(
|
|
"INSERT INTO outbox VALUES(?,?,?,?,?,?,?,?)",
|
|
(outbox_id, destination, digest, encoded, "PENDING", 0, time.time(), time.time()),
|
|
)
|
|
return True
|
|
|
|
def record_blocker(self, blocker_key: str, threshold: int) -> str:
|
|
now = time.time()
|
|
row = self.db.execute("SELECT occurrences FROM blockers WHERE blocker_key=?", (blocker_key,)).fetchone()
|
|
occurrences = (row["occurrences"] if row else 0) + 1
|
|
status = "DEAD_LETTER" if occurrences > threshold else "RETRY"
|
|
self.db.execute(
|
|
"INSERT INTO blockers VALUES(?,?,?,?) ON CONFLICT(blocker_key) DO UPDATE SET occurrences=excluded.occurrences,status=excluded.status,updated_at=excluded.updated_at",
|
|
(blocker_key, occurrences, status, now),
|
|
)
|
|
return status
|
|
|
|
def deliver_outbox(self, sender: Callable[[Mapping[str, Any]], None], max_attempts: int, now: float | None = None) -> tuple[int, int]:
|
|
now = time.time() if now is None else now
|
|
sent = 0
|
|
failed = 0
|
|
rows = self.db.execute(
|
|
"SELECT * FROM outbox WHERE status='PENDING' AND next_attempt<=? ORDER BY created_at,outbox_id LIMIT 100",
|
|
(now,),
|
|
).fetchall()
|
|
for row in rows:
|
|
payload = json.loads(row["payload_json"])
|
|
validate_metadata_only(payload)
|
|
try:
|
|
sender(payload)
|
|
except Exception:
|
|
attempts = row["attempts"] + 1
|
|
status = "DEAD_LETTER" if attempts >= max_attempts else "PENDING"
|
|
delay = min(300, 2 ** min(attempts, 8))
|
|
self.db.execute(
|
|
"UPDATE outbox SET attempts=?,status=?,next_attempt=? WHERE outbox_id=?",
|
|
(attempts, status, now + delay, row["outbox_id"]),
|
|
)
|
|
failed += 1
|
|
else:
|
|
self.db.execute(
|
|
"UPDATE outbox SET attempts=attempts+1,status='SENT',next_attempt=? WHERE outbox_id=?",
|
|
(now, row["outbox_id"]),
|
|
)
|
|
sent += 1
|
|
return sent, failed
|
|
|
|
def observe_ref(self, ref_name: str, oid: str) -> bool:
|
|
row = self.db.execute("SELECT oid FROM observed_refs WHERE ref_name=?", (ref_name,)).fetchone()
|
|
changed = row is not None and row["oid"] != oid
|
|
self.db.execute(
|
|
"INSERT INTO observed_refs VALUES(?,?,?) ON CONFLICT(ref_name) DO UPDATE SET oid=excluded.oid,updated_at=excluded.updated_at",
|
|
(ref_name, oid, time.time()),
|
|
)
|
|
return changed
|
|
|
|
|
|
def spool_after_http_failure(
|
|
sender: Callable[[Mapping[str, Any]], None],
|
|
payload: Mapping[str, Any],
|
|
spool_dir: Path,
|
|
event_id: str,
|
|
) -> bool:
|
|
validate_metadata_only(payload)
|
|
try:
|
|
sender(payload)
|
|
return True
|
|
except Exception:
|
|
atomic_json_write(spool_dir / f"{event_id}.json", {"event_id": event_id, "payload": dict(payload)})
|
|
return False
|
|
|
|
|
|
def poll_file_inbox(store: QuorumStore, inbox_dir: Path) -> int:
|
|
if not inbox_dir.exists():
|
|
return 0
|
|
validate_secure_directory(inbox_dir, expected_uid=os.geteuid() if os.name == "posix" else None)
|
|
count = 0
|
|
for path in sorted(inbox_dir.glob("*.json")):
|
|
if path.is_symlink():
|
|
raise ControllerError(f"symlink inbox entry rejected: {path}")
|
|
item = read_secure_json(path, expected_uid=os.geteuid() if os.name == "posix" else None)
|
|
event_id = item.get("event_id")
|
|
payload = item.get("payload")
|
|
if not isinstance(event_id, str) or not isinstance(payload, dict):
|
|
raise ControllerError(f"invalid inbox envelope: {path}")
|
|
if store.ingest_event(event_id, payload, "file-inbox"):
|
|
count += 1
|
|
path.unlink()
|
|
return count
|
|
|
|
|
|
def resolve_git_ref(repo: Path, ref_name: str) -> str:
|
|
if not repo.is_absolute():
|
|
raise ControllerError("Git repository path must be absolute")
|
|
info = repo.lstat()
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
|
|
raise ControllerError("Git repository must be a non-symlink directory")
|
|
if os.name == "posix" and repo.resolve(strict=True) != Path(os.path.abspath(repo)):
|
|
raise ControllerError("Git repository has a symlinked path component")
|
|
if os.name == "posix" and info.st_mode & 0o002:
|
|
raise ControllerError("world-writable Git repository rejected")
|
|
try:
|
|
completed = subprocess.run(
|
|
[
|
|
"git",
|
|
"-c",
|
|
f"safe.directory={repo}",
|
|
"-C",
|
|
str(repo),
|
|
"rev-parse",
|
|
"--verify",
|
|
ref_name,
|
|
],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
|
|
raise ControllerError("Git ref unavailable") from exc
|
|
oid = completed.stdout.strip()
|
|
if len(oid) != 40 or any(character not in "0123456789abcdef" for character in oid.lower()):
|
|
raise ControllerError("invalid Git object id")
|
|
return oid
|
|
|
|
|
|
def poll_git_refs(
|
|
store: QuorumStore,
|
|
repo: Path,
|
|
refs: Iterable[str],
|
|
resolver: Callable[[Path, str], str] = resolve_git_ref,
|
|
) -> int:
|
|
changed = 0
|
|
for ref_name in refs:
|
|
if not ref_name.startswith("refs/") or ".." in ref_name or ref_name.endswith("/"):
|
|
raise ControllerError(f"invalid Git ref: {ref_name}")
|
|
oid = resolver(repo, ref_name)
|
|
if store.observe_ref(ref_name, oid):
|
|
event_id = f"gitref:{ref_name}:{oid}"
|
|
if store.ingest_event(event_id, {"ref": ref_name, "oid": oid}, "git-ref-poll"):
|
|
changed += 1
|
|
return changed
|
|
|
|
|
|
def responses_fallback_allowed(
|
|
workspace_agent_ok: bool,
|
|
receipt: Mapping[str, Any],
|
|
*,
|
|
now: float | None = None,
|
|
) -> bool:
|
|
validate_metadata_only(receipt)
|
|
if workspace_agent_ok:
|
|
return False
|
|
responses = receipt.get("responses_api")
|
|
return (
|
|
validate_health_receipt(receipt, "central_gpt", now=now) is Health.HEALTHY
|
|
and isinstance(responses, Mapping)
|
|
and responses.get("validated_live") is True
|
|
)
|
|
|
|
|
|
def _validate_gateway_url(url: str) -> None:
|
|
parsed = urllib.parse.urlparse(url)
|
|
if (
|
|
parsed.scheme != "http"
|
|
or parsed.hostname != "127.0.0.1"
|
|
or parsed.port != 18920
|
|
or parsed.path != "/v1/orchestrator/events"
|
|
or parsed.params
|
|
or parsed.query
|
|
or parsed.fragment
|
|
):
|
|
raise ControllerError("gateway endpoint must be the fixed loopback endpoint")
|
|
|
|
|
|
def get_gateway_state(url: str, project_key: str, token: str, timeout: float = 5.0) -> int:
|
|
_validate_gateway_url(url)
|
|
state_url = url.removesuffix("/events") + "/state/" + urllib.parse.quote(project_key, safe="")
|
|
request = urllib.request.Request(
|
|
state_url,
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
payload = json.loads(response.read(262_145))
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 404:
|
|
return 0
|
|
raise ControllerError(f"gateway state rejected HTTP {exc.code}") from exc
|
|
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
|
|
raise ControllerError("gateway state unavailable") from exc
|
|
version = payload.get("state_version") if isinstance(payload, Mapping) else None
|
|
if not isinstance(version, int) or version < 0:
|
|
raise ControllerError("gateway state version invalid")
|
|
return version
|
|
|
|
|
|
def post_json(url: str, payload: Mapping[str, Any], token: str, timeout: float = 5.0) -> None:
|
|
validate_metadata_only(payload)
|
|
_validate_gateway_url(url)
|
|
request = urllib.request.Request(
|
|
url,
|
|
data=canonical_json(payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
if response.status < 200 or response.status >= 300:
|
|
raise ControllerError(f"gateway rejected HTTP {response.status}")
|
|
except urllib.error.HTTPError as exc:
|
|
raise ControllerError(f"gateway rejected HTTP {exc.code}") from exc
|
|
except (urllib.error.URLError, TimeoutError) as exc:
|
|
raise ControllerError("gateway unavailable") from exc
|
|
|
|
|
|
def run_once(config_path: Path) -> int:
|
|
config = read_secure_json(config_path, expected_uid=0 if os.name == "posix" else None)
|
|
gateway_url = config["gateway_event_url"]
|
|
token = read_gateway_token(Path(config["gateway_token_file"]))
|
|
state_version = get_gateway_state(gateway_url, config["project_key"], token)
|
|
store = QuorumStore(Path(config["database_path"]))
|
|
try:
|
|
receipts: dict[str, Health] = {}
|
|
for subject in ("central_gpt", "openclaw", "codex"):
|
|
receipt_path = Path(config["health_receipts"][subject])
|
|
try:
|
|
receipt = read_secure_json(receipt_path, expected_uid=0 if os.name == "posix" else None)
|
|
receipts[subject] = validate_health_receipt(
|
|
receipt,
|
|
subject,
|
|
max_age_seconds=int(config.get("max_health_age_seconds", 60)),
|
|
)
|
|
except (FileNotFoundError, ControllerError, json.JSONDecodeError):
|
|
receipts[subject] = Health.UNKNOWN
|
|
decision = evaluate_quorum(
|
|
receipts["central_gpt"], receipts["openclaw"], receipts["codex"], config.get("default_workload", "large_repetitive_infrastructure_monitoring")
|
|
)
|
|
quorum_event_id = f"quorum-{time.time_ns()}"
|
|
store.enqueue(
|
|
quorum_event_id,
|
|
"gateway",
|
|
{
|
|
"schema_version": 1,
|
|
"project_key": config["project_key"],
|
|
"contract_commit": config["contract_commit"],
|
|
"event_id": quorum_event_id,
|
|
"source_agent": "system",
|
|
"event_type": "executor_quorum_evaluated",
|
|
"work_unit_id": "A0-AUTOMATION-QUORUM",
|
|
"status": "READY" if decision.allowed else "BLOCKED",
|
|
"state_version": state_version,
|
|
"implementation_commit": None,
|
|
"result_commit": None,
|
|
"blocker": None if decision.allowed else decision.status,
|
|
"evidence_refs": [],
|
|
"payload": {
|
|
"allowed": decision.allowed,
|
|
"priority": decision.priority,
|
|
"quorum_status": decision.status,
|
|
"active_executor": decision.active_executor,
|
|
"standby_executor": decision.standby_executor,
|
|
},
|
|
},
|
|
)
|
|
poll_file_inbox(store, Path(config["file_inbox_path"]))
|
|
git_polling = config.get("git_ref_polling")
|
|
if isinstance(git_polling, Mapping):
|
|
poll_git_refs(store, Path(git_polling["repository"]), git_polling["refs"])
|
|
store.deliver_outbox(
|
|
lambda payload: post_json(gateway_url, payload, token),
|
|
max_attempts=int(config["blocker_retry_threshold"]),
|
|
)
|
|
return 0 if decision.allowed else 2
|
|
finally:
|
|
store.close()
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--config", type=Path, required=True)
|
|
parser.add_argument("--once", action="store_true", required=True)
|
|
return parser
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
return run_once(args.config)
|
|
except (ControllerError, KeyError, OSError, sqlite3.Error, json.JSONDecodeError) as error:
|
|
print(f"automation-quorum-controller: fail-closed: {type(error).__name__}", file=os.sys.stderr)
|
|
return 3
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|