atlas-iac/testing/tests/test_hermes_cli_routing.py

"""Switchyard routing and provider-health behavior."""

from __future__ import annotations

from testing.tests.test_hermes_cli_support import (
    Path,
    _SwitchyardResponse,
    json,
    lanes,
)


def test_auto_lane_uses_switchyard_worker_decision():
    observed = {}

    def route(request, timeout):
        observed["payload"] = json.loads(request.data)
        observed["timeout"] = timeout
        return _SwitchyardResponse("worker/claude/claude-opus-5/xhigh")

    route = lanes.select_route(
        "Perform the security review.",
        "cli-auto",
        open_request=route,
    )

    assert route.provider == "claude"
    assert route.model == "claude-opus-5"
    assert route.effort == "xhigh"
    assert route.classifier == "switchyard-classifier"
    assert observed["payload"]["model"] == "atlas/worker/auto/maximum"
    assert observed["timeout"] == 60


def test_manual_lane_is_still_enforced_by_switchyard():
    observed = {}

    def route(request, timeout):
        observed["payload"] = json.loads(request.data)
        return _SwitchyardResponse(
            "worker/claude/claude-sonnet-5/high", "manual route"
        )

    route = lanes.select_route(
        "Implement it.",
        "cli-claude-high",
        open_request=route,
    )

    assert route.provider == "claude"
    assert route.effort == "high"
    assert route.classifier == "switchyard-manual"
    assert route.reason == "manual route"
    assert observed["payload"]["model"] == "atlas/worker/manual/claude/high"


def test_cross_provider_retry_passes_failed_provider_to_switchyard():
    observed = {}

    def route(request, timeout):
        observed["payload"] = json.loads(request.data)
        return _SwitchyardResponse("worker/claude/claude-sonnet-5/high")

    route = lanes.select_route(
        "Retry after capacity exhaustion.",
        "cli-auto",
        exclude_provider="codex",
        open_request=route,
    )

    assert route.provider == "claude"
    assert route.classifier == "switchyard-classifier"
    content = observed["payload"]["messages"][0]["content"]
    assert "codex provider failed or exhausted capacity" in content


def test_classifier_cannot_select_a_freshly_excluded_provider():
    payloads = []

    def route(request, timeout):
        payload = json.loads(request.data)
        payloads.append(payload)
        if len(payloads) == 1:
            return _SwitchyardResponse("worker/claude/opus/xhigh")
        return _SwitchyardResponse("worker/codex/sol/xhigh", "healthy route")

    selected = lanes.select_route(
        "Perform a consequential review.",
        "cli-auto",
        exclude_provider="claude",
        exclude_reason="is unavailable according to fresh native health",
        open_request=route,
    )

    assert selected.provider == "codex"
    assert selected.effort == "xhigh"
    assert selected.classifier == "switchyard-classifier-health-guard"
    assert payloads[0]["model"] == "atlas/worker/auto/maximum"
    assert payloads[1]["model"] == "atlas/worker/manual/codex/advanced/xhigh"
    assert "claude provider is unavailable" in payloads[0]["messages"][0]["content"]


def test_fresh_native_health_excludes_only_one_proven_down_provider(
    tmp_path: Path, monkeypatch
):
    paths = {
        "codex": tmp_path / "codex.json",
        "claude": tmp_path / "claude.json",
    }
    paths["codex"].write_text('{"state":"available"}\n', encoding="utf-8")
    paths["claude"].write_text('{"state":"unavailable"}\n', encoding="utf-8")
    monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", paths)

    now = max(path.stat().st_mtime for path in paths.values())
    assert lanes.fresh_unavailable_provider(now=now) == "claude"

    paths["codex"].write_text('{"state":"unavailable"}\n', encoding="utf-8")
    now = max(path.stat().st_mtime for path in paths.values())
    assert lanes.fresh_unavailable_provider(now=now) is None


def test_explicit_auth_failure_survives_restart_health_gap(tmp_path: Path, monkeypatch):
    paths = {
        "codex": tmp_path / "codex.json",
        "claude": tmp_path / "claude.json",
    }
    paths["codex"].write_text(
        '{"state":"available","authenticated":true}\n', encoding="utf-8"
    )
    paths["claude"].write_text(
        '{"state":"unavailable","authenticated":false}\n', encoding="utf-8"
    )
    monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", paths)

    stale_during_rollout = paths["claude"].stat().st_mtime + 10 * 60
    assert lanes.fresh_unavailable_provider(now=stale_during_rollout) == "claude"

    paths["claude"].write_text(
        '{"state":"unavailable","authenticated":true}\n', encoding="utf-8"
    )
    transient_stale = paths["claude"].stat().st_mtime + 10 * 60
    assert lanes.fresh_unavailable_provider(now=transient_stale) is None


def test_worker_environment_preserves_vault_backed_cli_homes(monkeypatch):
    """Kanban workers must not fall back to credential-free persistent homes."""
    monkeypatch.setenv("CODEX_HOME", "/runtime-access/codex")
    monkeypatch.setenv("CLAUDE_CONFIG_DIR", "/runtime-access/claude")

    env = lanes._base_env()

    assert env["HOME"] == str(lanes.DATA_ROOT / "home")
    assert env["CODEX_HOME"] == "/runtime-access/codex"
    assert env["CLAUDE_CONFIG_DIR"] == "/runtime-access/claude"
    assert env["GIT_TERMINAL_PROMPT"] == "0"