From 52990c67ea7bf1d6aa411ecd0fc72a381408dc77 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:40:25 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20model=5Fcatalog/norma?= =?UTF-8?q?lize/guards/search=20=E2=80=94=20collapse=20manifest=20block=20?= =?UTF-8?q?accessors,=20dedupe=20deepseek=20fallbacks,=20compact=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/model_catalog.py | 262 +++++++------------------- hermes_cli/model_cost_guard.py | 35 +--- hermes_cli/model_data_policy_guard.py | 44 ++--- hermes_cli/model_normalize.py | 248 +++++------------------- hermes_cli/model_search.py | 19 +- hermes_cli/model_selection_guards.py | 31 +-- 6 files changed, 148 insertions(+), 491 deletions(-) diff --git a/hermes_cli/model_catalog.py b/hermes_cli/model_catalog.py index 029e2bcd72..1a76dd5e57 100644 --- a/hermes_cli/model_catalog.py +++ b/hermes_cli/model_catalog.py @@ -1,13 +1,9 @@ """Remote model catalog fetcher. -Pipeline -------- 1. ``get_catalog()`` — returns a parsed manifest dict. - Checks in-process cache -(invalidated by TTL). - Reads disk cache at ``~/.hermes/cache/model_catalog.json``. - Fetches the -master URL if disk cache is stale or missing. - On any fetch failure, keeps using the stale cache -(or empty dict). - -2. ``get_curated_openrouter_models()`` / ``get_curated_nous_models()`` — thin accessors returning -the shapes existing callers expect. Each falls back to the in-repo hardcoded list on any lookup -failure. +``get_catalog()`` returns the parsed manifest: in-process cache (TTL) → disk cache at +``~/.hermes/cache/model_catalog.json`` → master URL fetch; any fetch failure keeps the stale copy +(or ``{}``). ``get_curated_openrouter_models()`` / ``get_curated_nous_models()`` are thin accessors +whose callers fall back to the in-repo lists on ``None``. """ from __future__ import annotations @@ -26,43 +22,27 @@ from utils import atomic_replace logger = logging.getLogger(__name__) -# --------------------------------------------------------------------------- -# Constants -# --------------------------------------------------------------------------- - DEFAULT_CATALOG_URL = ( "https://hermes-agent.nousresearch.com/docs/api/model-catalog.json" ) -# Fallback fetch chain. The Docusaurus site is served through Vercel, which -# occasionally returns HTTP 403 + x-vercel-mitigated: challenge for non- -# browser clients (urllib, curl). When that happens the disk cache goes -# stale and new model releases never reach the picker. The raw GitHub URL -# is the same manifest published from the same repo and is not bot-gated, -# so we fall through to it whenever the primary URL fails. +# The Docusaurus site sits behind Vercel, which occasionally 403s non-browser clients (bot +# challenge); the raw GitHub copy is the same manifest and is not bot-gated. DEFAULT_CATALOG_FALLBACK_URLS: tuple[str, ...] = ( "https://raw.githubusercontent.com/NousResearch/hermes-agent/main/website/static/api/model-catalog.json", ) DEFAULT_TTL_MINUTES = 20 -# Legacy key. ``ttl_hours`` is honoured only when the user set it explicitly; -# the shipped default is ``ttl_minutes`` above. +# Legacy key, honoured only when the user set it explicitly; ``ttl_minutes`` is the shipped default. DEFAULT_TTL_HOURS = DEFAULT_TTL_MINUTES / 60.0 DEFAULT_FETCH_TIMEOUT = 8.0 SUPPORTED_SCHEMA_VERSION = 1 _HERMES_USER_AGENT = f"hermes-cli/{_HERMES_VERSION}" -# In-process cache to avoid repeated disk + parse work across multiple -# calls within the same session. Invalidated by TTL against the disk file's -# mtime, so calling code never has to think about this. +# In-process cache, invalidated against the disk file's mtime and TTL. _catalog_cache: dict[str, Any] | None = None _catalog_cache_source_mtime: float = 0.0 -# --------------------------------------------------------------------------- -# Config -# --------------------------------------------------------------------------- - - def _load_catalog_config() -> dict[str, Any]: """Load the ``model_catalog`` config block with defaults filled in.""" try: @@ -70,16 +50,12 @@ def _load_catalog_config() -> dict[str, Any]: cfg = load_config() or {} except Exception: cfg = {} - raw = cfg.get("model_catalog") if not isinstance(raw, dict): raw = {} - # ``ttl_minutes`` is the shipped default (20). ``ttl_hours`` is the legacy - # key: honoured when a user set it explicitly and ``ttl_minutes`` is still - # at its default (load_config() deep-merges the default in, so "present" - # alone doesn't mean "user-set"), so old customized configs keep their - # chosen window. + # ``ttl_hours`` (legacy) is honoured only when ``ttl_minutes`` is still at its default — + # load_config() deep-merges the default in, so "present" alone doesn't mean "user-set". ttl_minutes = raw.get("ttl_minutes") try: ttl_minutes = float(ttl_minutes) if ttl_minutes not in (None, "") else DEFAULT_TTL_MINUTES @@ -102,26 +78,15 @@ def _load_catalog_config() -> dict[str, Any]: def _cache_path() -> Path: - """Return the disk cache path. Import lazily so tests can monkeypatch home.""" + """Disk cache path; imported lazily so tests can monkeypatch home.""" from hermes_constants import get_hermes_home return get_hermes_home() / "cache" / "model_catalog.json" -# --------------------------------------------------------------------------- -# Fetch + validate + cache -# --------------------------------------------------------------------------- - - def _fetch_manifest(url: str, timeout: float) -> dict[str, Any] | None: - """HTTP GET the manifest URL and return a parsed dict, or None on failure.""" + """HTTP GET the manifest URL and return a validated dict, or None on failure.""" try: - req = urllib.request.Request( - url, - headers={ - "Accept": "application/json", - "User-Agent": _HERMES_USER_AGENT, - }, - ) + req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": _HERMES_USER_AGENT}) with urllib.request.urlopen(req, timeout=timeout) as resp: data = json.loads(resp.read().decode()) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc: @@ -130,11 +95,9 @@ def _fetch_manifest(url: str, timeout: float) -> dict[str, Any] | None: except Exception as exc: # pragma: no cover — defensive logger.info("model catalog fetch errored (%s): %s", url, exc) return None - if not _validate_manifest(data): logger.info("model catalog at %s failed schema validation", url) return None - return data @@ -143,12 +106,8 @@ def _fetch_manifest_with_fallback( timeout: float, fallback_urls: tuple[str, ...] = DEFAULT_CATALOG_FALLBACK_URLS, ) -> dict[str, Any] | None: - """Try ``primary_url`` first, then walk ``fallback_urls``. - - Returns the first manifest that fetches and validates, or None when every URL fails. Skips - fallback URLs identical to the primary so an operator who configured the catalog URL to point at - the raw GitHub copy doesn't double-fetch. - """ + """First manifest that fetches and validates from ``primary_url`` then ``fallback_urls`` (skipping + any equal to the primary so a raw-GitHub-configured operator doesn't double-fetch), or None.""" data = _fetch_manifest(primary_url, timeout) if data is not None: return data @@ -168,9 +127,7 @@ def _validate_manifest(data: Any) -> bool: return False version = data.get("version") if not isinstance(version, int) or version > SUPPORTED_SCHEMA_VERSION: - # Future schema version we don't understand — refuse rather than - # guess. Older schemas (version < 1) aren't supported either. - return False + return False # future schema we don't understand — refuse rather than guess providers = data.get("providers") if not isinstance(providers, dict): return False @@ -180,11 +137,8 @@ def _validate_manifest(data: Any) -> bool: models = pblock.get("models") if not isinstance(models, list): return False - for m in models: - if not isinstance(m, dict): - return False - if not isinstance(m.get("id"), str) or not m["id"].strip(): - return False + if not all(isinstance(m, dict) and isinstance(m.get("id"), str) and m["id"].strip() for m in models): + return False return True @@ -193,16 +147,11 @@ def _read_disk_cache() -> tuple[dict[str, Any] | None, float]: path = _cache_path() try: mtime = path.stat().st_mtime - except (OSError, FileNotFoundError): - return (None, 0.0) - try: with open(path, encoding="utf-8") as fh: data = json.load(fh) except (OSError, json.JSONDecodeError): return (None, 0.0) - if not _validate_manifest(data): - return (None, 0.0) - return (data, mtime) + return (data, mtime) if _validate_manifest(data) else (None, 0.0) def _write_disk_cache(data: dict[str, Any]) -> None: @@ -218,9 +167,8 @@ def _write_disk_cache(data: dict[str, Any]) -> None: logger.info("model catalog cache write failed: %s", exc) -# Stale-while-revalidate machinery: at most one background manifest refresh -# in flight per process. The refreshed manifest lands on disk; the NEXT -# get_catalog() call picks it up via the mtime check. +# Stale-while-revalidate: at most one background manifest refresh in flight per process. The +# refreshed manifest lands on disk; the NEXT get_catalog() call picks it up via the mtime check. _catalog_swr_lock = threading.Lock() _catalog_swr_inflight = False @@ -248,73 +196,42 @@ def _spawn_catalog_swr_refresh(url: str) -> None: threading.Thread(target=_refresh, daemon=True, name="model-catalog-swr").start() -# --------------------------------------------------------------------------- -# Public API -# --------------------------------------------------------------------------- +def _remember(data: dict[str, Any], mtime: float) -> dict[str, Any]: + global _catalog_cache, _catalog_cache_source_mtime + _catalog_cache, _catalog_cache_source_mtime = data, mtime + return data def get_catalog(*, force_refresh: bool = False) -> dict[str, Any]: - """Return the parsed model catalog manifest, or an empty dict on failure. - - Callers should treat a missing provider/model as "use the in-repo fallback" — never raise from - this function so the CLI keeps working offline. - """ - global _catalog_cache, _catalog_cache_source_mtime - + """Parsed model catalog manifest, or ``{}`` on failure — never raises, so the CLI works offline + (callers treat a missing provider/model as "use the in-repo fallback").""" cfg = _load_catalog_config() if not cfg["enabled"]: return {} - ttl_seconds = max(0.0, cfg["ttl_hours"] * 3600.0) - disk_data, disk_mtime = _read_disk_cache() now = time.time() disk_fresh = disk_data is not None and (now - disk_mtime) < ttl_seconds - # In-process cache hit: disk hasn't changed since we loaded it and still fresh. - if ( - not force_refresh - and _catalog_cache is not None - and disk_data is not None - and disk_mtime == _catalog_cache_source_mtime - and disk_fresh - ): - return _catalog_cache - - # Disk is fresh enough — use it without a network hit. - if not force_refresh and disk_fresh and disk_data is not None: - _catalog_cache = disk_data - _catalog_cache_source_mtime = disk_mtime - return disk_data - - # Stale-while-revalidate: an expired disk copy is served immediately and - # refreshed off-thread, so interactive surfaces (the /model picker calls - # this via get_curated_nous_model_ids on every open) never block on the - # manifest fetch. Only a cold cache (no disk copy at all) still blocks. if not force_refresh and disk_data is not None: - _catalog_cache = disk_data - _catalog_cache_source_mtime = disk_mtime - _spawn_catalog_swr_refresh(cfg["url"]) - return disk_data + if disk_fresh and _catalog_cache is not None and disk_mtime == _catalog_cache_source_mtime: + return _catalog_cache + if not disk_fresh: + # Stale-while-revalidate: serve the expired disk copy now and refresh off-thread so the + # /model picker (which calls this on every open) never blocks on the manifest fetch. + # Only a cold cache (no disk copy at all) still blocks. + _spawn_catalog_swr_refresh(cfg["url"]) + return _remember(disk_data, disk_mtime) - # Need to (re)fetch. If it fails, fall back to any stale disk copy. fetched = _fetch_manifest_with_fallback(cfg["url"], DEFAULT_FETCH_TIMEOUT) if fetched is not None: _write_disk_cache(fetched) new_disk_data, new_mtime = _read_disk_cache() if new_disk_data is not None: - _catalog_cache = new_disk_data - _catalog_cache_source_mtime = new_mtime - return new_disk_data - _catalog_cache = fetched - _catalog_cache_source_mtime = now - return fetched - + return _remember(new_disk_data, new_mtime) + return _remember(fetched, now) if disk_data is not None: - _catalog_cache = disk_data - _catalog_cache_source_mtime = disk_mtime - return disk_data - + return _remember(disk_data, disk_mtime) return {} @@ -324,12 +241,9 @@ def refresh_interval_seconds() -> float: def refresh_catalogs() -> bool: - """Force-refresh every remote model catalog the picker reads from. - - Fetches the curated manifest, the OpenRouter live list (tool-support / free-pricing filter) and - the Nous Portal recommendations, writing each to its disk cache so the next ``/model`` open in - ANY process on this machine sees the new lists. Blocking; run it off the event loop. - """ + """Force-refresh every remote catalog the picker reads (manifest, OpenRouter live list, Nous Portal + recommendations), writing each disk cache so the next ``/model`` open in ANY process sees them. + Blocking; run it off the event loop.""" if not _load_catalog_config()["enabled"]: return False catalog = get_catalog(force_refresh=True) @@ -354,96 +268,54 @@ def _fetch_provider_override(provider: str) -> dict[str, Any] | None: override_url = provider_cfg.get("url") if not isinstance(override_url, str) or not override_url.strip(): return None - # Override fetches skip the disk cache because they're usually - # third-party self-hosted. Re-request on every call but with a short - # timeout so they don't block the picker. + # Overrides are usually third-party self-hosted: skip the disk cache, re-request every call. return _fetch_manifest(override_url.strip(), DEFAULT_FETCH_TIMEOUT) +def _block_of(manifest: dict[str, Any] | None, provider: str) -> dict[str, Any] | None: + block = (manifest or {}).get("providers", {}).get(provider) + return block if isinstance(block, dict) else None + + def _get_provider_block(provider: str) -> dict[str, Any] | None: """Return the provider's manifest block, respecting per-provider overrides.""" - override = _fetch_provider_override(provider) - if override is not None: - block = override.get("providers", {}).get(provider) - if isinstance(block, dict): - return block + return _block_of(_fetch_provider_override(provider), provider) or _block_of(get_catalog(), provider) - catalog = get_catalog() - if not catalog: - return None - block = catalog.get("providers", {}).get(provider) - return block if isinstance(block, dict) else None + +def _block_ids(block: dict[str, Any] | None) -> list[tuple[str, dict[str, Any]]]: + """``(id, entry)`` for every model entry of ``block`` with a non-empty id.""" + return [(mid, m) for m in (block or {}).get("models", []) if isinstance(m, dict) and (mid := str(m.get("id") or "").strip())] def get_curated_openrouter_models() -> list[tuple[str, str]] | None: - """Return OpenRouter's curated ``[(id, description), ...]`` from the manifest.""" - block = _get_provider_block("openrouter") - if not block: - return None - out: list[tuple[str, str]] = [] - for m in block.get("models", []): - mid = str(m.get("id") or "").strip() - if not mid: - continue - desc = str(m.get("description") or "") - out.append((mid, desc)) - return out or None + """OpenRouter's curated ``[(id, description), ...]`` from the manifest.""" + return [(mid, str(m.get("description") or "")) for mid, m in _block_ids(_get_provider_block("openrouter"))] or None def get_curated_nous_models() -> list[str] | None: - """Return Nous Portal's curated list of model ids from the manifest.""" - block = _get_provider_block("nous") - if not block: - return None - out: list[str] = [] - for m in block.get("models", []): - mid = str(m.get("id") or "").strip() - if mid: - out.append(mid) - return out or None + """Nous Portal's curated model ids from the manifest.""" + return [mid for mid, _ in _block_ids(_get_provider_block("nous"))] or None def _default_model_from_block(block: dict[str, Any] | None) -> str | None: - """Return the id of the model entry labeled ``"default": true``, or None.""" - if not isinstance(block, dict): - return None - for m in block.get("models", []): - if isinstance(m, dict) and m.get("default"): - mid = str(m.get("id") or "").strip() - if mid: - return mid - return None + """Id of the model entry labeled ``"default": true``, or None.""" + return next((mid for mid, m in _block_ids(block) if m.get("default")), None) def get_default_model_from_cache(provider: str) -> str | None: - """Return the catalog's labeled default model for ``provider`` — cache only. - - The manifest marks exactly one model entry per provider with ``"default": true``; that entry is - the model Hermes silently lands on when the user never picked one. - """ - if _catalog_cache is not None: - block = _catalog_cache.get("providers", {}).get(provider) - found = _default_model_from_block(block) - if found: - return found + """The manifest's labeled default for ``provider`` (the model Hermes silently lands on when the + user never picked one) — in-process then disk cache only, never a fetch.""" + found = _default_model_from_block(_block_of(_catalog_cache, provider)) if _catalog_cache is not None else None + if found: + return found disk_data, _mtime = _read_disk_cache() - if disk_data is not None: - block = disk_data.get("providers", {}).get(provider) - return _default_model_from_block(block) - return None + return _default_model_from_block(_block_of(disk_data, provider)) if disk_data is not None else None def seed_cache_from_checkout(project_root: "Path | str") -> bool: - """Overwrite the disk cache with the catalog shipped in a local checkout. - - ``hermes update`` pulls the latest repo, so the freshly-pulled ``website/static/api/model- - catalog.json`` IS the newest catalog — no network round-trip needed. Copying it straight over - the disk cache keeps the model picker current even when the remote manifest fetch is bot-gated - or the Portal hiccups. - - Reads the shipped manifest, validates it against the schema, and writes it to - ``~/.hermes/cache/model_catalog.json`` via the same atomic writer the network path uses. - """ + """Overwrite the disk cache with the checkout's ``website/static/api/model-catalog.json``. + After ``hermes update`` that file IS the newest catalog, so the picker stays current even when + the remote fetch is bot-gated. Validated, then written via the same atomic writer.""" src = Path(project_root) / "website" / "static" / "api" / "model-catalog.json" try: with open(src, encoding="utf-8") as fh: diff --git a/hermes_cli/model_cost_guard.py b/hermes_cli/model_cost_guard.py index 16d4f0fd51..d317d830c4 100644 --- a/hermes_cli/model_cost_guard.py +++ b/hermes_cli/model_cost_guard.py @@ -28,10 +28,8 @@ class ExpensiveModelWarning: def _to_decimal(value: object) -> Optional[Decimal]: - if value is None: - return None try: - return Decimal(str(value)) + return None if value is None else Decimal(str(value)) except (InvalidOperation, ValueError): return None @@ -45,11 +43,7 @@ def _pricing_from_model_info( ) -> tuple[Optional[Decimal], Optional[Decimal], str]: if model_info is None or not model_info.has_cost_data(): return None, None, "" - return ( - _to_decimal(model_info.cost_input), - _to_decimal(model_info.cost_output), - "models.dev", - ) + return _to_decimal(model_info.cost_input), _to_decimal(model_info.cost_output), "models.dev" def _known_models_dev_provider(provider: Optional[str]) -> Optional[str]: @@ -64,7 +58,6 @@ def _can_trust_model_info_pricing( expected_provider = _known_models_dev_provider(provider) if not expected_provider or model_info is None: return False - actual_provider = str(getattr(model_info, "provider_id", "") or "").strip().lower() return not actual_provider or actual_provider == expected_provider @@ -92,11 +85,8 @@ def expensive_model_warning( api_key: Optional[str] = None, model_info: Optional[ModelInfo] = None, ) -> Optional[ExpensiveModelWarning]: - """Return a warning payload when known pricing exceeds safety thresholds. - - The guard only triggers when pricing is known. Callers should use this after model resolution so - aliases and provider-specific model IDs have settled. - """ + """Warning payload when KNOWN pricing exceeds the safety thresholds (never fires on unknown + pricing). Call after model resolution so aliases / provider-specific ids have settled.""" model = (model_name or "").strip() if not model: return None @@ -104,7 +94,6 @@ def expensive_model_warning( input_cost: Optional[Decimal] = None output_cost: Optional[Decimal] = None source = "" - if _can_trust_model_info_pricing(provider, model_info): input_cost, output_cost, source = _pricing_from_model_info(model_info) @@ -115,9 +104,7 @@ def expensive_model_warning( try: from agent.models_dev import get_model_info - input_cost, output_cost, source = _pricing_from_model_info( - get_model_info(provider, model) - ) + input_cost, output_cost, source = _pricing_from_model_info(get_model_info(provider, model)) except Exception: pass @@ -125,21 +112,13 @@ def expensive_model_warning( try: from agent.usage_pricing import get_pricing_entry - entry = get_pricing_entry( - model, - provider=provider, - base_url=base_url, - api_key=api_key, - ) + entry = get_pricing_entry(model, provider=provider, base_url=base_url, api_key=api_key) except Exception: entry = None if entry is not None: - input_cost = entry.input_cost_per_million - output_cost = entry.output_cost_per_million - source = entry.source + input_cost, output_cost, source = entry.input_cost_per_million, entry.output_cost_per_million, entry.source is_known_gpt55_pro_confusion = model.lower() == GPT55_PRO_OPENROUTER_ID - over_input = input_cost is not None and input_cost > INPUT_COST_WARNING_THRESHOLD over_output = output_cost is not None and output_cost > OUTPUT_COST_WARNING_THRESHOLD if not over_input and not over_output and not is_known_gpt55_pro_confusion: diff --git a/hermes_cli/model_data_policy_guard.py b/hermes_cli/model_data_policy_guard.py index 68c04d3e98..8735ff5a32 100644 --- a/hermes_cli/model_data_policy_guard.py +++ b/hermes_cli/model_data_policy_guard.py @@ -1,11 +1,8 @@ """Data-policy confirmation helpers for model selection surfaces. -Some inference tiers are cheap *because* the vendor trains future models on your prompts and -completions. Selecting one for the low price without realising the data trade-off is a real footgun. - -Why a static table (not a ProviderProfile hook): the guard runs inside core selection code -(``auth.py`` / ``web_server.py``), which never calls into the active provider profile for a -selection-time warning. +Some inference tiers are cheap *because* the vendor trains on your prompts. A static rule table (not +a ProviderProfile hook) because the guard runs inside core selection code (``auth.py`` / +``web_server.py``), which never calls into the active provider profile. """ from __future__ import annotations @@ -23,18 +20,12 @@ class DataTrainingWarning: message: str -# ── Rule table ──────────────────────────────────────────────────────────── -# Each rule: (human label, predicate over (model_lower, provider_lower), message). -# Extensible — new data-collection tiers from other vendors slot in here without -# touching the call sites. Predicates are intentionally conservative: match an -# explicit, vendor-documented id rather than guessing from price alone (price is -# only a corroborating signal and can change). +# Rule predicates are deliberately conservative: match an explicit, vendor-documented id rather +# than guessing from price (a corroborating signal that can change). def _is_meta_contributor(model_lower: str, provider_lower: str) -> bool: - # Meta Model API "contributor" tier (muse-spark-1.2-contributor and any - # future -contributor checkpoints). Match on the id suffix; do not require a - # specific provider id so it fires whether selected via the meta-ai plugin, - # a gateway, or a custom endpoint that serves the same model id. + # Meta "contributor" tier, matched on the id alone (no provider check) so it fires whether + # selected via the meta-ai plugin, a gateway, or a custom endpoint serving the same id. return model_lower.endswith("-contributor") or "contributor" in model_lower.split("-") @@ -71,27 +62,16 @@ def data_training_warning( provider: Optional[str] = None, base_url: Optional[str] = None, # noqa: ARG001 — reserved for host-scoped rules ) -> Optional[DataTrainingWarning]: - """Return a warning payload when *model_name* selects a data-training tier. - - Returns ``None`` when no rule matches (the common case). Callers should run this after model - resolution so aliases / provider-specific ids have settled, and surface ``.message`` as a - confirm prompt. - """ + """Warning payload when *model_name* selects a data-training tier, else ``None``. Call after model + resolution; surface ``.message`` as a confirm prompt.""" model = (model_name or "").strip() if not model: return None - model_lower = model.lower() - provider_lower = (provider or "").strip().lower() - + model_lower, provider_lower = model.lower(), (provider or "").strip().lower() for predicate, message in _RULES: try: if predicate(model_lower, provider_lower): - return DataTrainingWarning( - model=model, - provider=(provider or "").strip(), - message=message, - ) + return DataTrainingWarning(model=model, provider=(provider or "").strip(), message=message) except Exception: - # A misbehaving predicate must never break model selection. - continue + continue # a misbehaving predicate must never break model selection return None diff --git a/hermes_cli/model_normalize.py b/hermes_cli/model_normalize.py index 0838cb207b..d6b203bc3f 100644 --- a/hermes_cli/model_normalize.py +++ b/hermes_cli/model_normalize.py @@ -5,15 +5,8 @@ from __future__ import annotations import re from typing import Optional -# --------------------------------------------------------------------------- -# Vendor prefix mapping -# --------------------------------------------------------------------------- -# Maps the first hyphen-delimited token of a bare model name to the vendor -# slug used by aggregator APIs (OpenRouter, Nous, etc.). -# -# Example: "claude-sonnet-4.6" -> first token "claude" -> vendor "anthropic" -# -> aggregator slug: "anthropic/claude-sonnet-4.6" - +# First hyphen-delimited token of a bare model name → vendor slug used by aggregator APIs +# ("claude-sonnet-4.6" → "anthropic/claude-sonnet-4.6"). _VENDOR_PREFIXES: dict[str, str] = { "claude": "anthropic", "gpt": "openai", @@ -33,7 +26,6 @@ _VENDOR_PREFIXES: dict[str, str] = { "nemotron": "nvidia", "llama": "meta-llama", "step": "stepfun", - "trinity": "arcee-ai", } # Providers whose APIs consume vendor/model slugs. @@ -94,94 +86,46 @@ _CATALOGUE_PREFIX_REPAIR_PROVIDERS: frozenset[str] = frozenset({ "nvidia", }) -# Providers whose APIs require lowercase model IDs. Xiaomi's -# ``api.xiaomimimo.com`` rejects mixed-case names like ``MiMo-V2.5-Pro`` -# that users might copy from marketing docs — it only accepts -# ``mimo-v2.5-pro``. After stripping a matching provider prefix, these -# providers also get ``.lower()`` applied. +# Providers whose APIs require lowercase model IDs (Xiaomi rejects ``MiMo-V2.5-Pro`` copied from +# marketing docs; only ``mimo-v2.5-pro`` works). Applied after matching-prefix stripping. _LOWERCASE_MODEL_PROVIDERS: frozenset[str] = frozenset({ "xiaomi", }) -# --------------------------------------------------------------------------- -# DeepSeek special handling -# --------------------------------------------------------------------------- -# DeepSeek's direct API only accepts first-class V-series IDs after the -# 2026-07-24 cut-off. Legacy aliases and fuzzy names are remapped here so -# saved configs / picker leftovers cannot keep sending retired IDs. - -_DEEPSEEK_REASONER_KEYWORDS: frozenset[str] = frozenset({ - "reasoner", - "r1", - "think", - "reasoning", - "cot", -}) - -# Retired on 2026-07-24 15:59 UTC. Official docs: both aliases mapped to -# deepseek-v4-flash (chat = non-thinking, reasoner = thinking). Thinking -# mode itself is controlled by extra_body.thinking on the DeepSeek profile. +# DeepSeek's direct API only accepts first-class V-series IDs after the 2026-07-24 cut-off (HTTP 400 +# otherwise). Both retired aliases map to deepseek-v4-flash per the official docs (thinking mode is +# controlled by extra_body.thinking on the profile), so saved configs can't keep sending them. _DEEPSEEK_RETIRED_ALIASES: frozenset[str] = frozenset({ "deepseek-chat", "deepseek-reasoner", }) _DEEPSEEK_CANONICAL_MODELS: frozenset[str] = frozenset({ - "deepseek-v4-pro", # V4 Pro — first-class model ID - "deepseek-v4-flash", # V4 Flash — first-class model ID + "deepseek-v4-pro", + "deepseek-v4-flash", }) -# First-class V-series IDs (``deepseek-v4-pro``, ``deepseek-v4-flash``, -# future ``deepseek-v5-*``, dated variants like ``deepseek-v4-flash-20260423``). -# Verified empirically 2026-04-24: DeepSeek's Chat Completions API returns -# ``provider: DeepSeek`` / ``model: deepseek-v4-flash-20260423`` when called -# with ``model=deepseek/deepseek-v4-flash``, so these names are not aliases -# of ``deepseek-chat`` and must not be folded into it. +# First-class V-series IDs incl. future ``deepseek-v5-*`` and dated variants +# (``deepseek-v4-flash-20260423``): verified real model ids, NOT aliases of ``deepseek-chat``. _DEEPSEEK_V_SERIES_RE = re.compile(r"^deepseek-v\d+([-.].+)?$") def _normalize_for_deepseek(model_name: str) -> str: - """Map a model input to a DeepSeek-accepted identifier. - - Retired aliases ``deepseek-chat``/``deepseek-reasoner`` and known canonicals map as expected; - anything matching ``deepseek-v...`` passes through so future V-series ids work without a - release; reasoner keywords and everything else fall back to ``deepseek-v4-flash``. - """ + """Map a model input to a DeepSeek-accepted id: canonicals and ``deepseek-v…`` pass + through (future V-series work without a release); retired aliases and everything else become + ``deepseek-v4-flash``.""" bare = _strip_vendor_prefix(model_name).lower() - - # Retired aliases must rewrite — DeepSeek returns HTTP 400 after the - # 2026-07-24 cut-off if these IDs are sent on the wire. - if bare in _DEEPSEEK_RETIRED_ALIASES: - return "deepseek-v4-flash" - - if bare in _DEEPSEEK_CANONICAL_MODELS: + if bare in _DEEPSEEK_CANONICAL_MODELS or _DEEPSEEK_V_SERIES_RE.match(bare): return bare - - # V-series first-class IDs (v4-pro, v4-flash, future v5-*, dated variants) - if _DEEPSEEK_V_SERIES_RE.match(bare): - return bare - - # Check for reasoner-like keywords anywhere in the name - for keyword in _DEEPSEEK_REASONER_KEYWORDS: - if keyword in bare: - return "deepseek-v4-flash" - return "deepseek-v4-flash" -# --------------------------------------------------------------------------- -# Helper utilities -# --------------------------------------------------------------------------- - def _strip_vendor_prefix(model_name: str) -> str: """Remove a ``vendor/`` prefix if present.""" - if "/" in model_name: - return model_name.split("/", 1)[1] - return model_name + return model_name.split("/", 1)[1] if "/" in model_name else model_name def _dots_to_hyphens(model_name: str) -> str: - """Replace dots with hyphens in a model name.""" return model_name.replace(".", "-") @@ -199,108 +143,65 @@ def _normalize_provider_alias(provider_name: str) -> str: def _strip_matching_provider_prefix(model_name: str, target_provider: str) -> str: - """Strip ``provider/`` only when the prefix matches the target provider. - - Prevents arbitrary slash-bearing ids from being mangled on native providers while still - repairing config values like ``zai/glm-5.1`` for ``zai``. ``custom`` is a bucket, not a vendor: - an alias that merely resolves to it (e.g. ``ollama``) may be a real routing prefix required by a - proxy such as LiteLLM, so only a literal ``custom/`` prefix is treated as redundant. - """ + """Strip ``provider/`` only when the prefix matches the target provider, so arbitrary slash-bearing + ids aren't mangled while ``zai/glm-5.1`` is repaired for ``zai``. ``custom`` is a bucket, not a + vendor: an alias resolving to it (``ollama``) may be a real LiteLLM-style routing prefix, so only a + literal ``custom/`` prefix is redundant there.""" if "/" not in model_name: return model_name - prefix, remainder = model_name.split("/", 1) if not prefix.strip() or not remainder.strip(): return model_name - normalized_target = _normalize_provider_alias(target_provider) if normalized_target == "custom": - if prefix.strip().lower() == "custom": - return remainder.strip() - return model_name - + return remainder.strip() if prefix.strip().lower() == "custom" else model_name normalized_prefix = _normalize_provider_alias(prefix) - if normalized_prefix and normalized_prefix == normalized_target: - return remainder.strip() - return model_name + return remainder.strip() if normalized_prefix and normalized_prefix == normalized_target else model_name def detect_vendor(model_name: str) -> Optional[str]: - """Detect the vendor slug from a bare model name.""" + """Vendor slug from a bare model name: an existing ``vendor/`` prefix, the first hyphen token, + or a ``_VENDOR_PREFIXES`` key the name starts with (``qwen3.5-plus`` → ``qwen``).""" name = model_name.strip() if not name: return None - - # If there's already a vendor/ prefix, extract it if "/" in name: return name.split("/", 1)[0].lower() or None - name_lower = name.lower() - - # Try first hyphen-delimited token (exact match) first_token = name_lower.split("-")[0] if first_token in _VENDOR_PREFIXES: return _VENDOR_PREFIXES[first_token] - - # Handle patterns where the first token includes version digits, - # e.g. "qwen3.5-plus" -> first token "qwen3.5", but prefix is "qwen" - for prefix, vendor in _VENDOR_PREFIXES.items(): - if name_lower.startswith(prefix): - return vendor - - return None + return next((vendor for prefix, vendor in _VENDOR_PREFIXES.items() if name_lower.startswith(prefix)), None) def _prepend_vendor(model_name: str) -> str: - """Prepend the detected ``vendor/`` prefix if missing. - - For aggregators that require ``vendor/model``. Names already containing ``/`` or with no - detectable vendor are returned unchanged (the aggregator may still accept them). - """ + """Prepend the detected ``vendor/`` for aggregators; names with ``/`` or no detectable vendor + pass through (the aggregator may still accept them).""" if "/" in model_name: return model_name - vendor = detect_vendor(model_name) - if vendor: - return f"{vendor}/{model_name}" - return model_name + return f"{vendor}/{model_name}" if vendor else model_name def _repair_prefix_from_catalogue(model_name: str, provider: str) -> str: - """Restore a dropped ``vendor/`` prefix using the provider's catalogue. - - Unlike :func:`_prepend_vendor`, this never guesses from the model's name shape — it only repairs - a bare id that matches **exactly one** curated entry for this provider modulo the prefix. - """ + """Restore a dropped ``vendor/`` prefix only when the bare id matches **exactly one** curated + entry for this provider modulo the prefix — a lookup, never a guess from name shape.""" if "/" in model_name: return model_name try: from hermes_cli.models import _PROVIDER_MODELS except Exception: return model_name - - catalogue = _PROVIDER_MODELS.get(provider) or [] - # Compare against the catalogue's own suffix, tag included: a bare - # ``…:free`` id must resolve to the ``:free`` entry, not its paid sibling. + # Compare against the catalogue's own suffix, tag included: a bare ``…:free`` id must resolve to + # the ``:free`` entry, not its paid sibling. needle = model_name.strip().lower() - matches = { - entry - for entry in catalogue - if "/" in entry and entry.split("/", 1)[1].strip().lower() == needle - } - if len(matches) == 1: - return matches.pop() - return model_name + matches = {e for e in _PROVIDER_MODELS.get(provider) or [] if "/" in e and e.split("/", 1)[1].strip().lower() == needle} + return matches.pop() if len(matches) == 1 else model_name def suggest_prefixed_model_id(provider: str, model_name: str) -> Optional[str]: - """Return the prefixed catalogue id for a bare *model_name*, if unambiguous. - - Diagnostic counterpart to :func:`_repair_prefix_from_catalogue`, used to explain a provider's - content-free 404 when the configured id lost its ``vendor/`` prefix. Returns ``None`` when the - name already has a prefix, the provider has no catalogue, or nothing matches — so callers stay - silent rather than guess. - """ + """Prefixed catalogue id for a bare *model_name* if unambiguous, else ``None`` — the diagnostic + counterpart to :func:`_repair_prefix_from_catalogue` for explaining a content-free 404.""" name = (model_name or "").strip() if not name or "/" in name: return None @@ -312,59 +213,35 @@ def suggest_prefixed_model_id(provider: str, model_name: str) -> Optional[str]: return repaired if repaired != name else None -# --------------------------------------------------------------------------- -# Main normalisation entry point -# --------------------------------------------------------------------------- - def normalize_model_for_provider(model_input: str, target_provider: str) -> str: - """Translate a model name into the format the target provider's API expects. - - Primary entry point for model-name normalisation. Accepts bare, vendor-prefixed or native ids; - ``target_provider`` should already be normalised via ``normalize_provider()``. Never raises — - always returns a best-effort string. - """ + """Translate a model name (bare, vendor-prefixed or native) into what the target provider's API + expects. ``target_provider`` should already be canonical. Never raises.""" name = (model_input or "").strip() if not name: return name - provider = _normalize_provider_alias(target_provider) - # --- Aggregators: need vendor/model format --- if provider in _AGGREGATOR_PROVIDERS: return _prepend_vendor(name) - # --- OpenCode Zen / OpenCode Go: flat-namespace resellers. - # Their /v1/models API returns bare IDs only (no vendor prefix), and - # the inference endpoint rejects vendor-prefixed names with HTTP 401 - # "Model not supported". Strip ANY leading ``vendor/`` so config - # entries like ``minimax/minimax-m2.7`` or ``deepseek/deepseek-v4-flash`` - # — commonly copied from aggregator slugs into fallback_model lists — - # resolve to bare ``minimax-m2.7`` / ``deepseek-v4-flash`` the API - # actually serves. See PR reviewing opencode-go fallback 401s. --- + # OpenCode Zen / Go are flat-namespace resellers: /v1/models returns bare IDs and inference 401s + # vendor-prefixed names, so strip ANY leading ``vendor/`` (commonly copied from aggregator slugs). from hermes_cli.models import opencode_provider_family _oc_family = opencode_provider_family(provider) if _oc_family is not None: if "/" in name: - _, bare_after_slash = name.split("/", 1) - name = bare_after_slash.strip() or name + name = name.split("/", 1)[1].strip() or name if _oc_family == "opencode-zen" and name.lower().startswith("claude-"): return _dots_to_hyphens(name) return name - # --- Anthropic: strip matching provider prefix, dots -> hyphens --- if provider in _DOT_TO_HYPHEN_PROVIDERS: bare = _strip_matching_provider_prefix(name, provider) - if "/" in bare: - return bare - return _dots_to_hyphens(bare) + return bare if "/" in bare else _dots_to_hyphens(bare) - # --- Copilot / Copilot ACP: delegate to the Copilot-specific - # normalizer. It knows about the alias table (vendor-prefix - # stripping for Anthropic/OpenAI, dash-to-dot repair for Claude) - # and live-catalog lookups. Without this, vendor-prefixed or - # dash-notation Claude IDs survive to the Copilot API and hit - # HTTP 400 "model_not_supported". See issue #6879. + # Copilot's own normalizer knows the alias table (vendor stripping, dash-to-dot repair for Claude) + # and live-catalog lookups; without it dash-notation Claude ids hit HTTP 400 model_not_supported. if provider in {"copilot", "copilot-acp"}: try: from hermes_cli.models import normalize_copilot_model_id @@ -373,50 +250,25 @@ def normalize_model_for_provider(model_input: str, target_provider: str) -> str: if normalized: return normalized except Exception: - # Fall through to the generic strip-vendor behaviour below - # if the Copilot-specific path is unavailable for any reason. - pass + pass # fall through to the generic strip-vendor behaviour - # --- Copilot / Copilot ACP / openai-codex fallback: - # strip matching provider prefix, keep dots --- if provider in _STRIP_VENDOR_ONLY_PROVIDERS: stripped = _strip_matching_provider_prefix(name, provider) if stripped == name and name.startswith("openai/"): - # openai-codex maps openai/gpt-5.4 -> gpt-5.4 - return name.split("/", 1)[1] + return name.split("/", 1)[1] # openai-codex maps openai/gpt-5.4 -> gpt-5.4 return stripped - # --- DeepSeek: map to one of two canonical names --- if provider == "deepseek": bare = _strip_matching_provider_prefix(name, provider) - if "/" in bare: - return bare - return _normalize_for_deepseek(bare) + return bare if "/" in bare else _normalize_for_deepseek(bare) - # --- Direct providers: repair matching provider prefixes only --- if provider in _MATCHING_PREFIX_STRIP_PROVIDERS: result = _strip_matching_provider_prefix(name, provider) - # Some providers require lowercase model IDs (e.g. Xiaomi's API - # rejects "MiMo-V2.5-Pro" but accepts "mimo-v2.5-pro"). - if provider in _LOWERCASE_MODEL_PROVIDERS: - result = result.lower() - return result + return result.lower() if provider in _LOWERCASE_MODEL_PROVIDERS else result - # --- Catalogue-backed prefix repair: restore a dropped ``vendor/`` on a - # bare id that matches exactly one curated entry. Unknown names (a - # local NIM container, a proxied model) pass through untouched. --- + # Unknown names (a local NIM container, a proxied model) pass through untouched. if provider in _CATALOGUE_PREFIX_REPAIR_PROVIDERS: return _repair_prefix_from_catalogue(name, provider) - # --- Authoritative native providers: preserve user-facing slugs as-is --- - if provider in _AUTHORITATIVE_NATIVE_PROVIDERS: - return name - - # --- Custom & all others: pass through as-is --- + # Authoritative native providers, custom and all others: pass through as-is. return name - - -# --------------------------------------------------------------------------- -# Batch / convenience helpers -# --------------------------------------------------------------------------- - diff --git a/hermes_cli/model_search.py b/hermes_cli/model_search.py index 74bb8daa27..57673b2438 100644 --- a/hermes_cli/model_search.py +++ b/hermes_cli/model_search.py @@ -10,14 +10,10 @@ _MODEL_SEARCH_ALIASES: dict[str, tuple[str, ...]] = { "x-preview-f-free": ("ox-alpha", "ox"), } -# Lowercased wire id → canonical public slug it aliases. Used by picker -# dedup so a live bare id and its curated public slug (``k3`` / ``kimi-k3``) -# don't render as two rows for the same model. Derived from the FIRST alias -# entry, which by convention is the full public slug. +# Lowercased wire id → canonical public slug (the FIRST alias by convention), so picker dedup doesn't +# render a live bare id and its curated slug (``k3`` / ``kimi-k3``) as two rows. _MODEL_ALIAS_CANONICAL: dict[str, str] = { - wire_id: aliases[0].lower() - for wire_id, aliases in _MODEL_SEARCH_ALIASES.items() - if aliases + wire_id: aliases[0].lower() for wire_id, aliases in _MODEL_SEARCH_ALIASES.items() if aliases } @@ -28,14 +24,9 @@ def model_alias_canonical(model: str) -> str: def model_search_text(model: str) -> str: - """Return the haystack used for fuzzy/substring model search. - - Never changes the wire id passed to the provider. - """ + """Haystack for fuzzy/substring model search; never changes the wire id sent to the provider.""" mid = (model or "").strip() if not mid: return model or "" aliases = _MODEL_SEARCH_ALIASES.get(mid.lower()) - if not aliases: - return mid - return f"{mid} {' '.join(aliases)}" + return f"{mid} {' '.join(aliases)}" if aliases else mid diff --git a/hermes_cli/model_selection_guards.py b/hermes_cli/model_selection_guards.py index 8ce700f096..1f4f3e2291 100644 --- a/hermes_cli/model_selection_guards.py +++ b/hermes_cli/model_selection_guards.py @@ -24,11 +24,8 @@ class SelectionWarning: def _wrap(kind: str, title: str, warning, model_name: str, provider: Optional[str]): - """Lift a raw guard payload into a :class:`SelectionWarning` (None passes through). - - Duck-typed access: tests (and future guard payloads) may supply objects carrying only - ``.message``. - """ + """Lift a raw guard payload into a :class:`SelectionWarning` (None passes through). Duck-typed: + payloads may carry only ``.message``.""" if warning is None: return None return SelectionWarning( @@ -82,14 +79,8 @@ def selection_warnings( model_info: Optional[ModelInfo] = None, include_kinds: Optional[Iterable[str]] = None, ) -> List[SelectionWarning]: - """Run every registered selection guard and return the warnings that fired. - - Returns an empty list in the common case (no guard fired). Callers should run this after model - resolution so aliases / provider-specific ids have settled, then surface the messages as a - confirm step. ``include_kinds`` optionally restricts which guard kinds run (e.g. - - A misbehaving guard must never break model selection: individual guard exceptions are swallowed. - """ + """Warnings from every registered guard (empty in the common case). ``include_kinds`` restricts + which kinds are returned. Guard exceptions are swallowed — never break model selection.""" wanted = set(include_kinds) if include_kinds is not None else None results: List[SelectionWarning] = [] for guard in _GUARDS: @@ -103,11 +94,7 @@ def selection_warnings( def combined_message(warnings: List[SelectionWarning]) -> str: - """Join multiple warnings into one confirm-prompt body. - - Used by surfaces with a single confirm dialog when more than one guard fires (rare) — one - prompt showing both blocks beats two sequential prompts. - """ + """One confirm-prompt body for several warnings (one prompt beats two sequential ones).""" return "\n\n".join(w.message for w in warnings) @@ -119,12 +106,8 @@ def combined_selection_warning( api_key: Optional[str] = None, model_info: Optional[ModelInfo] = None, ) -> Optional[SelectionWarning]: - """Drop-in replacement for ``expensive_model_warning`` call sites. - - Returns ``None`` when no guard fired, the single :class:`SelectionWarning` when one fired, - or a merged ``kind="multiple"`` warning stacking every message — so surfaces rendering one - confirm dialog from ``warning.message`` can switch without reshaping control flow. - """ + """Drop-in for ``expensive_model_warning`` call sites: ``None``, the single warning, or a merged + ``kind="multiple"`` warning stacking every message.""" warnings = selection_warnings( model_name, provider=provider, base_url=base_url, api_key=api_key, model_info=model_info )