662 lines
26 KiB
Python
662 lines
26 KiB
Python
"""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 pytest
|
|
import yaml
|
|
|
|
from testing.tests.test_hermes_coordinator_support import (
|
|
SCRIPT,
|
|
_base_config,
|
|
catalog_resolver,
|
|
coordinator,
|
|
routing,
|
|
)
|
|
from model_evaluation_evidence import metadata_fingerprint
|
|
|
|
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.6-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.6-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_holds_unreviewed_releases_and_preserves_lkg():
|
|
"""New IDs stay pending until evidence while outages retain known routes."""
|
|
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)
|
|
|
|
with pytest.raises(ValueError, match="selector 'auto'"):
|
|
catalog_resolver.resolve_model("codex", "auto", "high", current)
|
|
with pytest.raises(ValueError, match="selector 'terra'"):
|
|
catalog_resolver.resolve_model("codex", "terra", "medium", current)
|
|
assert current["providers"]["codex"]["candidates"]["gpt-5.7-sol"]["proposed_role"] is None
|
|
with pytest.raises(ValueError, match="selector 'sonnet'"):
|
|
catalog_resolver.resolve_model("claude", "sonnet", "high", current)
|
|
|
|
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 == {effort: "" for effort in routing.EFFORTS}
|
|
with pytest.raises(ValueError, match="auto-frontier"):
|
|
catalog_resolver.resolve_model("codex", "auto-frontier", "xhigh", catalog)
|
|
assert "gpt-6.2-high-only" in catalog["providers"]["codex"]["candidates"]
|
|
assert catalog["providers"]["codex"]["candidates"]["gpt-6.2-high-only"]["proposed_role"] == "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"] == ""
|
|
assert catalog["providers"]["codex"]["candidates"]["gpt-7-nova"]["proposed_role"] == "frontier"
|
|
|
|
|
|
def test_provider_metadata_cannot_self_attest_evaluation():
|
|
"""Only the local evidence store may add an evaluated capability role."""
|
|
models, metadata = routing.model_records([{
|
|
"id": "gpt-9-mystery", "evaluated_capability_role": "frontier",
|
|
"evaluation_provenance": "untrusted", "description": "Routine tasks",
|
|
}])
|
|
catalog = routing.build_routing_catalog(
|
|
routing.Catalog("openai-codex", models, True, True, "connected", metadata),
|
|
routing.Catalog("anthropic", [], True, True, "connected"),
|
|
)
|
|
assert "evaluated_capability_role" not in metadata["gpt-9-mystery"]
|
|
assert catalog["providers"]["codex"]["capability_pools"]["frontier"] == []
|
|
|
|
|
|
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"] == ""
|
|
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_separates_advanced_sol_from_frontier_astra():
|
|
"""Actual Codex records preserve Sol AUTO and expose Astra explicitly."""
|
|
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-5.6-sol",
|
|
"xhigh": "gpt-5.6-sol",
|
|
}
|
|
assert catalog_resolver.resolve_model("codex", "auto-frontier", "xhigh", catalog) == "gpt-6-astra"
|
|
assert catalog["providers"]["codex"]["capability_pools"] == {
|
|
"economy": ["gpt-5.6-luna"],
|
|
"balanced": ["gpt-5.6-terra"],
|
|
"advanced": ["gpt-5.6-sol"],
|
|
"frontier": ["gpt-6-astra"],
|
|
}
|
|
|
|
|
|
def test_live_default_does_not_promote_an_unverified_frontier_candidate():
|
|
"""A default flag alone does not bypass the bounded evidence gate."""
|
|
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"] == ""
|
|
with pytest.raises(ValueError, match="auto-frontier"):
|
|
catalog_resolver.resolve_model("codex", "auto-frontier", "xhigh", catalog)
|
|
|
|
|
|
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": "opus", "xhigh": "opus",
|
|
}
|
|
assert catalog_resolver.resolve_model("claude", "auto-frontier", "xhigh", catalog) == "fable"
|
|
assert catalog["providers"]["claude"]["candidates"]["haiku"]["eligible"] is False
|
|
|
|
|
|
def test_live_unavailable_capability_does_not_promote_to_a_higher_role():
|
|
"""A fresh disabled economy model is an availability failure, not a tier gap."""
|
|
metadata = {
|
|
"gpt-5.6-luna": {"enabled": False},
|
|
"gpt-5.6-terra": {"supported_reasoning_efforts": ["low", "medium"]},
|
|
"gpt-5.6-sol": {"supported_reasoning_efforts": ["low", "medium", "high", "xhigh"]},
|
|
}
|
|
catalog = routing.build_routing_catalog(
|
|
routing.Catalog("openai-codex", list(metadata), True, True, "connected", metadata),
|
|
routing.Catalog("anthropic", [], True, True, "connected"),
|
|
)
|
|
assert catalog["providers"]["codex"]["resolved"]["low"] == ""
|
|
with pytest.raises(ValueError, match="selector 'auto'"):
|
|
catalog_resolver.resolve_model("codex", "auto", "low", catalog)
|
|
|
|
|
|
def test_legacy_aliases_are_exact_and_keep_fable_frontier():
|
|
"""Compatibility cannot admit new names merely because they contain a family word."""
|
|
unknown, _ = routing._select_tier_model(
|
|
"codex", ["gpt-9-luna-experimental"], {}, "economy", "low", "",
|
|
legacy_compat=True,
|
|
)
|
|
fable, _ = routing._select_tier_model(
|
|
"claude", ["claude-fable-5"], {}, "frontier", "xhigh", "",
|
|
legacy_compat=True,
|
|
)
|
|
assert unknown == ""
|
|
assert fable == "claude-fable-5"
|
|
|
|
|
|
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_frontier_selector_never_demotes_to_an_advanced_model():
|
|
"""A missing frontier route is explicit while advanced still resolves Sol."""
|
|
catalog = routing.build_routing_catalog(
|
|
routing.Catalog("openai-codex", ["gpt-5.6-sol"], True, True, "connected"),
|
|
routing.Catalog("anthropic", [], True, True, "connected"),
|
|
)
|
|
assert catalog_resolver.resolve_model("codex", "auto-advanced", "xhigh", catalog) == "gpt-5.6-sol"
|
|
with pytest.raises(ValueError, match="auto-frontier"):
|
|
catalog_resolver.resolve_model("codex", "auto-frontier", "xhigh", catalog)
|
|
|
|
|
|
def test_verified_evaluation_routes_an_unfamiliar_model_without_name_rules():
|
|
"""Only a current, positive evaluation may classify an ambiguous live model."""
|
|
catalog = routing.build_routing_catalog(
|
|
routing.Catalog("openai-codex", ["gpt-9-mystery"], True, True, "connected"),
|
|
routing.Catalog("anthropic", [], True, True, "connected"),
|
|
evaluations={"evaluations": {"codex": {"gpt-9-mystery": {
|
|
"proposed_role": "frontier", "result": "pass", "role_fit": "verified",
|
|
"eval_version": "capability-v1", "metadata_fingerprint": metadata_fingerprint("gpt-9-mystery", {}),
|
|
}}}},
|
|
)
|
|
assert catalog_resolver.resolve_model("codex", "auto-frontier", "high", catalog) == "gpt-9-mystery"
|
|
assert catalog["providers"]["codex"]["candidates"]["gpt-9-mystery"]["reason"] == "bounded-representative-eval-v1"
|
|
|
|
|
|
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"] == ""
|
|
|
|
|
|
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")
|