atlas-iac/services/hermes/scripts/hermes_model_routing.py

#!/usr/bin/env python3
"""Keep Hermes agent provider catalogs and managed profiles current."""

from __future__ import annotations

import copy
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable

import yaml


CODEX_BASELINE = "gpt-5.6-terra"
CLAUDE_BASELINE = "claude-opus-5"
EFFORTS = ("low", "medium", "high", "xhigh")
ATLAS_FALLBACK = {
    "provider": "custom",
    "model": "gpt-oss:20b",
    "base_url": "http://hermes-model-gate.hermes.svc.cluster.local:11434/v1",
    "api_key": "ollama",
}
# Backwards-compatible name used by the focused unit tests and status tooling.
LOCAL_FALLBACK = ATLAS_FALLBACK
MANAGED_ENV_KEYS = {
    "CLAUDE_CODE_OAUTH_TOKEN",
    "GITEA_TOKEN",
    "GITEA_USERNAME",
    "GIT_ASKPASS",
    "GIT_TERMINAL_PROMPT",
}


@dataclass(frozen=True)
class Catalog:
    """Non-secret provider discovery result."""

    provider: str
    models: list[str]
    live: bool
    connected: bool
    state: str


def _unique_models(models: Iterable[str]) -> list[str]:
    """Return normalized model IDs without changing provider order."""
    seen: set[str] = set()
    result: list[str] = []
    for model in models:
        value = str(model or "").strip()
        if value and value.lower() not in seen:
            result.append(value)
            seen.add(value.lower())
    return result


def model_version(model: str) -> tuple[int, ...]:
    """Extract a sortable model version while ignoring dated aliases."""
    value = re.sub(r"-\d{8}$", "", model.lower())
    groups = re.findall(r"\d+(?:\.\d+)*", value)
    parts: list[int] = []
    for group in groups:
        parts.extend(int(piece) for piece in group.split("."))
    return tuple(parts) or (0,)


def _codex_quality(model: str) -> int:
    """Rank known Codex capability tiers for delegated implementation work."""
    value = model.lower()
    if "sol" in value:
        return 60
    if "terra" in value:
        return 50
    if "codex" in value and "spark" not in value:
        return 45
    if re.fullmatch(r"gpt-\d+(?:\.\d+)*", value):
        return 40
    if "luna" in value:
        return 20
    if "mini" in value or "nano" in value:
        return 10
    if "spark" in value:
        return 5
    return 0


def choose_codex_model(
    models: Iterable[str], current: str = CODEX_BASELINE, *, balanced: bool = False
) -> str:
    """Choose a current full Codex model, retaining current when unknown."""
    candidates = [m for m in _unique_models(models) if m.lower().startswith("gpt-")]
    full = [m for m in candidates if _codex_quality(m) >= 40]
    pool = full or candidates
    if not pool:
        return current
    newest = max(model_version(model) for model in pool)
    latest = [model for model in pool if model_version(model) == newest]
    if balanced:
        terra = [model for model in latest if "terra" in model.lower()]
        if terra:
            return max(terra, key=lambda model: (_codex_quality(model), model))
    return max(latest, key=lambda model: (_codex_quality(model), model))


def _choose_by_hints(
    models: Iterable[str], hints: tuple[str, ...], current: str, prefix: str
) -> str:
    """Pick the newest model in the first available capability class."""
    candidates = [
        model
        for model in _unique_models(models)
        if model.lower().startswith(prefix)
    ]
    if not candidates:
        return current
    for hint in hints:
        tier = [model for model in candidates if hint in model.lower()]
        if tier:
            return max(tier, key=lambda model: (model_version(model), model))
    return max(candidates, key=lambda model: (model_version(model), model))


