mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-11-15 18:35:29 +03:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
hermes-agent/hermes_cli/models.py
T
Maciej Lisiewski 72620bd99c fix(models): Anthropic model discovery at a Foundry URL sends the Foundry key, not the OAuth token 
_fetch_anthropic_models filled a missing key with resolve_anthropic_token(),
so the /model picker and /model validation sent the Claude Code OAuth token
to <foundry>/v1/models whenever the key lived in key_env/AZURE_ANTHROPIC_KEY.
At an azure.com URL it now uses anthropic_foundry_key() and sends nothing
without one. Docs: Foundry known limits (auth-add rows, sovereign hosts). (#154242)
2026-11-14 15:35:26 +01:00

2654 lines
126 KiB
Python
Raw Permalink Blame History

"""Provider/model catalogs: discovery, caching, and identity helpers.
Origin module; cohesive clusters live in siblings and are re-imported here so
``hermes_cli.models.<name>`` stays the stable import/monkeypatch surface:
``models_catalog_static`` (curated tables, provider registry, aliases), ``models_reasoning_caps``,
``models_local`` (Ollama / LM Studio), ``models_pricing``, ``models_validate``.
"""
from __future__ import annotations
import contextvars
import copy
import gzip
import json
import logging
import os
import re
import sys
import threading
import urllib.parse
import urllib.request
import urllib.error
import time
from pathlib import Path
from typing import Any, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from typing import TypeGuard
from agent.proxy_bypass import is_loopback_host
from agent.bedrock_service_tiers import BEDROCK_ULTRAFAST_ROUTES, bedrock_fast_mode_overrides
from hermes_cli.route_identity import normalize_route_base_url
from hermes_cli.urllib_security import open_credentialed_url
from hermes_cli.version_info import hermes_user_agent
from hermes_cli.models_catalog_static import (
CuratedFallbackModels,
CANONICAL_PROVIDERS,
OPENROUTER_MODELS,
PREFERRED_SILENT_DEFAULT_MODEL,
VERCEL_AI_GATEWAY_MODELS,
_AGGREGATOR_PROVIDERS,
_AZURE_FOUNDRY_RESPONSES_PREFIXES,
_BORROWED_MODEL_PROVIDERS,
_LIVE_FIRST_PICKER_PROVIDERS,
_MODELS_DEV_PREFERRED,
_OPENAI_FAST_MODE_PREFIXES,
_OPENAI_ULTRAFAST_MODELS,
_PROVIDER_ALIASES,
_PROVIDER_LABELS,
_PROVIDER_MODELS,
_PROVIDER_RETIRED_ALIASES,
_SILENT_DEFAULT_PROVIDERS,
_xai_finalize_catalog)
from hermes_cli.models_reasoning_caps import (
_OPENROUTER_CATALOG_URL,
_seed_reasoning_caps)
from hermes_cli.models_local import (
_OLLAMA_LOCAL_MODELS_CACHE,
_OLLAMA_LOCAL_MODELS_CACHE_TTL,
_OLLAMA_LOCAL_PROBE_FAILURE_CACHE,
_OLLAMA_LOCAL_PROBE_REACHABLE,
_get_ollama_base_url,
_get_ollama_native_headers,
_ollama_local_catalog,
_ollama_probe_cache_key,
_root_for_ollama_native_api,
fetch_ollama_cloud_models)
logger = logging.getLogger(__name__)
def _urlopen_model_catalog_request(req: urllib.request.Request, *, timeout: float, ssl_context=None):
"""Open catalog requests without forwarding headers across origins."""
return open_credentialed_url(req, timeout=timeout, ssl_context=ssl_context)
def _get_json(
url: str, *, timeout: float, headers: Optional[dict[str, str]] = None, opener=None, **open_kwargs: Any
) -> Any:
"""GET ``url`` and parse the JSON body. ``opener`` defaults to the catalog opener (resolved at
call time so monkeypatching ``_urlopen_model_catalog_request`` still applies). Raises on failure."""
req = urllib.request.Request(url, headers=headers or {})
with (opener or _urlopen_model_catalog_request)(req, timeout=timeout, **open_kwargs) as resp:
body = resp.read()
if req.get_header("Accept-encoding") == "gzip" and resp.headers.get("Content-Encoding", "").lower() == "gzip":
body = gzip.decompress(body)
return json.loads(body.decode())
def _read_json_cache(path: Path, *, errors=Exception) -> Optional[dict]:
"""Load a JSON-object cache file; None when missing, unreadable, or not a dict."""
try:
with open(path, encoding="utf-8-sig") as fh:
data = json.load(fh)
except errors:
return None
return data if isinstance(data, dict) else None
def _write_json_cache(path: Path, data: Any, **dump_kwargs: Any) -> None:
"""Atomically persist a cache file (creating parents). Raises on failure — callers decide
whether a failed cache write is worth logging."""
from utils import atomic_json_write
from hermes_constants import mkdir_under_hermes_home
mkdir_under_hermes_home(path.parent)
atomic_json_write(path, data, **dump_kwargs)
def _merge_unique(primary: list[str], secondary: list[str], key=lambda m: str(m).lower()) -> list[str]:
"""``primary`` verbatim, then ``secondary`` entries whose ``key`` is new (deduped as it goes)."""
merged, seen = list(primary), {key(m) for m in primary}
for m in secondary:
k = key(m)
if k not in seen:
seen.add(k)
merged.append(m)
return merged
def _custom_provider_ssl_context(base_url: str):
"""Use the same trust decision for urllib catalogs and HTTPX metadata/chat."""
from agent.model_metadata_http import resolve_verify
verify = resolve_verify(base_url)
if verify is True:
return None
if verify is False:
import ssl
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context
return verify
# Process-lifetime picker lists refreshed from the live catalogs (see fetch_*_models).
_openrouter_catalog_cache: list[tuple[str, str]] | None = None
# The in-memory ``_openrouter_catalog_cache`` is per-process, so without a disk cache every cold
# picker open re-downloads the full ~686KB /api/v1/models catalog. The *curated* result
# (post-filter) is persisted under the same TTL the catalog manifest uses, so both layers go
# stale together.
def _openrouter_catalog_disk_ttl() -> float:
"""Same TTL as the catalog manifest this list is filtered from (honours ``model_catalog.ttl_minutes``)."""
from hermes_cli.model_catalog import refresh_interval_seconds
return refresh_interval_seconds()
def _openrouter_catalog_disk_path() -> Path:
from hermes_constants import get_hermes_home
return get_hermes_home() / "cache" / "openrouter_curated_catalog.json"
def _read_openrouter_catalog_disk(*, allow_stale: bool = False) -> list[tuple[str, str]] | None:
"""Fresh curated catalog from disk, or None (missing, corrupt, expired, or empty).
``allow_stale`` ignores the TTL — the cache-only read path prefers a stale copy over a live GET."""
obj = _read_json_cache(_openrouter_catalog_disk_path())
if obj is None:
return None
try:
if not allow_stale and time.time() - float(obj.get("fetched_at", 0)) > _openrouter_catalog_disk_ttl():
return None
except (TypeError, ValueError):
return None
items = obj.get("curated")
if not isinstance(items, list):
return None
out = [(str(it[0]), str(it[1])) for it in items if isinstance(it, (list, tuple)) and len(it) == 2]
return out or None
def _write_openrouter_catalog_disk(curated: list[tuple[str, str]]) -> None:
try:
_write_json_cache(
_openrouter_catalog_disk_path(),
{"fetched_at": time.time(), "curated": [list(c) for c in curated]})
except Exception as exc:
logger.debug("openrouter curated catalog disk write failed: %s", exc)
_ai_gateway_catalog_cache: list[tuple[str, str]] | None = None
# ---------------------------------------------------------------------------
# Nous Portal free-model helpers — the Portal models endpoint is the source of truth for what is
# offered (free or paid); we surface it as-is, no local allowlist filtering.
# ---------------------------------------------------------------------------
def _zero_priced(pricing: Any, keys: tuple[str, str], default: str) -> bool:
"""True when both pricing fields parse to 0 (missing fields read as ``default``)."""
if not isinstance(pricing, dict):
return False
try:
return all(float(pricing.get(k, default)) == 0 for k in keys)
except (TypeError, ValueError):
return False
def _is_subscription_billed(entry: Any) -> bool:
"""The gateway bills this catalog row to a subscription the account holds, not to credits."""
return isinstance(entry, dict) and entry.get("billing_mode") == "subscription"
def _is_model_free(model_id: str, pricing: dict[str, dict[str, str]]) -> bool:
"""Return True if *model_id* costs no credits: zero-cost prompt AND completion pricing, or a row
the gateway bills to a subscription."""
entry = pricing.get(model_id)
return bool(entry) and (_is_subscription_billed(entry) or _zero_priced(entry, ("prompt", "completion"), "1"))
def partition_nous_models_by_tier(
model_ids: list[str], pricing: dict[str, dict[str, str]], free_tier: bool
) -> tuple[list[str], list[str]]:
"""Split Nous models into (selectable, unavailable): free-tier users may only select free models
(paid ones are returned as unavailable, shown grayed out)."""
if not free_tier or not pricing: # no pricing → can't determine, show everything
return (model_ids, [])
selectable = [mid for mid in model_ids if _is_model_free(mid, pricing)]
return (selectable, [mid for mid in model_ids if mid not in selectable])
def _union_with_portal_recommendations(
tier_key: str, curated_ids: list[str], pricing: dict[str, dict[str, str]], portal_base_url: str,
*, force_refresh: bool, synthesize_free_pricing: bool,
) -> tuple[list[str], dict[str, dict[str, str]]]:
"""Append the Portal's ``<tier_key>`` recommendations missing from ``curated_ids``.
Curated models show first, Portal-only picks follow. Failures (network, parse, missing field)
silently return the inputs unchanged — never block the picker on a Portal-side hiccup.
"""
try:
payload = fetch_nous_recommended_models(portal_base_url, force_refresh=force_refresh)
except Exception:
payload = None
block = payload.get(tier_key) if isinstance(payload, dict) else None
entries = block if isinstance(block, list) else []
portal_ids = [name for entry in entries if (name := _extract_model_name(entry))]
if not portal_ids:
return (list(curated_ids), dict(pricing))
augmented_pricing = dict(pricing)
if synthesize_free_pricing:
for mid in portal_ids:
augmented_pricing.setdefault(mid, {"prompt": "0", "completion": "0"})
seen = set(curated_ids)
return (list(curated_ids) + [mid for mid in portal_ids if mid not in seen], augmented_pricing)
def union_with_portal_free_recommendations(
curated_ids: list[str], pricing: dict[str, dict[str, str]], portal_base_url: str = "", *,
force_refresh: bool = False) -> tuple[list[str], dict[str, dict[str, str]]]:
"""Curated list + pricing plus the Portal's ``freeRecommendedModels``; Portal-only free picks get a
synthetic $0 pricing entry so tier partitioning sees them as free."""
return _union_with_portal_recommendations(
"freeRecommendedModels", curated_ids, pricing, portal_base_url,
force_refresh=force_refresh, synthesize_free_pricing=True)
def union_with_portal_paid_recommendations(
curated_ids: list[str], pricing: dict[str, dict[str, str]], portal_base_url: str = "", *,
force_refresh: bool = False) -> tuple[list[str], dict[str, dict[str, str]]]:
"""Curated list plus the Portal's ``paidRecommendedModels``; ``pricing`` is deliberately left untouched."""
return _union_with_portal_recommendations(
"paidRecommendedModels", curated_ids, pricing, portal_base_url,
force_refresh=force_refresh, synthesize_free_pricing=False)
def union_with_nous_on_sale_models(curated_ids: list[str], pricing: dict[str, dict[str, Any]]) -> list[str]:
"""Curated list plus every paid Nous model the gateway is discounting right now, deepest
discount first. A sale lives only in ``/v1/models`` ``pricing.original``, so without this the
picker badges discounts on curated rows but never shows a discounted model the curated list and
Portal recommendations omit. Free rows stay with ``freeRecommendedModels``; rows the gateway
marks tool-less (``"tools": False``) are skipped because Hermes is tool-calling-first, and
image/video generation rows are skipped because they are not chat models."""
from math import isfinite
from hermes_cli.models_pricing import _price_float, compute_sale_discount
seen = set(curated_ids)
on_sale: list[tuple[int, str]] = []
for mid, entry in (pricing or {}).items():
if mid in seen or not isinstance(entry, dict) or entry.get("tools") is False or entry.get("generation"):
continue
sale = compute_sale_discount(entry.get("prompt", ""), entry.get("completion", ""), entry.get("original"))
# Badge percentages are rounded: a nearly-free paid row can display 100% off.
paid = any(
(rate := _price_float(entry.get(key), positive=True)) is not None and isfinite(rate)
for key in ("prompt", "completion")
)
if sale is not None and isinstance(entry.get("original"), dict) and paid:
on_sale.append((-sale[0], mid))
return list(curated_ids) + [mid for _, mid in sorted(on_sale)]
# Free-tier detection cache, per profile — short so an account upgrade shows within minutes.
_FREE_TIER_CACHE_TTL: int = 180 # seconds
_free_tier_cache: dict[str, tuple[bool, float]] = {} # profile key -> (result, timestamp)
def _pricing_profile_key() -> str:
"""Stable profile identity for process-local pricing caches."""
from hermes_constants import hermes_home_key
return hermes_home_key()
def get_cached_nous_free_tier() -> Optional[bool]:
"""This profile's live cached entitlement, or ``None`` if unknown/expired."""
cached = _free_tier_cache.get(_pricing_profile_key())
if cached is None or time.monotonic() - cached[1] >= _FREE_TIER_CACHE_TTL:
return None
return cached[0]
def check_nous_free_tier(*, force_fresh: bool = False, cached_only: bool = False) -> bool:
"""True only when the Nous Portal user is KNOWN to be free-tier (unknown/error → False so this
never blocks users). Cached ``_FREE_TIER_CACHE_TTL`` seconds so an upgrade shows within minutes.
``cached_only`` returns the live cached answer or the fail-open ``False`` without contacting Portal."""
now = time.monotonic()
profile_key = _pricing_profile_key()
if not force_fresh:
cached_result = get_cached_nous_free_tier()
if cached_result is not None:
return cached_result
if cached_only:
return False
try:
from hermes_cli.nous_account import get_nous_portal_account_info
result = get_nous_portal_account_info(force_fresh=force_fresh).is_free_tier
except Exception:
result = False # default to paid on error — don't block users
_free_tier_cache[profile_key] = (result, now)
return result
# ---------------------------------------------------------------------------
# Nous Portal recommended models — curated paid/free suggestions plus dedicated compaction (aux)
# and vision picks, TTL-cached per process. Fields read: {paid,free}RecommendedModels:
# [{modelName}], {paid,free}Recommended{Compaction,Vision}Model: {modelName} | null
# ---------------------------------------------------------------------------
NOUS_RECOMMENDED_MODELS_MIRROR = "https://web-assets.nousresearch.com/dynamic"
_NOUS_MIRROR_TIMEOUT: float = 2.0 # seconds: a hanging mirror delays the Portal fallback by at most this
_NOUS_RECOMMENDED_CACHE_TTL: int = 600 # seconds (10 minutes)
# (result_dict, monotonic timestamp), scoped to the profile and portal.
_nous_recommended_cache: dict[tuple[str, str], tuple[dict[str, Any], float]] = {}
def _nous_recommended_disk_path() -> Path:
from hermes_constants import get_hermes_home
return get_hermes_home() / "cache" / "nous_recommended_cache.json"
def _read_nous_recommended_disk(base: str) -> tuple[dict[str, Any], float] | None:
"""Return the last good payload and its age for the freshness check."""
blob = _read_json_cache(_nous_recommended_disk_path(), errors=(OSError, json.JSONDecodeError, UnicodeDecodeError))
entry = (blob or {}).get(base)
data = entry.get("data") if isinstance(entry, dict) else None
if not isinstance(data, dict) or not data:
return None
try:
age = time.time() - float(entry.get("ts", 0))
except (TypeError, ValueError, OverflowError):
age = float("inf")
return data, age
def _write_nous_recommended_disk(base: str, data: dict[str, Any]) -> None:
"""Merge ``data`` into the per-base disk map atomically; failures are debug-logged (the in-process
cache still works)."""
if not data:
return
path = _nous_recommended_disk_path()
try:
blob = _read_json_cache(path, errors=(OSError, json.JSONDecodeError, UnicodeDecodeError)) or {}
blob[base] = {"data": data, "ts": time.time()}
_write_json_cache(path, blob, indent=2)
except OSError as exc:
logger.debug("nous recommended-models disk cache write failed: %s", exc)
def fetch_nous_recommended_models(
portal_base_url: str = "", timeout: float = 5.0, *, force_refresh: bool = False
) -> dict[str, Any]:
"""Fetch the Portal's public recommended models (no auth): its R2 mirror, else the Portal itself.
Results are reused for ``_NOUS_RECOMMENDED_CACHE_TTL`` s across restarts unless ``force_refresh``;
stale disk data is the fallback on failure, and reading it never renews its freshness."""
base = (portal_base_url or "https://portal.nousresearch.com").rstrip("/")
now = time.monotonic()
cache_key = (_pricing_profile_key(), base)
cached = _nous_recommended_cache.get(cache_key)
if not force_refresh and cached is not None and now - cached[1] < _NOUS_RECOMMENDED_CACHE_TTL:
return cached[0]
disk = _read_nous_recommended_disk(base)
if not force_refresh and disk is not None and 0 <= disk[1] < _NOUS_RECOMMENDED_CACHE_TTL:
data, age = disk
_nous_recommended_cache[cache_key] = (data, now - age)
return data
parts = urllib.parse.urlsplit(base)
mirror = [] if is_loopback_host(parts.hostname) else [
(f"{NOUS_RECOMMENDED_MODELS_MIRROR}/{parts.netloc}", min(timeout, _NOUS_MIRROR_TIMEOUT))]
for origin, origin_timeout in [*mirror, (base, timeout)]: # the Portal itself when its mirror is missing (404) or failing
try:
data = _get_json(f"{origin}/api/nous/recommended-models", timeout=origin_timeout,
headers={"Accept": "application/json", "Accept-Encoding": "gzip"})
except Exception:
continue
if isinstance(data, dict) and data:
_write_nous_recommended_disk(base, data)
break
else:
data = disk[0] if disk is not None else {}
_nous_recommended_cache[cache_key] = (data, now)
return data
def _resolve_nous_portal_url() -> str:
"""Best-effort lookup of the Portal base URL the user is authed against."""
try:
from hermes_cli.auth import DEFAULT_NOUS_PORTAL_URL, get_provider_auth_state
state = get_provider_auth_state("nous") or {}
portal = str(state.get("portal_base_url") or "").strip()
return (portal or str(DEFAULT_NOUS_PORTAL_URL)).rstrip("/")
except Exception:
return "https://portal.nousresearch.com"
def _extract_model_name(entry: Any) -> Optional[str]:
"""Pull the ``modelName`` field from a recommended-model entry, else None."""
model_name = entry.get("modelName") if isinstance(entry, dict) else None
return model_name.strip() if isinstance(model_name, str) and model_name.strip() else None
def get_nous_recommended_aux_model(
*, vision: bool = False, free_tier: Optional[bool] = None, portal_base_url: str = "",
force_refresh: bool = False) -> Optional[str]:
"""The Portal's recommended model for an auxiliary task: free tier → free pick only; paid tier →
paid pick, falling back to the free one when the Portal returned ``null`` (staged rollouts)."""
base = portal_base_url or _resolve_nous_portal_url()
payload = fetch_nous_recommended_models(base, force_refresh=force_refresh)
if not payload:
return None
if free_tier is None:
try:
free_tier = check_nous_free_tier()
except Exception:
free_tier = False # assume paid on detection error — paid users see both fields anyway
kind = "Vision" if vision else "Compaction"
tiers = ("free",) if free_tier else ("paid", "free")
return next((n for t in tiers if (n := _extract_model_name(payload.get(f"{t}Recommended{kind}Model")))), None)
def get_preferred_silent_default_model(provider: str = "openrouter") -> str:
"""Silent-default model id: the cached remote catalog's ``"default": true`` label (never hits the
network — safe on hot paths), else :data:`PREFERRED_SILENT_DEFAULT_MODEL`."""
try:
from hermes_cli.model_catalog import get_default_model_from_cache
labeled = get_default_model_from_cache(provider)
if labeled:
return labeled
except Exception:
pass
return PREFERRED_SILENT_DEFAULT_MODEL
def _is_anthropic_frontier_tier(model_id: Optional[str]) -> bool:
"""Return True for Anthropic frontier-tier models (Opus + Fable).
Both tiers are the priciest Anthropic offerings and are unsafe defaults
for a freshly-authenticated provider picker or any other silent fallback:
paid users landing on a frontier tier have no opportunity to opt out
before the choice pins their main model and inherits into every cron job
that doesn't override ``model.provider``.
Anchored on the ``claude-`` family prefix, then the ``opus-`` / ``fable-``
tier token (after vendor-strip and lowercase normalization), so the
predicate survives future Anthropic releases (``claude-opus-5-0``,
``claude-fable-6``, etc.) and rejects community / distill models whose
slug merely *contains* the substring ``opus`` (e.g. ``qwopus3.6-27b-coder``).
Sonnet and Haiku are deliberately NOT classified as frontier here — they
remain reasonable auto-default candidates.
"""
raw = _strip_vendor_prefix(str(model_id or ""))
base = raw.split(":")[0].lower()
if not base.startswith("claude-"):
return False
return ("opus-" in base) or ("fable-" in base)
def pick_silent_default_model(model_ids: list[str], provider: str = "openrouter") -> str:
"""Pick a cost-safe silent default from an available-models list.
Shared policy for every surface that must choose a model on the user's
behalf without an interactive confirmation (GUI onboarding
recommended-default, empty-model runtime fallback, provider-set-but-
model-missing resolution):
1. If the catalog-labeled preferred default (see
:func:`get_preferred_silent_default_model`) is present in
``model_ids``, return it.
2. Else, return the first entry that is NOT an Anthropic frontier tier
(Opus / Fable — see :func:`_is_anthropic_frontier_tier`). Catalog
ordering is preserved, so a freshly-released cheaper model still
wins over a leftover Sonnet.
3. Else (every entry is a frontier tier), return ``model_ids[0]`` so the
picker is never empty — defensive last resort only.
4. Else (empty list), return ``\"\"``.
This hardened fallback closes the hole where a Portal-augmented list
that didn't carry the catalog label fell through to ``model_ids[0]``
(currently ``anthropic/claude-fable-5``) and billed the most expensive
Anthropic flagship for traffic the user never opted into.
"""
if not model_ids:
return ""
preferred = get_preferred_silent_default_model(provider)
if preferred in model_ids:
return preferred
non_frontier = [mid for mid in model_ids if not _is_anthropic_frontier_tier(mid)]
if non_frontier:
return non_frontier[0]
return model_ids[0]
def recommended_nous_default_model() -> dict[str, Any]:
"""The model a Nous account lands on without choosing one, honouring the account's tier.
Curated catalog plus the Portal's recommendations for the tier, narrowed to the org's policy,
then (free tier) to the rows the tier may select, then :func:`pick_silent_default_model`.
Contacts the Portal for a fresh tier read, so never call it on a hot path. Returns
``{"provider": "nous", "model": str, "free_tier": bool}``; ``model`` may be ``""`` when nothing
is selectable (callers degrade). Shared by ``GET /api/model/recommended-default`` and the
sign-in completion in ``hermes_cli.anon_auth`` so both land on the same model.
"""
from hermes_cli import model_picker_usage, models_pricing as mp
from hermes_cli.auth import get_provider_auth_state
model_ids = get_curated_nous_model_ids()
pricing = mp.get_pricing_for_provider("nous") or {}
free_tier = check_nous_free_tier(force_fresh=True)
try:
portal_url = (get_provider_auth_state("nous") or {}).get("portal_base_url", "") or ""
except Exception:
portal_url = ""
# Narrow to policy BEFORE the tier split, so a rescued id still has to pass the free/paid predicate.
policy_allowed = mp.nous_policy_allowed_ids()
union = union_with_portal_free_recommendations if free_tier else union_with_portal_paid_recommendations
model_ids, pricing = union(model_ids, pricing, portal_url)
model_ids = mp.restrict_to_nous_policy(model_ids, policy_allowed, rescue_empty=True)
if free_tier:
model_ids, _unavailable = partition_nous_models_by_tier(model_ids, pricing, free_tier=True)
# Never default onto a subscription-billed row: spending that plan is the user's call.
model_ids = [mid for mid in model_ids if not _is_subscription_billed(pricing.get(mid))] or model_ids
model = "" if free_tier else model_picker_usage.window_plan_default(model_ids, pricing)
return {"provider": "nous", "model": model or pick_silent_default_model(model_ids, provider="nous"),
"free_tier": bool(free_tier)}
def static_catalog(provider: str) -> list[str]:
"""Curated ids for *provider*: its built-in row, else the bound home's profile ``static_models``.
Resolved per lookup, never copied into the shared table: two served homes may register the same
provider name with different models (or only one of them has the plugin)."""
if provider in _PROVIDER_MODELS:
return _PROVIDER_MODELS[provider]
from providers import get_provider_profile
profile = get_provider_profile(provider) if provider else None
return list(profile.static_models) if profile is not None else []
def static_catalog_providers() -> list[str]:
"""Every provider with a static catalog row in the bound home: built-ins, then plugin profiles."""
from providers import list_providers
plugin_rows = [p.name for p in list_providers() if p.static_models and p.name not in _PROVIDER_MODELS]
return [*_PROVIDER_MODELS, *dict.fromkeys(plugin_rows)]
def get_default_model_for_provider(provider: str) -> str:
"""Cost-safe default model for a provider, or "" — the NON-INTERACTIVE fallback when a provider
is configured but no model was ever selected."""
models = static_catalog(provider)
if provider in _SILENT_DEFAULT_PROVIDERS:
preferred = get_preferred_silent_default_model(provider)
# Trust the preferred default even without a static catalog (OpenRouter's picker list is
# fetched live; its curated snapshot carries the default).
if preferred and (preferred in models or not models):
return preferred
return models[0] if models else ""
def _openrouter_model_is_free(pricing: Any) -> bool:
return _zero_priced(pricing, ("prompt", "completion"), "0")
def _openrouter_model_supports_tools(item: Any) -> bool:
"""True when ``supported_parameters`` advertises ``tools`` (hermes-agent is tool-calling-first).
Permissive when the field is absent/malformed: some OpenRouter-compatible gateways (Nous Portal,
private mirrors) don't populate it, and the picker must not silently empty for them.
Ported from Kilo-Org/kilocode#9068.
"""
params = item.get("supported_parameters") if isinstance(item, dict) else None
return "tools" in params if isinstance(params, list) else True
# Reasoning-capability cache slots, one set per catalog (OpenRouter, Nous Portal). The logic
# lives in models_reasoning_caps and reads/writes these by name so tests can reset them here.
# ``*_cache``: model id → parsed caps for the process lifetime; ``*_failed_at``: monotonic time
# of the last failed fetch (60s re-fetch suppression); the flags are once-per-process guards.
_openrouter_reasoning_caps_cache: dict[str, Optional[dict[str, Any]]] | None = None
_openrouter_reasoning_caps_failed_at: float | None = None
_openrouter_caps_disk_checked = False
_openrouter_caps_warm_started = False
_nous_reasoning_caps_cache: dict[str, Optional[dict[str, Any]]] | None = None
_nous_reasoning_caps_failed_at: float | None = None
_nous_caps_disk_checked = False
_nous_caps_warm_started = False
from agent.reasoning_effort import CODEX_ASTRA_EFFORTS, clamp_effort as _clamp_effort, is_astra_model
def clamp_reasoning_effort_to_supported(
effort: Optional[str], supported_efforts: Optional[list[str]]) -> Optional[str]:
"""Thin wrapper over :func:`agent.reasoning_effort.clamp_effort`: keep a supported level verbatim,
else the nearest WEAKER supported level (never silently escalate cost), else the weakest; unknown
supported-sets and bespoke level names pass through unchanged."""
return _clamp_effort(effort, supported_efforts)
def _fetch_live_catalog_index(url: str, timeout: float, opener) -> Optional[tuple[list, dict[str, dict[str, Any]]]]:
"""GET an OpenAI-style ``/models`` listing → ``(raw data array, {id: item})``, or None when the
endpoint is unreachable or the payload has no ``data`` list."""
try:
payload = _get_json(url, timeout=timeout, headers={"Accept": "application/json"}, opener=opener)
except Exception:
return None
live_items = payload.get("data", [])
if not isinstance(live_items, list):
return None
live_by_id = {
mid: item for item in live_items if isinstance(item, dict) and (mid := str(item.get("id") or "").strip())
}
return live_items, live_by_id
def fetch_openrouter_models(
timeout: float = 8.0, *, force_refresh: bool = False, cache_only: bool = False) -> list[tuple[str, str]]:
"""Return the curated OpenRouter picker list, refreshed from the live catalog when possible.
``cache_only`` never opens a socket: memory, then disk (stale accepted), then the in-repo snapshot."""
# The curated list is filtered from this profile's manifest (``model_catalog.*`` config, its
# ``<home>/cache`` copy), so a routed profile keeps its own slot instead of the module one.
from hermes_cli.models_profile_cache import profile_slot_get, profile_slot_set
_me = sys.modules[__name__]
cached = profile_slot_get(_me, "_openrouter_catalog_cache")
if cached is not None and not force_refresh:
return list(cached)
# Cold process: serve from the persisted disk cache when fresh so the
# picker doesn't re-download the full ~686KB catalog on every open.
if not force_refresh:
disk = _read_openrouter_catalog_disk(allow_stale=cache_only)
if disk:
if not cache_only: # a stale copy is served, never memoized as fresh
profile_slot_set(_me, "_openrouter_catalog_cache", disk)
return list(disk)
if cache_only:
return list(OPENROUTER_MODELS)
# Remote catalog manifest first, in-repo snapshot when unreachable; the live /v1/models filter
# (tool support, free pricing) is applied on top either way.
try:
from hermes_cli.model_catalog import get_curated_openrouter_models
remote = get_curated_openrouter_models()
except Exception:
remote = None
fallback = list(remote) if remote else list(OPENROUTER_MODELS)
live = _fetch_live_catalog_index(_OPENROUTER_CATALOG_URL, timeout, _urlopen_model_catalog_request)
if live is None:
return list(cached or fallback)
live_items, live_by_id = live
# Free warm-up for the reasoning-capability cache: same payload the caps fetch would pull.
global _openrouter_reasoning_caps_cache
seeded = _seed_reasoning_caps(_OPENROUTER_CATALOG_URL, live_items)
if _openrouter_reasoning_caps_cache is None and seeded is not None:
_openrouter_reasoning_caps_cache = seeded
curated: list[tuple[str, str]] = []
silent_default = get_preferred_silent_default_model("openrouter")
for preferred_id, _ in fallback:
live_item = live_by_id.get(preferred_id)
# Hide models without tool-calling support — selecting one fails at the first tool call.
if live_item is None or not _openrouter_model_supports_tools(live_item):
continue
# Hide models that don't advertise tool-calling support — hermes-agent requires it and surfacing
# them leads to immediate runtime failures when the user selects them. Ported from
# Kilo-Org/kilocode#9068.
if preferred_id == silent_default:
desc = "default" # keep the silent-default badge through the live refresh
else:
desc = "free" if _openrouter_model_is_free(live_item.get("pricing")) else ""
curated.append((preferred_id, desc))
if not curated:
return list(cached or fallback)
if not curated[0][1]:
curated[0] = (curated[0][0], "recommended")
profile_slot_set(_me, "_openrouter_catalog_cache", curated)
_write_openrouter_catalog_disk(curated)
return list(curated)
def model_ids(*, force_refresh: bool = False) -> list[str]:
"""Return just the OpenRouter model-id strings."""
return [mid for mid, _ in fetch_openrouter_models(force_refresh=force_refresh)]
def get_curated_nous_model_ids() -> list[str]:
"""Curated Nous Portal model ids: the remote catalog manifest, else the in-repo
``_PROVIDER_MODELS["nous"]`` snapshot. Always a list."""
try:
from hermes_cli.model_catalog import get_curated_nous_models
remote = get_curated_nous_models()
except Exception:
remote = None
return list(remote or _PROVIDER_MODELS.get("nous", []))
def _ai_gateway_model_is_free(pricing: Any) -> bool:
return _zero_priced(pricing, ("input", "output"), "0")
def fetch_ai_gateway_models(
timeout: float = 8.0, *, force_refresh: bool = False) -> list[tuple[str, str]]:
"""Return the curated AI Gateway picker list, refreshed from the live catalog when possible."""
global _ai_gateway_catalog_cache
if _ai_gateway_catalog_cache is not None and not force_refresh:
return list(_ai_gateway_catalog_cache)
from hermes_constants import AI_GATEWAY_BASE_URL
fallback = list(VERCEL_AI_GATEWAY_MODELS)
live = _fetch_live_catalog_index(f"{AI_GATEWAY_BASE_URL.rstrip('/')}/models", timeout, _urlopen_model_catalog_request)
if live is None:
return list(_ai_gateway_catalog_cache or fallback)
_, live_by_id = live
curated = [
(pid, "free" if _ai_gateway_model_is_free(live_by_id[pid].get("pricing")) else "")
for pid, _ in fallback if pid in live_by_id]
if not curated:
return list(_ai_gateway_catalog_cache or fallback)
# A free Moonshot model in the live catalog is auto-promoted to #1 as "recommended".
free_moonshot = next(
(mid for mid, item in live_by_id.items()
if mid.startswith("moonshotai/") and _ai_gateway_model_is_free(item.get("pricing"))),
None)
if free_moonshot:
curated = [(free_moonshot, "recommended")] + [(mid, desc) for mid, desc in curated if mid != free_moonshot]
else:
curated[0] = (curated[0][0], "recommended")
_ai_gateway_catalog_cache = curated
return list(curated)
def ai_gateway_model_ids(*, force_refresh: bool = False) -> list[str]:
"""Return just the AI Gateway model-id strings."""
return [mid for mid, _ in fetch_ai_gateway_models(force_refresh=force_refresh)]
# ---------------------------------------------------------------------------
# Provider identity: ``provider:model`` parsing, auto-detection, labels
# ---------------------------------------------------------------------------
# All provider IDs and aliases valid on the left of the ``provider:model`` syntax.
_KNOWN_PROVIDER_NAMES: set[str] = set(_PROVIDER_LABELS) | set(_PROVIDER_ALIASES) | {"openrouter", "custom"}
_CONFIG_ERRORS = (ImportError, OSError, RuntimeError, TypeError, ValueError, AttributeError)
def _configured_custom_provider_ids() -> set[str]:
"""Return routable custom-provider IDs configured by the user."""
ids = {"custom"}
try:
from hermes_cli.config import load_config
from hermes_cli.providers import custom_provider_slug
config = load_config()
providers = config.get("providers", {})
if isinstance(providers, dict):
ids.update(custom_provider_slug(str(entry.get("name") or key), str(key))
for key, entry in providers.items() if isinstance(entry, dict))
legacy = config.get("custom_providers", [])
if isinstance(legacy, list):
ids.update(
custom_provider_slug(str(entry.get("name") or "")) for entry in legacy if isinstance(entry, dict))
except _CONFIG_ERRORS:
pass
return ids
def _provider_has_credentials(pid: str) -> bool:
try:
from hermes_cli.auth import get_auth_status, has_usable_secret
if pid == "custom":
return bool((_get_custom_base_url() or "").strip())
if pid == "openrouter":
from hermes_cli.model_switch import _scoped_key_env
return has_usable_secret(_scoped_key_env("OPENROUTER_API_KEY"))
status = get_auth_status(pid)
return bool(status.get("logged_in") or status.get("configured"))
except Exception:
return False
def list_available_providers() -> list[dict[str, str]]:
"""``{id, label, aliases, authenticated}`` per listed provider (shared with ``hermes model`` / ``/model``)."""
from hermes_cli.models_catalog_static import listed_canonical_providers
aliases_for: dict[str, list[str]] = {}
for alias, canonical in _PROVIDER_ALIASES.items():
aliases_for.setdefault(canonical, []).append(alias)
return [
{
"id": pid,
"label": _PROVIDER_LABELS.get(pid, pid),
"aliases": aliases_for.get(pid, []),
"authenticated": _provider_has_credentials(pid)}
for pid in [p.slug for p in listed_canonical_providers()] + ["custom"]]
def parse_model_input(
raw: str, current_provider: str, *, custom_ids: Optional[set[str]] = None) -> tuple[str, str]:
"""Parse ``/model`` input into ``(provider, model)``. The colon is a provider delimiter only when
the left side is a known provider/alias, so ``anthropic/claude-3.5-sonnet:beta`` stays a model.
``custom_ids`` is the caller's already-loaded set of configured ``custom:<name>`` ids (default:
read from config) so one decision never consults two config sources."""
stripped = raw.strip()
colon = stripped.find(":")
if colon > 0:
provider_part = stripped[:colon].strip().lower()
model_part = stripped[colon + 1:].strip()
if provider_part and model_part and provider_part in _KNOWN_PROVIDER_NAMES:
if provider_part == "custom":
configured = _configured_custom_provider_ids() if custom_ids is None else custom_ids
# Longest configured ``custom:<name>`` id that prefixes the input wins.
lowered = stripped.lower()
for custom_id in sorted(configured - {"custom"}, key=len, reverse=True):
if lowered.startswith(f"{custom_id.lower()}:"):
return custom_id, stripped[len(custom_id) + 1 :].strip()
# ``custom:local:qwen`` → ("custom:local", "qwen") for a configured named provider;
# single-colon ``custom:qwen`` → ("custom", "qwen") as before.
if ":" in model_part:
custom_name, actual_model = (part.strip() for part in model_part.split(":", 1))
if custom_name and actual_model:
if f"custom:{custom_name.lower()}" in configured:
return (f"custom:{custom_name.lower()}", actual_model)
return ("custom", model_part)
return (normalize_provider(provider_part), model_part)
return (current_provider, stripped)
def _get_custom_base_url() -> str:
"""The custom endpoint ``model.base_url`` from config.yaml."""
return str(_get_model_config_dict().get("base_url", "")).strip()
def _get_provider_config_dict(provider: str) -> dict[str, Any]:
"""Return config.yaml providers.<provider>, or an empty dict."""
key = str(provider or "").strip()
if not key:
return {}
try:
from hermes_cli.config import load_config
providers_cfg = load_config().get("providers", {})
if isinstance(providers_cfg, dict):
entry = providers_cfg.get(key) or providers_cfg.get(key.lower())
if isinstance(entry, dict):
return entry
except _CONFIG_ERRORS:
pass
return {}
def _get_model_config_dict() -> dict[str, Any]:
"""Return the main model config mapping, or an empty dict."""
try:
from hermes_cli.config import load_config
model_cfg = load_config().get("model", {})
if isinstance(model_cfg, dict):
return model_cfg
except Exception:
pass
return {}
def _base_url_looks_like_anthropic_messages(base_url: str) -> bool:
normalized = str(base_url or "").strip().lower().rstrip("/")
if not normalized:
return False
return urllib.parse.urlparse(normalized).path.rstrip("/").endswith(("/anthropic", "/anthropic/v1"))
def _anthropic_models_url(base_url: Optional[str] = None, *, after_id: Optional[str] = None) -> str:
"""Anthropic ``/v1/models`` page URL. The endpoint is cursor-paginated with a default page of
20 (smaller than the live catalog), so every request asks for the maximum page size and
``after_id`` continues from a previous page's ``last_id``."""
endpoint = str(base_url or "https://api.anthropic.com").strip().rstrip("/")
url = endpoint + ("/models" if endpoint.endswith("/v1") else "/v1/models")
params = {"limit": "1000"}
if after_id:
params["after_id"] = after_id
return url + ("&" if "?" in url else "?") + urllib.parse.urlencode(params)
_ANTHROPIC_MODELS_MAX_PAGES = 20
def _anthropic_next_cursor(page: Any, seen_cursors: set[str]) -> Optional[str]:
"""``last_id`` to continue from, or None when the page is final or the server repeats a
cursor (which would otherwise loop forever)."""
if not isinstance(page, dict) or page.get("has_more") is not True:
return None
last_id = page.get("last_id")
if not isinstance(last_id, str) or not last_id or last_id in seen_cursors:
return None
seen_cursors.add(last_id)
return last_id
def curated_models_for_provider(
provider: Optional[str],
*,
force_refresh: bool = False,
) -> list[tuple[str, str]]:
"""Return ``(model_id, description)`` tuples for a provider's model list.
Tries to fetch the live model list from the provider's API first,
falling back to the static ``_PROVIDER_MODELS`` catalog if the API
is unreachable.
"""
normalized = normalize_provider(provider)
if normalized == "openrouter":
return fetch_openrouter_models(force_refresh=force_refresh)
# Try live API first (Codex, Nous, etc. all support /models)
live = provider_model_ids(normalized)
if live:
# StepFun's Step Plan /models merge now lives in its provider fetcher
# (``_stepfun_catalog``) so every picker surface sees the same list.
return [(m, "") for m in live]
# Fallback to static catalog
models = static_catalog(normalized)
return [(m, "") for m in models]
def _provider_keys(provider: str) -> set[str]:
key = (provider or "").strip().lower()
normalized = normalize_provider(provider)
return {k for k in (key, normalized) if k}
def _provider_catalog_names(provider: str) -> tuple[str, ...]:
"""Active picker models plus retired aliases recognized for detection."""
return tuple(static_catalog(provider)) + _PROVIDER_RETIRED_ALIASES.get(provider, ())
def _model_in_provider_catalog(name_lower: str, providers: set[str]) -> bool:
return any(
name_lower == model.lower()
for provider in providers
for model in _provider_catalog_names(provider))
def _resolve_static_model_alias(
name_lower: str, current_keys: set[str]) -> Optional[tuple[str, str]]:
"""Resolve short aliases (e.g. sonnet/opus) using static catalogs only."""
try:
from hermes_cli.model_switch import MODEL_ALIASES
except Exception:
return None
identity = MODEL_ALIASES.get(name_lower)
if identity is None:
return None
def _match(provider: str) -> Optional[str]:
prefix = f"{identity.vendor}/{identity.family}" if provider in _AGGREGATOR_PROVIDERS else identity.family
prefix = prefix.lower()
return next((m for m in static_catalog(provider) if m.lower().startswith(prefix)), None)
# Current provider first, then native vendors, then aggregators / borrow-list providers the user
# is already on — so `sonnet` resolves to anthropic before any re-exposing provider.
skip = current_keys | _AGGREGATOR_PROVIDERS | _BORROWED_MODEL_PROVIDERS
candidates = [
*current_keys, *(p for p in static_catalog_providers() if p not in skip),
*(p for p in _AGGREGATOR_PROVIDERS if p in current_keys),
*(p for p in _BORROWED_MODEL_PROVIDERS if p in current_keys)]
for provider in candidates:
if matched := _match(provider):
return provider, matched
return None
def detect_static_provider_for_model(
model_name: str, current_provider: str) -> Optional[tuple[str, str]]:
"""Auto-detect a provider from static catalogs only → ``(provider_id, model_name)`` (the name may
be remapped by a static alias or a bare provider name), or ``None`` without a confident match."""
name = (model_name or "").strip()
if not name:
return None
name_lower = name.lower()
current_keys = _provider_keys(current_provider)
alias_match = _resolve_static_model_alias(name_lower, current_keys)
if alias_match:
return alias_match
# Step 0: a bare provider name typed as the model (`/model nous`) is a provider switch to that
# provider's default. Skip "custom" (no catalog) and "openrouter" (needs an explicit model).
resolved_provider = _PROVIDER_ALIASES.get(name_lower, name_lower)
if resolved_provider not in {"custom", "openrouter"}:
default_models = static_catalog(resolved_provider)
if resolved_provider in _PROVIDER_LABELS and default_models and resolved_provider not in current_keys:
# Cost-safe default, not ``default_models[0]``: metered aggregators list most-capable-first,
# so [0] would silently escalate `/model nous` to the priciest flagship.
return (resolved_provider, get_default_model_for_provider(resolved_provider) or default_models[0])
# A model in the current provider's own catalog never suggests switching.
if _model_in_provider_catalog(name_lower, current_keys):
return None
return next(_static_catalog_matches(name, current_provider), None)
def _static_catalog_matches(name: str, current_provider: str):
"""Yield every ``(provider_id, name)`` whose static catalog lists *name*, in ladder order.
Several first-party providers list the same slug (``gpt-5.6-luna`` on ``openai-api`` AND
``openai-codex``); the first is only a guess, so callers that gate on credentials need the
siblings too (#102775)."""
name_lower = name.lower()
current_keys = _provider_keys(current_provider)
# Step 1: direct static-catalog match. Aggregators list other vendors' models — never
# auto-switch TO them. A custom endpoint (custom / custom:*) is never auto-switched away
# from: the user configured it deliberately and may serve the same model name there.
if current_provider != "custom" and not current_provider.startswith("custom:"):
for pid in static_catalog_providers():
if pid in current_keys or pid in _AGGREGATOR_PROVIDERS or pid in _BORROWED_MODEL_PROVIDERS:
continue
if _model_in_provider_catalog(name_lower, {pid}):
yield (pid, name)
# Borrow-list providers (re-expose other vendors' models) only after every native-vendor
# catalog, and only when one is the current provider.
for pid in _BORROWED_MODEL_PROVIDERS:
if pid not in current_keys and _model_in_provider_catalog(name_lower, {pid}):
yield (pid, name)
def _configured_provider_ids() -> set[str]:
"""Provider ids (incl. ``custom:*``) from the user's ``providers:`` config block; empty when config
is unreadable (callers fall through to built-in catalogs)."""
try:
from hermes_cli.config import load_config
providers = (load_config() or {}).get("providers")
if not isinstance(providers, dict):
return set()
return {key for pid in providers if (key := str(pid).strip().lower())}
except Exception:
return set()
def _resolve_provider_prefix(model_name: str) -> Optional[tuple[str, str]]:
"""Route an explicit ``vendor/model`` prefix (``nous/deepseek-v4-pro``, ``ollama/qwen3.5:4b``) to
a provider the user defined in ``providers:`` (by raw name or alias) instead of the default.
``nous/deepseek-v4-pro`` or ``ollama/qwen3.5:4b`` should route to the named provider instead of falling
back to the configured default (which silently sends non-default models to the wrong endpoint, #87189).
"""
if "/" not in model_name:
return None
vendor, model = model_name.split("/", 1)
vendor, model = vendor.strip().lower(), model.strip()
if not vendor or not model:
return None
configured = _configured_provider_ids()
# An explicitly named provider block (``ollama:``) wins over the alias table, which may
# canonicalize the same name elsewhere (``ollama`` → ``custom``).
for candidate in (vendor, _PROVIDER_ALIASES.get(vendor, vendor)):
if candidate in configured:
return (candidate, model)
return None
def detect_provider_for_model(
model_name: str, current_provider: str) -> Optional[tuple[str, str]]:
"""Auto-detect the best provider for a model name: the current provider's live catalog, static
catalogs (bare provider name → its default; direct catalog match), then the OpenRouter catalog,
then a configured ``vendor/`` prefix.
Never hands back a provider the user holds no credentials for: an unauthenticated guess is
skipped and the ladder continues (``None`` = stay on the current provider). Exceptions: the user
NAMED the provider (``/model nous``), or there is no current provider yet (``auto``) — then the
first guess is returned so the credential step fails loudly instead of silently ignoring input."""
from hermes_cli.models_detect import (
current_provider_catalog_match, current_provider_owns_vendor, provider_has_credentials)
name = (model_name or "").strip()
if not name:
return None
# The current provider's LIVE catalog outranks every static guess: a model it already serves
# (Codex early-access ids, Portal-only slugs, Ollama Cloud models absent from _PROVIDER_MODELS)
# must never re-route the session to another vendor or to metered OpenRouter.
served = current_provider_catalog_match(name, current_provider)
if served is not None:
return (current_provider, served) if served != name else None
# Live catalog unavailable or lagging: the vendor's own id on the vendor's first-party provider
# is still a selection — an aggregator relisting it is not grounds to switch.
if current_provider_owns_vendor(name, current_provider):
return None
no_selection = (current_provider or "").strip().lower() in {"", "auto"}
first_guess = None
for candidate in _detection_candidates(name, current_provider):
if candidate is None:
return None # the current catalog owns this name
if candidate[0] == current_provider or provider_has_credentials(candidate[0]):
return candidate
if _PROVIDER_ALIASES.get(name.lower(), name.lower()) == candidate[0]:
return candidate # explicitly named provider: let the credential step report it
first_guess = first_guess or candidate
logger.debug("Skipping auto-switch of '%s' to %s: no credentials configured", name, candidate[0])
if no_selection and first_guess:
return first_guess # nothing usable anywhere: fail loudly on the first guess
# A ``vendor/model`` prefix naming a provider the user DECLARED in ``providers:`` is a selection,
# not a guess — hand it back even before its key is wired up.
return _resolve_provider_prefix(name)
def _detection_candidates(name: str, current_provider: str):
"""Yield ``(provider, model)`` guesses in ladder order; ``None`` means the current provider's own
catalog owns the name (stop, stay)."""
static_match = detect_static_provider_for_model(name, current_provider)
if static_match:
yield static_match
# Sibling catalogs listing the same slug (openai-api / openai-codex share the gpt-5.6
# family): the credential gate downstream takes the first one the user can actually use.
for sibling in _static_catalog_matches(name, current_provider):
if sibling != static_match:
yield sibling
if _model_in_provider_catalog(name.lower(), _provider_keys(current_provider)):
yield None
return
# OpenRouter catalog (exact slug, then bare model part).
or_slug = _find_openrouter_slug(name)
if or_slug:
if current_provider == "openrouter" and or_slug == name:
yield None # already on openrouter with matching name
return
yield ("openrouter", or_slug)
# Explicit ``vendor/model`` prefix naming a configured provider — AFTER the OpenRouter lookup so
# aggregator-native slugs (``deepseek/deepseek-chat``) keep their routing.
prefixed = _resolve_provider_prefix(name)
if prefixed:
yield prefixed
def _find_openrouter_slug(model_name: str) -> Optional[str]:
"""Full OpenRouter slug for a bare or partial model name (exact slug first, then bare part)."""
name_lower = model_name.strip().lower()
if not name_lower:
return None
ids = model_ids()
return (
next((mid for mid in ids if name_lower == mid.lower()), None)
or next((mid for mid in ids if "/" in mid and name_lower == mid.split("/", 1)[1].lower()), None)
)
def normalize_provider(provider: Optional[str]) -> str:
"""Normalize provider aliases to canonical ids. ``"auto"`` passes through — use
``hermes_cli.auth.resolve_provider()`` to resolve it from credentials."""
normalized = (provider or "openrouter").strip().lower()
return _PROVIDER_ALIASES.get(normalized, normalized)
def provider_label(provider: Optional[str]) -> str:
"""Return a human-friendly label for a provider id or alias."""
original = (provider or "openrouter").strip()
normalized = original.lower()
if normalized == "auto":
return "Auto"
normalized = normalize_provider(normalized)
return _PROVIDER_LABELS.get(normalized, original or "OpenRouter")
def _is_openai_fast_model(model_id: Optional[str]) -> bool:
"""OpenAI flagship eligible for Priority Processing. Codex-series excluded — the Codex Responses
API doesn't accept ``service_tier``."""
base = _strip_vendor_prefix(str(model_id or "")).split(":")[0]
return bool(base) and "codex" not in base and base.startswith(tuple(_OPENAI_FAST_MODE_PREFIXES))
def _strip_vendor_prefix(model_id: str) -> str:
"""Lowercase and strip a ``vendor/`` prefix (``anthropic/claude-opus-4-6`` → ``claude-opus-4-6``)."""
raw = str(model_id or "").strip().lower()
return raw.split("/", 1)[1] if "/" in raw else raw
def model_supports_fast_mode(model_id: Optional[str], *, provider: Optional[str] = None, base_url: Optional[str] = None) -> bool:
"""Whether to expose /fast for this model; a Bedrock-only one only on the route its request builder tiers."""
from agent.model_metadata import is_grok_46_family
if model_id in BEDROCK_ULTRAFAST_ROUTES:
return normalize_provider(provider) == "bedrock" and bedrock_fast_mode_overrides(model_id, base_url, "ultrafast") is not None
return (
_is_anthropic_fast_model(model_id)
or _is_openai_fast_model(model_id)
or is_grok_46_family(str(model_id or "")))
def _is_anthropic_fast_model(model_id: Optional[str]) -> bool:
"""Accepts the Anthropic Fast Mode ``speed`` param (Opus 4.8 / Opus 5 / Opus 5.5 only) —
deliberately NOT a general "fast model" check. The list lives in ``agent.model_metadata``."""
from agent.model_metadata import is_anthropic_fast_mode_model
return is_anthropic_fast_mode_model(model_id)
def _fast_mode_route_supported(
model_id: Optional[str], provider: Optional[str], base_url: Optional[str]) -> bool:
"""Only the first-party endpoint that bills for fast mode may receive its params."""
from urllib.parse import urlparse
from agent.model_metadata import is_grok_46_family
if _is_anthropic_fast_model(model_id):
allowed = {"anthropic": "api.anthropic.com"}
elif is_grok_46_family(str(model_id or "")):
allowed = {"xai": "api.x.ai"}
else:
from hermes_cli.models_fast_regions import openai_fast_region_allows
if openai_fast_region_allows(base_url, model_id):
base_url = "https://api.openai.com" # Eligibility only; the runtime URL stays regional.
allowed = {"openai": "api.openai.com", "openai-api": "api.openai.com", "openai-codex": "chatgpt.com"}
if provider and normalize_provider(provider) not in allowed:
return False
host = (urlparse(str(base_url or "")).hostname or "").lower()
return not host or host in allowed.values()
def model_supports_ultrafast(model_id: Optional[str]) -> bool:
"""OpenAI Ultrafast (``service_tier: "ultrafast"``) is published per model, not per family."""
from agent.model_metadata import strip_codex_context_variant_suffix
base = _strip_vendor_prefix(strip_codex_context_variant_suffix(str(model_id or ""))).split(":")[0]
return base in _OPENAI_ULTRAFAST_MODELS
def resolve_fast_mode_overrides(
model_id: Optional[str], *, provider: Optional[str] = None, base_url: Optional[str] = None,
tier: Optional[str] = None,
) -> dict[str, Any] | None:
"""Route/model-gated fast overrides for /fast and bounded windows; None if unsupported.
Ultrafast never silently downgrades to Priority. Bedrock has its own capability matrix."""
if normalize_provider(provider) == "bedrock":
return bedrock_fast_mode_overrides(model_id, base_url, tier)
if not model_supports_fast_mode(model_id):
return None
if (provider or base_url) and not _fast_mode_route_supported(model_id, provider, base_url):
return None
if tier == "ultrafast":
return {"service_tier": "ultrafast"} if model_supports_ultrafast(model_id) else None
return {"speed": "fast"} if _is_anthropic_fast_model(model_id) else {"service_tier": "priority"}
def _model_dedup_key(model_id: str) -> str:
"""Case-insensitive dedup key folded through the picker-search alias table, so a bare live wire
id and its curated public slug (Kimi ``k3`` / ``kimi-k3``) don't both survive a merge."""
key = str(model_id).strip().lower()
try:
from hermes_cli.model_search import model_alias_canonical
return model_alias_canonical(key)
except Exception:
return key
def _merge_with_models_dev(provider: str, curated: list[str]) -> list[str]:
"""models.dev entries first (their order), then curated-only extras, case-insensitively deduped
while preserving curated casing. Curated unchanged when models.dev is unreachable/empty."""
try:
from agent.models_dev import list_agentic_models
mdev = list_agentic_models(provider)
except Exception:
mdev = []
if not mdev:
return list(curated)
return _merge_unique(_merge_unique([], mdev), curated)
def _openai_discovery_base_url(provider: str) -> str:
"""OpenAI endpoint for model discovery, mirroring runtime precedence so discovery probes the SAME
endpoint inference uses: ``$OPENAI_BASE_URL`` → config ``model.base_url`` (when the configured
provider matches) → the canonical default."""
env_raw = os.getenv("OPENAI_BASE_URL", "").strip().rstrip("/")
if env_raw:
return env_raw
try:
model_cfg = _get_model_config_dict()
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
same_provider = normalize_provider(provider) == normalize_provider(cfg_provider)
if cfg_provider in ("openai", "openai-api") and same_provider:
cfg_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
if cfg_url:
return cfg_url
except Exception:
pass
return "https://api.openai.com/v1"
def _codex_catalog(normalized: str, force_refresh: bool) -> list[str]:
from hermes_cli.codex_models import get_codex_model_ids
# Live OAuth token so the picker matches what ChatGPT lists for this account; hardcoded
# catalog without a token / when unreachable. Read-only (#68004): a picker never imports,
# refreshes or persists a credential, so an expired stored token means the hardcoded catalog
# until the runtime lease refreshes it.
# The token and the host it is routed to come from the same resolution (#121486): a pooled
# gateway key is only ever sent to that gateway, never to the chatgpt.com default.
base_url = None
try:
from hermes_cli.auth import _codex_access_token_is_expiring, resolve_codex_runtime_credentials
creds = resolve_codex_runtime_credentials(read_only=True)
access_token, base_url = creds.get("api_key"), creds.get("base_url")
if _codex_access_token_is_expiring(access_token, 0):
access_token = None
except Exception:
access_token = None
return get_codex_model_ids(access_token=access_token, base_url=base_url)
def _nous_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
try:
from hermes_cli.anon_challenge import background_caller
from hermes_cli.auth import fetch_nous_models, resolve_nous_runtime_credentials
with background_caller(): # never waits on or opens a challenge
creds = resolve_nous_runtime_credentials()
if creds:
live = fetch_nous_models(api_key=creds.get("api_key", ""), inference_base_url=creds.get("base_url", ""))
if live:
return live
except Exception:
pass
# Live failed / no creds: the docs-hosted manifest, so new Portal models surface without a release.
return get_curated_nous_model_ids() or None
def _api_key_credentials(normalized: str) -> tuple[str, str]:
"""``(api_key, base_url)`` from ``resolve_api_key_provider_credentials``; empty strings on any miss."""
try:
from hermes_cli.auth import resolve_api_key_provider_credentials
creds = resolve_api_key_provider_credentials(normalized)
return str(creds.get("api_key") or "").strip(), str(creds.get("base_url") or "").strip()
except Exception:
return "", ""
def _api_key_provider_live(normalized: str, force_refresh: bool) -> Optional[list[str]]:
"""Live /v1/models for a simple api-key provider (stepfun, gmi); None on any miss."""
api_key, base_url = _api_key_credentials(normalized)
if not (api_key and base_url):
return None
try:
return fetch_api_models(api_key, base_url) or None
except Exception:
return None
def _stepfun_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
"""Step Plan live list merged with the curated catalog (the ``_anthropic_catalog`` pattern).
The StepFun inference endpoint is the Step Plan API, whose ``/models`` returns a subset of
the full catalog: it omits models served by the Standard API (e.g. ``step-3.7-flash``),
so a healthy live response would shadow curated-only models from every picker surface
(#41147). Live rows first, then curated-only additions — and when the probe declines
(no credentials / failure) ``None`` falls through to the generic profile path, whose
``merge_profile_catalog`` already serves the curated floor as a placeholder.
"""
live = _api_key_provider_live(normalized, force_refresh)
if not live:
return live
curated = list(static_catalog(normalized))
return _merge_unique(live, curated, key=_model_dedup_key) if curated else live
def _anthropic_catalog(normalized: str, force_refresh: bool) -> list[str]:
model_cfg = _get_model_config_dict()
cfg_base_url = cfg_api_key = ""
if normalize_provider(str(model_cfg.get("provider", "") or "")) == "anthropic":
cfg_base_url = str(model_cfg.get("base_url", "") or "").strip()
cfg_api_key = str(model_cfg.get("api_key", "") or "").strip()
live = _fetch_anthropic_models(base_url=cfg_base_url or None, api_key=cfg_api_key or None)
curated = list(_PROVIDER_MODELS.get("anthropic", []))
if not live:
# A placeholder for the outage, not this account/proxy's catalog: the disk cache must
# never pin it over a same-credentials live row (#107391).
return CuratedFallbackModels(curated)
# The live /v1/models dump lags newly-routed curated aliases (reachable before enumerated):
# curated first, then live-only extras, so a fresh curated model never disappears.
return live if cfg_base_url else _merge_unique(curated, live)
def _openai_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
api_key = os.getenv("OPENAI_API_KEY", "").strip()
if not api_key:
return None
base = _openai_discovery_base_url(normalized)
# Custom OpenAI-compatible endpoints serve a small curated catalog — use it verbatim. Official
# OpenAI hosts (canonical and data-residency regional) return 120+ embeddings/whisper/tts/…
# entries, so intersect with the curated agentic catalog so ``/model`` matches ``hermes model``.
# Model not in live /v1/models — check the curated catalog before rejecting. Providers may omit models
# from their live listing that are still valid (stale cache, partial rollout, gated previews). Use the
# pure-catalog helper (no extra live fetch) so we only accept models Hermes actually ships. (#46850)
# Their /v1/models listing is access-scoped and authoritative — a model absent from it is one this key
# CANNOT serve, so the curated soft-accept would manufacture a selection that 400s at first use. Custom
# OpenAI-compatible proxies keep the fallback (incomplete listings are common there).
from hermes_cli.providers import is_official_openai_host
try:
live = fetch_api_models(api_key, base)
except Exception:
live = None
if not live:
return None
if not is_official_openai_host(base):
return live
live_lower = {m.lower() for m in live}
curated = list(static_catalog(normalized))
# Curated order, only models the account has access to; an account serving none of them (rare)
# falls back to curated so the picker still offers sane defaults.
discovered = [m for m in curated if m.lower() in live_lower]
# Astra is intentionally absent from offline/static catalogs: the official API's
# account-scoped /models response is the only source that may advertise it.
discovered.extend(m for m in live if is_astra_model(m))
return discovered or curated or live
def _custom_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
base_url = _get_custom_base_url()
if not base_url:
return None
model_cfg = _get_model_config_dict()
from hermes_cli.auth_declared_providers import custom_catalog_probe_api_key
api_key = custom_catalog_probe_api_key(model_cfg, base_url)
api_mode = "anthropic_messages" if _base_url_looks_like_anthropic_messages(base_url) else None
return fetch_api_models(api_key, base_url, api_mode=api_mode) or None
def _bedrock_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
# Live discovery keyed by the resolved AWS region so EU/AP users see eu.*/ap.* ids.
try:
from agent.bedrock_adapter import bedrock_model_ids_or_none
return bedrock_model_ids_or_none()
except Exception:
return None
def _azure_foundry_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
"""Live ``GET <base>/models`` of the configured Azure Foundry resource (#27989).
Deployments are per-resource, so the static catalog is intentionally empty and the plugin
profile ships ``base_url=""`` — which is why the generic profile fetch never fires. Resolve
through the runtime resolver so the picker targets the same resource inference hits
(``model.base_url`` / ``AZURE_FOUNDRY_BASE_URL``) with the same credential: an API key string,
or the Entra ID token-provider callable that ``azure_detect`` already accepts. Anthropic-style
``/anthropic`` routes serve no ``/models``; the probe never raises, so any miss keeps ``[]``.
"""
try:
from hermes_cli.azure_detect import _probe_openai_models
from hermes_cli.runtime_provider import _resolve_azure_foundry_runtime
runtime = _resolve_azure_foundry_runtime(requested_provider=normalized, model_cfg=_get_model_config_dict())
base_url = str(runtime.get("base_url") or "").strip().rstrip("/")
credential = runtime.get("api_key")
if not (base_url and credential):
return None
ok, ids = _probe_openai_models(base_url, credential)
return ids if ok and ids else None
except Exception:
return None
# Per-provider catalog sources tried before the generic profile fetch (a profile's ``picker_catalog``
# hook fills the same slot, see ``_catalog_fetcher``). A fetcher returning None falls through to the
# profile/curated path; a list is returned as-is (even empty).
_PROVIDER_CATALOG_FETCHERS: dict[str, Any] = {
"openrouter": lambda normalized, force_refresh: model_ids(force_refresh=force_refresh),
"openai-codex": _codex_catalog,
"nous": _nous_catalog,
"stepfun": _stepfun_catalog,
"gmi": _api_key_provider_live,
"anthropic": _anthropic_catalog,
"ai-gateway": lambda normalized, force_refresh: _fetch_ai_gateway_models() or None,
# DeepInfra's generic /models mixes chat, image, video, speech and embedding models; the tagged
# catalog helper is the only safe source for the chat picker, including its empty/failure result.
"deepinfra": lambda normalized, force_refresh: _fetch_deepinfra_models(force_refresh=force_refresh) or [],
"ollama-cloud": lambda normalized, force_refresh: fetch_ollama_cloud_models(force_refresh=force_refresh) or None,
"openai": _openai_catalog,
"openai-api": _openai_catalog,
"custom": _custom_catalog,
"bedrock": _bedrock_catalog,
"azure-foundry": _azure_foundry_catalog}
# ``-free`` slugs the relay still LISTS but no longer serves: the Go-only twin (``ox-alpha-free``)
# and the promo it delisted without removing from ``/models`` (``deepseek-v4-flash-free``). The
# live-first keyed Zen/Go pickers filter through this so a stale live listing can never route
# into a 400/403 (#111749).
_OPENCODE_FREE_EXCLUDED_MODELS = frozenset(
{"ox-alpha-free", "deepseek-v4-flash-free", "x-preview-f-free"}
)
def _catalog_fetcher(normalized: str):
"""The built-in fetcher for *normalized*, else its profile's ``picker_catalog`` hook, else None."""
fetcher = _PROVIDER_CATALOG_FETCHERS.get(normalized)
if fetcher is not None:
return fetcher
from providers import get_provider_profile
from providers.base import ProviderProfile
profile = get_provider_profile(normalized)
hook = profile.picker_catalog if isinstance(profile, ProviderProfile) else None
return (lambda _normalized, force_refresh: hook(force_refresh)) if hook is not None else None
def _profile_live_catalog(normalized: str) -> Optional[list[str]]:
"""Generic live fetch for any provider registered in providers/ with ``auth_type="api_key"``.
Live results are merged with the curated list so models the live endpoint omits still appear:
curated-first by default so the newest curated models lead when the live API lags;
``_LIVE_FIRST_PICKER_PROVIDERS`` (OpenCode Zen/Go, authoritative live API) live-first so stale
curated entries stop polluting the top. Plugin providers without a static entry use the
profile's ``fallback_models`` as the curated list (Fireworks lists an image model first).
"""
from providers import get_provider_profile
profile = get_provider_profile(normalized)
if not profile:
return None
# external_process providers (ACP agent CLIs) have no api_key/base_url credentials: the
# profile's fetch_models drives its own subprocess (kwargs are ignored per the base contract).
# Every non-api-key profile falls back to its own fallback_models (OAuth plugins have no
# static _PROVIDER_MODELS row), exactly as api_key plugins do below.
if profile.auth_type == "external_process":
try:
live = profile.fetch_models()
except Exception as exc: # a failed subprocess launch degrades to the curated list, like api_key below
logger.debug("external_process catalog fetch failed for %s: %s", normalized, exc)
live = None
# Same merge as setup (`_model_flow_plugin_provider`) so /model, the Desktop picker and
# `hermes model` offer one list: live ids plus any pinned id the probe omitted.
return merge_profile_catalog(normalized, profile, list(live) if live else None)
if not (profile.auth_type == "api_key" and profile.base_url):
return list(profile.fallback_models) or None
api_key, base_url = _api_key_credentials(normalized)
return probe_profile_catalog(normalized, profile, api_key, base_url or profile.base_url or None)
def probe_profile_catalog(normalized: str, profile, api_key: Optional[str], base_url: Optional[str]) -> Optional[list[str]]:
"""``profile.fetch_models`` gated on a key (no key → no doomed probe) and merged with the curated
list; a raising catalog override degrades like a None return — fallback_models, not an empty picker."""
live = None
if api_key:
try:
live = profile.fetch_models(api_key=api_key, base_url=base_url)
except Exception:
live = None
return merge_profile_catalog(normalized, profile, live)
def merge_profile_catalog(normalized: str, profile, live: Optional[list[str]]) -> Optional[list[str]]:
"""Combine a profile's live catalog with its curated list the way the ``/model`` picker does, so
first-time setup (``model_setup_flows._api_key_provider_model_list``) offers the same rows the
picker will later show. Empty live → ``fallback_models`` (None when the profile has none)."""
if not live:
rows = CuratedFallbackModels(profile.fallback_models) if profile.fallback_models else None
else:
curated = list(static_catalog(normalized)) or list(profile.fallback_models or ())
if not curated:
rows = live
else:
primary, secondary = (live, curated) if normalized in _LIVE_FIRST_PICKER_PROVIDERS else (curated, live)
rows = _merge_unique(primary, secondary, key=_model_dedup_key)
return _drop_delisted_opencode_models(normalized, rows)
def _drop_delisted_opencode_models(normalized: str, rows: Optional[list[str]]) -> Optional[list[str]]:
"""The relay still LISTS delisted ids it no longer serves, and the curated floor (merged back in
as the secondary half, or served alone when there is no key) carries retired ids too. Filter the
FINAL rows for the live-first Zen/Go pickers so no path can offer a slug that 401s (#111749,
#115496)."""
if rows and normalized in _LIVE_FIRST_PICKER_PROVIDERS:
return type(rows)(m for m in rows if str(m).lower() not in _OPENCODE_FREE_EXCLUDED_MODELS)
return rows
def _chat_catalog_rows(models):
"""Drop generation ids from a chat-catalog list, keeping its list subclass."""
from hermes_cli.chat_catalog import without_generation_models
return without_generation_models(models)
def _configured_relay_base_url(provider: str) -> str:
"""``model.base_url`` when it points the *configured* provider at a relay/proxy, else "".
Discovery must probe the same endpoint inference uses (#121387): when ``model.base_url``
differs from the provider's own endpoint, the vendor's canonical host is NOT the catalog to list.
Mirrors the ``$OPENAI_BASE_URL`` -> ``model.base_url`` -> canonical precedence of
``_openai_discovery_base_url`` for every built-in provider, not just OpenAI.
"""
try:
model_cfg = _get_model_config_dict()
except Exception:
return ""
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
if not cfg_provider or not provider:
return ""
try:
normalized = normalize_provider(provider)
if normalized != normalize_provider(cfg_provider):
return ""
except Exception:
return ""
base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
if not base_url:
return ""
# A base_url equal to the provider's own endpoint (the profile's or the registry row's: TokenHub's
# profile leaves it empty; OpenRouter has no registry row) is not a relay: setup persists
# canonical URLs, and OAuth providers (Codex) need native discovery.
try:
from hermes_cli.auth import PROVIDER_REGISTRY
from providers import get_provider_profile
profile_url = getattr(get_provider_profile(normalized), "base_url", "")
registry_url = getattr(PROVIDER_REGISTRY.get(normalized), "inference_base_url", "")
canonical = {normalize_route_base_url(u) for u in (profile_url, registry_url) if u}
except Exception:
return base_url # lookup failed: stay a relay, never widening where credentials go
return "" if normalize_route_base_url(base_url) in canonical else base_url
def _relay_model_catalog(normalized: str, relay: str) -> Optional[list[str]]:
"""Live catalog probed at a configured ``model.base_url`` relay, or None to fall through.
Returns only the relay's live ids (no curated merge): a relay user must see the relay's
catalog, and a failed/empty probe degrades to the canonical fetchers untouched.
"""
try:
from providers import get_provider_profile
profile = get_provider_profile(normalized)
if profile is None or getattr(profile, "auth_type", "") != "api_key":
return None
from hermes_cli.auth_declared_providers import relay_probe_api_key
api_key = relay_probe_api_key(normalized, relay, *_api_key_credentials(normalized))
live = profile.fetch_models(api_key=api_key, base_url=relay)
return [str(m) for m in (live or []) if m] or None
except Exception:
return None
# Canonical fetchers that already resolve `model.base_url` themselves for the configured
# provider AND degrade to their curated list when that relay fails — `_anthropic_catalog`,
# `_custom_catalog`, `_openai_catalog` (via `_openai_discovery_base_url`) and the simple
# api-key fetchers (via `resolve_api_key_provider_credentials`). They already satisfy the
# "no vendor egress when a relay is configured" invariant, so intercepting them would only
# override correct, better-merged behaviour. Everything else is vendor-pinned (#121387).
_RELAY_AWARE_CATALOG_FETCHERS = frozenset(
{"anthropic", "custom", "openai", "openai-api", "stepfun", "gmi"}
)
def _static_catalog(normalized: str, fetcher: Any) -> list[str]:
"""The local, no-egress catalog tail: curated static list (+ models.dev merge where preferred).
Shared by the normal path's final fallback and by the configured-relay degrade path, which
must never reach a live vendor fetcher (#121387).
"""
# Merge static curated list with live API results so models that the live endpoint omits (stale cache,
# partial rollout) still appear in the picker. Single providers (kimi, zai) use curated-first (commit
# 658ac1d86) to surface newest models even when live API lags (#46309). OpenCode Zen / Go are different:
# their live API is the authoritative catalog, so they merge live-first — live entries lead and stale
# curated entries no longer pollute the top of the picker. (#49129) Plugin providers with no static
# _PROVIDER_MODELS entry fall back to the profile's curated fallback_models so their agentic picks lead
# the picker instead of whatever the live catalog happens to return first (e.g. Fireworks lists an image
# model, flux-*, ahead of its chat models).
# A provider with a live fetcher that declined is serving a placeholder; one without any live
# source is serving its authoritative catalog.
curated_static = (CuratedFallbackModels if fetcher is not None else list)(static_catalog(normalized))
if normalized not in _MODELS_DEV_PREFERRED:
return _chat_catalog_rows(_drop_delisted_opencode_models(normalized, curated_static))
# models.dev keeps listing retired Zen ids too: filter after the merge, not before.
merged = _drop_delisted_opencode_models(normalized, _merge_with_models_dev(normalized, curated_static))
return _chat_catalog_rows(_xai_finalize_catalog(merged) if normalized in {"xai", "xai-oauth"} else merged)
def provider_model_ids(provider: Optional[str], *, force_refresh: bool = False) -> list[str]:
"""Best known model catalog for a provider: per-provider live fetchers, then the generic profile
fetch, then the static list (merged with models.dev for ``_MODELS_DEV_PREFERRED`` providers)."""
requested = str(provider or "").strip().lower()
if requested == "ollama":
return _ollama_local_catalog(force_refresh)
normalized = normalize_provider(provider)
# A configured `model.base_url` relay is TERMINAL for live catalog egress: the picker must
# list what the configured endpoint serves and must never touch the vendor host (#121387).
# A failed or empty probe degrades to the local curated list — falling through to the
# canonical fetchers would send the provider credential to exactly the host the user
# deliberately routed away from, recreating the bug on the failure path.
relay = _configured_relay_base_url(provider or "")
if relay and normalized not in _RELAY_AWARE_CATALOG_FETCHERS:
relayed = _relay_model_catalog(normalized, relay)
if relayed:
return _chat_catalog_rows(relayed)
return _static_catalog(normalized, _catalog_fetcher(normalized))
fetcher = _catalog_fetcher(normalized)
if fetcher is not None:
models = fetcher(normalized, force_refresh)
if models is not None:
return _chat_catalog_rows(models)
try:
models = _profile_live_catalog(normalized)
except Exception:
models = None
if models is not None:
return _chat_catalog_rows(models)
return _static_catalog(normalized, fetcher)
# ---------------------------------------------------------------------------
# Disk cache for provider_model_ids() — keeps /model picker fast (otherwise every open re-fetches
# every authed provider's /v1/models). One JSON file at $HERMES_HOME/provider_models_cache.json;
# entries keyed by credential fingerprint (rotate OPENAI_API_KEY → entry invalidates); 1h TTL;
# only NON-EMPTY results are cached so a transient failure is never pinned; any read/write error
# degrades silently to a live fetch.
# ---------------------------------------------------------------------------
_PROVIDER_MODELS_CACHE_TTL = 3600 # 1h
# Stale-while-revalidate window: an expired same-credentials entry is served IMMEDIATELY while a
# daemon thread refreshes the disk cache; beyond this bound the caller blocks on a live fetch.
# Catalogs change on release timescales, so hour-old data beats stalling every picker surface.
_PROVIDER_MODELS_STALE_SERVE_MAX = 7 * 24 * 3600 # 7d
# A curated fallback row is a placeholder for an outage, not a catalog: re-probe soon and never
# serve it through the stale window.
_PROVIDER_MODELS_FALLBACK_TTL = 60
# Cache keys with a background SWR refresh in flight — dedupes concurrent refreshes.
_swr_refresh_inflight: set = set()
_swr_refresh_lock = threading.Lock()
def _cache_entry(fp: str, models: list[str], at: Optional[float] = None) -> dict:
"""One provider row of the disk cache: credential fingerprint, write time, model ids."""
return {"fp": fp, "at": time.time() if at is None else at, "models": list(models)}
def _live_result_entry(fp: str, live: list[str], existing: Any, at: Optional[float] = None) -> Optional[dict]:
"""Row to store for a ``provider_model_ids`` result, or ``None`` to keep *existing*: a curated
fallback never replaces the account's real catalog for the same credentials, and when it is
stored it is flagged so it expires on the short fallback TTL."""
if not isinstance(live, CuratedFallbackModels):
return _cache_entry(fp, live, at)
if _cache_entry_valid(existing, fp) and not existing.get("fallback"):
return None
return {**_cache_entry(fp, live, at), "fallback": True}
def _ollama_native_probe_reachable() -> bool:
"""Whether the configured local Ollama root answered the native ``/api/tags`` probe (an empty
catalog from a reachable server is authoritative; a failed probe is not)."""
base_url = _get_ollama_base_url()
headers = _get_ollama_native_headers(base_url) or None
probe_key = _ollama_probe_cache_key(_root_for_ollama_native_api(base_url), headers)
return _OLLAMA_LOCAL_PROBE_REACHABLE.get(probe_key) is True
def _spawn_swr_refresh(cache_key: str, refresh_fn=None) -> None:
"""Fire-and-forget daemon refresh of *cache_key*'s cache entry, at most one in flight per key.
Failures are swallowed — the stale entry stays served until a later refresh succeeds.
``refresh_fn`` (no-args → fresh entry dict or None) lets ``custom:<base_url>`` keys from
:func:`cached_fetch_api_models` reuse the same inflight-dedupe scaffolding."""
# Under a routed profile the inflight key includes the home: the same provider slug names a
# different disk cache and credential set per profile, so one profile's refresh must not
# suppress another's. Unscoped keeps the bare key (tests inspect the set by slug).
from hermes_constants import get_hermes_home_override, hermes_home_key
inflight_key = cache_key if get_hermes_home_override() is None else (hermes_home_key(), cache_key)
with _swr_refresh_lock:
if inflight_key in _swr_refresh_inflight:
return
_swr_refresh_inflight.add(inflight_key)
def _default_refresh():
live = provider_model_ids(cache_key, force_refresh=True)
if live or (cache_key == "ollama" and _ollama_native_probe_reachable()):
fp = _credential_fingerprint(cache_key)
return _live_result_entry(fp, live or [], _load_provider_models_cache().get(cache_key))
return None
def _refresh() -> None:
try:
entry = (refresh_fn or _default_refresh)()
if entry:
# _store_cache_entry re-reads under the write lock: the GUI read path spawns one of these
# per stale provider, so a plain load-modify-save would let concurrent warms drop rows.
_store_cache_entry(cache_key, entry)
except Exception:
logger.debug("SWR refresh failed for %s", cache_key, exc_info=True)
finally:
with _swr_refresh_lock:
_swr_refresh_inflight.discard(inflight_key)
# copy_context: the refresh must read the calling profile's credentials and write ITS disk cache.
ctx = contextvars.copy_context()
threading.Thread(target=lambda: ctx.run(_refresh), daemon=True, name=f"model-cache-swr-{cache_key}").start()
def _provider_models_cache_path() -> Path:
from hermes_constants import get_hermes_home
return get_hermes_home() / "provider_models_cache.json"
def _credential_fingerprint(provider: str) -> str:
"""Short hash of the credentials ``provider_model_ids(provider)`` would see right now.
API-key providers include their configured values and credential-file mtimes. Codex uses the
stable principal selected by its read-only resolver: routine token and pool-state writes must
not discard an account-scoped catalog, while a real account switch must invalidate it.
"""
import hashlib
parts: list[str] = []
try:
from hermes_cli.auth import PROVIDER_REGISTRY
pcfg = PROVIDER_REGISTRY.get(provider)
if pcfg is not None:
for ev in getattr(pcfg, "api_key_env_vars", ()) or ():
parts.append(f"{ev}={os.environ.get(ev, '')}")
bev = getattr(pcfg, "base_url_env_var", "") or ""
if bev:
parts.append(f"{bev}={os.environ.get(bev, '')}")
except Exception:
pass
from hermes_cli.models_process_fingerprint import process_catalog_identity
process_profile, process_parts = process_catalog_identity(provider)
parts += process_parts
# config.yaml's model.base_url changes the endpoint discovery probes (data-residency hosts)
# without touching any env var, so it must change the fingerprint too.
if provider in ("openai", "openai-api"):
try:
parts.append(f"effective_base={_openai_discovery_base_url(provider)}")
except Exception:
pass
# Azure Foundry deployments are per-resource and the wizard writes only model.base_url, so a
# resource switch under the same key must not serve the previous resource's catalog (#27989).
if provider == "azure-foundry":
try:
from hermes_cli.runtime_provider import _config_base_url_for_provider
parts.append(f"effective_base={_config_base_url_for_provider(_get_model_config_dict(), 'azure-foundry')}")
except Exception:
pass
if provider == "ollama":
provider_cfg = _get_provider_config_dict("ollama")
key_env = provider_cfg.get("key_env") or provider_cfg.get("api_key_env") or ""
model_cfg = _get_model_config_dict()
parts += [
f"OLLAMA_HOST={os.environ.get('OLLAMA_HOST', '')}",
"providers.ollama.base_url="
f"{provider_cfg.get('base_url', '') or provider_cfg.get('api', '') or provider_cfg.get('url', '')}",
f"providers.ollama.api_key={provider_cfg.get('api_key', '')}",
f"providers.ollama.key_env={key_env}",
]
if key_env:
parts.append(f"{key_env}={os.environ.get(str(key_env), '')}")
parts += [
f"model.provider={model_cfg.get('provider', '')}|model.base_url={model_cfg.get('base_url', '')}",
"providers.ollama.extra_headers="
+ json.dumps(provider_cfg.get("extra_headers", {}), sort_keys=True, default=str),
]
def _mtime_part(label: str, path) -> None:
from utils import file_cache_version # hashed every time: keys a disk cache later processes read
try:
parts.append(f"{label}@{file_cache_version(path, always_digest=True)}")
except FileNotFoundError:
parts.append(f"{label}@missing")
except Exception:
pass
if provider == "openai-codex":
from hermes_cli.codex_models import codex_catalog_credential_identity
parts.append(f"codex_identity={codex_catalog_credential_identity()}")
elif process_profile is not None:
# Only the profile's own files: the vendor CLI rewrites its token file on every refresh (#129094).
for rel in process_profile.catalog_credential_files:
path = os.path.expanduser(rel)
_mtime_part(path, path)
else:
try:
from hermes_constants import get_hermes_home
for rel in ("auth.json", "credentials.json"):
_mtime_part(rel, get_hermes_home() / rel)
except Exception:
pass
from providers import list_providers
profile_files = [rel for profile in list_providers() for rel in profile.catalog_credential_files]
for rel in ("~/.codex/auth.json", "~/.claude/.credentials.json", *profile_files, "~/.minimax/credentials.json"):
path = os.path.expanduser(rel)
_mtime_part(path, path)
blob = "|".join(parts).encode("utf-8", errors="replace")
# blake2b, not sha256: fingerprint only (collisions = a harmless cache miss), and CodeQL's
# weak-sensitive-data-hashing rule flags sha256 over env vars named *API_KEY*/*TOKEN*.
return hashlib.blake2b(blob, digest_size=8).hexdigest()
def _load_provider_models_cache() -> dict:
"""Return the full cache dict, or {} on any error."""
try:
return _read_json_cache(_provider_models_cache_path()) or {}
except Exception:
return {}
def _save_provider_models_cache(data: dict) -> None:
"""Persist the cache dict. Best-effort — silent on any error."""
try:
_write_json_cache(_provider_models_cache_path(), data, indent=None)
except Exception:
pass
def _store_cache_entry(cache_key: str, entry: dict) -> None:
"""Write one row into the disk cache, re-reading it under the thread + cross-process locks: a slow
fetch worker or another process must never write back the snapshot it loaded before fetching."""
from hermes_cli.models_cache_lock import write_under_cache_lock
def write_row() -> None:
cache = _load_provider_models_cache()
cache[cache_key] = entry
_save_provider_models_cache(cache)
write_under_cache_lock(_provider_models_cache_path(), write_row)
def _normalized_cache_slug(provider: Optional[str]) -> str:
"""``ollama`` stays a raw slug (its alias would canonicalize to ``custom``); everything else normalizes."""
requested = str(provider or "").strip().lower()
return requested if requested == "ollama" else (normalize_provider(provider) or (provider or ""))
def _model_requires_account_discovery(provider: Optional[str], model: str) -> bool:
"""Astra names cannot confer API/OAuth entitlement through picker state."""
return _normalized_cache_slug(provider) in {"openai", "openai-api", "openai-codex"} and is_astra_model(model)
def cached_provider_model_ids(
provider: Optional[str], *, force_refresh: bool = False,
ttl_seconds: int = _PROVIDER_MODELS_CACHE_TTL, non_blocking: bool = False) -> list[str]:
"""Disk-cached :func:`provider_model_ids`: fresh cache hit, else live fetch persisting a non-empty
result. Always returns a list.
``non_blocking`` marks the GUI read path (``model.options``): it NEVER waits on a provider probe.
A same-credentials row of any age is served as-is and a daemon thread warms the next open; a
cold/mismatched row returns ``[]`` so the caller keeps its curated list. One degraded provider
(hanging or timing-out ``/v1/models``) therefore delays nothing but itself (#114215)."""
normalized = _normalized_cache_slug(provider)
if not normalized:
return []
is_ollama = normalized == "ollama"
cache = _load_provider_models_cache()
fp = _credential_fingerprint(normalized)
entry = cache.get(normalized)
now = time.time()
if not force_refresh:
tier = _disk_serve_tier(entry, fp, now, is_ollama=is_ollama, ttl_seconds=ttl_seconds)
if tier is not None:
if tier == "stale":
_spawn_swr_refresh(normalized)
return _chat_catalog_rows(list(entry["models"]))
if non_blocking and not force_refresh:
# Read path: never touch the network in the caller's thread. A same-credentials row past the
# SWR window is still served (hour-old catalog beats an empty picker) while a daemon thread
# warms the next open; a cold row returns [] and the caller falls back to its curated list.
_spawn_swr_refresh(normalized)
if _cache_entry_valid(entry, fp, allow_empty=is_ollama):
return _chat_catalog_rows([
model for model in entry["models"]
if not _model_requires_account_discovery(normalized, model)])
return []
live = provider_model_ids(normalized, force_refresh=force_refresh)
if live:
fresh = _live_result_entry(fp, live, entry, now)
if fresh is None:
# The live fetch degraded to the curated list; the account's real catalog is on disk.
return _chat_catalog_rows([model for model in entry["models"] if not _model_requires_account_discovery(normalized, model)])
_store_cache_entry(normalized, fresh)
return _chat_catalog_rows(list(live))
if is_ollama:
if _ollama_native_probe_reachable():
# A reachable empty native catalog is authoritative; do not resurrect a stale disk catalog.
_store_cache_entry(normalized, _cache_entry(fp, [], now))
return []
# A failed/non-native probe is not authoritative: keep a stale catalog rather than blanking
# the picker during a transient outage.
same_creds = isinstance(entry, dict) and entry.get("fp") == fp
if same_creds and isinstance(entry.get("models"), list) and entry["models"]:
return _chat_catalog_rows(list(entry["models"]))
return []
# Live returned nothing: a stale same-fingerprint entry beats an empty result — minus account-gated
# models, which only a successful discovery may advertise (the entry itself is untouched, so the
# next successful fetch restores them).
if _cache_entry_valid(entry, fp):
return _chat_catalog_rows([model for model in entry["models"] if not _model_requires_account_discovery(normalized, model)])
return []
def clear_provider_models_cache(provider: Optional[str] = None) -> None:
"""Drop one provider's cache entry, or wipe the whole cache (``provider=None``). Used by
``/model --refresh`` and ``hermes model --refresh``."""
try:
from hermes_cli.model_catalog import clear_fetch_backoff
clear_fetch_backoff() # a recovered catalog is fetched now, not after the offline backoff
# Native Ollama tags are keyed by root URL, not provider slug: clear them on every refresh.
_OLLAMA_LOCAL_MODELS_CACHE.clear()
_OLLAMA_LOCAL_PROBE_FAILURE_CACHE.clear()
_OLLAMA_LOCAL_PROBE_REACHABLE.clear()
# Provider-owned in-process memos (e.g. an external CLI's session catalog) must re-probe so a
# fresh login shows on the next switch.
from providers import list_providers
for profile in list_providers():
profile.invalidate_catalog_cache()
from hermes_cli.models_cache_lock import write_under_cache_lock
path = _provider_models_cache_path()
# Under the cache lock like every other writer: a locked writer that loaded the cache
# before this clear would otherwise save its snapshot after it, bringing the row back.
def clear_file() -> None:
if provider is None:
if path.exists():
path.unlink()
return
cache = _load_provider_models_cache()
normalized = _normalized_cache_slug(provider)
if normalized in cache:
del cache[normalized]
_save_provider_models_cache(cache)
write_under_cache_lock(path, clear_file)
except Exception:
pass
def _resolve_anthropic_pool_catalog_credentials() -> tuple[str, str]:
"""Read-only API-key pool credential for model discovery (``resolve_anthropic_token()`` ignores
``api_key`` pool entries — its runtime contract is OAuth-oriented)."""
try:
from agent.credential_pool import AUTH_TYPE_API_KEY
from hermes_cli.auth import read_credential_pool
for entry in read_credential_pool("anthropic"):
if not isinstance(entry, dict) or entry.get("auth_type") != AUTH_TYPE_API_KEY:
continue
token = str(entry.get("access_token") or "").strip()
if token:
return token, str(entry.get("base_url") or entry.get("inference_base_url") or "").strip()
except Exception:
pass
return "", ""
def _fetch_anthropic_models(
timeout: float = 5.0, *, base_url: Optional[str] = None, api_key: Optional[str] = None
) -> Optional[list[str]]:
"""Sorted model ids from the Anthropic /v1/models endpoint, or None. Credentials: explicit
``api_key``, else ``anthropic_discovery_token`` (the Foundry key at azure.com, else env / OAuth /
Claude Code), else (off Foundry) a read-only API-key credential_pool entry."""
try:
from agent.anthropic_credentials import _is_oauth_token
except ImportError:
return None
from hermes_cli.runtime_provider import anthropic_discovery_token # Foundry: never the OAuth token (#154242)
resolved_base_url, token = base_url, anthropic_discovery_token(api_key, base_url)
if token == "":
# A pool credential and its endpoint are one security boundary — never pair the pool key
# with a caller-provided endpoint.
token, resolved_base_url = _resolve_anthropic_pool_catalog_credentials()
if not token:
return None
headers: dict[str, str] = {"anthropic-version": "2023-06-01"}
is_oauth = _is_oauth_token(token)
if is_oauth:
headers["Authorization"] = f"Bearer {token}"
from agent.anthropic_adapter import _COMMON_BETAS, _OAUTH_ONLY_BETAS, _CONTEXT_1M_BETA
headers["anthropic-beta"] = ",".join(_COMMON_BETAS + _OAUTH_ONLY_BETAS)
else:
headers["x-api-key"] = token
url = _anthropic_models_url(resolved_base_url)
try:
try:
data = _get_json(url, timeout=timeout, headers=headers)
except urllib.error.HTTPError as http_err:
# OAuth subscriptions that 400 the 1M context beta ("long context beta is not yet
# available for this subscription"): retry once without it; re-raise anything else.
if not (is_oauth and http_err.code == 400):
raise
try:
body_text = http_err.read().decode(errors="ignore").lower()
except Exception:
body_text = ""
if not ("long context beta" in body_text and "not yet available" in body_text):
raise
headers["anthropic-beta"] = ",".join(
[b for b in _COMMON_BETAS if b != _CONTEXT_1M_BETA] + list(_OAUTH_ONLY_BETAS)
)
data = _get_json(url, timeout=timeout, headers=headers)
models = [m["id"] for m in data.get("data", []) if m.get("id")]
seen_cursors: set[str] = set()
for _page in range(_ANTHROPIC_MODELS_MAX_PAGES):
cursor = _anthropic_next_cursor(data, seen_cursors)
if cursor is None:
break
data = _get_json(_anthropic_models_url(resolved_base_url, after_id=cursor), timeout=timeout, headers=headers)
models.extend(m["id"] for m in data.get("data", []) if m.get("id"))
models = list(dict.fromkeys(models))
# opus, then sonnet, then haiku; alphabetical within tier.
return sorted(models, key=lambda m: ("opus" not in m, "sonnet" not in m, "haiku" not in m, m))
except Exception as e:
logger.debug("Failed to fetch Anthropic models: %s", e)
return None
def azure_foundry_model_api_mode(model_name: Optional[str]) -> Optional[str]:
"""``"codex_responses"`` for families that only accept the Responses API on Azure Foundry (GPT-5.x
incl. gpt-5-mini, codex, o1/o3/o4), else None. Any ``vendor/`` prefix is stripped first."""
raw = str(model_name or "").strip().lower().rsplit("/", 1)[-1]
return "codex_responses" if raw and raw.startswith(tuple(_AZURE_FOUNDRY_RESPONSES_PREFIXES)) else None
_OPENCODE_FAMILIES = ("opencode-go", "opencode-zen")
def opencode_provider_family(provider_id: Optional[str]) -> Optional[str]:
"""Resolve a provider id (canonical or prefixed) to its OpenCode family, or None.
Returns ``"opencode-zen"`` or ``"opencode-go"`` for the built-in providers AND for custom providers
whose name extends a family slug (e.g. ``opencode-go-bridge`` pointing at
``https://opencode.ai/zen/go/v1``, issue #85589). Matching is case-insensitive. Custom family providers
need the same per-model api_mode routing and /v1 base-url normalization as the built-ins — this
predicate is the single owner of that family-membership question; do not re-implement it inline.
"""
raw = str(provider_id or "").strip().lower()
if not raw:
return None
canonical = normalize_provider(provider_id)
if canonical in _OPENCODE_FAMILIES:
return canonical
return next((f for f in _OPENCODE_FAMILIES if raw.startswith(f)), None)
def normalize_opencode_model_id(provider_id: Optional[str], model_id: Optional[str]) -> str:
"""Normalize OpenCode config IDs to the bare model slug used in API requests."""
family = opencode_provider_family(provider_id)
current = str(model_id or "").strip()
if not current or family is None:
return current
for prefix in (f"{provider_id or family}/", f"{family}/"):
if current.lower().startswith(prefix.lower()):
return current[len(prefix):]
return current
# Per-family (model-id prefix → api_mode) routing from OpenCode's published Zen/Go endpoint
# tables, checked in order. GPT/Codex/Grok and Muse Spark use /v1/responses (Muse Spark 503s on
# chat/completions); Claude, MiniMax (Go), Union Alpha, and Qwen use /v1/messages;
# everything else falls through to /v1/chat/completions.
_OPENCODE_API_MODE_PREFIXES: dict[str, tuple[tuple[tuple[str, ...], str], ...]] = {
"opencode-go": (
(("gpt-", "grok-", "muse-spark"), "codex_responses"),
(("claude-", "minimax-", "qwen", "union-alpha"), "anthropic_messages")),
"opencode-zen": (
(("claude-", "union-alpha"), "anthropic_messages"), (("gpt-", "grok-", "muse-spark"), "codex_responses"),
(("qwen",), "anthropic_messages"))}
def opencode_model_api_mode(provider_id: Optional[str], model_id: Optional[str]) -> str:
"""Determine the API mode for an OpenCode Zen / Go model (see ``_OPENCODE_API_MODE_PREFIXES``)."""
family = opencode_provider_family(provider_id)
normalized = normalize_opencode_model_id(provider_id, model_id).lower()
if normalized:
for prefixes, mode in _OPENCODE_API_MODE_PREFIXES.get(family or "", ()):
if normalized.startswith(prefixes):
return mode
return "chat_completions"
# Relay path per OpenCode family on opencode.ai hosts.
_OPENCODE_FAMILY_PATHS = {"opencode-zen": "/zen", "opencode-go": "/zen/go"}
def normalize_opencode_base_url(
provider_id: Optional[str], api_mode: Optional[str], base_url: Optional[str]) -> str:
"""Normalize an OpenCode Zen / Go base URL for the API mode. Must be SYMMETRIC: the anthropic-
stripped URL gets persisted to ``model.base_url`` after switching into an anthropic-routed model,
and chat/codex modes heal it by re-adding ``/v1`` — but only on opencode.ai hosts, so custom
``OPENCODE_*_BASE_URL`` proxies are left alone. On those hosts the relay path segment follows
the resolved family too (``/zen`` vs ``/zen/go``): the two relays serve different model sets,
so a ``model.base_url`` carried over from the other family 401s ("Model ... is not supported").
The family heal applies to the BUILT-IN providers only: a custom provider merely named after a
family (``opencode-go-bridge``) declared its relay path explicitly in ``providers:`` and keeps it.
Only the path is edited, so a port, userinfo, query or fragment round-trips untouched."""
url = str(base_url or "").strip().rstrip("/")
family = opencode_provider_family(provider_id)
if not url or family is None:
return url
parsed = urllib.parse.urlparse(url)
host = (parsed.hostname or "").lower()
official = host == "opencode.ai" or host.endswith(".opencode.ai")
path = parsed.path.rstrip("/")
if official and normalize_provider(provider_id) in _OPENCODE_FAMILIES and re.fullmatch(r"/zen(/go)?(/v1)?", path):
path = _OPENCODE_FAMILY_PATHS[family] + ("/v1" if path.endswith("/v1") else "")
if api_mode == "anthropic_messages":
path = re.sub(r"/v1$", "", path)
elif official and not path.endswith("/v1"):
path += "/v1"
return urllib.parse.urlunparse(parsed._replace(path=path))
# Negative cache: monotonic timestamp of the last timed-out probe, keyed
# by ``host:port`` so both URL candidates (``/v1`` + root) share one entry.
# Without this, an unreachable endpoint (TCP blackhole — SYN draws no reply,
# so every attempt burns its full connect timeout) makes every picker open /
# chat turn re-pay the timeout per candidate, and the sequential stalls stack
# past 10s while the Desktop sits on a spinner with no error (#81123). Short
# TTL collapses the burst but still picks up recovery without a restart.
# Mirrors _deepinfra_catalog_neg_cache.
_probe_neg_cache: dict[str, float] = {}
_PROBE_NEG_TTL = 60.0 # seconds
def _probe_neg_key(base_url: str) -> Optional[str]:
"""``host:port`` for *base_url* (both URL candidates share one entry), or None without a host."""
from utils import base_url_origin
_, host, port = base_url_origin(base_url)
return f"{host}:{port}" if host else None
def _probe_result(
models, probed_url, resolved_base_url, suggested_base_url=None, used_fallback=False
) -> dict[str, Any]:
return {
"models": models,
"probed_url": probed_url,
"resolved_base_url": resolved_base_url,
"suggested_base_url": suggested_base_url,
"used_fallback": used_fallback}
def probe_api_models(
api_key: Optional[str], base_url: Optional[str], timeout: float = 5.0,
api_mode: Optional[str] = None, request_headers: Optional[dict[str, str]] = None,
) -> dict[str, Any]:
"""Probe a ``/models`` endpoint with light URL heuristics (``base`` then ``base±/v1``).
``anthropic_messages`` mode sends ``x-api-key`` + ``anthropic-version`` instead of a bearer; the
``data[].id`` response shape is identical. ``models`` is None when no candidate answered."""
normalized = (base_url or "").strip().rstrip("/")
if not normalized:
return _probe_result(None, None, "")
from providers import profile_for_host
owner = profile_for_host(normalized, metadata=True)
owned = owner.probe_models(base_url=normalized, api_key=api_key, timeout=timeout) if owner else None
if owned is not None:
return _probe_result(*owned)
alternate_base = normalized[:-3].rstrip("/") if normalized.endswith("/v1") else normalized + "/v1"
candidates: list[tuple[str, bool]] = [(normalized, False)]
if alternate_base and alternate_base != normalized:
candidates.append((alternate_base, True))
tried: list[str] = []
_neg_key = _probe_neg_key(normalized)
if _neg_key is not None:
_neg_seen = _probe_neg_cache.get(_neg_key)
if _neg_seen is not None and (time.monotonic() - _neg_seen) < _PROBE_NEG_TTL:
return _probe_result(
None, normalized.rstrip("/") + "/models", normalized,
alternate_base if alternate_base != normalized else None)
headers: dict[str, str] = {"User-Agent": hermes_user_agent()}
if urllib.parse.urlparse(normalized).hostname == "generativelanguage.googleapis.com":
headers["X-Goog-Api-Client"] = hermes_user_agent("hermes-agent")
if api_key and api_mode == "anthropic_messages":
headers["x-api-key"] = api_key
headers["anthropic-version"] = "2023-06-01"
elif api_key:
headers["Authorization"] = f"Bearer {api_key}"
if isinstance(request_headers, dict):
# Per-provider custom headers can contain secrets: merge last so endpoint config wins; never log.
from hermes_cli.config import normalize_extra_headers
headers.update(normalize_extra_headers(request_headers))
# Only thread ssl_context when a per-provider TLS override applies; public endpoints keep the
# original 2-arg call so existing call-seam mocks stay valid.
_open_kwargs: dict[str, Any] = {}
_ssl_context = _custom_provider_ssl_context(normalized)
if _ssl_context is not None:
_open_kwargs["ssl_context"] = _ssl_context
all_timed_out = True
for candidate_base, is_fallback in candidates:
url = candidate_base.rstrip("/") + "/models"
tried.append(url)
try:
data = _get_json(url, timeout=timeout, headers=headers, **_open_kwargs)
except Exception as exc:
# TLS, authentication and parsing failures must not hide corrected settings.
cause = exc.reason if isinstance(exc, urllib.error.URLError) else exc
all_timed_out = all_timed_out and isinstance(cause, TimeoutError)
continue
if _neg_key is not None:
_probe_neg_cache.pop(_neg_key, None)
from hermes_cli.chat_catalog import note_catalog_item
probed = []
for item in data.get("data", []):
if isinstance(item, dict) and note_catalog_item(item):
continue
probed.append(item.get("id", ""))
return _probe_result(
probed, url, candidate_base.rstrip("/"),
alternate_base if alternate_base != candidate_base else normalized, is_fallback)
if _neg_key is not None and all_timed_out:
_probe_neg_cache[_neg_key] = time.monotonic()
return _probe_result(
None, tried[0] if tried else normalized.rstrip("/") + "/models", normalized,
alternate_base if alternate_base != normalized else None)
# Legacy id-regex filter for items with no surface tag; unreachable (deletable) once every catalog
# entry carries an explicit ``chat``/``embed``/``image-gen``/``tts``/``stt`` tag.
_DEEPINFRA_EXCLUDE_RE = re.compile(
r"(?i)(embed|rerank|whisper|stable-diffusion|flux|sdxl|"
r"tts|bark|speech|image-gen|clip|vit-|dpt-)")
# Surface tags say *what kind of model* this is. Absent all of them, the tags array only carries
# capability tags (``reasoning``, ``vision``, …) and the chat surface falls back to id-regex inference.
_DEEPINFRA_SURFACE_TAGS: frozenset[str] = frozenset({
"chat", "embed", "image-gen", "tts", "stt", "video-gen"})
_DEEPINFRA_DEFAULT_BASE_URL = "https://api.deepinfra.com/v1/openai"
_DEEPINFRA_MODELS_QUERY = "filter=true&sort_by=hermes"
# Full tagged catalog keyed by base URL; every surface filter reads it so one round-trip serves all.
_deepinfra_catalog_cache: dict[str, list[dict]] = {}
# Negative cache (monotonic time of the last failed fetch per base URL) so an unreachable catalog
# doesn't make every surface helper eat the full timeout in turn. Short TTL so connectivity recovers.
_deepinfra_catalog_neg_cache: dict[str, float] = {}
_DEEPINFRA_CATALOG_NEG_TTL = 60.0 # seconds
def _deepinfra_env(key: str) -> str:
"""Profile-scoped ``.env``/environ read: under a multiplexed turn the launch env is not this profile's."""
from hermes_cli.config import get_env_value_prefer_dotenv
return (get_env_value_prefer_dotenv(key) or "").strip()
def _deepinfra_catalog_url() -> tuple[str, str]:
"""Return ``(cache_key, full_url)`` for the DeepInfra catalog endpoint. The key carries the
api-key fingerprint: the catalog is user-scoped (private fine-tunes), so two profiles with
different keys must not share an entry."""
base = (_deepinfra_env("DEEPINFRA_BASE_URL") or _DEEPINFRA_DEFAULT_BASE_URL).rstrip("/")
from agent.credential_persistence import fingerprint_secret_value
fp = fingerprint_secret_value(_deepinfra_env("DEEPINFRA_API_KEY")) or "anon"
return f"{base}#{fp}", f"{base}/models?{_DEEPINFRA_MODELS_QUERY}"
def _fetch_deepinfra_catalog(
*, timeout: float = 5.0, force_refresh: bool = False) -> Optional[list[dict]]:
"""Raw DeepInfra catalog list (chat, embed, image-gen, TTS, STT in one response), cached per base
URL. A Bearer token is attached when available so user-scoped catalogs (private fine-tunes) show."""
cache_key, url = _deepinfra_catalog_url()
if not force_refresh:
if cache_key in _deepinfra_catalog_cache:
return _deepinfra_catalog_cache[cache_key]
last_fail = _deepinfra_catalog_neg_cache.get(cache_key)
if last_fail is not None and (time.monotonic() - last_fail) < _DEEPINFRA_CATALOG_NEG_TTL:
return None
headers: dict[str, str] = {"User-Agent": hermes_user_agent()}
api_key = _deepinfra_env("DEEPINFRA_API_KEY")
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
try:
payload = _get_json(url, timeout=timeout, headers=headers)
except Exception:
_deepinfra_catalog_neg_cache[cache_key] = time.monotonic()
return None
data = payload.get("data")
if not isinstance(data, list):
_deepinfra_catalog_neg_cache[cache_key] = time.monotonic()
return None
_deepinfra_catalog_cache[cache_key] = data
_deepinfra_catalog_neg_cache.pop(cache_key, None)
return data
def _fetch_deepinfra_models_by_tag(
tag: str, *, timeout: float = 5.0, force_refresh: bool = False) -> Optional[list[dict]]:
"""DeepInfra ``{"id", "metadata"}`` items whose ``metadata.tags`` includes *tag*. Items with no
surface tag fall through to the legacy id-regex exclusion (chat surface only — embed/image-gen/
tts/stt cannot be inferred from an id). ``None`` on network failure."""
data = _fetch_deepinfra_catalog(timeout=timeout, force_refresh=force_refresh)
if data is None:
return None
matched: list[dict] = []
for item in data:
mid = item.get("id")
raw_metadata = item.get("metadata")
if not mid or raw_metadata is None: # metadata None = listed-but-not-served stub
continue
metadata = raw_metadata if isinstance(raw_metadata, dict) else {}
raw_tags = metadata.get("tags")
tags = raw_tags if isinstance(raw_tags, list) else []
if any(t in _DEEPINFRA_SURFACE_TAGS for t in tags):
hit = tag in tags
else:
hit = tag == "chat" and not _DEEPINFRA_EXCLUDE_RE.search(mid)
if hit:
matched.append({"id": mid, "metadata": metadata})
return matched
def _fetch_deepinfra_models(
timeout: float = 5.0, *, force_refresh: bool = False) -> Optional[list[str]]:
"""DeepInfra chat-model ids (string-list contract for :func:`provider_model_ids`); ``None`` on
network failure or when no chat-tagged id exists."""
items = _fetch_deepinfra_models_by_tag("chat", timeout=timeout, force_refresh=force_refresh)
return ([item["id"] for item in items] or None) if items is not None else None
def deepinfra_model_ids(tag: str, *, force_refresh: bool = False) -> list[str]:
"""Return DeepInfra model ids carrying surface *tag* (``[]`` on failure)."""
items = _fetch_deepinfra_models_by_tag(tag, force_refresh=force_refresh)
return [item["id"] for item in items] if items else []
def deepinfra_base_url(section: Optional[dict] = None) -> str:
"""DeepInfra base URL: config-section ``base_url`` → ``DEEPINFRA_BASE_URL`` env → default; stripped."""
candidate = section.get("base_url") if isinstance(section, dict) else None
value = candidate or _deepinfra_env("DEEPINFRA_BASE_URL") or _DEEPINFRA_DEFAULT_BASE_URL
return str(value).strip().rstrip("/")
def _fetch_ai_gateway_models(timeout: float = 5.0) -> Optional[list[str]]:
"""Fetch available language models with tool-use from AI Gateway."""
api_key = os.getenv("AI_GATEWAY_API_KEY", "").strip()
if not api_key:
return None
base_url = os.getenv("AI_GATEWAY_BASE_URL", "").strip()
if not base_url:
from hermes_constants import AI_GATEWAY_BASE_URL
base_url = AI_GATEWAY_BASE_URL
headers = {"Authorization": f"Bearer {api_key}", "User-Agent": hermes_user_agent()}
try:
url = base_url.rstrip("/") + "/models"
data = _get_json(url, timeout=timeout, headers=headers)
return [
m["id"] for m in data.get("data", [])
if m.get("id") and m.get("type") == "language" and "tool-use" in (m.get("tags") or [])]
except Exception:
return None
def fetch_api_models(
api_key: Optional[str], base_url: Optional[str], timeout: float = 5.0,
api_mode: Optional[str] = None, headers: Optional[dict[str, str]] = None,
) -> Optional[list[str]]:
"""Fetch the list of available model IDs from the provider's ``/models`` endpoint."""
result = probe_api_models(api_key, base_url, timeout=timeout, api_mode=api_mode, request_headers=headers)
return result.get("models")
def _custom_endpoint_fingerprint(
api_key: Any, api_mode: Optional[str], headers: Optional[dict[str, str]]) -> str:
"""Custom endpoints have no ``PROVIDER_REGISTRY`` slug, so hash exactly what callers pass to
:func:`fetch_api_models`: a rotated ``api_key``, changed ``api_mode`` or edited ``extra_headers``
each bust the cache entry. blake2b for the same CodeQL rationale as ``_credential_fingerprint``."""
import hashlib
from agent.command_token_source import CommandTokenSource
identity = api_key.cache_identity if isinstance(api_key, CommandTokenSource) else api_key
blob = "|".join((identity or "", api_mode or "", json.dumps(headers or {}, sort_keys=True)))
return hashlib.blake2b(blob.encode("utf-8", errors="replace"), digest_size=8).hexdigest()
def _cache_entry_valid(
entry: Any, fp: str, *, allow_empty: bool = False) -> TypeGuard[dict[str, Any]]:
"""Well-formed cache row for fingerprint *fp*. Requires a numeric ``at`` so corrupt disk state
degrades to a cache miss instead of raising; empty model lists are valid only when the caller
opts into an authoritative empty catalog."""
return (
isinstance(entry, dict)
and entry.get("fp") == fp
and isinstance(entry.get("models"), list)
and (allow_empty or bool(entry["models"]))
and isinstance(entry.get("at"), (int, float))
and not isinstance(entry.get("at"), bool))
def _disk_serve_tier(entry: Any, fp: str, now: float, *, is_ollama: bool,
ttl_seconds: int = _PROVIDER_MODELS_CACHE_TTL) -> Optional[str]:
"""How :func:`cached_provider_model_ids` serves *entry* without the network.
``"fresh"`` inside the row's TTL (a curated fallback row only for
``_PROVIDER_MODELS_FALLBACK_TTL``), ``"stale"`` for a non-empty, non-fallback row inside
``_PROVIDER_MODELS_STALE_SERVE_MAX`` (served while an SWR thread revalidates), else ``None``:
the call would block on a live fetch. Empty native catalogs are authoritative only inside the
short native TTL, never through the stale window."""
if is_ollama:
ttl_seconds = min(ttl_seconds, _OLLAMA_LOCAL_MODELS_CACHE_TTL)
if not _cache_entry_valid(entry, fp, allow_empty=is_ollama):
return None
age = now - entry["at"]
if age < (_PROVIDER_MODELS_FALLBACK_TTL if entry.get("fallback") else ttl_seconds):
return "fresh"
if entry["models"] and not entry.get("fallback") and age < _PROVIDER_MODELS_STALE_SERVE_MAX:
return "stale"
return None
def cached_fetch_api_models(
api_key: Any, base_url: Optional[str], *, timeout: float = 5.0,
api_mode: Optional[str] = None, headers: Optional[dict[str, str]] = None,
force_refresh: bool = False, cache_only: bool = False,
fetch_models=None,
ttl_seconds: int = _PROVIDER_MODELS_CACHE_TTL) -> Optional[list[str]]:
"""Disk-cached :func:`fetch_api_models` for custom endpoints. ``cache_only`` callers (GUI picker
opens that must not block on a stopped local endpoint) still get a warm catalog instead of
collapsing to the config-declared subset. ``fetch_models`` supplies native-aware discovery
without minting a command token before cache admission."""
from hermes_cli.model_switch_providers import _NativePickerModelList
def _catalog(entry):
rows = (_NativePickerModelList if entry.get("native_catalog") else list)(entry["models"])
return _chat_catalog_rows(rows)
def _entry(live, at=None):
return {**_cache_entry(fp, live, at), "native_catalog": isinstance(live, _NativePickerModelList)}
def _live():
if fetch_models is not None:
return fetch_models()
from agent.command_token_source import materialize_probe_api_key
return fetch_api_models(materialize_probe_api_key(api_key), base_url, timeout=timeout, api_mode=api_mode, headers=headers)
normalized_url = str(base_url or "").strip().rstrip("/").lower()
if not normalized_url: # nothing to key the cache on
return None if cache_only else _chat_catalog_rows(_live())
# Key on URL AND credential fingerprint: N ``custom_providers`` rows can share one proxy URL
# with distinct keys (#106184). A URL-only key let the last probe overwrite its siblings'
# slot, so every other same-URL row failed the fingerprint check, got an empty catalog and
# vanished from the no-probe pickers.
fp = _custom_endpoint_fingerprint(api_key, api_mode, headers)
cache_key = f"custom:{normalized_url}#{fp}"
cache = _load_provider_models_cache()
entry = cache.get(cache_key)
now = time.time()
native_row = isinstance(entry, dict) and entry.get("native_catalog") is True
valid = not force_refresh and _cache_entry_valid(entry, fp, allow_empty=native_row)
if valid:
age = now - entry["at"]
if age < ttl_seconds:
return _catalog(entry)
# An empty native catalog is authoritative only inside the TTL (as in
# cached_provider_model_ids): never stale-serve it, or an Ollama that was model-less at
# first open keeps an empty row for the whole stale window after models are pulled.
if entry["models"] and age < _PROVIDER_MODELS_STALE_SERVE_MAX:
# Stale-while-revalidate: serve now, refresh off-thread for the next open. cache_only
# opens (GUI pickers that must not block on a stopped local server) take the same
# non-blocking refresh: without it a locally loaded model stayed invisible for the
# whole 7-day stale window unless the user found "Refresh Models" (#71169 class).
def _refresh_custom():
live = _live()
return _entry(live) if live or isinstance(live, _NativePickerModelList) else None
_spawn_swr_refresh(cache_key, _refresh_custom)
return _catalog(entry)
if cache_only:
return None
live = _live()
if live or isinstance(live, _NativePickerModelList):
stored = _entry(live, now)
_store_cache_entry(cache_key, stored)
return _catalog(stored)
# Live returned nothing (offline, timeout, auth hiccup): a stale same-fingerprint entry beats it
# (non-empty only: an empty native row is not worth resurrecting over the generic fallback).
if _cache_entry_valid(entry, fp):
return _catalog(entry)
return _chat_catalog_rows(live)
Reference in New Issue View Git Blame Copy Permalink