"""Contracts for the thin Hermes-to-Switchyard boundary adapter.""" from __future__ import annotations import importlib.util import json import sys from pathlib import Path from types import SimpleNamespace SOURCE = Path(__file__).parents[2] / "services/hermes/plugins/auto-router/__init__.py" PLUGIN_ROOT = SOURCE.parent sys.path.insert(0, str(PLUGIN_ROOT.parent)) SPEC = importlib.util.spec_from_file_location("hermes_auto_router", SOURCE) assert SPEC and SPEC.loader router = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = router SPEC.loader.exec_module(router) def _agent(**overrides): values = { "provider": router.SWITCHYARD_PROVIDER, "model": "atlas/auto/maximum", "base_url": "http://hermes-switchyard:9005/v1", "api_key": "atlas-switchyard", "api_mode": "chat_completions", "reasoning_config": {"effort": "high"}, "_fallback_chain": [{"provider": "anthropic"}], "_fallback_index": 1, "_fallback_activated": True, "_fallback_model": {"provider": "anthropic"}, "_hermes_explicit_model_pick": False, "_hermes_explicit_reasoning_effort": "", "_hermes_routing_priority": "", } values.update(overrides) return SimpleNamespace(**values) def test_profile_defaults_are_distinct_and_quality_ordered(): assert router.PROFILE_ROUTE == { "chat": "atlas/auto/fast", "triage": "atlas/auto/deep", "agent": "atlas/auto/maximum", } assert router.PROFILE_ROUTE[router.ROUTER_PROFILE] in router.AUTO_ROUTES def test_auto_boundary_selects_only_a_public_switchyard_route(tmp_path, monkeypatch): monkeypatch.setattr(router, "POLICY_PATH", tmp_path / "route-policy.json") agent = _agent() route, effort, source = router._boundary_selection(agent) assert route == router.PROFILE_ROUTE[router.ROUTER_PROFILE] assert effort == "" assert source == "auto" def test_ui_priority_changes_auto_posture_without_selecting_a_target( tmp_path, monkeypatch ): monkeypatch.setattr(router, "POLICY_PATH", tmp_path / "route-policy.json") agent = _agent(_hermes_routing_priority="deep") route, effort, source = router._boundary_selection(agent) assert (route, effort, source) == ("atlas/auto/deep", "", "ui-auto") assert route in router.AUTO_ROUTES def test_ui_manual_model_and_effort_are_forwarded_as_constraints( tmp_path, monkeypatch ): monkeypatch.setattr(router, "POLICY_PATH", tmp_path / "route-policy.json") agent = _agent( model="atlas/manual/claude/opus", _hermes_explicit_model_pick=True, _hermes_explicit_reasoning_effort="xhigh", ) assert router._boundary_selection(agent) == ( "atlas/manual/claude/opus", "xhigh", "ui-manual", ) def test_manual_command_persists_a_switchyard_route_not_a_direct_provider( tmp_path, monkeypatch ): path = tmp_path / "route-policy.json" monkeypatch.setattr(router, "POLICY_PATH", path) ctx = SimpleNamespace(_manager=SimpleNamespace(_cli_ref=None)) message = router._route_command(ctx, "manual codex xhigh sol") policy = json.loads(path.read_text(encoding="utf-8")) assert "Manual Switchyard constraint enabled" in message assert policy["manual"] == { "route": "atlas/manual/codex/sol", "effort": "xhigh", } assert "provider" not in policy["manual"] def test_effort_above_xhigh_is_rejected(tmp_path, monkeypatch): monkeypatch.setattr(router, "POLICY_PATH", tmp_path / "route-policy.json") ctx = SimpleNamespace(_manager=SimpleNamespace(_cli_ref=None)) message = router._route_command(ctx, "manual claude max opus") assert "Effort must be" in message def test_switchyard_owns_fallbacks_and_auto_clears_static_effort(): agent = _agent() ctx = SimpleNamespace(_manager=SimpleNamespace(_cli_ref=None)) router._switch_agent(ctx, agent, "atlas/auto/maximum", "") assert agent.reasoning_config is None assert agent._fallback_chain == [] assert agent._fallback_index == 0 assert agent._fallback_activated is False assert agent._fallback_model is None def test_registers_every_model_call_boundary_and_route_command(): hooks = {} commands = {} class Context: def register_hook(self, name, callback): hooks[name] = callback def register_command(self, name, callback, **metadata): commands[name] = (callback, metadata) router.register(Context()) assert set(hooks) == { "pre_turn_route", "pre_internal_route", "pre_subagent_route", } assert "route" in commands assert "providers" in commands def test_adapter_contains_no_content_classifier_or_direct_ollama_call(): source = SOURCE.read_text(encoding="utf-8") assert "jetson_decision" not in source assert "ollama.ai.svc" not in source def test_provider_status_separates_observed_activity_from_plan_quota(monkeypatch): status = sys.modules["hermes_auto_router"].provider_status_text.__globals__ module = sys.modules[status["provider_status_payload"].__module__] monkeypatch.setattr(module, "_get_json", lambda url, timeout=3.0: ( {"status": "ok"} if url.endswith("/health") else { "total_requests": 3, "total_errors": 1, "total_tokens": {"total": 900}, "models": { "route/codex/sol/xhigh": { "calls": 1, "errors": 0, "total_tokens": 600, "avg_latency_ms": 1200, }, "route/claude/sonnet/high": { "calls": 0, "errors": 1, "total_tokens": 0, }, "route/local/qwen2.5-14b/medium": { "calls": 1, "errors": 0, "total_tokens": 300, }, }, "routing_fallbacks": {"unavailable": 1}, "classifier": {"total_requests": 3, "total_errors": 0}, } )) monkeypatch.setattr(module, "_codex_account", lambda: { "authenticated": True, "plan": "plus", "quota_reported": False, }) monkeypatch.setattr(module, "_claude_account", lambda: { "authenticated": True, "plan": "max", "quota_reported": False, }) payload = module.provider_status_payload() assert payload["router"]["state"] == "available" assert payload["providers"]["codex"]["calls"] == 1 assert payload["providers"]["claude"]["state"] == "unavailable" assert payload["providers"]["local"]["total_tokens"] == 300 assert payload["providers"]["codex"]["account"]["quota_reported"] is False assert "not exposed" in payload["quota_note"] def test_provider_status_accepts_iso_and_epoch_credential_expiry(): status = sys.modules["hermes_auto_router"].provider_status_text.__globals__ module = sys.modules[status["provider_status_payload"].__module__] iso_value, iso_live = module._timestamp("2999-01-01T00:00:00Z") epoch_value, epoch_live = module._timestamp(32_472_192_000) assert iso_value == "2999-01-01T00:00:00+00:00" assert iso_live is True assert epoch_value and epoch_value.startswith("2999-01-01T") assert epoch_live is True def test_claude_account_treats_a_live_refresh_token_as_refreshable( tmp_path, monkeypatch ): status = sys.modules["hermes_auto_router"].provider_status_text.__globals__ module = sys.modules[status["provider_status_payload"].__module__] path = tmp_path / ".credentials.json" path.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "expired-access", "refreshToken": "live-refresh", "expiresAt": 1, "refreshTokenExpiresAt": 32_472_192_000, "subscriptionType": "max", } }), encoding="utf-8") monkeypatch.setattr(module, "CLAUDE_AUTH_PATH", path) account = module._claude_account() assert account["authenticated"] is True assert account["access_token_live"] is False assert account["refreshable"] is True def test_agent_mounts_provider_status_dashboard_into_auto_router_plugin(): import yaml root = Path(__file__).parents[2] deployment = yaml.safe_load( (root / "services/hermes/agent-deployment.yaml").read_text(encoding="utf-8") ) volume = next( item for item in deployment["spec"]["template"]["spec"]["volumes"] if item["name"] == "auto-router-plugin" ) paths = {item["path"] for item in volume["configMap"]["items"]} assert "provider_status.py" in paths assert "dashboard/manifest.json" in paths assert "dashboard/plugin_api.py" in paths assert "dashboard/dist/index.js" in paths