"""Unit tests for Hermes coordinator model routing and profile generation.""" from __future__ import annotations import json import subprocess import tomllib from pathlib import Path import yaml from testing.tests.test_hermes_coordinator_support import ( SCRIPT, _base_config, catalog_resolver, coordinator, routing, ) 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", ) stale_profile = tmp_path / "profiles/codex-high" stale_profile.mkdir(parents=True) (stale_profile / ".env").write_text( "CLAUDE_CODE_OAUTH_TOKEN=stale-claude\n" "GITEA_TOKEN=stale-gitea\n" "HERMES_IMAGE_BROKER_KEY=stale-relay\n" "USER_SETTING=preserve\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" not in profile_env assert "GITEA_TOKEN" not in profile_env assert "HERMES_IMAGE_BROKER_KEY" not in profile_env assert "API_SERVER_KEY" not in profile_env assert "USER_SETTING=preserve" 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_state", lambda root: {"state": "ready"} ) 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 status["projects"]["cassandra"]["board_state"] == {"state": "ready"} assert "do-not-report" not in status_path.read_text(encoding="utf-8")