from __future__ import annotations import json import os import stat from pathlib import Path import tempfile import time import unittest import urllib.error from unittest import mock from services.aios_orchestrator_gateway.clients import ResponsesClient, WorkspaceAgentClient from services.aios_orchestrator_gateway.config import Config from services.aios_orchestrator_gateway.dispatch import DispatchSink from services.aios_orchestrator_gateway.errors import ConflictError, DispatchError, TriggerError, ValidationError from services.aios_orchestrator_gateway.models import Decision, Event from services.aios_orchestrator_gateway.service import Runtime from services.aios_orchestrator_gateway.securefs import ensure_shared_directory from services.aios_orchestrator_gateway.store import Store from services.aios_orchestrator_gateway.sync import OpenClawSynchronizer, lifecycle_record from scripts import report_orchestrator_event CONTRACT = "cbe33ecab8e9794ba5f318bea43aeaf447a56fbc" BASE = "8d9c6da90fc8caa69b691d243c6fefb40b6b0575" def event_payload(**overrides): value = { "schema_version": 1, "project_key": "v2r1-personal-weixin", "contract_commit": CONTRACT, "event_id": "evt-001", "source_agent": "openclaw", "event_type": "work_unit_completed", "work_unit_id": "P1-REACK", "status": "PASS", "state_version": 0, "implementation_commit": None, "result_commit": None, "blocker": None, "evidence_refs": [], "payload": {}, } value.update(overrides) return value def decision_payload(**overrides): value = { "schema_version": 1, "project_key": "v2r1-personal-weixin", "contract_commit": CONTRACT, "event_id": "evt-001", "expected_state_version": 0, "verdict": "DISPATCH", "state_updates": [ {"key": "P1_REPAIR3_CODE", "value": "PASS_CANDIDATE_PREPARED"}, {"key": "current_gate", "value": "R0_RUNNER_LIVE_RECOVERY"}, ], "next_work_units": [ { "id": "R0-DISCOVER", "track_id": "R_RUNNER", "gate_id": "R0_RUNNER_LIVE_RECOVERY", "primary_agent": "openclaw", "fallback_agent": "codex", "clean_base_commit": BASE, "allowed_paths": [], "required_checks": ["secret_provider_metadata"], "forbidden_operations": ["force_push"], "instruction": "Discover provider metadata without reading secrets.", } ], "requires_human_input": False, "stop_reason": None, } value.update(overrides) return value class ModelTests(unittest.TestCase): def test_event_accepts_valid_payload(self): event = Event.from_dict(event_payload()) self.assertEqual(event.event_id, "evt-001") self.assertEqual(len(event.digest()), 64) def test_event_rejects_extra_fields(self): with self.assertRaises(ValidationError): Event.from_dict(event_payload(extra=True)) def test_event_rejects_invalid_contract_oid(self): with self.assertRaises(ValidationError): Event.from_dict(event_payload(contract_commit="abc")) def test_decision_requires_work_unit_for_dispatch(self): with self.assertRaises(ValidationError): Decision.from_dict(decision_payload(next_work_units=[])) def test_decision_rejects_same_primary_fallback(self): value = decision_payload() value["next_work_units"][0]["fallback_agent"] = "openclaw" with self.assertRaises(ValidationError): Decision.from_dict(value) def test_non_dispatch_rejects_work_units(self): with self.assertRaises(ValidationError): Decision.from_dict(decision_payload(verdict="BLOCKED")) class ClientTests(unittest.TestCase): def test_workspace_trigger_accepts_202_without_run_id(self): event = Event.from_dict(event_payload()) response = mock.MagicMock() response.status = 202 response.read.return_value = b"" response.__enter__.return_value = response response.__exit__.return_value = False with mock.patch( "services.aios_orchestrator_gateway.clients.urllib.request.urlopen", return_value=response, ) as call: WorkspaceAgentClient("agtch_test", "token").trigger(event) request = call.call_args.args[0] self.assertEqual(request.headers["Idempotency-key"], "evt-001") def test_responses_background_start_and_poll(self): event = Event.from_dict(event_payload()) decision = decision_payload() client = ResponsesClient("key", "gpt-test") with mock.patch.object( client, "_request", side_effect=[ {"id": "resp_123", "status": "queued"}, { "id": "resp_123", "status": "completed", "output": [ { "content": [ { "type": "output_text", "text": json.dumps(decision), } ] } ], }, ], ): response_id = client.start(event) status, parsed = client.poll(response_id) self.assertEqual(response_id, "resp_123") self.assertEqual(status, "completed") self.assertEqual(parsed.event_id, "evt-001") def test_responses_poll_keeps_queued_state(self): client = ResponsesClient("key", "gpt-test") with mock.patch.object( client, "_request", return_value={"id": "resp_123", "status": "in_progress"}, ): status, parsed = client.poll("resp_123") self.assertEqual(status, "in_progress") self.assertIsNone(parsed) def test_responses_poll_queued_in_progress_completed_sequence(self): client = ResponsesClient("key", "gpt-test") completed = { "status": "completed", "output": [{"content": [{"type": "output_text", "text": json.dumps(decision_payload())}]}], } with mock.patch.object( client, "_request", side_effect=[{"status": "queued"}, {"status": "in_progress"}, completed], ): self.assertEqual(client.poll("resp_123"), ("queued", None)) self.assertEqual(client.poll("resp_123"), ("in_progress", None)) status, decision = client.poll("resp_123") self.assertEqual(status, "completed") self.assertEqual(decision.event_id, "evt-001") def test_responses_invalid_completed_decision_fails_closed(self): client = ResponsesClient("key", "gpt-test") payload = {"status": "completed", "output": [{"content": [{"type": "output_text", "text": "{}"}]}]} with mock.patch.object(client, "_request", return_value=payload): with self.assertRaises(ValidationError): client.poll("resp_123") class StoreTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.store = Store(Path(self.temp.name) / "state.sqlite3") def tearDown(self): self.store.close() self.temp.cleanup() def test_event_idempotency(self): event = Event.from_dict(event_payload()) created, _ = self.store.register_event(event) duplicate, _ = self.store.register_event(event) self.assertTrue(created) self.assertFalse(duplicate) def test_event_status_is_metadata_only_and_maps_internal_states(self): self.store.register_event(Event.from_dict(event_payload())) received = self.store.event_status("evt-001") self.assertEqual( set(received), {"event_id", "status", "trigger_channel", "decision_applied", "state_version_after"}, ) self.assertEqual(received["status"], "RECEIVED") self.store.update_event_trigger( "evt-001", status="POLLING_RESPONSE", channel="responses_api", remote_id="resp-1" ) waiting = self.store.event_status("evt-001") self.assertEqual(waiting["status"], "AWAITING_DECISION") self.assertEqual(waiting["trigger_channel"], "responses_api") def test_event_status_decided_has_applied_version(self): self.store.register_event(Event.from_dict(event_payload())) decision = Decision.from_dict( decision_payload(verdict="NOOP", next_work_units=[]) ) version = self.store.apply_decision( decision, max_hops_per_gate=20, max_same_blocker_retries=3 ) status = self.store.event_status("evt-001") self.assertEqual(status["status"], "DECIDED") self.assertTrue(status["decision_applied"]) self.assertEqual(status["state_version_after"], version) def test_event_payload_conflict_rejected(self): self.store.register_event(Event.from_dict(event_payload())) with self.assertRaises(ConflictError): self.store.register_event(Event.from_dict(event_payload(status="FAIL"))) def test_stale_state_version_rejected(self): self.store.register_event(Event.from_dict(event_payload())) with self.assertRaises(ConflictError): self.store.register_event( Event.from_dict(event_payload(event_id="evt-002", state_version=1)) ) def test_lease_claim_prevents_double_claim(self): self.store.register_event(Event.from_dict(event_payload())) first = self.store.claim_events( owner="one", statuses=("RECEIVED",), limit=5, lease_seconds=60 ) second_store = Store(Path(self.temp.name) / "state.sqlite3") try: second = second_store.claim_events( owner="two", statuses=("RECEIVED",), limit=5, lease_seconds=60 ) finally: second_store.close() self.assertEqual(len(first), 1) self.assertEqual(second, []) def test_health_probe_claim_has_priority_over_older_normal_event(self): self.store.register_event(Event.from_dict(event_payload(event_id="evt-normal"))) self.store.register_event( Event.from_dict( event_payload( event_id="health-central-priority", event_type="central_gpt_health_probe", ) ) ) claimed = self.store.claim_events( owner="health-worker", statuses=("RECEIVED",), limit=1, lease_seconds=60 ) self.assertEqual([row["event_id"] for row in claimed], ["health-central-priority"]) def test_newest_health_probe_claims_before_stale_health_backlog(self): for event_id in ("health-central-old", "health-central-new"): self.store.register_event( Event.from_dict( event_payload(event_id=event_id, event_type="central_gpt_health_probe") ) ) self.store._conn().execute( "UPDATE events SET created_at=1 WHERE event_id='health-central-old'" ) self.store._conn().execute( "UPDATE events SET created_at=2 WHERE event_id='health-central-new'" ) claimed = self.store.claim_events( owner="health-worker", statuses=("RECEIVED",), limit=1, lease_seconds=60 ) self.assertEqual([row["event_id"] for row in claimed], ["health-central-new"]) def test_event_status_distinguishes_dead_letter_from_retryable_failure(self): self.store.register_event(Event.from_dict(event_payload())) self.store.update_event_trigger( "evt-001", status="TRIGGER_FAILED", channel=None, remote_id=None ) self.assertEqual(self.store.event_status("evt-001")["status"], "TRIGGER_FAILED") self.store.update_event_trigger( "evt-001", status="DEAD_LETTER", channel=None, remote_id=None ) self.assertEqual(self.store.event_status("evt-001")["status"], "DEAD_LETTER") def test_workspace_timeout_claim_prevents_double_worker(self): self.store.register_event(Event.from_dict(event_payload())) self.store.update_event_trigger("evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None) self.store._conn().execute("UPDATE events SET updated_at=0 WHERE event_id='evt-001'") first = self.store.claim_workspace_timeouts(owner="one", older_than=1, limit=5, lease_seconds=60) second = self.store.claim_workspace_timeouts(owner="two", older_than=1, limit=5, lease_seconds=60) self.assertEqual(len(first), 1) self.assertEqual(second, []) def test_response_poll_claim_prevents_double_worker(self): self.store.register_event(Event.from_dict(event_payload())) self.store.update_event_trigger("evt-001", status="POLLING_RESPONSE", channel="responses_api", remote_id="resp_123") first = self.store.claim_response_polls(owner="one", limit=5, lease_seconds=60) second = self.store.claim_response_polls(owner="two", limit=5, lease_seconds=60) self.assertEqual(len(first), 1) self.assertEqual(second, []) def test_max_hops_limit(self): self.store.register_event(Event.from_dict(event_payload())) self.store._conn().execute( """ INSERT INTO gate_hops(project_key, gate_id, hops, updated_at) VALUES ('v2r1-personal-weixin', 'R0_RUNNER_LIVE_RECOVERY', 1, 0) """ ) with self.assertRaisesRegex(ConflictError, "max_hops_exceeded"): self.store.apply_decision( Decision.from_dict(decision_payload()), max_hops_per_gate=1, max_same_blocker_retries=3, ) def test_hop_limit_is_scoped_per_gate(self): self.store.register_event(Event.from_dict(event_payload())) first = Decision.from_dict(decision_payload()) self.assertEqual( self.store.apply_decision( first, max_hops_per_gate=1, max_same_blocker_retries=3 ), 1, ) self.store.register_event( Event.from_dict(event_payload(event_id="evt-002", state_version=1)) ) second_value = decision_payload( event_id="evt-002", expected_state_version=1, state_updates=[{"key": "current_gate", "value": "P2_REAL_B3_EVIDENCE"}], ) second_value["next_work_units"][0]["id"] = "P2-REVIEW" second_value["next_work_units"][0]["gate_id"] = "P2_REAL_B3_EVIDENCE" self.assertEqual( self.store.apply_decision( Decision.from_dict(second_value), max_hops_per_gate=1, max_same_blocker_retries=3, ), 2, ) self.store.register_event( Event.from_dict(event_payload(event_id="evt-003", state_version=2)) ) third_value = decision_payload( event_id="evt-003", expected_state_version=2, state_updates=[{"key": "current_gate", "value": "P2_REAL_B3_EVIDENCE"}], ) third_value["next_work_units"][0]["id"] = "P2-REVIEW-RETRY" third_value["next_work_units"][0]["gate_id"] = "P2_REAL_B3_EVIDENCE" with self.assertRaisesRegex(ConflictError, "max_hops_exceeded"): self.store.apply_decision( Decision.from_dict(third_value), max_hops_per_gate=1, max_same_blocker_retries=3, ) def test_decision_advances_state_and_creates_outbox(self): self.store.register_event(Event.from_dict(event_payload())) version = self.store.apply_decision( Decision.from_dict(decision_payload()), max_hops_per_gate=20, max_same_blocker_retries=3, ) self.assertEqual(version, 1) state = self.store.project_state("v2r1-personal-weixin") self.assertEqual(state["state_version"], 1) rows = self.store.claim_outbox(owner="x", limit=10, lease_seconds=60) self.assertEqual(len(rows), 1) def test_decision_idempotency(self): self.store.register_event(Event.from_dict(event_payload())) decision = Decision.from_dict(decision_payload()) first = self.store.apply_decision( decision, max_hops_per_gate=20, max_same_blocker_retries=3 ) second = self.store.apply_decision( decision, max_hops_per_gate=20, max_same_blocker_retries=3 ) self.assertEqual(first, second) def test_same_blocker_limit(self): self.store.register_event(Event.from_dict(event_payload())) decision = Decision.from_dict( decision_payload( verdict="BLOCKED", next_work_units=[], stop_reason="SAME_BLOCKER", ) ) self.store.apply_decision( decision, max_hops_per_gate=20, max_same_blocker_retries=1 ) # New event at next state with the same blocker exceeds the configured limit. self.store.register_event( Event.from_dict(event_payload(event_id="evt-002", state_version=1)) ) decision2 = Decision.from_dict( decision_payload( event_id="evt-002", expected_state_version=1, verdict="BLOCKED", next_work_units=[], stop_reason="SAME_BLOCKER", ) ) with self.assertRaises(ConflictError): self.store.apply_decision( decision2, max_hops_per_gate=20, max_same_blocker_retries=1 ) def test_project_summary_contains_counts(self): self.store.register_event(Event.from_dict(event_payload())) summary = self.store.project_summary("v2r1-personal-weixin") self.assertEqual(summary["event_counts"]["RECEIVED"], 1) class DispatchTests(unittest.TestCase): def test_existing_shared_directory_is_verified_without_mkdir(self): with tempfile.TemporaryDirectory() as temp: queue = Path(temp) / "queue" queue.mkdir(mode=0o2750) queue.chmod(0o2750) with mock.patch.object(Path, "mkdir", side_effect=PermissionError("sandbox")): ensure_shared_directory(queue) def test_codex_dispatch_uses_watcher_command_schema_at_active_root(self): with tempfile.TemporaryDirectory() as temp: outbox = Path(temp) / "outbox" queue = Path(temp) / "queue" outbox.mkdir(mode=0o700) queue.mkdir(mode=0o2770) queue.chmod(0o2770) sink = DispatchSink(outbox, queue, CONTRACT) work_unit = decision_payload()["next_work_units"][0] work_unit.update( primary_agent="codex", fallback_agent="openclaw", required_checks=["acceptance_probe"], ) _, queue_path = sink.dispatch( { "work_unit_id": work_unit["id"], "event_id": "evt-001", "assigned_agent": "codex", "payload_json": json.dumps(work_unit), "project_key": "v2r1-personal-weixin", "state_version": 1, } ) self.assertEqual(queue_path.parent, queue) command = json.loads(queue_path.read_text()) self.assertEqual(command["execution"]["kind"], "acceptance_probe") self.assertEqual(command["state_version"], 1) self.assertEqual(command["contract_commit"], CONTRACT) def test_dispatch_is_idempotent(self): with tempfile.TemporaryDirectory() as temp: outbox = Path(temp) / "outbox" queue = Path(temp) / "queue" outbox.mkdir(mode=0o700) queue.mkdir(mode=0o2770) queue.chmod(0o2770) sink = DispatchSink(outbox, queue, CONTRACT) record = { "work_unit_id": "R0-DISCOVER", "event_id": "evt-001", "assigned_agent": "openclaw", "payload_json": json.dumps(decision_payload()["next_work_units"][0]), } first = sink.dispatch(record) second = sink.dispatch(record) self.assertEqual(first, second) self.assertTrue(first[0].exists()) self.assertTrue(first[1].exists()) self.assertEqual(stat.S_IMODE(first[1].stat().st_mode), 0o640) def test_dispatch_rejects_path_escape(self): with tempfile.TemporaryDirectory() as temp: outbox = Path(temp) / "outbox" queue = Path(temp) / "queue" outbox.mkdir(mode=0o700) queue.mkdir(mode=0o2770) queue.chmod(0o2770) sink = DispatchSink(outbox, queue, CONTRACT) with self.assertRaises(DispatchError): sink.dispatch( { "work_unit_id": "../escape", "event_id": "evt-001", "assigned_agent": "openclaw", "payload_json": "{}", } ) def test_dispatch_rejects_existing_symlink_target(self): with tempfile.TemporaryDirectory() as temp: outbox = Path(temp) / "outbox" queue = Path(temp) / "queue" outbox.mkdir(mode=0o700) queue.mkdir(mode=0o2770) queue.chmod(0o2770) sink = DispatchSink(outbox, queue, CONTRACT) target = Path(temp) / "target" target.write_text("x") pending = outbox / "dispatched" pending.mkdir(mode=0o700) (pending / "R0-DISCOVER.json").symlink_to(target) with self.assertRaises(DispatchError): sink.dispatch( { "work_unit_id": "R0-DISCOVER", "event_id": "evt-001", "assigned_agent": "openclaw", "payload_json": json.dumps(decision_payload()["next_work_units"][0]), } ) class SyncTests(unittest.TestCase): def test_lifecycle_record_is_stable(self): one = lifecycle_record( contract_commit=CONTRACT, project_key="v2r1", kind="TASK_STARTED", work_unit_id="R0", payload={"status": "RUNNING"}, ) two = lifecycle_record( contract_commit=CONTRACT, project_key="v2r1", kind="TASK_STARTED", work_unit_id="R0", payload={"status": "RUNNING"}, ) self.assertEqual(one["sync_id"], two["sync_id"]) def test_file_sync_is_idempotent(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) root.chmod(0o2750) sync = OpenClawSynchronizer(root) record = lifecycle_record( contract_commit=CONTRACT, project_key="v2r1", kind="PROJECT_SUMMARY", work_unit_id=None, payload={"done": 1}, ) first = sync.emit(record) second = sync.emit(record) self.assertEqual(first, second) def test_http_failure_keeps_file_for_retry(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) root.chmod(0o2750) sync = OpenClawSynchronizer(root, endpoint="https://openclaw.invalid/events") record = lifecycle_record( contract_commit=CONTRACT, project_key="v2r1", kind="PROJECT_COMPLETED", work_unit_id=None, payload={"done": 1}, ) with mock.patch( "services.aios_orchestrator_gateway.sync.urllib.request.urlopen", side_effect=OSError("offline"), ): with self.assertRaises(DispatchError): sync.emit(record) pending = root / "pending" / f"{record['sync_id']}.json" self.assertTrue(pending.exists()) response = mock.MagicMock(status=202) response.__enter__.return_value = response response.__exit__.return_value = False with mock.patch( "services.aios_orchestrator_gateway.sync.urllib.request.urlopen", return_value=response, ): self.assertEqual(sync.emit(record), pending) class ReporterTests(unittest.TestCase): def test_http_failure_falls_back_to_file_inbox(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) inbox = root / "inbox" inbox.mkdir(mode=0o2770) inbox.chmod(0o2770) token = root / "token" token.write_text("not-a-real-secret", encoding="utf-8") token.chmod(0o600) argv = [ "report_orchestrator_event.py", "--project-key", "v2r1-personal-weixin", "--contract-commit", CONTRACT, "--event-id", "evt-http-fallback", "--source-agent", "codex", "--event-type", "work_unit_completed", "--work-unit-id", "P1-REACK", "--status", "PASS", "--state-version", "0", "--endpoint", "https://gateway.invalid", "--token-file", str(token), "--inbox-dir", str(inbox), ] with mock.patch("sys.argv", argv), mock.patch( "scripts.report_orchestrator_event.urllib.request.urlopen", side_effect=urllib.error.URLError("offline"), ): self.assertEqual(report_orchestrator_event.main(), 0) pending = inbox / "pending/evt-http-fallback.json" self.assertTrue(pending.exists()) self.assertEqual(stat.S_IMODE(pending.stat().st_mode), 0o640) def test_payload_symlink_is_rejected(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) target = root / "payload.json" target.write_text("{}", encoding="utf-8") target.chmod(0o600) link = root / "payload-link.json" link.symlink_to(target) argv = [ "report_orchestrator_event.py", "--project-key", "v2r1-personal-weixin", "--contract-commit", CONTRACT, "--event-id", "evt-symlink", "--source-agent", "codex", "--event-type", "work_unit_completed", "--work-unit-id", "P1-REACK", "--status", "PASS", "--state-version", "0", "--payload-file", str(link), "--inbox-dir", str(root / "inbox"), ] with mock.patch("sys.argv", argv), self.assertRaisesRegex( Exception, "symlink_rejected" ): report_orchestrator_event.main() class RuntimePollingTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) for name in ("events", "decisions", "outbox", "etc"): (self.root / name).mkdir(mode=0o700) for name in ("sync", "commands"): (self.root / name).mkdir(mode=0o2750) (self.root / name).chmod(0o2750) for name, value in ( ("inbound.token", "inbound"), ("callback.token", "callback"), ("workspace.token", "workspace"), ): path = self.root / "etc" / name path.write_text(value) path.chmod(0o600) config = { "host": "127.0.0.1", "port": 18920, "project_key": "v2r1-personal-weixin-production", "database_path": str(self.root / "state.sqlite3"), "event_inbox_dir": str(self.root / "events"), "decision_inbox_dir": str(self.root / "decisions"), "dispatch_outbox_dir": str(self.root / "outbox"), "openclaw_sync_dir": str(self.root / "sync"), "command_queue_dir": str(self.root / "commands"), "inbound_token_file": str(self.root / "etc/inbound.token"), "callback_token_file": str(self.root / "etc/callback.token"), "contract_commit": CONTRACT, "workspace_agent": { "trigger_id": "agtch_test", "access_token_file": str(self.root / "etc/workspace.token"), }, "responses_fallback": { "api_key_file": None, "model": None, }, "polling": { "interval_seconds": 1, "summary_interval_seconds": 10, }, "loop_controls": { "callback_timeout_seconds": 30, "lease_seconds": 10, "retry_base_seconds": 1, "max_retries": 2, "max_same_blocker_retries": 2, "max_hops_per_gate": 20, }, "openclaw_sync": {"endpoint": None, "token_file": None}, "git_poll": { "enabled": False, "bare_repo": None, "ref_prefixes": [], "interval_seconds": 30, }, } config_path = self.root / "etc/config.json" config_path.write_text(json.dumps(config)) config_path.chmod(0o600) os.environ["AIOS_ORCHESTRATOR_CONFIG_OWNER_UID"] = str(os.getuid()) os.environ["AIOS_ORCHESTRATOR_SECRET_OWNER_UID"] = str(os.getuid()) self.runtime = Runtime(Config.from_file(config_path)) def tearDown(self): self.runtime.store.close() self.temp.cleanup() os.environ.pop("AIOS_ORCHESTRATOR_CONFIG_OWNER_UID", None) os.environ.pop("AIOS_ORCHESTRATOR_SECRET_OWNER_UID", None) def test_file_event_polling_and_workspace_trigger(self): pending = self.root / "events/pending" pending.mkdir(mode=0o2770) pending.chmod(0o2770) (pending / "evt-001.json").write_text(json.dumps(event_payload())) (pending / "evt-001.json").chmod(0o640) with mock.patch.object(self.runtime.workspace, "trigger") as trigger: self.runtime.poll_once() trigger.assert_called_once() processed = self.root / "events/processed/evt-001.json" self.assertTrue(processed.exists()) def test_workspace_timeout_uses_responses_fallback(self): self.runtime.submit_event(event_payload()) self.runtime.store.update_event_trigger( "evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None ) self.runtime.store._conn().execute("UPDATE events SET updated_at=0 WHERE event_id='evt-001'") fallback = mock.Mock() fallback.start.return_value = "resp_fallback" self.runtime.responses = fallback self.runtime.process_workspace_timeouts() row = self.runtime.store._conn().execute( "SELECT status, channel, remote_id FROM events WHERE event_id='evt-001'" ).fetchone() self.assertEqual(tuple(row), ("POLLING_RESPONSE", "responses_api", "resp_fallback")) fallback.start.assert_called_once() def test_workspace_start_failure_uses_configured_responses_fallback(self): self.runtime.submit_event(event_payload()) self.runtime.workspace = mock.Mock() self.runtime.workspace.trigger.side_effect = TriggerError("workspace_unavailable") fallback = mock.Mock() fallback.start.return_value = "resp_start_fallback" self.runtime.responses = fallback self.runtime.process_new_events() row = self.runtime.store._conn().execute( "SELECT status, channel, remote_id, attempts FROM events WHERE event_id='evt-001'" ).fetchone() self.assertEqual( tuple(row), ("POLLING_RESPONSE", "responses_api", "resp_start_fallback", 1), ) def test_workspace_start_failure_without_responses_fails_closed(self): self.runtime.submit_event(event_payload()) self.runtime.workspace = mock.Mock() self.runtime.workspace.trigger.side_effect = TriggerError("workspace_unavailable") self.assertIsNone(self.runtime.responses) self.runtime.process_new_events() row = self.runtime.store._conn().execute( "SELECT status, channel, remote_id, attempts FROM events WHERE event_id='evt-001'" ).fetchone() self.assertEqual(tuple(row), ("TRIGGER_FAILED", None, None, 1)) def test_workspace_callback_timeout_reaches_dead_letter_without_fallback(self): self.runtime.submit_event(event_payload()) self.runtime.store.update_event_trigger( "evt-001", status="AWAITING_DECISION", channel="workspace_agent", remote_id=None ) self.runtime.store._conn().execute( "UPDATE events SET attempts=2, updated_at=0 WHERE event_id='evt-001'" ) self.assertIsNone(self.runtime.responses) self.runtime.process_workspace_timeouts() row = self.runtime.store._conn().execute( "SELECT status, attempts FROM events WHERE event_id='evt-001'" ).fetchone() self.assertEqual(tuple(row), ("DEAD_LETTER", 3)) def test_response_poll_limit_dead_letters_queued_response(self): self.runtime.submit_event(event_payload()) self.runtime.store.update_event_trigger( "evt-001", status="POLLING_RESPONSE", channel="responses_api", remote_id="resp-stuck", ) self.runtime.store._conn().execute( "UPDATE events SET attempts=2, next_attempt_at=0 WHERE event_id='evt-001'" ) fallback = mock.Mock() fallback.poll.return_value = ("queued", None) self.runtime.responses = fallback self.runtime.poll_background_responses() row = self.runtime.store._conn().execute( "SELECT status, attempts FROM events WHERE event_id='evt-001'" ).fetchone() self.assertEqual(tuple(row), ("DEAD_LETTER", 3)) def test_callback_rejects_wrong_contract(self): self.runtime.submit_event(event_payload()) with self.assertRaisesRegex(ConflictError, "contract_version_mismatch"): self.runtime.submit_decision( decision_payload(contract_commit="f" * 40) ) def test_task_start_and_end_are_synced(self): start = Event.from_dict( event_payload( event_id="evt-start", event_type="work_unit_started", status="RUNNING", ) ) self.runtime.submit_event(start.to_dict()) self.runtime.synchronize_openclaw() pending = list((self.root / "sync/pending").glob("*.json")) self.assertTrue(pending) kinds = {json.loads(p.read_text())["kind"] for p in pending} self.assertIn("TASK_STARTED", kinds) def test_decision_dispatch_and_summary_sync(self): self.runtime.submit_event(event_payload()) self.runtime.submit_decision(decision_payload()) self.runtime.dispatch_work_units() self.runtime.synchronize_openclaw() self.assertTrue(list((self.root / "commands/openclaw/pending").glob("*.json"))) kinds = { json.loads(p.read_text())["kind"] for p in (self.root / "sync/pending").glob("*.json") } self.assertIn("TASK_DISPATCHED", kinds) self.assertIn("PROJECT_SUMMARY", kinds) def test_poll_once_ingests_decision_file(self): self.runtime.submit_event(event_payload()) pending = self.root / "decisions/pending" pending.mkdir(mode=0o2770) pending.chmod(0o2770) (pending / "decision.json").write_text(json.dumps(decision_payload())) (pending / "decision.json").chmod(0o640) self.runtime.poll_once() self.assertTrue((self.root / "decisions/processed/decision.json").exists()) def test_git_ref_polling_baselines_then_emits_change(self): object.__setattr__(self.runtime.config, "git_poll_enabled", True) object.__setattr__(self.runtime.config, "git_poll_bare_repo", self.root / "repo.git") object.__setattr__(self.runtime.config, "git_poll_ref_prefixes", ("refs/heads/main",)) oid1 = "1" * 40 oid2 = "2" * 40 first = mock.Mock(returncode=0, stdout=f"refs/heads/main {oid1}\n", stderr="") second = mock.Mock(returncode=0, stdout=f"refs/heads/main {oid2}\n", stderr="") with mock.patch( "services.aios_orchestrator_gateway.service.subprocess.run", side_effect=[first, second], ): self.runtime._next_git_poll = 0 self.runtime.poll_git_refs() self.runtime._next_git_poll = 0 self.runtime.poll_git_refs() summary = self.runtime.store.project_summary( "v2r1-personal-weixin-production" ) self.assertIsNotNone(summary) self.assertEqual(summary["event_counts"]["RECEIVED"], 1) def test_retry_backoff_is_bounded(self): self.assertEqual(self.runtime._backoff(0), 1) self.assertLessEqual(self.runtime._backoff(100), 3600) if __name__ == "__main__": unittest.main()