""" Tests for Phase 5 of krow_talent_app/docs/agent-platform-plan.md: the engine reads operator settings from the backend's agent registry, records its model decisions, and the six agent-to-agent message bugs are fixed. No network, no NATS, no Redis — every external edge is faked. Run: python -m unittest discover -s tests """ import json import unittest from datetime import datetime from unittest.mock import patch import core.decisions as decisions import core.llm as llm from core.message_bus import message_bus from core.registry import RegistryClient, registry from core.types import AgentMessage, AgentTask, MessageType SNAPSHOT = { "agents": [ {"agentid": "EXCEPTION_AGENT", "autonomous": True, "model": "claude-sonnet-5-5"}, {"agentid": "DISPATCH_AGENT", "autonomous": False, "model": ""}, ], "skills": [ {"skillid": "stall_response", "enabled": True, "thresholds": {"stallMinutes": 15, "reassignConfidence": 0.9, "bogus": "x", "flag": True}}, {"skillid": "customer_notifications", "enabled": False}, ], } def _task(task_type, **data): return AgentTask(task_id="t", agent_type="x", task_type=task_type, data=data) class _RegistryIsolation: """Save and restore the shared registry singleton the agents read.""" def _save_registry(self): self._saved = (registry._agents, registry._skills, registry._etag, registry.loaded) def _restore_registry(self): registry._agents, registry._skills, registry._etag, registry.loaded = self._saved # ── RegistryClient: precedence ─────────────────────────────────────────────── class TestRegistryPrecedence(unittest.TestCase): def test_env_defaults_before_load(self): r = RegistryClient() self.assertTrue(r.autonomous("EXCEPTION_AGENT", True)) self.assertFalse(r.autonomous("EXCEPTION_AGENT", False)) self.assertIsNone(r.model("EXCEPTION_AGENT")) self.assertTrue(r.skill_enabled("customer_notifications")) self.assertFalse(r.skill_enabled("anything", default=False)) self.assertEqual(r.threshold("stall_response", "stallMinutes", 10), 10) def test_registry_wins_once_loaded(self): r = RegistryClient() r.apply(SNAPSHOT) self.assertTrue(r.loaded) self.assertTrue(r.autonomous("EXCEPTION_AGENT", False)) self.assertFalse(r.autonomous("DISPATCH_AGENT", True)) self.assertEqual(r.model("EXCEPTION_AGENT"), "claude-sonnet-5-5") self.assertIsNone(r.model("DISPATCH_AGENT")) # "" means engine default self.assertFalse(r.skill_enabled("customer_notifications")) self.assertEqual(r.threshold("stall_response", "stallMinutes", 10), 15) self.assertEqual(r.threshold("stall_response", "reassignConfidence", 0.75), 0.9) def test_unknown_ids_fall_back(self): r = RegistryClient() r.apply(SNAPSHOT) self.assertTrue(r.autonomous("HUB_AGENT", True)) self.assertTrue(r.skill_enabled("not_in_registry")) self.assertEqual(r.threshold("not_in_registry", "k", 3), 3) def test_non_numeric_thresholds_ignored(self): r = RegistryClient() r.apply(SNAPSHOT) self.assertEqual(r.threshold("stall_response", "bogus", 7), 7) self.assertEqual(r.threshold("stall_response", "flag", 7), 7) # bool is not a number here def test_rows_without_ids_skipped(self): r = RegistryClient() r.apply({"agents": [{"autonomous": True}], "skills": [{"enabled": False}]}) self.assertTrue(r.loaded) self.assertEqual(r._agents, {}) self.assertEqual(r._skills, {}) # ── RegistryClient: polling ────────────────────────────────────────────────── class FakeResp: def __init__(self, status, body=None, etag=None, raise_on_json=False): self.status = status self._body = body self.headers = {"ETag": etag} if etag else {} self._raise = raise_on_json async def json(self): if self._raise: raise ValueError("bad json") return self._body async def __aenter__(self): return self async def __aexit__(self, *exc): return False class FakeSession: def __init__(self, *responses, exc=None): self._responses = list(responses) self._exc = exc self.calls = [] def get(self, url, headers=None, timeout=None): self.calls.append((url, dict(headers or {}))) if self._exc: raise self._exc return self._responses.pop(0) class TestRegistryFetch(unittest.IsolatedAsyncioTestCase): async def test_200_then_304_uses_etag(self): r = RegistryClient() s = FakeSession(FakeResp(200, {"data": SNAPSHOT}, etag='"v1"'), FakeResp(304)) self.assertEqual(await r.fetch_once(s, "http://api", "k"), "updated") self.assertTrue(r.loaded) self.assertEqual(await r.fetch_once(s, "http://api", "k"), "unchanged") url, first = s.calls[0] self.assertEqual(url, "http://api/api/v1/internal/ai/registry") self.assertEqual(first["X-Internal-Key"], "k") self.assertNotIn("If-None-Match", first) self.assertEqual(s.calls[1][1]["If-None-Match"], '"v1"') async def test_refused_and_unavailable_keep_last_copy(self): r = RegistryClient() r.apply(SNAPSHOT) for resp, want in ((FakeResp(401), "refused"), (FakeResp(403), "refused"), (FakeResp(500), "unavailable"), (FakeResp(200, raise_on_json=True), "unavailable")): self.assertEqual(await r.fetch_once(FakeSession(resp), "http://api", "k"), want) self.assertEqual(await r.fetch_once(FakeSession(exc=OSError("down")), "http://api", "k"), "unavailable") # Still the operator's settings, not the env defaults. self.assertTrue(r.autonomous("EXCEPTION_AGENT", False)) self.assertFalse(r.skill_enabled("customer_notifications")) async def test_run_without_key_returns_at_once(self): r = RegistryClient() await r.run("http://api", "", poll_seconds=0) # must not loop forever self.assertFalse(r.loaded) # ── Decision log payload ───────────────────────────────────────────────────── class TestDecisions(unittest.IsolatedAsyncioTestCase): def test_build_payload_shape(self): p = decisions.build_payload("stall_response", "42", {"eta": 5}, "reassign", 0.87654, "slow", "m") self.assertEqual(p, { "decision_type": "stall_response", "booking_id": 42, "context": {"facts": {"eta": 5}, "model": "m"}, "decision": {"action": "reassign", "confidence": 0.877}, "reasoning": "slow", }) def test_bad_booking_ids_omitted(self): for bad in (None, "", "abc", 0, -3): self.assertIsNone(decisions.build_payload("t", bad, None, "a", 0, None, "m")["booking_id"]) self.assertEqual(decisions.build_payload("t", 7, None, "a", 0, None, "m")["context"]["facts"], {}) def test_no_running_loop_returns_none(self): with patch.object(decisions, "INTERNAL_API_KEY", "k"): self.assertIsNone(decisions.record_decision("t", 1, {}, "a", 0.5, "r", "m")) async def test_no_key_returns_none(self): with patch.object(decisions, "INTERNAL_API_KEY", ""): self.assertIsNone(decisions.record_decision("t", 1, {}, "a", 0.5, "r", "m")) async def test_posts_to_decision_log(self): sent = [] async def fake_post(url, json=None, headers=None, **kw): sent.append((url, json, headers)) with patch.object(decisions, "INTERNAL_API_KEY", "k"), \ patch.object(decisions, "GO_API_BASE_URL", "http://api"), \ patch.object(decisions, "api_post", fake_post): task = decisions.record_decision("assignment_failure", 9, {}, "notify", 0.5, "r", "m") self.assertIsNotNone(task) await task url, body, headers = sent[0] self.assertEqual(url, "http://api/api/v1/internal/agent-decisions") self.assertEqual(body["booking_id"], 9) self.assertEqual(headers, {"X-Internal-Key": "k"}) # ── Model-dependent request params ────────────────────────────────────────── class TestRequestParams(unittest.TestCase): def test_default_model_has_thinking_and_effort(self): p = llm.request_params() self.assertEqual(p["model"], llm.LLM_MODEL) self.assertEqual(p["thinking"], {"type": "adaptive"}) self.assertIn("effort", p["output_config"]) def test_override_model_used(self): self.assertEqual(llm.request_params("claude-sonnet-5-5")["model"], "claude-sonnet-5-5") def test_haiku_omits_thinking_and_effort(self): p = llm.request_params("claude-haiku-4-5") self.assertNotIn("thinking", p) self.assertNotIn("effort", p["output_config"]) def test_calls_do_not_share_output_config(self): a, b = llm.request_params(), llm.request_params() a["output_config"]["format"] = {} self.assertNotIn("format", b["output_config"]) # ── ExceptionAgent: registry gates and the LLM-down fallback ──────────────── class FakeStallMsg: def __init__(self, payload): self.data = json.dumps(payload).encode() class TestExceptionAgentRegistry(_RegistryIsolation, unittest.IsolatedAsyncioTestCase): def setUp(self): from agents import exception_agent as mod self.mod = mod self._save_registry() self._orig_decide = mod.decide_stall_response self._orig_record = mod.record_decision def tearDown(self): self._restore_registry() self.mod.decide_stall_response = self._orig_decide self.mod.record_decision = self._orig_record def _agent(self, decision=None): from agents.exception_agent import ExceptionAgent agent = ExceptionAgent.__new__(ExceptionAgent) agent.acts = [] async def claim(_b): return True async def facts(_m, _b): return {} async def reassign(b, reason): agent.acts.append("reassign") return True async def notify(b, m): agent.acts.append("notify") return True async def escalate(m, b, mins, d): agent.acts.append(("escalate", d.action)) agent._claim_stall_handled = claim agent._gather_stall_facts = facts agent._reassign = reassign agent._notify_customer = notify agent._escalate_to_human = escalate agent._record_stall_exception = lambda *a, **k: None self.models = [] self.recorded = [] async def fake_decide(context, model=None): self.models.append(model) return decision self.mod.decide_stall_response = fake_decide self.mod.record_decision = lambda *a: self.recorded.append(a) return agent def _load(self, autonomous, enabled=True, confidence=0.75): registry.apply({ "agents": [{"agentid": "EXCEPTION_AGENT", "autonomous": autonomous, "model": "claude-sonnet-5-5"}], "skills": [{"skillid": "stall_response", "enabled": enabled, "thresholds": {"reassignConfidence": confidence}}], }) async def _stall(self, agent): await agent._on_miler_stalled(FakeStallMsg({"miler_id": "m1", "booking_id": "5", "minutes_stalled": 12})) async def test_llm_down_not_autonomous_escalates(self): self._load(autonomous=False) agent = self._agent(decision=None) await self._stall(agent) self.assertEqual(agent.acts, [("escalate", "escalate")]) async def test_llm_down_autonomous_reassigns(self): self._load(autonomous=True) agent = self._agent(decision=None) await self._stall(agent) self.assertEqual(agent.acts, ["reassign", "notify"]) async def test_skill_disabled_does_nothing(self): self._load(autonomous=True, enabled=False) agent = self._agent(decision=None) await self._stall(agent) self.assertEqual(agent.acts, []) self.assertEqual(self.models, []) # the model was never asked async def test_model_pin_and_decision_recorded(self): self._load(autonomous=True, confidence=0.95) agent = self._agent(decision=llm.StallDecision("reassign", "stuck", 0.9)) await self._stall(agent) self.assertEqual(self.models, ["claude-sonnet-5-5"]) # 0.9 is below the registry's 0.95, so the reassign goes to a human. self.assertEqual(agent.acts, [("escalate", "reassign")]) self.assertEqual(self.recorded[0][0], "stall_response") self.assertEqual(self.recorded[0][-1], "claude-sonnet-5-5") async def test_stall_minutes_from_registry(self): registry.loaded = False self.assertEqual(self.mod.stall_minutes(), self.mod.STALL_MINUTES) registry.apply({"skills": [{"skillid": "stall_response", "enabled": True, "thresholds": {"stallMinutes": 20}}]}) self.assertEqual(self.mod.stall_minutes(), 20) # ── CustomerAgent: skill gate and recorded events ─────────────────────────── class TestCustomerAgent(_RegistryIsolation, unittest.IsolatedAsyncioTestCase): def setUp(self): self._save_registry() async def asyncTearDown(self): self._restore_registry() message_bus.unregister_agent("CUSTOMER_AGENT") async def test_notifications_disabled_skips_send(self): from agents.customer_agent import CustomerAgent registry.apply({"skills": [{"skillid": "customer_notifications", "enabled": False}]}) result = await CustomerAgent()._send_notification(_task("send_notification", order_id="O1")) self.assertEqual(result["status"], "skipped") async def test_cancel_and_notification_events_recorded(self): from agents.customer_agent import CustomerAgent agent = CustomerAgent() for mt in (MessageType.ORDER_CANCELLED, MessageType.NOTIFICATION_SENT, MessageType.HEARTBEAT): await agent.handle_message(AgentMessage("id", "EXCEPTION_AGENT", "CUSTOMER_AGENT", mt, {"order_id": "O9"}, datetime.now())) self.assertEqual([e["type"] for e in agent._received_events], ["ORDER_CANCELLED", "NOTIFICATION_SENT"]) self.assertEqual(agent._received_events[0]["order_id"], "O9") # ── Message-bug fixes ──────────────────────────────────────────────────────── class TestMessageFixes(unittest.IsolatedAsyncioTestCase): async def asyncTearDown(self): for agent_id in ("FLEET_AGENT", "ORDER_AGENT", "JARVIS", "FAKE_ORDER", "FAKE_CUSTOMER"): message_bus.unregister_agent(agent_id) async def test_fleet_release_for_cancel_finds_vehicle_by_order(self): from agents.fleet_agent import FleetAgent, VehicleAssignment fleet = FleetAgent() vehicle_id = next(iter(fleet._vehicles)) fleet._vehicles[vehicle_id]["status"] = "in_transit" fleet._assignments["A1"] = VehicleAssignment( "A1", vehicle_id, "R1", ["O1", "O2"], datetime.now(), datetime.now(), {}) missing = await fleet.handle_task(_task("release_vehicle_for_cancel", order_id="NOPE")) self.assertEqual(missing["status"], "no_assignment") await fleet.handle_task(_task("release_vehicle_for_cancel", order_id="O2")) self.assertEqual(fleet._vehicles[vehicle_id]["status"], "available") self.assertEqual(fleet._assignments["A1"].status, "returned") async def test_order_agent_refuses_backend_calls(self): from agents.order_agent import OrderAgent agent = OrderAgent() with patch("core.http_client.api_post") as post, patch("core.http_client.api_get") as get: self.assertIsNone(await agent._api_post("/api/v1/admin/crmbooking", {})) self.assertIsNone(await agent._api_get("/api/v1/admin/x")) self.assertIsNone(await agent._api_patch("/api/v1/admin/x", {})) post.assert_not_called() get.assert_not_called() async def test_jarvis_validate_payload_carries_order_id(self): from core.agent import MasterAgent class Sink: def __init__(self, agent_id): self.agent_id = agent_id self.tasks = [] async def submit_task(self, task): self.tasks.append(task) jarvis = MasterAgent() order, customer = Sink("ORDER_AGENT"), Sink("CUSTOMER_AGENT") jarvis._sub_agents = {"ORDER_AGENT": order, "CUSTOMER_AGENT": customer} await jarvis.handle_task(_task("orchestrate_order", order={"order_id": "O7"})) self.assertEqual(order.tasks[0].task_type, "validate_order") self.assertEqual(order.tasks[0].data["order_id"], "O7") def test_order_status_update_message_type_exists(self): self.assertEqual(MessageType.ORDER_STATUS_UPDATE.value, "ORDER_STATUS_UPDATE") if __name__ == "__main__": unittest.main()