def choose_codex_for_effort(
    models: Iterable[str], effort: str, current: str = CODEX_BASELINE
) -> str:
    """Choose the account-visible Codex tier for an allowed effort level."""
    hints = {
        "low": ("luna", "mini", "spark", "terra", "sol", "codex"),
        "medium": ("terra", "codex", "sol", "luna", "mini", "spark"),
        "high": ("sol", "codex", "terra", "luna", "mini", "spark"),
        "xhigh": ("sol", "codex", "terra", "luna", "mini", "spark"),
    }
    if effort not in EFFORTS:
        raise ValueError(f"unsupported effort: {effort}")
    return _choose_by_hints(models, hints[effort], current, "gpt-")


def _claude_quality(model: str) -> int:
    """Rank known Claude capability tiers for architecture and review work."""
    value = model.lower()
    if "opus" in value:
        return 40
    if "sonnet" in value:
        return 30
    if "fable" in value:
        return 20
    if "haiku" in value:
        return 10
    return 0


def choose_claude_model(models: Iterable[str], current: str = CLAUDE_BASELINE) -> str:
    """Choose the newest full Claude reasoning model visible to the account."""
    candidates = [m for m in _unique_models(models) if m.lower().startswith("claude-")]
    full = [m for m in candidates if _claude_quality(m) >= 30]
    pool = full or candidates
    if not pool:
        return current
    return max(
        pool,
        key=lambda model: (model_version(model), _claude_quality(model), model),
    )


def choose_claude_for_effort(
    models: Iterable[str], effort: str, current: str = CLAUDE_BASELINE
) -> str:
    """Choose the account-visible Claude tier for an allowed effort level."""
    hints = {
        "low": ("haiku", "fable", "sonnet", "opus"),
        "medium": ("sonnet", "fable", "opus", "haiku"),
        "high": ("opus", "sonnet", "fable", "haiku"),
        "xhigh": ("opus", "sonnet", "fable", "haiku"),
    }
    if effort not in EFFORTS:
        raise ValueError(f"unsupported effort: {effort}")
    return _choose_by_hints(models, hints[effort], current, "claude-")


def _read_yaml(path: Path) -> dict[str, Any]:
    """Read a mapping from YAML, returning an empty mapping when unavailable."""
    if not path.is_file():
        return {}
    try:
        value = yaml.safe_load(path.read_text(encoding="utf-8"))
    except (OSError, yaml.YAMLError):
        return {}
    return value if isinstance(value, dict) else {}


def _atomic_write(path: Path, content: str, mode: int | None = None) -> bool:
    """Replace a file only when its content changes."""
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        if path.read_text(encoding="utf-8") == content:
            if mode is not None:
                path.chmod(mode)
            return False
    except OSError:
        pass
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    temporary.write_text(content, encoding="utf-8")
    if mode is not None:
        temporary.chmod(mode)
    os.replace(temporary, path)
    return True


def _write_yaml(path: Path, value: dict[str, Any], mode: int | None = None) -> bool:
    """Serialize a mapping and atomically update the target YAML file."""
    return _atomic_write(path, yaml.safe_dump(value, sort_keys=False), mode)


def _read_env(path: Path) -> dict[str, str]:
    """Read the small dotenv subset used by Hermes provider credentials."""
    values: dict[str, str] = {}
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError:
        return values
    for line in lines:
        value = line.strip()
        if not value or value.startswith("#") or "=" not in value:
            continue
        key, raw = value.removeprefix("export ").split("=", 1)
        raw = raw.strip()
        if len(raw) >= 2 and raw[0] == raw[-1] and raw[0] in "\"'":
            raw = raw[1:-1]
        values[key.strip()] = raw
    return values


def _update_profile_env(path: Path, source: dict[str, str]) -> None:
    """Refresh managed credentials while preserving user-owned environment keys."""
    try:
        old_lines = path.read_text(encoding="utf-8").splitlines()
    except OSError:
        old_lines = []
    kept = [
        line
        for line in old_lines
        if not any(line.lstrip().startswith(f"{key}=") for key in MANAGED_ENV_KEYS)
    ]
    kept.extend(
        f"{key}={source[key]}" for key in sorted(MANAGED_ENV_KEYS) if source.get(key)
    )
    _atomic_write(path, "\n".join(kept).rstrip() + "\n", 0o600)


