"""Audit fixes: fallback_models storage, capability fallback audit, registry skill check without re-parse, shadow dedup sorting, and gateway-model capability check. Run: python3 -m unittest discover -s cloud/tests -p "test_audit_fixes.py" -v """ import os import sqlite3 import sys import tempfile import time import unittest from types import SimpleNamespace sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) from cloud.agent_def import default_agents_dir, resolve_registry_agent from cloud.models import decide_routing, parse_routing_policy from cloud.routing_shadow import evaluate_shadow from cloud.server import init_db from cloud.worker import Worker class FallbackModelsTestCase(unittest.TestCase): def test_policy_stores_fallback_models(self): policy = parse_routing_policy({ "mode": "shadow", "tiers": {"default": "m1"}, "fallback_models": ["m2", "m2", " ", "m3"], }) # Previously unreachable dead code: the attribute did # not exist at all on a RoutingPolicy. self.assertEqual(policy.fallback_models, ["m2", "m3"]) class CapabilityFallbackAuditTestCase(unittest.TestCase): def test_decision_flags_capability_fallback(self): policy = parse_routing_policy({ "mode": "active", "tiers": {"plan": {"model": "no-tools", "requires": {"tool_calling": True}}}, "model_capabilities": { "no-tools": {"tool_calling": False}, "worker-model": {"tool_calling": True}, }, }) decision = decide_routing( {"id": "t1", "plan": True}, "worker-model", policy=policy) self.assertEqual(decision.source, "worker_default") self.assertTrue(decision.capability_fallback) self.assertEqual(decision.chosen_model, "worker-model") def test_normal_worker_default_is_not_flagged(self): policy = parse_routing_policy({ "mode": "active", "tiers": {"plan": "plan-model"}}) decision = decide_routing( {"id": "t1", "plan": True}, "worker-model", policy=policy) self.assertFalse(decision.capability_fallback) class CapabilityFallbackWorkerAuditTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.calls = [] def tearDown(self): self.tmp.cleanup() def test_fallback_is_audited(self): _insert_task(self.db_path, "F1") def recording(cmd): self.calls.append(list(cmd)) return _completed_runner(cmd) worker = Worker( self.db_path, "/fake/host.py", runner=recording, env={"XCM_API_KEY": "test-placeholder"}, model="worker-model", model_explicit=False, routing_policy={ "mode": "active", "tiers": {"default": {"model": "no-tools", "requires": {"tool_calling": True}}}, "model_capabilities": { "no-tools": {"tool_calling": False}, "worker-model": {"tool_calling": True}, }, }) result = worker.run_one() self.assertEqual(result["status"], "completed") from cloud.auth import AuthStore store = AuthStore(self.db_path) try: entries = [e for e in store.list_audit("p1") if e["action"] == "worker_capability_fallback"] finally: store.conn.close() self.assertTrue(entries) self.assertIn("no-tools", entries[0]["detail"]) class RegistrySkillCheckTestCase(unittest.TestCase): def test_registry_definition_keeps_all_fields(self): # The old re-parse dropped any field not copied into # its hand-built dict; the direct check must return # the loaded definition unchanged, including # tool_approvals, and still reject unknown skills. definition = resolve_registry_agent( "research-brief-approval", agents_dir=default_agents_dir(), known_skills=["live-info"]) self.assertEqual(definition.tool_approvals, {"websearch": "ask"}) self.assertEqual(definition.tools, ["websearch"]) with self.assertRaises(ValueError): resolve_registry_agent( "research-brief-approval", agents_dir=default_agents_dir(), known_skills=["other-skill"]) class ShadowSortTestCase(unittest.TestCase): def test_dedup_uses_earliest_row_regardless_of_input_order(self): rows = [ {"detail": "t1 tier=default source=shadow applied=False" " suggested=new-model chosen=worker-model", "created_at": 200, "id": 2}, {"detail": "t1 tier=default source=shadow applied=False" " suggested=old-model chosen=worker-model", "created_at": 100, "id": 1}, ] report = evaluate_shadow( rows, {"t1": {"status": "completed"}}, {}) self.assertEqual( report["tiers"]["default"]["suggested_models"], {"old-model": 1}) def _completed_runner(cmd): if "dispatch" in cmd: name = cmd[cmd.index("--name") + 1] else: name = cmd[cmd.index("--task") + 1] return SimpleNamespace( returncode=0, stdout='{"task": "%s", "status": "completed", "text": "done"}' % name, stderr="") def _insert_task(db_path, task_id): now = time.time() conn = sqlite3.connect(db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," " created_at, updated_at, last_heartbeat)" " VALUES(?, 'p1', 'do work', 'queued', NULL, '[]', 0," " 'tester', ?, ?, NULL)", (task_id, now, now)) conn.commit() finally: conn.close() class GatewayCapabilityTestCase(unittest.TestCase): """Gateway check is exercised through the worker's routing policy path by stubbing _ensure_gateway: the worker's run_one gateway step is the only place the sub2api default model is filled, so the check must live there.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.calls = [] def tearDown(self): self.tmp.cleanup() def _worker(self, policy_doc, gateway_model): def recording(cmd): self.calls.append(list(cmd)) return _completed_runner(cmd) worker = Worker(self.db_path, "/fake/host.py", runner=recording, env={"XCM_API_KEY": "test-placeholder"}, routing_policy=policy_doc) # Stub the gateway to behave like sub2api filling a # default model, without touching real gateway code. worker._ensure_gateway = lambda task: ("sub2api", {}) worker._sub2api_default_model = lambda env: gateway_model return worker def test_active_gateway_incapable_blocks_before_runner(self): _insert_task(self.db_path, "G1") worker = self._worker({ "mode": "active", "tiers": {"default": {"model": "policy-model", "requires": {"tool_calling": True}}}, "model_capabilities": { "gateway-model": {"tool_calling": False}, }, }, gateway_model="gateway-model") result = worker.run_one() self.assertTrue(result["dispatched"]) self.assertEqual(result["status"], "failed") self.assertIn("tool_calling", result["error"]) self.assertEqual(self.calls, []) def test_shadow_gateway_incapable_only_audits(self): _insert_task(self.db_path, "G2") worker = self._worker({ "mode": "shadow", "tiers": {"default": {"model": "policy-model", "requires": {"tool_calling": True}}}, "model_capabilities": { "gateway-model": {"tool_calling": False}, }, }, gateway_model="gateway-model") result = worker.run_one() self.assertEqual(result["status"], "completed") from cloud.auth import AuthStore store = AuthStore(self.db_path) try: entries = [e for e in store.list_audit("p1") if e["action"] == "worker_capability_gateway_shadow"] finally: store.conn.close() self.assertTrue(entries)