from __future__ import annotations import copy import json import tempfile import unittest from datetime import timedelta from pathlib import Path from agent_call.core import ( AgentCallService, BrokerError, ConflictError, InMemoryBroker, ServiceError, ValidationError, iso, routing_key, utcnow, ) ROOT = Path(__file__).resolve().parents[1] def fixture() -> dict: try: return json.loads( (ROOT / "docs/contracts/examples/call.execute.json").read_text( encoding="utf-8" ) ) except (OSError, json.JSONDecodeError) as exc: raise AssertionError("valid command fixture cannot be loaded") from exc class AgentCallTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.broker = InMemoryBroker() self.service = AgentCallService( db_path=Path(self.temp.name) / "executor.sqlite3", object_dir=Path(self.temp.name) / "objects", broker=self.broker, ) self.service.profile["limits"]["tenant_cps"] = 100 self.service.profile["limits"]["global_cps"] = 100 def tearDown(self) -> None: self.service.stop() self.temp.cleanup() def command(self, command_id: str, execution_id: str | None = None) -> dict: body = copy.deepcopy(fixture()) body["command_id"] = command_id body["trace_id"] = f"trace_{command_id}" body["payload"]["execution_id"] = execution_id or f"exec_{command_id}" body["issued_at"] = iso(utcnow()) return body def run_command(self, body: dict) -> dict: self.service.publish_execute(body) self.service.wait_for_idle() return self.service.get_command(body["tenant_id"], body["command_id"]) def test_success_is_durable_and_has_verified_recording(self) -> None: command = self.command("cmd_success") snapshot = self.run_command(command) self.assertEqual(snapshot["status"], "completed") call = self.service.get_call("tenant-demo", snapshot["call_id"]) self.assertEqual(call["call_state"], "ended") self.assertEqual(call["outcome"], "completed") self.assertGreaterEqual(len(call["transcript"]["segments"]), 4) self.assertEqual(call["recordings"][0]["status"], "verified") self.assertTrue(call["recordings"][0]["oss_id"].startswith("oss://mock/")) journal = self.service.journal.read() invite = next( item for item in journal if item["protocol"] == "sip" and item["event"] == "invite" ) self.assertEqual(invite["callee"], "18601013734") self.assertEqual(invite["target_uri"], "708918601013734") self.assertEqual(invite["caller"], "BD93205882") self.assertTrue( any( item["event"] == "cleanup" for item in journal if item["protocol"] == "ari" ) ) event_count = self.service.store.one("SELECT COUNT(*) AS n FROM events")["n"] inbox_count = self.service.store.one("SELECT COUNT(*) AS n FROM inbox")["n"] self.assertEqual(event_count, inbox_count) self.assertIn("agent_call_outbox_pending 0", self.service.metrics()) def test_execution_id_idempotency_never_dials_twice(self) -> None: first = self.command("cmd_first", "exec_once") first_snapshot = self.run_command(first) duplicate = self.command("cmd_duplicate", "exec_once") duplicate_snapshot = self.run_command(duplicate) self.assertEqual(duplicate_snapshot["status"], "completed") self.assertEqual(duplicate_snapshot["reason_code"], "DUPLICATE_EXECUTION") self.assertEqual(duplicate_snapshot["call_id"], first_snapshot["call_id"]) calls = self.service.store.one("SELECT COUNT(*) AS n FROM calls")["n"] attempts = self.service.store.one("SELECT COUNT(*) AS n FROM attempts")["n"] self.assertEqual(calls, 1) self.assertEqual(attempts, 1) self.assertEqual( self.service.store.one( "SELECT status FROM executions WHERE tenant_id=? AND execution_id=?", ("tenant-demo", "exec_once"), )["status"], "completed", ) def test_unhealthy_cell_isolated_without_blocking_healthy_capacity(self) -> None: self.service.set_cell_health("cell-a", False) self.assertTrue(self.service.ready()) snapshot = self.run_command(self.command("cmd_healthy_cell")) call = self.service.get_call("tenant-demo", snapshot["call_id"]) self.assertEqual(call["call_state"], "ended") self.assertEqual( self.service.store.one( "SELECT cell_id FROM calls WHERE call_id=?", (snapshot["call_id"],) )["cell_id"], "cell-b", ) def test_missing_tenant_route_closes_readiness(self) -> None: self.service._unrouteable_tenants.add("tenant-demo") snapshot = self.service.readiness() self.assertFalse(snapshot["checks"]["routes"]) self.assertFalse(snapshot["ready"]) def test_expired_scheduler_lease_fences_old_instance(self) -> None: body = self.command("cmd_lease_fence") self.service.publish_execute(body) self.service.ingest_commands() second = AgentCallService( db_path=Path(self.temp.name) / "executor.sqlite3", object_dir=Path(self.temp.name) / "objects-second", broker=self.broker, ) try: with self.service.store.tx() as con: con.execute( "UPDATE leases SET expires_at=? WHERE lease_name='scheduler'", (iso(utcnow() - timedelta(seconds=1)),), ) self.assertEqual(second.schedule_once(), 1) self.assertEqual(self.service.schedule_once(), 0) second.wait_for_idle() finally: second.stop() def test_controls_use_cas_and_old_commands_do_not_resume(self) -> None: pause = self.service.control_task( "tenant-demo", "task-demo", { "command_id": "ctrl_pause", "action": "pause", "expected_task_revision": 1, "reason": "maintenance", }, "ctrl_pause", ) self.assertEqual(pause["status"], "accepted") self.service.process_once() self.assertEqual( self.service.get_command("tenant-demo", "ctrl_pause")["status"], "applied" ) rejected = self.command("cmd_paused") rejected["payload"]["task_revision"] = 2 with self.assertRaises(ConflictError) as blocked: self.service.publish_execute(rejected) self.assertEqual(blocked.exception.code, "TASK_PUBLISH_BLOCKED") with self.assertRaises(ConflictError): self.service.control_task( "tenant-demo", "task-demo", { "command_id": "ctrl_stale", "action": "resume", "expected_task_revision": 1, "reason": "stale", }, "ctrl_stale", ) self.service.control_task( "tenant-demo", "task-demo", { "command_id": "ctrl_resume", "action": "resume", "expected_task_revision": 2, "reason": "ready", }, "ctrl_resume", ) self.service.process_once() resumed = self.command("cmd_resumed") resumed["payload"]["task_revision"] = 3 self.run_command(resumed) self.assertEqual( self.service.get_command("tenant-demo", "cmd_resumed")["status"], "completed", ) def test_stop_hangup_requires_scope_and_ends_active_call(self) -> None: self.service.profile["limits"]["hold_ms"] = 200 body = self.command("cmd_hangup") self.service.publish_execute(body) self.service.ingest_commands() self.service.schedule_once() denied = { "command_id": "ctrl_denied", "action": "stop", "expected_task_revision": 1, "active_call_policy": "hangup", "reason": "test", } with self.assertRaises(ServiceError) as caught: self.service.control_task( "tenant-demo", "task-demo", denied, "ctrl_denied", allow_hangup=False ) self.assertEqual(caught.exception.status, 403) allowed = { "command_id": "ctrl_stop", "action": "stop", "expected_task_revision": 1, "active_call_policy": "hangup", "reason": "test", } self.service.control_task( "tenant-demo", "task-demo", allowed, "ctrl_stop", allow_hangup=True ) self.service.apply_controls() self.service.wait_for_idle() self.assertEqual( self.service.get_command("tenant-demo", "ctrl_stop")["status"], "applied" ) call_id = self.service.get_command("tenant-demo", "cmd_hangup")["call_id"] self.assertEqual( self.service.get_call("tenant-demo", call_id)["reason_code"], "CONTROL_STOPPED", ) def test_broker_outage_retains_publication_for_bounded_retry(self) -> None: command = self.command("cmd_outage") self.broker.outage = True with self.assertRaises(BrokerError): self.service.publish_execute(command) self.assertEqual( self.service.store.one( "SELECT status FROM published_commands WHERE command_id=?", ("cmd_outage",), )["status"], "retained", ) self.broker.outage = False self.assertEqual( self.service.publish_execute(command)["status"], "broker_confirmed" ) self.assertEqual(self.run_command(command)["status"], "completed") def test_replay_uses_cutoff_and_original_event_ids(self) -> None: command = self.command("cmd_replay_source") snapshot = self.run_command(command) original = self.service.store.all( "SELECT event_id FROM events WHERE tenant_id=?", ("tenant-demo",) ) result = self.service.create_replay( "tenant-demo", "call", snapshot["call_id"], {"command_id": "replay_call", "reason": "audit"}, "replay_call", ) self.assertEqual(result["status"], "accepted") self.service.process_replays() replay = self.service.store.one( "SELECT * FROM replays WHERE command_id=?", ("replay_call",) ) self.assertEqual(replay["status"], "completed") replay_ids = { row["event_id"] for row in self.service.store.all( "SELECT event_id FROM replay_items WHERE replay_id=?", (replay["replay_id"],), ) } self.assertTrue(replay_ids) self.assertTrue(replay_ids.issubset({row["event_id"] for row in original})) self.assertEqual( self.service.get_command("tenant-demo", "replay_call")["status"], "completed", ) def test_bad_recording_checksum_emits_failure_without_ready(self) -> None: command = self.command("cmd_bad_recording") self.service.set_failure(command["command_id"], "upload_bad_checksum") snapshot = self.run_command(command) call = self.service.get_call("tenant-demo", snapshot["call_id"]) self.assertEqual(call["recordings"][0]["status"], "failed") self.assertEqual( call["recordings"][0]["failure_code"], "OBJECT_VERIFICATION_FAILED" ) ready = self.service.store.one( "SELECT COUNT(*) AS n FROM events WHERE event_type='recording.ready'" )["n"] failed = self.service.store.one( "SELECT COUNT(*) AS n FROM events WHERE event_type='recording.failed'" )["n"] self.assertEqual(ready, 0) self.assertEqual(failed, 1) def test_tenant_routing_is_exact_and_byte_bounded(self) -> None: self.assertEqual( routing_key("tenant#c"), "agent-call.tenant.tenant#c.call.execute" ) self.assertEqual(len(routing_key("a" * 224).encode("utf-8")), 255) too_long = "a" * 225 with self.assertRaises(ValidationError) as caught: routing_key(too_long) self.assertEqual(caught.exception.code, "TENANT_KEY_UNSUPPORTED") self.assertEqual( self.broker._routes[routing_key("tenant#c")], "agent-call.executor.tenant#c.v1", ) bounded = InMemoryBroker(max_messages=1) bounded.declare_tenant("tenant#c") bounded.publish( "agent-call.commands.v1", routing_key("tenant#c"), {"message": 1} ) with self.assertRaises(BrokerError): bounded.publish( "agent-call.commands.v1", routing_key("tenant#c"), {"message": 2} ) def test_published_sip_directory_overrides_static_route(self) -> None: class Directory: def get_trunk(self, trunk_id: str) -> dict: return { "trunk_id": trunk_id, "status": "published", "revision": 7, "config": { "enabled": True, "codec_profile": { "allowed": ["PCMA"], "preferred": "PCMA", }, "caller_ids": ["BD93205882"], "dial_prefix": "9001", "egress_pool_id": "egress-mock", "max_concurrency": 12, "max_cps": 4, "sip": {"transport": "udp"}, }, } self.service.sip_readonly = Directory() route = self.service._route_config("route_policy_test") self.assertEqual(route["trunk_revision"], 7) self.assertEqual(route["dial_prefix"], "9001") self.assertEqual(route["max_concurrency"], 12) cell = self.service._resource_cell(route) self.assertIsNotNone(cell) if cell is not None: self.assertEqual(cell[0], "cell-a") def test_profile_never_claims_real_ai(self) -> None: metrics = self.service.metrics() self.assertIn( 'agent_call_provider_mode_info{component="llm",mode="mock"} 1', metrics ) self.assertIn( 'agent_call_provider_mode_info{component="tts",mode="mock"} 1', metrics ) self.assertNotIn('mode="real"', metrics) def test_scale_profile_covers_warmup_and_observation(self) -> None: from scripts.scale_mock import build_profile profile = build_profile(Path(self.temp.name) / "scale-profile.json", 1) self.assertEqual(profile["limits"]["hold_ms"], (120 + 60 + 5) * 1000) if __name__ == "__main__": unittest.main()