493 lines
16 KiB
Python
493 lines
16 KiB
Python
#!/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,
|
|
}
|
|
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,
|
|
}
|
|
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/qwen2.5:14b-instruct-q4_0",
|
|
"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/qwen2.5:14b-instruct-q4_0",
|
|
"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/qwen2.5:14b-instruct-q4_0",
|
|
"custom/gpt-oss:20b",
|
|
]
|
|
return routes
|