"""Unit tests for Hermes coordinator model routing and profile generation.""" from __future__ import annotations import importlib.util import json import subprocess import sys import tomllib from pathlib import Path import yaml SCRIPT = ( Path(__file__).parents[2] / "services/hermes/scripts/hermes_coordinator.py" ) sys.path.insert(0, str(SCRIPT.parent)) routing = importlib.import_module("hermes_model_routing") catalog_resolver = importlib.import_module("routing_catalog") SPEC = importlib.util.spec_from_file_location("hermes_coordinator", SCRIPT) assert SPEC and SPEC.loader coordinator = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = coordinator SPEC.loader.exec_module(coordinator) def _base_config() -> dict: """Return the minimal coordinator configuration used by routing tests.""" return { "model": { "provider": "openai-codex", "default": "gpt-5.6-terra", "model": "gpt-5.6-terra", }, "fallback_providers": [ {"provider": "anthropic", "model": "claude-opus-5"}, dict(routing.LOCAL_FALLBACK), ], "toolsets": ["kanban"], "terminal": {"cwd": "/opt/data/workspace"}, } def test_model_version_and_quality_selection_handle_new_and_small_models(): assert routing.model_version("claude-3-5-sonnet-20241022") == (3, 5) assert routing.model_version("gpt-5.7-terra") == (5, 7) codex = ["gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.7-luna", "gpt-5.3-codex-spark"] assert routing.choose_codex_model(codex) == "gpt-5.6-sol" assert routing.choose_codex_model(codex, balanced=True) == "gpt-5.6-terra" assert routing.choose_codex_model(codex + ["gpt-5.7-terra"]) == "gpt-5.7-terra" claude = ["claude-opus-4.8", "claude-haiku-5", "claude-sonnet-5"] assert routing.choose_claude_model(claude) == "claude-sonnet-5" assert routing.choose_codex_for_effort(codex, "low") == "gpt-5.7-luna" assert routing.choose_codex_for_effort(codex, "medium") == "gpt-5.6-terra" assert routing.choose_codex_for_effort(codex, "xhigh") == "gpt-5.6-sol" assert routing.choose_claude_for_effort(claude, "low") == "claude-haiku-5" assert routing.choose_claude_for_effort(claude, "medium") == "claude-sonnet-5" assert routing.choose_claude_for_effort(claude, "xhigh") == "claude-opus-4.8" def test_empty_catalog_retains_current_models(): assert routing.choose_codex_model([], "gpt-5.6-terra") == "gpt-5.6-terra" assert routing.choose_claude_model([], "claude-opus-5") == "claude-opus-5" def test_dynamic_catalog_resolves_new_models_and_preserves_last_known_good(): """Stable Switchyard aliases follow live releases and survive catalog outages.""" codex = routing.Catalog( "openai-codex", ["gpt-5.7-luna", "gpt-5.7-terra", "gpt-5.7-sol"], True, True, "connected", ) claude = routing.Catalog( "anthropic", ["claude-haiku-5", "claude-sonnet-6", "claude-opus-6"], True, True, "connected", ) current = routing.build_routing_catalog(codex, claude) assert catalog_resolver.resolve_model("codex", "auto", "high", current) == "gpt-5.7-sol" assert catalog_resolver.resolve_model("codex", "terra", "medium", current) == "gpt-5.7-terra" assert catalog_resolver.resolve_model("claude", "auto", "xhigh", current) == "claude-opus-6" assert catalog_resolver.resolve_model("claude", "sonnet", "high", current) == "claude-sonnet-6" unavailable = routing.Catalog("openai-codex", [], False, True, "degraded") preserved = routing.build_routing_catalog(unavailable, unavailable, current) assert preserved["providers"]["codex"]["resolved"] == current["providers"]["codex"]["resolved"] assert preserved["providers"]["claude"]["tiers"] == current["providers"]["claude"]["tiers"] def test_live_catalog_never_routes_to_a_removed_model_tier(): """A live provider catalog must replace a retired tier with a live model.""" previous = { "providers": { "codex": { "models": ["gpt-5.6-luna", "gpt-5.6-terra", "gpt-5.6-sol"], "resolved": { "low": "gpt-5.6-luna", "medium": "gpt-5.6-terra", "high": "gpt-5.6-sol", "xhigh": "gpt-5.6-sol", }, "tiers": { "luna": "gpt-5.6-luna", "terra": "gpt-5.6-terra", "sol": "gpt-5.6-sol", }, } } } codex = routing.Catalog( "openai-codex", ["gpt-5.7-terra", "gpt-5.7-sol"], True, True, "connected" ) claude = routing.Catalog("anthropic", ["claude-sonnet-6"], True, True, "connected") current = routing.build_routing_catalog(codex, claude, previous) assert current["providers"]["codex"]["tiers"]["luna"] == "gpt-5.7-terra" assert current["providers"]["codex"]["tiers"]["luna"] in codex.models assert current["providers"]["claude"]["tiers"]["opus"] == "claude-sonnet-6" def test_switchyard_targets_have_unique_upstream_identities(): """Switchyard drops duplicate client/model pairs, so reject them in Git.""" manifest = yaml.safe_load( (SCRIPT.parents[1] / "switchyard-configmap.yaml").read_text(encoding="utf-8") ) config = tomllib.loads(manifest["data"]["routes.toml"]) target_names = set(config["targets"]) identities: set[tuple[str, str]] = set() for target in config["targets"].values(): identity = (target["llm_client"], target["id"]) assert identity not in identities identities.add(identity) for route in config["routes"].values(): for target_name in route.get("targets", []): assert target_name in target_names if route.get("target"): assert route["target"] in target_names def test_worker_alias_resolves_before_cli_launch(): document = { "providers": { "codex": { "resolved": {"high": "gpt-5.8-sol"}, "tiers": {"sol": "gpt-5.8-sol"}, } } } assert ( catalog_resolver.resolve_worker_route("worker/codex/auto/high", document) == "worker/codex/gpt-5.8-sol/high" ) def test_codex_cli_login_counts_as_connected_runtime(monkeypatch): """AUTO routing must recognize the authenticated app-server CLI lane.""" monkeypatch.setattr(routing.shutil, "which", lambda name: "/usr/bin/codex") monkeypatch.setattr( routing.subprocess, "run", lambda *args, **kwargs: subprocess.CompletedProcess( args[0], 0, stdout="Logged in using ChatGPT\n", stderr="", ), ) assert routing.codex_cli_authenticated() is True def test_configure_routes_keeps_every_profile_on_switchyard(tmp_path: Path): (tmp_path / "config.yaml").write_text( yaml.safe_dump(_base_config()), encoding="utf-8" ) (tmp_path / ".env").write_text( "API_SERVER_KEY=keep-root-only\n" "CLAUDE_CODE_OAUTH_TOKEN=claude-secret\n" "GITEA_TOKEN=gitea-secret\n" "GIT_ASKPASS=/opt/coordinator/gitea_askpass.sh\n", encoding="utf-8", ) codex = routing.Catalog( "openai-codex", ["gpt-5.6-sol", "gpt-5.6-terra"], True, True, "connected" ) claude = routing.Catalog( "anthropic", ["claude-sonnet-5", "claude-opus-5"], True, True, "connected" ) routes = routing.configure_routes(tmp_path, codex, claude) root = yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8")) codex_profile = yaml.safe_load( (tmp_path / "profiles/codex-high/config.yaml").read_text(encoding="utf-8") ) codex_xhigh_profile = yaml.safe_load( (tmp_path / "profiles/codex-xhigh/config.yaml").read_text(encoding="utf-8") ) claude_profile = yaml.safe_load( (tmp_path / "profiles/claude-high/config.yaml").read_text(encoding="utf-8") ) assert root["model"] == { "provider": "atlas-switchyard", "default": "atlas/auto/maximum", "model": "atlas/auto/maximum", } assert root["fallback_providers"] == [] assert root["toolsets"] == ["kanban"] assert codex_profile["model"]["model"] == "atlas/manual/codex/sol" assert codex_profile["model"]["provider"] == "atlas-switchyard" assert claude_profile["model"]["model"] == "atlas/manual/claude/sonnet" assert claude_profile["model"]["provider"] == "atlas-switchyard" assert codex_profile["fallback_providers"] == [] assert claude_profile["fallback_providers"] == [] assert codex_profile["toolsets"] == [] assert codex_profile["agent"]["reasoning_effort"] == "high" assert codex_xhigh_profile["fallback_providers"] == [] assert routes["codex-xhigh"] == ["atlas/manual/codex/sol"] assert routes["claude-xhigh"] == ["atlas/manual/claude/opus"] assert routes["synthesis-xhigh"] == ["atlas/auto/maximum"] assert routes["coordinator"] == ["atlas/auto/maximum"] assert all( "max" not in profile_name for profile_name in routes if profile_name != "catalog" ) profile_env = (tmp_path / "profiles/codex-high/.env").read_text(encoding="utf-8") assert "CLAUDE_CODE_OAUTH_TOKEN=claude-secret" in profile_env assert "GITEA_TOKEN=gitea-secret" in profile_env assert "API_SERVER_KEY" not in profile_env assert "claude-secret" not in json.dumps(routes) assert (tmp_path / "profiles/codex-high/.env").stat().st_mode & 0o777 == 0o600 def test_degraded_catalog_does_not_replace_switchyard_authority(tmp_path: Path): base = _base_config() base["model"]["model"] = base["model"]["default"] = "gpt-5.6-sol" (tmp_path / "config.yaml").write_text(yaml.safe_dump(base), encoding="utf-8") (tmp_path / ".env").write_text("", encoding="utf-8") codex = routing.Catalog("openai-codex", ["gpt-5.4"], False, True, "degraded") claude = routing.Catalog("anthropic", ["claude-haiku-4.5"], False, True, "degraded") routing.configure_routes(tmp_path, codex, claude) current = yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8")) assert current["model"]["model"] == "atlas/auto/maximum" assert current["model"]["provider"] == "atlas-switchyard" assert current["fallback_providers"] == [] def test_degraded_refresh_preserves_switchyard_worker_preference(tmp_path: Path): """A catalog outage must not bypass a managed worker's Switchyard route.""" (tmp_path / "config.yaml").write_text( yaml.safe_dump(_base_config()), encoding="utf-8" ) (tmp_path / ".env").write_text("", encoding="utf-8") live_codex = routing.Catalog( "openai-codex", ["gpt-5.6-sol", "gpt-5.6-terra"], True, True, "connected" ) live_claude = routing.Catalog( "anthropic", ["claude-opus-5"], True, True, "connected" ) routing.configure_routes(tmp_path, live_codex, live_claude) degraded_codex = routing.Catalog( "openai-codex", ["gpt-5.4"], False, True, "degraded" ) routing.configure_routes(tmp_path, degraded_codex, live_claude) worker = yaml.safe_load( (tmp_path / "profiles/codex-high/config.yaml").read_text(encoding="utf-8") ) assert worker["model"]["model"] == "atlas/manual/codex/sol" assert worker["model"]["provider"] == "atlas-switchyard" def test_refresh_writes_non_secret_routing_status(tmp_path: Path, monkeypatch): (tmp_path / "config.yaml").write_text( yaml.safe_dump(_base_config()), encoding="utf-8" ) (tmp_path / ".env").write_text("GITEA_TOKEN=do-not-report\n", encoding="utf-8") codex = routing.Catalog("openai-codex", ["gpt-5.6-terra"], True, True, "connected") claude = routing.Catalog("anthropic", ["claude-opus-5"], True, True, "connected") monkeypatch.setattr(coordinator, "discover_codex_models", lambda: codex) monkeypatch.setattr(coordinator, "discover_claude_models", lambda: claude) monkeypatch.setattr(coordinator, "bootstrap_cassandra", lambda root: None) monkeypatch.setattr(coordinator, "sync_cassandra_repo", lambda env: "ready") status = coordinator.refresh_once(tmp_path) status_path = tmp_path / "workspace/coordinator/model-routing.json" assert status_path.is_file() assert status["projects"]["cassandra"]["state"] == "ready" assert "do-not-report" not in status_path.read_text(encoding="utf-8") def test_cassandra_workspace_uses_valid_configured_worktree( tmp_path: Path, monkeypatch ): """The coordinator selects an explicit Git worktree but rejects stale paths.""" active = tmp_path / "cassandra-v69" active.mkdir() (active / ".git").write_text("gitdir: elsewhere\n", encoding="utf-8") monkeypatch.setenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", str(active)) assert coordinator.cassandra_workspace() == active (active / ".git").unlink() assert coordinator.cassandra_workspace() == coordinator.CASSANDRA_BASE_PATH def test_migrate_open_cassandra_tasks_preserves_running_and_done_tasks(): """A project switch moves queued work without relocating active evidence.""" class Task: def __init__(self, task_id: str, status: str): self.id = task_id self.status = status self.workspace_kind = "dir" self.workspace_path = str(coordinator.CASSANDRA_BASE_PATH) class Closing: def __enter__(self): return object() def __exit__(self, *_args): return None class FakeKanban: tasks = [Task("queued", "todo"), Task("active", "running"), Task("done", "done")] moved: list[tuple[str, str]] = [] @staticmethod def connect_closing(*, board: str): assert board == "cassandra" return Closing() @classmethod def list_tasks(cls, _connection, *, include_archived: bool): assert include_archived is False return cls.tasks @classmethod def set_workspace_path(cls, _connection, task_id: str, path: Path): cls.moved.append((task_id, str(path))) active = Path("/opt/data/workspace/projects/cassandra-hermes-v69") coordinator._migrate_open_cassandra_tasks(FakeKanban, active) assert FakeKanban.moved == [("queued", str(active))]