446 lines
20 KiB
Python
446 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "automation" / "executor_quorum_controller.py"
|
|
SPEC = importlib.util.spec_from_file_location("executor_quorum_controller", MODULE_PATH)
|
|
assert SPEC and SPEC.loader
|
|
gateway = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = gateway
|
|
SPEC.loader.exec_module(gateway)
|
|
|
|
|
|
class QuorumDecisionTests(unittest.TestCase):
|
|
def test_gpt_and_openclaw_two_unattended_cycles(self):
|
|
for _ in range(2):
|
|
decision = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY,
|
|
gateway.Health.HEALTHY,
|
|
gateway.Health.UNHEALTHY,
|
|
"large_repetitive_infrastructure_monitoring",
|
|
)
|
|
self.assertTrue(decision.allowed)
|
|
self.assertEqual("openclaw", decision.active_executor)
|
|
self.assertFalse(decision.repair_required)
|
|
|
|
def test_gpt_and_codex_two_unattended_cycles(self):
|
|
for _ in range(2):
|
|
decision = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY,
|
|
gateway.Health.UNHEALTHY,
|
|
gateway.Health.HEALTHY,
|
|
"complex_code_security_architecture",
|
|
)
|
|
self.assertTrue(decision.allowed)
|
|
self.assertEqual("codex", decision.active_executor)
|
|
|
|
def test_both_healthy_route_by_workload(self):
|
|
infrastructure = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.HEALTHY,
|
|
"large_repetitive_infrastructure_monitoring",
|
|
)
|
|
code = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.HEALTHY,
|
|
"complex_code_security_architecture",
|
|
)
|
|
self.assertEqual("openclaw", infrastructure.active_executor)
|
|
self.assertEqual("codex", code.active_executor)
|
|
|
|
def test_standby_failure_does_not_block_or_request_repair(self):
|
|
decision = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.UNHEALTHY, "architecture"
|
|
)
|
|
self.assertTrue(decision.allowed)
|
|
self.assertEqual("P3", decision.priority)
|
|
self.assertFalse(decision.repair_required)
|
|
|
|
def test_gpt_unhealthy_blocks_even_with_two_executors(self):
|
|
decision = gateway.evaluate_quorum(
|
|
gateway.Health.UNHEALTHY, gateway.Health.HEALTHY, gateway.Health.HEALTHY, "architecture"
|
|
)
|
|
self.assertFalse(decision.allowed)
|
|
self.assertEqual("P0", decision.priority)
|
|
|
|
def test_gpt_unknown_blocks(self):
|
|
decision = gateway.evaluate_quorum(
|
|
gateway.Health.UNKNOWN, gateway.Health.HEALTHY, gateway.Health.HEALTHY, "architecture"
|
|
)
|
|
self.assertFalse(decision.allowed)
|
|
self.assertIn("UNKNOWN", decision.status)
|
|
|
|
def test_both_executors_unhealthy_block(self):
|
|
decision = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.UNHEALTHY, gateway.Health.UNHEALTHY, "architecture"
|
|
)
|
|
self.assertFalse(decision.allowed)
|
|
self.assertEqual("BLOCKED_NO_HEALTHY_EXECUTOR", decision.status)
|
|
|
|
def test_unknown_executor_does_not_count_as_healthy(self):
|
|
blocked = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.UNKNOWN, gateway.Health.UNHEALTHY, "architecture"
|
|
)
|
|
allowed = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.UNKNOWN, gateway.Health.HEALTHY, "architecture"
|
|
)
|
|
self.assertFalse(blocked.allowed)
|
|
self.assertTrue(allowed.allowed)
|
|
self.assertEqual("codex", allowed.active_executor)
|
|
|
|
def test_health_receipt_requires_live_decision_channel(self):
|
|
base = {
|
|
"subject": "central_gpt",
|
|
"health": "HEALTHY",
|
|
"live_probe_succeeded": True,
|
|
"observed_at_epoch": 1,
|
|
}
|
|
self.assertEqual(gateway.Health.UNKNOWN, gateway.validate_health_receipt(base, "central_gpt", now=1))
|
|
base["workspace_agent"] = {"validated_live": True}
|
|
self.assertEqual(gateway.Health.HEALTHY, gateway.validate_health_receipt(base, "central_gpt", now=1))
|
|
|
|
def test_stale_health_receipt_is_unknown(self):
|
|
receipt = {
|
|
"subject": "codex", "health": "HEALTHY", "live_probe_succeeded": True,
|
|
"observed_at_epoch": 1,
|
|
}
|
|
self.assertEqual(
|
|
gateway.Health.UNKNOWN,
|
|
gateway.validate_health_receipt(receipt, "codex", now=100, max_age_seconds=60),
|
|
)
|
|
|
|
def test_workspace_fallback_requires_validated_responses(self):
|
|
receipt = {
|
|
"subject": "central_gpt", "health": "HEALTHY", "live_probe_succeeded": True,
|
|
"observed_at_epoch": 10, "responses_api": {"validated_live": True},
|
|
}
|
|
unvalidated = dict(receipt, responses_api={"configured": True})
|
|
self.assertFalse(gateway.responses_fallback_allowed(False, unvalidated, now=10))
|
|
self.assertTrue(gateway.responses_fallback_allowed(False, receipt, now=10))
|
|
self.assertFalse(gateway.responses_fallback_allowed(True, receipt, now=10))
|
|
|
|
|
|
class StoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
os.chmod(self.root, 0o700)
|
|
self.database = self.root / "controller.sqlite3"
|
|
self.store = gateway.QuorumStore(self.database)
|
|
self.decision_openclaw = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.HEALTHY, gateway.Health.UNHEALTHY,
|
|
"large_repetitive_infrastructure_monitoring",
|
|
)
|
|
self.decision_codex = gateway.evaluate_quorum(
|
|
gateway.Health.HEALTHY, gateway.Health.UNHEALTHY, gateway.Health.HEALTHY,
|
|
"complex_code_security_architecture",
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.store.close()
|
|
self.temporary.cleanup()
|
|
|
|
def create_task(self):
|
|
self.store.ingest_event("event-1", {"work_unit": "WU-1"}, "http")
|
|
return self.store.upsert_task(
|
|
work_unit="WU-1",
|
|
contract_commit="c" * 40,
|
|
event_id="event-1",
|
|
state_version=1,
|
|
workload="large_repetitive_infrastructure_monitoring",
|
|
decision=self.decision_openclaw,
|
|
checkpoint="checkpoint-7",
|
|
)
|
|
|
|
def test_same_event_id_same_payload_is_idempotent(self):
|
|
self.assertTrue(self.store.ingest_event("event", {"value": 1}, "http"))
|
|
self.assertFalse(self.store.ingest_event("event", {"value": 1}, "file-inbox"))
|
|
count = self.store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"]
|
|
self.assertEqual(1, count)
|
|
|
|
def test_same_dispatch_is_not_emitted_twice(self):
|
|
self.store.ingest_event("event-3", {"work_unit": "WU-3"}, "http")
|
|
kwargs = dict(
|
|
work_unit="WU-3", contract_commit="e" * 40, event_id="event-3", state_version=1,
|
|
workload="large_repetitive_infrastructure_monitoring", decision=self.decision_openclaw,
|
|
)
|
|
self.store.dispatch_task(**kwargs)
|
|
self.store.dispatch_task(**kwargs)
|
|
lifecycle = self.store.db.execute(
|
|
"SELECT COUNT(*) AS n FROM lifecycle WHERE work_unit='WU-3' AND event_type='TASK_DISPATCHED'"
|
|
).fetchone()["n"]
|
|
outbox = self.store.db.execute(
|
|
"SELECT COUNT(*) AS n FROM outbox WHERE outbox_id LIKE 'dispatch:WU-3:%'"
|
|
).fetchone()["n"]
|
|
self.assertEqual(1, lifecycle)
|
|
self.assertEqual(1, outbox)
|
|
|
|
def test_same_event_id_different_payload_dead_letters(self):
|
|
self.store.ingest_event("event", {"value": 1}, "http")
|
|
with self.assertRaises(gateway.EventConflict):
|
|
self.store.ingest_event("event", {"value": 2}, "file-inbox")
|
|
status = self.store.db.execute("SELECT status FROM events WHERE event_id='event'").fetchone()["status"]
|
|
self.assertEqual("DEAD_LETTER_CONFLICT", status)
|
|
|
|
def test_stale_state_version_rejected(self):
|
|
self.create_task()
|
|
with self.assertRaises(gateway.StaleStateVersion):
|
|
self.store.upsert_task(
|
|
work_unit="WU-1", contract_commit="c" * 40, event_id="event-1", state_version=0,
|
|
workload="architecture", decision=self.decision_codex,
|
|
)
|
|
|
|
def test_two_workers_compete_for_lease(self):
|
|
self.create_task()
|
|
self.assertTrue(self.store.claim_lease("WU-1", "openclaw", 30, now=100))
|
|
self.assertFalse(self.store.claim_lease("WU-1", "codex", 30, now=101))
|
|
|
|
def test_stale_lease_is_recovered(self):
|
|
self.create_task()
|
|
self.assertTrue(self.store.claim_lease("WU-1", "openclaw", 5, now=100))
|
|
self.store.upsert_task(
|
|
work_unit="WU-1", contract_commit="c" * 40, event_id="event-1", state_version=1,
|
|
workload="complex_code_security_architecture", decision=self.decision_codex,
|
|
)
|
|
self.assertTrue(self.store.claim_lease("WU-1", "codex", 5, now=106))
|
|
|
|
def test_active_failover_preserves_identity_checkpoint_and_effect(self):
|
|
self.create_task()
|
|
self.assertTrue(self.store.record_effect("effect-1", "WU-1", {"kind": "write_ref", "oid": "a" * 40}))
|
|
status = self.store.upsert_task(
|
|
work_unit="WU-1", contract_commit="c" * 40, event_id="event-1", state_version=1,
|
|
workload="complex_code_security_architecture", decision=self.decision_codex,
|
|
)
|
|
task = self.store.db.execute("SELECT * FROM tasks WHERE work_unit='WU-1'").fetchone()
|
|
self.assertEqual("INTERRUPTED_AGENT_FAILOVER", status)
|
|
self.assertEqual("openclaw", task["interrupted_executor"])
|
|
self.assertEqual("codex", task["active_executor"])
|
|
self.assertEqual("P2", task["recovery_priority"])
|
|
self.assertEqual("event-1", task["event_id"])
|
|
self.assertEqual("c" * 40, task["contract_commit"])
|
|
self.assertEqual("checkpoint-7", task["checkpoint"])
|
|
self.assertFalse(self.store.record_effect("effect-1", "WU-1", {"kind": "write_ref", "oid": "a" * 40}))
|
|
|
|
def test_lifecycle_and_summary_persist_without_openclaw(self):
|
|
self.create_task()
|
|
for event_type in (
|
|
"TASK_DISPATCHED", "TASK_STARTED", "TASK_ENDED", "DECISION_APPLIED",
|
|
"PROJECT_SUMMARY", "PROJECT_COMPLETED",
|
|
):
|
|
self.store.append_lifecycle("WU-1", event_type, "codex", 1, {"result": "PASS"})
|
|
self.store.persist_summary("WU-1", {"executor": "codex", "result": "PASS"})
|
|
events = self.store.db.execute("SELECT COUNT(*) AS n FROM lifecycle").fetchone()["n"]
|
|
summaries = self.store.db.execute("SELECT COUNT(*) AS n FROM summaries").fetchone()["n"]
|
|
self.assertEqual(6, events)
|
|
self.assertEqual(1, summaries)
|
|
|
|
def test_lifecycle_and_summary_persist_without_codex(self):
|
|
self.create_task()
|
|
self.store.append_lifecycle("WU-1", "TASK_STARTED", "openclaw", 1, {"result": "RUNNING"})
|
|
self.store.append_lifecycle("WU-1", "TASK_ENDED", "openclaw", 1, {"result": "PASS"})
|
|
self.store.persist_summary("WU-1", {"executor": "openclaw", "result": "PASS"})
|
|
self.assertEqual(
|
|
"openclaw",
|
|
json.loads(self.store.db.execute("SELECT summary_json FROM summaries").fetchone()["summary_json"])["executor"],
|
|
)
|
|
|
|
def test_restart_recovers_event_lease_and_outbox(self):
|
|
self.create_task()
|
|
self.store.claim_lease("WU-1", "openclaw", 5, now=100)
|
|
self.store.enqueue("outbox-1", "gateway", {"event_type": "TASK_STARTED"})
|
|
self.store.close()
|
|
self.store = gateway.QuorumStore(self.database)
|
|
self.assertEqual(1, self.store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"])
|
|
self.assertEqual(1, self.store.db.execute("SELECT COUNT(*) AS n FROM outbox").fetchone()["n"])
|
|
self.assertTrue(self.store.claim_lease("WU-1", "openclaw", 5, now=106))
|
|
|
|
def test_lifecycle_helpers_enforce_active_executor_and_complete(self):
|
|
self.store.ingest_event("event-2", {"work_unit": "WU-2"}, "http")
|
|
self.store.dispatch_task(
|
|
work_unit="WU-2", contract_commit="d" * 40, event_id="event-2", state_version=3,
|
|
workload="complex_code_security_architecture", decision=self.decision_codex,
|
|
)
|
|
self.assertFalse(self.store.start_task("WU-2", "openclaw", 3, 30))
|
|
self.assertTrue(self.store.start_task("WU-2", "codex", 3, 30))
|
|
self.store.end_task(
|
|
"WU-2", "codex", 3, "PASS", {"executor": "codex", "result": "PASS"}, project_completed=True,
|
|
)
|
|
task = self.store.db.execute("SELECT status,lease_owner FROM tasks WHERE work_unit='WU-2'").fetchone()
|
|
self.assertEqual("COMPLETED", task["status"])
|
|
self.assertIsNone(task["lease_owner"])
|
|
event_types = [row["event_type"] for row in self.store.db.execute("SELECT event_type FROM lifecycle ORDER BY sequence")]
|
|
self.assertEqual(
|
|
["TASK_DISPATCHED", "TASK_STARTED", "TASK_ENDED", "DECISION_APPLIED", "PROJECT_SUMMARY", "PROJECT_COMPLETED"],
|
|
event_types,
|
|
)
|
|
self.assertEqual(6, self.store.db.execute("SELECT COUNT(*) AS n FROM outbox").fetchone()["n"])
|
|
|
|
def test_blocker_threshold_enters_dead_letter(self):
|
|
self.assertEqual("RETRY", self.store.record_blocker("same", 2))
|
|
self.assertEqual("RETRY", self.store.record_blocker("same", 2))
|
|
self.assertEqual("DEAD_LETTER", self.store.record_blocker("same", 2))
|
|
|
|
def test_outbox_retries_then_dead_letters_and_success_is_persisted(self):
|
|
self.store.enqueue("failure", "gateway", {"event_type": "TASK_STARTED"})
|
|
self.store.enqueue("success", "gateway", {"event_type": "TASK_ENDED"})
|
|
|
|
def sender(payload):
|
|
if payload["event_type"] == "TASK_STARTED":
|
|
raise OSError("gateway unavailable")
|
|
|
|
now = time.time() + 1
|
|
self.assertEqual((1, 1), self.store.deliver_outbox(sender, max_attempts=2, now=now))
|
|
self.assertEqual((0, 1), self.store.deliver_outbox(sender, max_attempts=2, now=now + 3))
|
|
statuses = {
|
|
row["outbox_id"]: row["status"] for row in self.store.db.execute("SELECT outbox_id,status FROM outbox")
|
|
}
|
|
self.assertEqual("DEAD_LETTER", statuses["failure"])
|
|
self.assertEqual("SENT", statuses["success"])
|
|
|
|
def test_git_ref_polling_detects_missed_webhook(self):
|
|
self.assertFalse(self.store.observe_ref("refs/heads/main", "a" * 40))
|
|
self.assertTrue(self.store.observe_ref("refs/heads/main", "b" * 40))
|
|
self.assertFalse(self.store.observe_ref("refs/heads/main", "b" * 40))
|
|
|
|
def test_git_ref_poller_creates_one_event_per_change(self):
|
|
values = iter(("a" * 40, "b" * 40, "b" * 40))
|
|
|
|
def resolver(_repo, _ref):
|
|
return next(values)
|
|
|
|
self.assertEqual(0, gateway.poll_git_refs(self.store, Path("/repo"), ["refs/heads/main"], resolver))
|
|
self.assertEqual(1, gateway.poll_git_refs(self.store, Path("/repo"), ["refs/heads/main"], resolver))
|
|
self.assertEqual(0, gateway.poll_git_refs(self.store, Path("/repo"), ["refs/heads/main"], resolver))
|
|
self.assertEqual(1, self.store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"])
|
|
|
|
|
|
class RecoveryAndSafetyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
os.chmod(self.root, 0o700)
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def test_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)
|
|
store = gateway.QuorumStore(self.root / "store.sqlite3")
|
|
try:
|
|
def fail(_payload):
|
|
raise OSError("network unavailable")
|
|
|
|
self.assertFalse(gateway.spool_after_http_failure(fail, {"kind": "dispatch"}, inbox, "event-1"))
|
|
self.assertEqual(1, gateway.poll_file_inbox(store, inbox))
|
|
self.assertEqual(1, store.db.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"])
|
|
finally:
|
|
store.close()
|
|
|
|
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"
|
|
safe.mkdir(mode=0o700)
|
|
receipt = safe / "receipt.json"
|
|
receipt.write_text('{"subject":"codex"}', encoding="utf-8")
|
|
receipt.chmod(0o600)
|
|
link = safe / "link.json"
|
|
link.symlink_to(receipt)
|
|
with self.assertRaises(gateway.ControllerError):
|
|
gateway.read_secure_json(link)
|
|
safe.chmod(0o777)
|
|
with self.assertRaises(gateway.ControllerError):
|
|
gateway.read_secure_json(receipt)
|
|
|
|
@unittest.skipIf(os.name == "nt", "POSIX ownership and symlink semantics required")
|
|
def test_database_symlink_is_fail_closed(self):
|
|
safe = self.root / "safe-db"
|
|
safe.mkdir(mode=0o700)
|
|
target = safe / "target.sqlite3"
|
|
target.write_bytes(b"")
|
|
target.chmod(0o600)
|
|
link = safe / "controller.sqlite3"
|
|
link.symlink_to(target)
|
|
with self.assertRaises(gateway.ControllerError):
|
|
gateway.QuorumStore(link)
|
|
|
|
def test_secret_and_process_output_fields_rejected(self):
|
|
for payload in ({"token": "x"}, {"nested": {"stderr": "x"}}, {"env": {"X": "x"}}):
|
|
with self.assertRaises(gateway.ControllerError):
|
|
gateway.validate_metadata_only(payload)
|
|
|
|
def test_invalid_lifecycle_event_rejected(self):
|
|
store = gateway.QuorumStore(self.root / "store.sqlite3")
|
|
try:
|
|
with self.assertRaises(gateway.ControllerError):
|
|
store.append_lifecycle("WU", "UNAPPROVED", None, 1, {})
|
|
finally:
|
|
store.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|