361 lines
14 KiB
Python
361 lines
14 KiB
Python
"""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))]
|