def _existing_model(config: dict[str, Any], provider: str, default: str) -> str:
    """Return the last configured model for a provider or a safe baseline."""
    model_cfg = config.get("model", {})
    if isinstance(model_cfg, dict) and model_cfg.get("provider") == provider:
        return str(model_cfg.get("model") or model_cfg.get("default") or default)
    fallbacks = config.get("fallback_providers", [])
    if isinstance(fallbacks, list):
        for fallback in fallbacks:
            if isinstance(fallback, dict) and fallback.get("provider") == provider:
                return str(fallback.get("model") or default)
    return default


def discover_codex_models() -> Catalog:
    """Use the authenticated Codex endpoint; catalogs are status-only fallback."""
    token = ""
    try:
        from hermes_cli.auth import resolve_codex_runtime_credentials

        credentials = resolve_codex_runtime_credentials(refresh_if_expiring=True) or {}
        token = str(credentials.get("api_key") or "").strip()
    except Exception:
        token = ""
    live: list[str] = []
    if token:
        try:
            from hermes_cli.codex_models import _fetch_models_from_api

            live = _unique_models(_fetch_models_from_api(token))
        except Exception:
            live = []
    if live:
        return Catalog("openai-codex", live, True, True, "connected")
    try:
        from hermes_cli.models import provider_model_ids

        known = _unique_models(provider_model_ids("openai-codex", force_refresh=True))
    except Exception:
        known = []
    state = "degraded" if token else "not-configured"
    return Catalog("openai-codex", known, False, bool(token), state)


def discover_claude_models() -> Catalog:
    """Use Anthropic's authenticated model endpoint when configured."""
    token = str(os.environ.get("CLAUDE_CODE_OAUTH_TOKEN") or "").strip()
    live: list[str] = []
    if token:
        try:
            from hermes_cli.models import _fetch_anthropic_models

            live = _unique_models(_fetch_anthropic_models(timeout=10.0) or [])
        except Exception:
            live = []
    if live:
        return Catalog("anthropic", live, True, True, "connected")
    try:
        from hermes_cli.models import provider_model_ids

        known = _unique_models(provider_model_ids("anthropic", force_refresh=True))
    except Exception:
        known = []
    state = "degraded" if token else "not-configured"
    return Catalog("anthropic", known, False, bool(token), state)


def _profile_config(
    base: dict[str, Any],
    primary_provider: str,
    primary_model: str,
    fallback: dict[str, str],
    effort: str,
) -> dict[str, Any]:
    """Derive a worker configuration with an explicit cross-provider fallback."""
    config = copy.deepcopy(base)
    config["model"] = {
        "provider": primary_provider,
        "default": primary_model,
        "model": primary_model,
        "openai_runtime": "codex_app_server",
    }
    config["fallback_providers"] = [
        fallback,
        copy.deepcopy(ATLAS_FALLBACK),
    ]
    agent = config.setdefault("agent", {})
    if isinstance(agent, dict):
        agent["reasoning_effort"] = effort
    config["toolsets"] = []
    return config


def _write_profile(
    root: Path,
    name: str,
    description: str,
    soul: str,
    config: dict[str, Any],
    env_values: dict[str, str],
) -> None:
    """Create or refresh a managed Hermes worker profile."""
    profile = root / "profiles" / name
    for directory in ("logs", "sessions", "skills", "workspace", "home"):
        (profile / directory).mkdir(parents=True, exist_ok=True)
    _write_yaml(profile / "config.yaml", config)
    _write_yaml(
        profile / "profile.yaml",
        {"description": description, "description_auto": False},
    )
    _atomic_write(profile / "SOUL.md", soul.rstrip() + "\n")
    _update_profile_env(profile / ".env", env_values)


