140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
"""Contracts for Agent Hermes automatic route selection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
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 _status() -> dict:
|
|
return {
|
|
"providers": {
|
|
"openai-codex": {"connected": True},
|
|
"anthropic": {"connected": True},
|
|
},
|
|
"routes": {
|
|
"codex-low": [
|
|
"openai-codex/gpt-5.6-luna",
|
|
"anthropic/claude-haiku-4-5-20251001",
|
|
],
|
|
"codex-medium": [
|
|
"openai-codex/gpt-5.6-terra",
|
|
"anthropic/claude-sonnet-5",
|
|
],
|
|
"claude-medium": [
|
|
"anthropic/claude-sonnet-5",
|
|
"openai-codex/gpt-5.6-terra",
|
|
],
|
|
"claude-xhigh": [
|
|
"anthropic/claude-opus-5",
|
|
"openai-codex/gpt-5.6-sol",
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def test_heuristics_keep_simple_questions_cheap_and_risky_work_capped():
|
|
simple = router.heuristic_decision("Who is the current provider?")
|
|
risky = router.heuristic_decision("Migrate production Vault credentials safely")
|
|
|
|
assert (simple.shape, simple.effort, simple.provider) == (
|
|
"question",
|
|
"low",
|
|
"codex",
|
|
)
|
|
assert (risky.shape, risky.effort, risky.provider) == (
|
|
"review",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
|
|
|
|
def test_jetson_calibrates_effort_without_overriding_shape_or_provider(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "high", "claude", "jetson", "test", 42
|
|
),
|
|
)
|
|
|
|
decision = router.classify_task("Implement and test the new API handler")
|
|
|
|
assert decision.shape == "implementation"
|
|
assert decision.provider == "codex"
|
|
assert decision.effort == "high"
|
|
assert decision.classifier == "jetson"
|
|
|
|
|
|
def test_route_uses_managed_models_and_connected_provider_fallback():
|
|
status = _status()
|
|
decision = router.Decision(
|
|
"question", "low", "codex", "heuristic", "short question"
|
|
)
|
|
plan = router.select_route(status, decision)
|
|
assert plan["profile"] == "codex-low"
|
|
assert plan["model"] == "gpt-5.6-luna"
|
|
|
|
status["providers"]["openai-codex"]["connected"] = False
|
|
status["routes"]["claude-low"] = [
|
|
"anthropic/claude-haiku-4-5-20251001",
|
|
"openai-codex/gpt-5.6-luna",
|
|
]
|
|
fallback = router.select_route(status, decision)
|
|
assert fallback["profile"] == "claude-low"
|
|
assert fallback["provider"] == "anthropic"
|
|
|
|
|
|
def test_local_classifier_accepts_only_bounded_effort_codes():
|
|
assert router._validated_local_effort("X", 1) is None
|
|
assert router._validated_local_effort("M", 1).effort == "medium"
|
|
|
|
|
|
def test_deterministic_low_and_xhigh_routes_skip_local_latency(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: (_ for _ in ()).throw(AssertionError("Jetson should be skipped")),
|
|
)
|
|
|
|
assert router.classify_task("Who is the provider?").effort == "low"
|
|
assert router.classify_task("Migrate production Vault credentials").effort == "xhigh"
|
|
|
|
|
|
def test_manual_policy_is_reapplied_on_every_non_command_turn(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "manual",
|
|
"manual": {"provider": "claude", "effort": "medium", "model": ""},
|
|
},
|
|
)
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_plan", lambda policy, plan: None)
|
|
|
|
class Agent:
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
router._pre_turn_route(object(), agent=agent, user_message="Continue the task")
|
|
|
|
assert plans[0]["profile"] == "claude-medium"
|
|
assert plans[0]["model"] == "claude-sonnet-5"
|
|
assert agent.message.startswith("MANUAL")
|