574 lines
18 KiB
Python
574 lines
18 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_selects_provider_while_deterministic_policy_preserves_work_shape(
|
|
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 == "claude"
|
|
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_route_circuit_breaker_skips_recently_failed_provider():
|
|
status = _status()
|
|
status["routes"]["codex-medium"] = [
|
|
"openai-codex/gpt-5.6-terra",
|
|
"anthropic/claude-sonnet-5",
|
|
]
|
|
decision = router.Decision(
|
|
"question", "medium", "claude", "jetson", "local vote"
|
|
)
|
|
policy = {
|
|
"mode": "auto",
|
|
"provider_cooldowns": {
|
|
"anthropic": {"until_epoch": router.time.time() + 300}
|
|
},
|
|
}
|
|
|
|
plan = router.select_route(status, decision, policy=policy)
|
|
|
|
assert plan["profile"] == "codex-medium"
|
|
assert plan["provider"] == "openai-codex"
|
|
|
|
|
|
def test_local_classifier_accepts_only_bounded_route_decisions():
|
|
assert router._validated_local_route("?", "?", 1) is None
|
|
decision = router._validated_local_route("A", "H", 1)
|
|
assert decision is not None
|
|
assert (decision.shape, decision.provider, decision.effort) == (
|
|
"question",
|
|
"claude",
|
|
"high",
|
|
)
|
|
partial = router._validated_local_route("?", "M", 1)
|
|
assert partial is not None
|
|
assert (partial.provider, partial.effort) == ("codex", "medium")
|
|
|
|
|
|
def test_scalar_vote_accepts_raw_and_json_strings_but_remains_bounded():
|
|
codes = ("C", "A")
|
|
|
|
assert router._parse_scalar_vote("C", codes) == "C"
|
|
assert router._parse_scalar_vote('"A"', codes) == "A"
|
|
assert router._parse_scalar_vote(" codex ", codes) is None
|
|
assert router._parse_scalar_vote("C\nA", codes) is None
|
|
|
|
|
|
def test_jetson_requests_separate_bounded_provider_and_effort_votes(monkeypatch):
|
|
calls = []
|
|
|
|
def scalar(text, prompt, codes, timeout):
|
|
calls.append((text, codes))
|
|
return (("A" if codes == ("C", "A") else "H"), 12)
|
|
|
|
monkeypatch.setattr(router, "_jetson_scalar", scalar)
|
|
|
|
decision = router.jetson_decision("Review the architecture")
|
|
|
|
assert (decision.provider, decision.effort, decision.latency_ms) == (
|
|
"claude",
|
|
"high",
|
|
24,
|
|
)
|
|
assert [codes for _, codes in calls] == [("C", "A"), ("L", "M", "H", "X")]
|
|
|
|
|
|
def test_every_auto_classification_consults_jetson_and_keeps_safety_floors(monkeypatch):
|
|
calls = []
|
|
|
|
def classify(text):
|
|
calls.append(text)
|
|
return router.Decision("question", "low", "codex", "jetson", "test", 5)
|
|
|
|
monkeypatch.setattr(router, "jetson_decision", classify)
|
|
|
|
simple = router.classify_task("Who is the provider?")
|
|
risky = router.classify_task("Migrate production Vault credentials")
|
|
|
|
assert len(calls) == 2
|
|
assert (simple.effort, simple.provider) == ("low", "codex")
|
|
assert (risky.shape, risky.effort, risky.provider) == (
|
|
"review",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
|
|
|
|
def test_trivial_prompt_ignores_one_step_jetson_effort_wobble(monkeypatch):
|
|
calls = []
|
|
|
|
def classify(text):
|
|
calls.append(text)
|
|
return router.Decision("question", "medium", "codex", "jetson", "test", 5)
|
|
|
|
monkeypatch.setattr(router, "jetson_decision", classify)
|
|
|
|
decision = router.classify_task(
|
|
"Reply with exactly ROUTE_SMOKE_OK. Do not call tools."
|
|
)
|
|
|
|
assert calls == ["Reply with exactly ROUTE_SMOKE_OK. Do not call tools."]
|
|
assert (decision.effort, decision.provider, decision.classifier) == (
|
|
"low",
|
|
"codex",
|
|
"jetson",
|
|
)
|
|
|
|
|
|
def test_trivial_prompt_can_still_escalate_on_strong_jetson_signal(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "high", "claude", "jetson", "test", 5
|
|
),
|
|
)
|
|
|
|
decision = router.classify_task("Check this.")
|
|
|
|
assert (decision.effort, decision.provider) == ("high", "claude")
|
|
|
|
|
|
def test_architecture_and_review_fail_upward_to_claude(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "low", "codex", "jetson", "test", 5
|
|
),
|
|
)
|
|
|
|
architecture = router.classify_task("Design the service architecture")
|
|
review = router.classify_task("Review this change for regressions")
|
|
|
|
assert (architecture.provider, architecture.effort) == ("claude", "medium")
|
|
assert (review.provider, review.effort) == ("claude", "medium")
|
|
|
|
|
|
def test_referential_outstanding_work_never_uses_low_route(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "low", "codex", "jetson", "test", 10
|
|
),
|
|
)
|
|
|
|
decision = router.classify_task(
|
|
"Look, loop through all of the still outstanding work that you identified. "
|
|
"Do it to the best of your abilities."
|
|
)
|
|
|
|
assert (decision.shape, decision.effort, decision.provider) == (
|
|
"implementation",
|
|
"high",
|
|
"codex",
|
|
)
|
|
|
|
|
|
def test_referential_followup_uses_recent_context_for_risk_floor(monkeypatch):
|
|
calls = []
|
|
|
|
def classify(text):
|
|
calls.append(text)
|
|
return router.Decision("question", "low", "codex", "jetson", "test", 5)
|
|
|
|
monkeypatch.setattr(router, "jetson_decision", classify)
|
|
history = [
|
|
{"role": "user", "content": "Is the work complete?"},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": (
|
|
"Still outstanding: production deployment retry, provider "
|
|
"switching, runtime experiments, and a full repository test pass."
|
|
),
|
|
}
|
|
],
|
|
},
|
|
]
|
|
|
|
decision = router.classify_task(
|
|
"Loop through all outstanding work and finish it.", history
|
|
)
|
|
|
|
assert (decision.shape, decision.effort, decision.provider) == (
|
|
"review",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
assert decision.classifier == "jetson-context"
|
|
assert "recent assistant context" in decision.reason
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_internal_prompt_contains_objective_tools_and_results():
|
|
text = router._internal_task_text(
|
|
"Finish the production deployment safely",
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "I will inspect the failed rollout.",
|
|
"tool_calls": [
|
|
{
|
|
"function": {
|
|
"name": "terminal",
|
|
"arguments": '{"command":"kubectl get pods"}',
|
|
}
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "content": "deployment is degraded"},
|
|
],
|
|
)
|
|
|
|
assert "Finish the production deployment safely" in text
|
|
assert "planned tool terminal" in text
|
|
assert "deployment is degraded" in text
|
|
|
|
|
|
def test_every_internal_auto_prompt_is_reclassified_and_applied(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "auto"})
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision(
|
|
"question", "medium", "claude", "jetson", "test", 7
|
|
),
|
|
)
|
|
applied = []
|
|
recorded = []
|
|
monkeypatch.setattr(
|
|
router, "_apply_route", lambda ctx, agent, plan: applied.append(plan)
|
|
)
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_record_internal_plan",
|
|
lambda policy, plan, count: recorded.append((plan, count)),
|
|
)
|
|
|
|
class Agent:
|
|
provider = "openai-codex"
|
|
model = "gpt-5.6-luna"
|
|
reasoning_config = {"effort": "low"}
|
|
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
router._pre_internal_route(
|
|
object(),
|
|
agent=agent,
|
|
user_message="Continue",
|
|
conversation_history=[
|
|
{"role": "tool", "content": "The architecture review found a risk."}
|
|
],
|
|
api_call_count=3,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert applied[0]["profile"] == "claude-medium"
|
|
assert applied[0]["classifier"] == "jetson-internal"
|
|
assert recorded[0][1] == 3
|
|
assert agent.message.startswith("AUTO internal #3")
|
|
|
|
|
|
def test_manual_route_audits_internal_prompt_without_overriding_user_choice(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "manual",
|
|
"manual": {"provider": "claude", "effort": "medium", "model": ""},
|
|
},
|
|
)
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision("implementation", "xhigh", "codex", "jetson", "audit", 5),
|
|
)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_internal_plan", lambda *args: None)
|
|
|
|
router._pre_internal_route(
|
|
object(),
|
|
agent=object(),
|
|
user_message="Continue",
|
|
conversation_history=[{"role": "tool", "content": "done"}],
|
|
api_call_count=2,
|
|
)
|
|
assert len(calls) == 1
|
|
assert plans[0]["profile"] == "claude-medium"
|
|
assert plans[0]["classifier"] == "manual-jetson-internal"
|
|
|
|
|
|
def test_every_native_subagent_is_classified_and_routed_independently(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "auto"})
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision(
|
|
"implementation", "medium", "codex", "jetson", "test", 9
|
|
),
|
|
)
|
|
applied = []
|
|
recorded = []
|
|
monkeypatch.setattr(
|
|
router, "_apply_route", lambda ctx, agent, plan: applied.append((agent, plan))
|
|
)
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_record_subagent_plan",
|
|
lambda policy, plan, goal, index: recorded.append((plan, goal, index)),
|
|
)
|
|
|
|
class Parent:
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
child = object()
|
|
parent = Parent()
|
|
router._pre_subagent_route(
|
|
object(),
|
|
agent=child,
|
|
parent_agent=parent,
|
|
goal="Implement the bounded parser fix",
|
|
context="Run the focused tests.",
|
|
task_index=2,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert "Run the focused tests" in calls[0]
|
|
assert applied[0][0] is child
|
|
assert applied[0][1]["profile"] == "codex-medium"
|
|
assert applied[0][1]["classifier"] == "jetson-subagent"
|
|
assert recorded[0][1:] == ("Implement the bounded parser fix", 2)
|
|
assert parent.message.startswith("AUTO child #3")
|
|
|
|
|
|
def test_manual_route_audits_and_applies_override_to_native_subagent(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "manual",
|
|
"manual": {"provider": "claude", "effort": "medium", "model": ""},
|
|
},
|
|
)
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision("implementation", "high", "codex", "jetson", "audit", 5),
|
|
)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_subagent_plan", lambda *args: None)
|
|
|
|
router._pre_subagent_route(
|
|
object(), agent=object(), parent_agent=object(), goal="Review the diff"
|
|
)
|
|
assert len(calls) == 1
|
|
assert plans[0]["profile"] == "claude-medium"
|
|
assert plans[0]["classifier"] == "manual-jetson-subagent"
|
|
|
|
|
|
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())
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text, history=None: router.Decision(
|
|
"implementation", "high", "codex", "jetson", "audit", 5
|
|
),
|
|
)
|
|
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 target")
|
|
assert plans[0]["classifier"] == "manual-jetson"
|
|
|
|
|
|
def test_post_turn_records_and_announces_capacity_fallback(monkeypatch):
|
|
policy = {
|
|
"mode": "auto",
|
|
"last_decision": {
|
|
"provider": "anthropic",
|
|
"model": "claude-sonnet-5",
|
|
"effort": "medium",
|
|
"classifier": "jetson",
|
|
},
|
|
}
|
|
written = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: policy)
|
|
monkeypatch.setattr(router, "_write_policy", lambda value: written.append(value))
|
|
|
|
class Agent:
|
|
provider = "openai-codex"
|
|
model = "gpt-5.6-terra"
|
|
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
cli = type("CLI", (), {"agent": agent})()
|
|
manager = type("Manager", (), {"_cli_ref": cli})()
|
|
ctx = type("Context", (), {"_manager": manager})()
|
|
|
|
router._post_turn_route(ctx, model="gpt-5.6-terra")
|
|
|
|
outcome = written[-1]["last_decision"]
|
|
assert outcome["fallback_used"] is True
|
|
assert outcome["actual_provider"] == "openai-codex"
|
|
assert outcome["actual_model"] == "gpt-5.6-terra"
|
|
assert written[-1]["provider_cooldowns"]["anthropic"]["until_epoch"] > router.time.time()
|
|
assert agent.message.startswith("FALLBACK USED")
|
|
|
|
|
|
def test_status_distinguishes_requested_route_from_actual_outcome(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "auto",
|
|
"last_decision": {
|
|
"provider": "anthropic",
|
|
"model": "claude-sonnet-5",
|
|
"effort": "medium",
|
|
"classifier": "jetson",
|
|
"actual_provider": "openai-codex",
|
|
"actual_model": "gpt-5.6-terra",
|
|
"fallback_used": True,
|
|
},
|
|
},
|
|
)
|
|
manager = type("Manager", (), {"_cli_ref": None})()
|
|
ctx = type("Context", (), {"_manager": manager})()
|
|
|
|
status = router._status_text(ctx)
|
|
|
|
assert "Last requested route: anthropic/claude-sonnet-5" in status
|
|
assert "Last actual outcome: fallback: openai-codex/gpt-5.6-terra" in status
|