def configure_routes(root: Path, codex: Catalog, claude: Catalog) -> dict[str, Any]:
    """Update the coordinator and managed worker profiles."""
    config_path = root / "config.yaml"
    base = _read_yaml(config_path)
    old_coordinator = _existing_model(base, "openai-codex", CODEX_BASELINE)
    old_claude_root = _existing_model(base, "anthropic", CLAUDE_BASELINE)
    codex_models: dict[str, str] = {}
    claude_models: dict[str, str] = {}
    for effort in EFFORTS:
        old_codex = _existing_model(
            _read_yaml(root / "profiles" / f"codex-{effort}" / "config.yaml"),
            "openai-codex",
            old_coordinator,
        )
        old_claude = _existing_model(
            _read_yaml(root / "profiles" / f"claude-{effort}" / "config.yaml"),
            "anthropic",
            old_claude_root,
        )
        codex_models[effort] = (
            choose_codex_for_effort(codex.models, effort, old_codex)
            if codex.live
            else old_codex
        )
        claude_models[effort] = (
            choose_claude_for_effort(claude.models, effort, old_claude)
            if claude.live
            else old_claude
        )

    codex_coordinator = codex_models["medium"]
    claude_coordinator = claude_models["medium"]

    base["model"] = {
        "provider": "openai-codex",
        "default": codex_coordinator,
        "model": codex_coordinator,
        "openai_runtime": "codex_app_server",
    }
    base["fallback_providers"] = [
        {"provider": "anthropic", "model": claude_coordinator},
        copy.deepcopy(ATLAS_FALLBACK),
    ]
    base["model_catalog"] = {"enabled": True, "ttl_hours": 1}
    _write_yaml(config_path, base)

    env_values = _read_env(root / ".env")
    routes: dict[str, list[str]] = {}
    for effort in EFFORTS:
        codex_model = codex_models[effort]
        claude_model = claude_models[effort]
        codex_name = f"codex-{effort}"
        claude_name = f"claude-{effort}"
        _write_profile(
            root,
            codex_name,
            f"Codex implementation worker at {effort} effort, with Claude and local fallback.",
            "You are an implementation worker. Make focused, tested changes for the assigned task, preserve unrelated work, and report evidence and blockers to the coordinator.",
            _profile_config(
                base,
                "openai-codex",
                codex_model,
                {"provider": "anthropic", "model": claude_model},
                effort,
            ),
            env_values,
        )
        _write_profile(
            root,
            claude_name,
            f"Claude architecture and review worker at {effort} effort, with Codex and local fallback.",
            "You are an architecture and review worker. Analyze the assigned task deeply, change files only when asked, and return concise conclusions, evidence, and risks to the coordinator.",
            _profile_config(
                base,
                "anthropic",
                claude_model,
                {"provider": "openai-codex", "model": codex_model},
                effort,
            ),
            env_values,
        )
        local = ["custom/gpt-oss:20b"]
        routes[codex_name] = [
            f"openai-codex/{codex_model}",
            f"anthropic/{claude_model}",
            *local,
        ]
        routes[claude_name] = [
            f"anthropic/{claude_model}",
            f"openai-codex/{codex_model}",
            *local,
        ]

    _write_profile(
        root,
        "synthesis-xhigh",
        "Cross-provider synthesis and critical review, capped at xhigh effort.",
        "Synthesize the worker evidence into one answer. Resolve disagreements explicitly, verify high-risk claims, and never claim completion without cited validation.",
        _profile_config(
            base,
            "anthropic",
            claude_models["xhigh"],
            {"provider": "openai-codex", "model": codex_models["xhigh"]},
            "xhigh",
        ),
        env_values,
    )
    routes["synthesis-xhigh"] = [
        f"anthropic/{claude_models['xhigh']}",
        f"openai-codex/{codex_models['xhigh']}",
        "custom/gpt-oss:20b",
    ]
    _write_yaml(
        root / "profile.yaml",
        {
            "description": "Coordinator for project objectives, delegating implementation to Codex and architecture or review to Claude through isolated Kanban boards.",
            "description_auto": False,
        },
    )
    routes["coordinator"] = [
        f"openai-codex/{codex_coordinator}",
        f"anthropic/{claude_coordinator}",
        "custom/gpt-oss:20b",
    ]
    return routes