atlas-iac/testing/tests/test_hermes_auto_router.py

"""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 target")


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 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