156 lines
4.7 KiB
Python
156 lines
4.7 KiB
Python
"""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"
|
|
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
|
|
|
|
|
|
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 "urllib.request" not in source
|
|
assert "ollama.ai.svc" not in source
|