"""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.6-sol" claude = ["claude-opus-4.8", "claude-haiku-5", "claude-sonnet-5"] assert routing.choose_claude_model(claude) == "claude-opus-4.8" 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_dynamic_catalog_uses_metadata_for_unfamiliar_models_and_efforts(): """Future IDs route by declared capability, price, and effort support.""" codex = routing.Catalog( "openai-codex", [ "gpt-6.2-orbit", "gpt-6.2-balanced", "gpt-6.2-quick", "gpt-6.2-high-only", "gpt-6.2-xhigh", ], True, True, "connected", { "gpt-6.2-orbit": { "description": "A strongest upgrade for difficult, complex tasks", "upgrade": "strongest", "supported_reasoning_efforts": ["low", "medium", "high", "xhigh"], "cost": 10, }, "gpt-6.2-balanced": { "capability_tier": "balanced", "description": "General purpose model for everyday tasks", "supported_reasoning_efforts": ["low", "medium", "high", "xhigh"], "cost": 2, }, "gpt-6.2-quick": { "capability_tier": "economy", "description": "Fast and affordable general purpose model", "supported_reasoning_efforts": ["low", "medium"], "cost": 1, }, "gpt-6.2-high-only": { "capability_tier": "advanced", "supported_reasoning_efforts": ["high"], "cost": 1, }, "gpt-6.2-xhigh": { "capability_tier": "advanced", "supported_reasoning_efforts": ["xhigh"], "cost": 2, }, }, ) claude = routing.Catalog("anthropic", [], True, True, "connected") catalog = routing.build_routing_catalog(codex, claude) resolved = catalog["providers"]["codex"]["resolved"] assert resolved["xhigh"] == "gpt-6.2-orbit" assert resolved["low"] == "gpt-6.2-quick" assert resolved["medium"] == "gpt-6.2-balanced" assert resolved["high"] == "gpt-6.2-orbit" assert "gpt-6.2-high-only" in catalog["providers"]["codex"]["candidates"] assert catalog["providers"]["codex"]["candidates"]["gpt-6.2-high-only"]["tier"] == "advanced" def test_catalog_accepts_codex_effort_records_and_generalist_image_input(): """Codex app-server effort records and multimodal generalists stay routable.""" codex = routing.Catalog( "openai-codex", ["gpt-7-nova"], True, True, "connected-app-server", { "gpt-7-nova": { "description": "Strongest general purpose model with text and image inputs", "supportedReasoningEfforts": [ {"reasoningEffort": "low"}, {"reasoningEffort": "medium"}, {"reasoningEffort": "high"}, {"reasoningEffort": "xhigh"}, ], } }, ) catalog = routing.build_routing_catalog( codex, routing.Catalog("anthropic", [], True, True, "connected") ) assert catalog["providers"]["codex"]["resolved"]["xhigh"] == "gpt-7-nova" def test_dynamic_catalog_excludes_hidden_internal_and_specialist_records(): """Provider listings must not make private or modality-only products routable.""" models = [ "gpt-6.2-hidden", "gpt-6.2-internal", "gpt-6.2-image", "gpt-6.2-public", ] metadata = { "gpt-6.2-hidden": { "hidden": True, "capability_tier": "advanced", }, "gpt-6.2-internal": { "visibility": "internal", "description": "Internal evaluation model", "capability_tier": "advanced", }, "gpt-6.2-image": { "description": "Image generation specialist", "capability_tier": "advanced", }, "gpt-6.2-public": { "description": "Strongest general purpose model", "capability_tier": "advanced", }, } codex = routing.Catalog( "openai-codex", models, True, True, "connected", metadata ) claude = routing.Catalog("anthropic", [], True, True, "connected") catalog = routing.build_routing_catalog(codex, claude) provider = catalog["providers"]["codex"] assert provider["resolved"]["xhigh"] == "gpt-6.2-public" assert provider["candidates"]["gpt-6.2-hidden"]["eligible"] is False assert provider["candidates"]["gpt-6.2-internal"]["eligible"] is False assert provider["candidates"]["gpt-6.2-image"]["eligible"] is False def test_live_codex_metadata_selects_luna_terra_and_astra_by_effort(): """Actual Codex model/list fields route generic effort without name guesses.""" fixture = Path(__file__).parents[1] / "fixtures/hermes/codex-0.154-visible-models.json" records = json.loads(fixture.read_text(encoding="utf-8")) models, metadata = routing.model_records(records) catalog = routing.build_routing_catalog( routing.Catalog("openai-codex", models, True, True, "connected", metadata), routing.Catalog("anthropic", [], True, True, "connected"), ) assert catalog["providers"]["codex"]["resolved"] == { "low": "gpt-5.6-luna", "medium": "gpt-5.6-terra", "high": "gpt-6-astra", "xhigh": "gpt-6-astra", } def test_live_default_flagship_replaces_previous_advanced_route(): """A provider's current default wins over an equally strongest old model.""" previous = { "providers": {"codex": {"models": ["gpt-astra"], "resolved": { effort: "gpt-astra" for effort in routing.EFFORTS }, "model_metadata": {"gpt-astra": { "description": "Most capable for complex work", "isDefault": False, }}}} } metadata = { "gpt-astra": {"description": "Most capable for complex work", "isDefault": False}, "gpt-nova": {"description": "Most capable for complex work", "isDefault": True}, } catalog = routing.build_routing_catalog( routing.Catalog("openai-codex", list(metadata), True, True, "connected", metadata), routing.Catalog("anthropic", [], True, True, "connected"), previous, ) assert catalog["providers"]["codex"]["resolved"]["xhigh"] == "gpt-nova" def test_live_claude_effort_gap_uses_nearest_higher_declared_tier(): """An effortless economy alias cannot silently take a stronger route.""" metadata = { "haiku": {"description": "Fastest for quick answers", "supportsEffort": False}, "sonnet": { "description": "Efficient for routine tasks", "supportsEffort": True, "supportedEffortLevels": ["low", "medium", "high", "xhigh"], }, "opus": { "description": "Best for everyday, complex tasks", "supportsEffort": True, "supportedEffortLevels": ["low", "medium", "high", "xhigh"], }, "fable": { "description": "Most capable for your hardest and longest-running tasks", "supportsEffort": True, "supportedEffortLevels": ["low", "medium", "high", "xhigh"], }, } catalog = routing.build_routing_catalog( routing.Catalog("openai-codex", [], True, True, "connected"), routing.Catalog("anthropic", list(metadata), True, True, "connected", metadata), ) assert catalog["providers"]["claude"]["resolved"] == { "low": "sonnet", "medium": "sonnet", "high": "fable", "xhigh": "fable", } assert catalog["providers"]["claude"]["candidates"]["haiku"]["eligible"] is False def test_ambiguous_live_catalog_exposes_an_unresolved_route(): """An unfamiliar live model cannot revive a retired LKG model.""" previous = { "providers": { "codex": { "models": ["gpt-5.6-sol"], "resolved": {effort: "gpt-5.6-sol" for effort in routing.EFFORTS}, "tiers": {"advanced": "gpt-5.6-sol"}, "model_metadata": {}, } } } codex = routing.Catalog( "openai-codex", ["gpt-7.0-mystery"], True, True, "connected" ) claude = routing.Catalog("anthropic", [], False, True, "degraded") catalog = routing.build_routing_catalog(codex, claude, previous) assert catalog["providers"]["codex"]["resolved"]["xhigh"] == "" assert catalog["providers"]["codex"]["candidates"]["gpt-7.0-mystery"]["tier"] is None def test_live_removal_clears_legacy_selector_instead_of_mapping_to_new_family(): """A live authoritative list cannot revive removed legacy selectors.""" previous = { "providers": { "codex": { "models": ["gpt-5.6-sol"], "resolved": {effort: "gpt-5.6-sol" for effort in routing.EFFORTS}, "tiers": {"sol": "gpt-5.6-sol", "advanced": "gpt-5.6-sol"}, } } } codex = routing.Catalog( "openai-codex", ["gpt-6.2-astra"], True, True, "connected", {"gpt-6.2-astra": {"capability_tier": "advanced"}}, ) claude = routing.Catalog("anthropic", [], False, True, "degraded") catalog = routing.build_routing_catalog(codex, claude, previous) assert catalog["providers"]["codex"]["tiers"]["sol"] == "" assert catalog["providers"]["codex"]["resolved"]["xhigh"] == "gpt-6.2-astra" def test_degraded_catalog_preserves_lkg_after_live_removal(): """A provider outage retains the prior route and does not treat it as removal.""" previous = { "providers": { "codex": { "models": ["gpt-5.6-sol"], "resolved": {effort: "gpt-5.6-sol" for effort in routing.EFFORTS}, "tiers": {"sol": "gpt-5.6-sol"}, "model_metadata": {}, } } } codex = routing.Catalog("openai-codex", [], False, True, "degraded") claude = routing.Catalog("anthropic", [], False, True, "degraded") catalog = routing.build_routing_catalog(codex, claude, previous) assert catalog["providers"]["codex"]["resolved"]["xhigh"] == "gpt-5.6-sol" assert catalog["providers"]["codex"]["tiers"]["sol"] == "gpt-5.6-sol" def test_explicit_legacy_selector_never_maps_to_astra(): """The old sol selector is exact and cannot silently follow a new family.""" catalog = { "providers": { "codex": { "tiers": {"sol": "gpt-6.2-astra"}, "resolved": {"xhigh": "gpt-6.2-astra"}, } } } assert catalog_resolver.resolve_model("codex", "sol", "xhigh", catalog) == ( "gpt-5.6-sol" ) 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"] == "" assert current["providers"]["claude"]["tiers"]["opus"] == "" 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/auto/high" assert codex_profile["model"]["provider"] == "atlas-switchyard" assert claude_profile["model"]["model"] == "atlas/manual/claude/auto/high" 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/auto/xhigh"] assert routes["claude-xhigh"] == ["atlas/manual/claude/auto/xhigh"] 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/auto/high" 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")