Merge branch 'simp/r2m-models' into simp/integration2

hermes_cli/models_local.py

@@ -0,0 +1,824 @@
+"""Local / self-hosted model servers.
+
+Ollama (native ``/api/tags`` probe, request headers, base-url resolution), LM Studio
+(``/api/v1/models``, load-on-demand), and Ollama Cloud (live + models.dev merged catalog with a
+disk cache).
+
+Split out of ``hermes_cli.models``; every moved name is re-imported there, so
+``hermes_cli.models.<name>`` keeps resolving (and monkeypatching) as before.
+"""
+
+from __future__ import annotations
+
+import http.client
+import json
+import logging
+import os
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any, NamedTuple, Optional
+from hermes_cli.urllib_security import url_origin
+
+# Log-record parity with the origin module.
+logger = logging.getLogger("hermes_cli.models")
+
+
+def _root_for_ollama_native_api(base_url: str) -> str:
+    """Convert an OpenAI-style Ollama base URL to the native API root."""
+    root = str(base_url or "").strip().rstrip("/")
+    if root.startswith(":"):
+        root = "http://127.0.0.1" + root
+    elif root and "://" not in root:
+        root = "http://" + root
+    for suffix in ("/api/tags", "/v1/models", "/api", "/v1"):
+        if root.endswith(suffix):
+            root = root[: -len(suffix)].rstrip("/")
+            break
+    return root
+
+
+def _normalize_openai_base_url(base_url: Optional[str]) -> str:
+    """Add a usable HTTP scheme without changing an OpenAI API path."""
+    value = str(base_url or "").strip()
+    if value.startswith(":"):
+        return "http://127.0.0.1" + value
+    if value and "://" not in value:
+        return "http://" + value
+    return value
+
+
+def _configured_ollama_base_url() -> str:
+    """``providers.ollama.base_url`` (legacy keys ``api`` / ``url``), or ``""``."""
+    from hermes_cli.models import _get_provider_config_dict
+
+    cfg = _get_provider_config_dict("ollama")
+    return str(cfg.get("base_url") or cfg.get("api") or cfg.get("url") or "").strip()
+
+
+def _get_ollama_base_url() -> str:
+    """Resolve the local Ollama-compatible endpoint URL.
+
+    Prefer explicit config under ``providers.ollama.base_url`` because this is how local Ollama-
+    compatible endpoints can be wired without changing the active model provider. Fall back to
+    active ``model.base_url`` only when the active provider is ollama/custom, then to Ollama's local
+    default.
+    """
+    from hermes_cli.models import _get_model_config_dict, should_use_ollama_native_catalog
+    configured = _configured_ollama_base_url()
+    if configured:
+        return configured
+
+    model_cfg = _get_model_config_dict()
+    model_provider = str(model_cfg.get("provider", "") or "").strip().lower()
+    model_base = str(model_cfg.get("base_url", "") or "").strip()
+    if model_provider == "ollama" and model_base:
+        return model_base
+    if model_provider == "custom" and model_base:
+        # Only reuse the active bare custom endpoint when it is actually Ollama-compatible;
+        # otherwise the Ollama picker would probe an unrelated endpoint's /api/tags and hide the
+        # local Ollama catalog.
+        try:
+            if should_use_ollama_native_catalog("custom", model_base):
+                return model_base
+        except (OSError, RuntimeError, TypeError, ValueError):
+            pass
+
+    env_host = os.getenv("OLLAMA_HOST", "").strip()
+    if env_host:
+        if env_host.startswith(":") and not env_host.startswith("::"):
+            env_host = "127.0.0.1" + env_host
+        elif env_host.startswith("[") and env_host.endswith("]"):
+            env_host = f"{env_host}:11434"
+        elif "://" in env_host:
+            try:
+                parsed = urllib.parse.urlsplit(env_host)
+                if parsed.hostname and parsed.port is None:
+                    hostname = parsed.hostname
+                    if ":" in hostname and not hostname.startswith("["):
+                        hostname = f"[{hostname}]"
+                    userinfo = (
+                        parsed.netloc.rsplit("@", 1)[0] + "@"
+                        if "@" in parsed.netloc
+                        else ""
+                    )
+                    env_host = parsed._replace(
+                        netloc=f"{userinfo}{hostname}:11434"
+                    ).geturl()
+            except ValueError:
+                pass
+        elif env_host.count(":") > 1 and not env_host.startswith("["):
+            env_host = f"[{env_host}]:11434"
+        elif ":" not in env_host:
+            env_host = f"{env_host}:11434"
+        return env_host
+    return "http://localhost:11434"
+
+
+def _get_ollama_request_headers() -> dict[str, str]:
+    """Return configured headers and credentials for native Ollama requests."""
+    from hermes_cli.models import _get_provider_config_dict
+    entry = _get_provider_config_dict("ollama")
+    raw = entry.get("extra_headers")
+    try:
+        from hermes_cli.config import normalize_extra_headers
+
+        result = normalize_extra_headers(raw)
+    except (ImportError, OSError, RuntimeError, TypeError, ValueError):
+        result = {}
+
+    api_key = str(entry.get("api_key") or "").strip()
+    if not api_key:
+        key_env = str(entry.get("key_env") or entry.get("api_key_env") or "").strip()
+        api_key = os.getenv(key_env, "").strip() if key_env else ""
+    if api_key and not any(key.lower() == "authorization" for key in result):
+        result["Authorization"] = f"Bearer {api_key}"
+    return result
+
+
+def _get_ollama_native_headers(
+    base_url: Optional[str],
+    *,
+    api_key: Optional[str] = None,
+) -> dict[str, str]:
+    """Resolve Ollama credentials and headers for one endpoint origin."""
+    from hermes_cli.models import _get_ollama_request_headers
+    configured_base = _configured_ollama_base_url()
+    explicit_key = str(api_key or "").strip()
+    configured_matches = bool(configured_base and base_url and _same_ollama_native_root(base_url, configured_base))
+    if not configured_matches and not explicit_key:
+        return {}
+    headers = _get_ollama_request_headers() if configured_matches else {}
+    if explicit_key:
+        # A provider-specific key must not inherit any configured Authorization
+        # variant from the Ollama origin when both share a native root.
+        for key in tuple(headers):
+            if key.lower() == "authorization":
+                del headers[key]
+        headers["Authorization"] = f"Bearer {explicit_key}"
+    return headers
+
+
+# Native /api/tags probe caches, keyed by root (+ header fingerprint): successful catalogs,
+# failure timestamps (short negative TTL), and whether the root answered the native probe.
+_OLLAMA_LOCAL_MODELS_CACHE_TTL: int = 300  # seconds
+_OLLAMA_LOCAL_MODELS_CACHE: dict[str, tuple[tuple[str, ...], float]] = {}
+_OLLAMA_LOCAL_PROBE_FAILURE_CACHE: dict[str, float] = {}
+_OLLAMA_LOCAL_PROBE_REACHABLE: dict[str, bool] = {}
+_OLLAMA_LOCAL_PROBE_FAILURE_TTL: int = 30
+
+
+_OLLAMA_LOCAL_CACHE_MAX_ENTRIES: int = 256
+
+
+def _evict_related_ollama_cache_entries(key: str) -> None:
+    _OLLAMA_LOCAL_MODELS_CACHE.pop(key, None)
+    _OLLAMA_LOCAL_PROBE_REACHABLE.pop(key, None)
+    for failure_key in list(_OLLAMA_LOCAL_PROBE_FAILURE_CACHE):
+        if failure_key == key or failure_key.startswith(f"{key}|timeout:"):
+            _OLLAMA_LOCAL_PROBE_FAILURE_CACHE.pop(failure_key, None)
+
+
+def _remember_ollama_cache(cache: dict[str, Any], key: str, value: Any) -> None:
+    if key not in cache and len(cache) >= _OLLAMA_LOCAL_CACHE_MAX_ENTRIES:
+        oldest_key = next(iter(cache))
+        _evict_related_ollama_cache_entries(
+            oldest_key.split("|timeout:", 1)[0]
+        )
+    cache[key] = value
+
+
+def _ollama_probe_cache_key(root: str, headers: Optional[dict[str, str]]) -> str:
+    cache_key = root
+    if headers:
+        import hashlib
+
+        normalized_headers = sorted(
+            (str(key).lower(), str(value)) for key, value in headers.items()
+        )
+        header_blob = json.dumps(
+            normalized_headers, ensure_ascii=False, separators=(",", ":")
+        ).encode("utf-8", errors="replace")
+        header_fingerprint = hashlib.blake2b(header_blob, digest_size=8).hexdigest()
+        cache_key = f"{root}|headers:{header_fingerprint}"
+    return cache_key
+
+
+def _parse_ollama_tags(payload: Any) -> Optional[list[str]]:
+    """Model ids from an ``/api/tags`` payload; None when the shape is not Ollama's."""
+    raw_models = payload.get("models") if isinstance(payload, dict) else None
+    if not isinstance(raw_models, list):
+        return None
+    models: list[str] = []
+    seen: set[str] = set()
+    for item in raw_models:
+        if not isinstance(item, dict):
+            return None
+        model_id = str(item.get("model") or item.get("name") or "").strip()
+        if model_id and model_id not in seen:
+            seen.add(model_id)
+            models.append(model_id)
+    if raw_models and not models:
+        return None
+    return models
+
+
+def probe_ollama_local_models(
+    base_url: Optional[str] = None,
+    timeout: float = 2.0,
+    headers: Optional[dict[str, str]] = None,
+) -> Optional[list[str]]:
+    """Probe local Ollama-compatible models from native ``/api/tags``.
+
+    Returns ``None`` when the endpoint cannot be reached or returns malformed data, and a list
+    (possibly empty) when ``/api/tags`` was reachable. Stock Ollama exposes its authoritative local
+    model catalog at ``/api/tags``; OpenAI-compatible ``/v1/models`` is not required for local
+    Ollama servers.
+    """
+    from hermes_cli.models import _HERMES_USER_AGENT, _get_ollama_base_url, _urlopen_model_catalog_request
+    root = _root_for_ollama_native_api(base_url or _get_ollama_base_url())
+    if not root:
+        return None
+    cache_key = _ollama_probe_cache_key(root, headers)
+    failure_key = f"{cache_key}|timeout:{float(timeout):.3f}"
+    cached = _OLLAMA_LOCAL_MODELS_CACHE.get(cache_key)
+    if cached is not None:
+        cached_models, cached_at = cached
+        if time.monotonic() - cached_at < _OLLAMA_LOCAL_MODELS_CACHE_TTL:
+            return list(cached_models)
+    failed_at = _OLLAMA_LOCAL_PROBE_FAILURE_CACHE.get(failure_key)
+    if failed_at is not None:
+        if time.monotonic() - failed_at < _OLLAMA_LOCAL_PROBE_FAILURE_TTL:
+            return None
+        _OLLAMA_LOCAL_PROBE_FAILURE_CACHE.pop(failure_key, None)
+
+    def _unreachable() -> None:
+        _remember_ollama_cache(_OLLAMA_LOCAL_PROBE_REACHABLE, cache_key, False)
+        _remember_ollama_cache(_OLLAMA_LOCAL_PROBE_FAILURE_CACHE, failure_key, time.monotonic())
+
+    try:
+        request_headers = {"User-Agent": _HERMES_USER_AGENT, **(headers or {})}
+        req = urllib.request.Request(root.rstrip("/") + "/api/tags", headers=request_headers)
+        with _urlopen_model_catalog_request(req, timeout=timeout) as resp:
+            payload = json.loads(resp.read().decode())
+    except (ValueError, OSError, TimeoutError, http.client.HTTPException, urllib.error.URLError,
+            json.JSONDecodeError, UnicodeDecodeError):
+        _unreachable()
+        return None
+
+    models = _parse_ollama_tags(payload)
+    if models is None:
+        _unreachable()
+        return None
+    _remember_ollama_cache(_OLLAMA_LOCAL_PROBE_REACHABLE, cache_key, True)
+    _OLLAMA_LOCAL_PROBE_FAILURE_CACHE.pop(failure_key, None)
+    _remember_ollama_cache(_OLLAMA_LOCAL_MODELS_CACHE, cache_key, (tuple(models), time.monotonic()))
+    return models
+
+
+def fetch_ollama_local_models(
+    base_url: Optional[str] = None,
+    timeout: float = 2.0,
+    headers: Optional[dict[str, str]] = None,
+) -> Optional[list[str]]:
+    """Fetch local Ollama-compatible models, preserving probe failure as ``None``."""
+    from hermes_cli.models import probe_ollama_local_models
+    return probe_ollama_local_models(base_url, timeout, headers=headers)
+
+
+def _same_ollama_native_root(left: str, right: str) -> bool:
+    """Return True when two Ollama/OpenAI-style base URLs share an API root."""
+    left_root = _root_for_ollama_native_api(left).rstrip("/")
+    right_root = _root_for_ollama_native_api(right).rstrip("/")
+    if not left_root or not right_root:
+        return False
+    try:
+        left_parts = urllib.parse.urlsplit(left_root)
+        right_parts = urllib.parse.urlsplit(right_root)
+        return (
+            url_origin(left_root) == url_origin(right_root)
+            and left_parts.path.rstrip("/") == right_parts.path.rstrip("/")
+        )
+    except (AttributeError, ValueError):
+        return False
+
+
+def should_use_ollama_native_catalog(
+    provider: Optional[str],
+    base_url: Optional[str],
+    headers: Optional[dict[str, str]] = None,
+) -> bool:
+    """Return True when model discovery should use local Ollama ``/api/tags``.
+
+    Bare ``ollama`` is normalized to ``custom`` elsewhere so runtime paths share the OpenAI-
+    compatible client, but local Ollama's authoritative model list is ``/api/tags``. Use it when
+    the caller asked for Ollama explicitly, the base URL matches ``providers.ollama.base_url``,
+    or an ambiguous custom URL on Ollama's default port actually serves ``/api/tags``; other
+    custom endpoints keep the ``/models`` probe.
+    """
+    from hermes_cli.models import probe_ollama_local_models
+    requested = str(provider or "").strip().lower()
+    root = _root_for_ollama_native_api(base_url or "")
+    if root:
+        try:
+            host = (urllib.parse.urlparse(root).hostname or "").lower()
+            if host == "ollama.com" or host.endswith(".ollama.com"):
+                return False
+        except ValueError:
+            pass
+
+    if requested in {"openrouter", "nous", "anthropic", "openai", "openai-codex", "gemini", "ollama-cloud"}:
+        return False
+
+    configured_base = _configured_ollama_base_url()
+    if requested == "ollama":
+        if not root:
+            return False
+        if configured_base and not _same_ollama_native_root(root, configured_base):
+            return probe_ollama_local_models(root, timeout=0.5, headers=headers) is not None
+        return True
+
+    if configured_base and _same_ollama_native_root(root, configured_base):
+        return True
+
+    if not root:
+        return False
+
+    local_like_providers = {"", "custom", "local", "llamacpp", "llama.cpp", "llama-cpp", "vllm"}
+    if requested not in local_like_providers and not requested.startswith("custom:"):
+        return False
+
+    if requested == "custom:ollama" or requested.endswith("-ollama"):
+        return True
+
+    try:
+        parsed = urllib.parse.urlparse(root)
+        if parsed.port != 11434:
+            return False
+    except ValueError:
+        return False
+
+    return probe_ollama_local_models(root, timeout=0.5, headers=headers) is not None
+
+
+def _ollama_local_catalog(force_refresh: bool) -> list[str]:
+    """Catalog for the raw ``ollama`` provider: native ``/api/tags`` when the endpoint is a real
+    Ollama server, else the OpenAI-style ``/v1/models`` of the configured gateway."""
+    from hermes_cli.models import _get_ollama_base_url, _get_ollama_native_headers, _get_provider_config_dict, fetch_api_models, fetch_ollama_local_models, should_use_ollama_native_catalog
+    if force_refresh:
+        _OLLAMA_LOCAL_MODELS_CACHE.clear()
+        _OLLAMA_LOCAL_PROBE_FAILURE_CACHE.clear()
+        _OLLAMA_LOCAL_PROBE_REACHABLE.clear()
+    base_url = _get_ollama_base_url()
+    headers = _get_ollama_native_headers(base_url)
+    if should_use_ollama_native_catalog("ollama", base_url, headers=headers):
+        if headers:
+            native_models = fetch_ollama_local_models(base_url, headers=headers)
+        else:
+            native_models = fetch_ollama_local_models(base_url)
+        native_key = _ollama_probe_cache_key(_root_for_ollama_native_api(base_url), headers or None)
+        if native_models or _OLLAMA_LOCAL_PROBE_REACHABLE.get(native_key) is True:
+            return native_models or []
+    # Non-native Ollama-compatible endpoints (incl. Ollama Cloud) and gateways exposing only
+    # OpenAI-style /v1/models.
+    config = _get_provider_config_dict("ollama")
+    fallback_key = str(config.get("api_key") or "").strip()
+    if not fallback_key:
+        key_env = str(config.get("key_env") or "").strip()
+        fallback_key = os.getenv(key_env, "").strip() if key_env else ""
+    fallback_base = _normalize_openai_base_url(config.get("base_url") or base_url)
+    fallback_headers = _get_ollama_native_headers(fallback_base, api_key=fallback_key)
+    return fetch_api_models(fallback_key, fallback_base, headers=fallback_headers or None) or []
+
+
+def _lmstudio_server_root(base_url: Optional[str]) -> Optional[str]:
+    """Return the LM Studio server root for native ``/api/v1`` endpoints.
+
+    Users commonly copy either the OpenAI-compatible runtime URL (``.../v1``) or the native API
+    prefix (``.../api`` / ``.../api/v1``). Native probes append ``/api/v1/...`` themselves, so
+    normalize all accepted forms back to the bare server root to avoid ``/api/api/v1`` requests.
+    """
+    root = (base_url or "").strip().rstrip("/")
+    for suffix in ("/api/v1", "/api", "/v1"):
+        if root.endswith(suffix):
+            root = root[: -len(suffix)].rstrip("/")
+            break
+    return root or None
+
+
+def _lmstudio_request_headers(api_key: Optional[str] = None) -> dict:
+    """Build HTTP headers for LM Studio native API requests."""
+    from hermes_cli.models import _HERMES_USER_AGENT
+    headers = {"User-Agent": _HERMES_USER_AGENT}
+    token = str(api_key or "").strip()
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    return headers
+
+
+def _lmstudio_fetch_raw_models(
+    api_key: Optional[str] = None,
+    base_url: Optional[str] = None,
+    timeout: float = 5.0,
+) -> Optional[list[dict]]:
+    """Fetch the raw model list from LM Studio's ``/api/v1/models``."""
+    from hermes_cli.models import _urlopen_model_catalog_request
+    server_root = _lmstudio_server_root(base_url)
+    if not server_root:
+        return None
+
+    headers = _lmstudio_request_headers(api_key)
+    request = urllib.request.Request(server_root + "/api/v1/models", headers=headers)
+    try:
+        with _urlopen_model_catalog_request(request, timeout=timeout) as resp:
+            payload = json.loads(resp.read().decode())
+    except urllib.error.HTTPError as exc:
+        if exc.code in {401, 403}:
+            from hermes_cli.auth import AuthError
+            raise AuthError(
+                f"LM Studio rejected the request with HTTP {exc.code}.",
+                provider="lmstudio",
+                code="auth_rejected",
+            ) from exc
+        logger.debug("LM Studio probe at %s failed with HTTP %s", server_root, exc.code)
+        return None
+    except Exception as exc:
+        logger.debug("LM Studio probe at %s failed: %s", server_root, exc)
+        return None
+
+    raw_models = payload.get("models") if isinstance(payload, dict) else None
+    if not isinstance(raw_models, list):
+        logger.debug("LM Studio probe at %s returned malformed payload (no `models` list)", server_root)
+        return None
+    return raw_models
+
+
+def probe_lmstudio_models(
+    api_key: Optional[str] = None,
+    base_url: Optional[str] = None,
+    timeout: float = 5.0,
+) -> Optional[list[str]]:
+    """Probe LM Studio's model listing.
+
+    Returns chat-capable model keys, including a valid empty list when the server is reachable
+    but has no non-embedding models; returns ``None`` on network errors, malformed responses, or
+    bad base URLs. Raises ``AuthError`` on HTTP 401/403 so token issues surface separately from
+    reachability.
+    """
+    from hermes_cli.models import _lmstudio_fetch_raw_models
+    raw_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=timeout)
+    if raw_models is None:
+        return None
+
+    keys: list[str] = []
+    for raw in raw_models:
+        if not isinstance(raw, dict):
+            continue
+        if str(raw.get("type") or "").strip().lower() == "embedding":
+            continue
+        key = str(raw.get("key") or raw.get("id") or "").strip()
+        if key and key not in keys:
+            keys.append(key)
+    return keys
+
+
+def fetch_lmstudio_models(
+    api_key: Optional[str] = None,
+    base_url: Optional[str] = None,
+    timeout: float = 5.0,
+) -> list[str]:
+    """Fetch LM Studio chat-capable model keys from native ``/api/v1/models``.
+
+    Embedding models are filtered out; network errors, malformed responses, and bad base URLs
+    yield an empty list. Raises ``AuthError`` on HTTP 401/403 so callers can distinguish a
+    missing or wrong ``LM_API_KEY`` from an unreachable server — the most common LM Studio
+    support case.
+    """
+    from hermes_cli.models import probe_lmstudio_models
+    models = probe_lmstudio_models(api_key=api_key, base_url=base_url, timeout=timeout)
+    return models or []
+
+
+class LMStudioLoadResult(NamedTuple):
+    """Verified LM Studio runtime plus load-attempt provenance."""
+
+    context_length: Optional[int]
+    load_attempted: bool = False
+    rejected: bool = False
+
+
+def ensure_lmstudio_model_loaded(
+    model: str,
+    base_url: Optional[str],
+    api_key: Optional[str],
+    target_context_length: Optional[int],
+    timeout: float = 120.0,
+    *,
+    return_load_result: bool = False,
+) -> Optional[int] | LMStudioLoadResult:
+    """Ensure ``model`` is loaded and return verified runtime context.
+
+    Existing loaded-instance context is authoritative. Cold loads omit ``context_length`` unless the
+    caller supplied an explicit override; the returned context must come from LM Studio's echoed or
+    refreshed state.
+    """
+    from hermes_cli.models import _lmstudio_fetch_raw_models, _urlopen_model_catalog_request
+
+    def _result(
+        context_length: Optional[int],
+        *,
+        load_attempted: bool = False,
+        rejected: bool = False,
+    ) -> Optional[int] | LMStudioLoadResult:
+        value = LMStudioLoadResult(context_length, load_attempted, rejected)
+        return value if return_load_result else context_length
+
+    def _positive_int(value: Any) -> Optional[int]:
+        if isinstance(value, int) and not isinstance(value, bool) and value > 0:
+            return value
+        return None
+
+    def _loaded_context(entry: dict) -> Optional[int]:
+        instances = entry.get("loaded_instances")
+        if not isinstance(instances, list):
+            return None
+        for instance in instances:
+            config = instance.get("config") if isinstance(instance, dict) else None
+            context = config.get("context_length") if isinstance(config, dict) else None
+            parsed = _positive_int(context)
+            if parsed is not None:
+                return parsed
+        return None
+
+    def _find_entry(raw_models: list[dict]) -> Optional[dict]:
+        for raw in raw_models:
+            if isinstance(raw, dict) and (raw.get("key") == model or raw.get("id") == model):
+                return raw
+        return None
+
+    server_root = _lmstudio_server_root(base_url)
+    if not server_root:
+        return _result(None)
+
+    explicit_context = _positive_int(target_context_length)
+    if target_context_length is not None and explicit_context is None:
+        return _result(None)
+
+    headers = _lmstudio_request_headers(api_key)
+
+    try:
+        raw_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=10)
+    except Exception:
+        raw_models = None
+    if raw_models is None:
+        return _result(None)
+
+    target_entry = _find_entry(raw_models)
+    if target_entry is None:
+        return _result(None)
+
+    max_ctx = _positive_int(target_entry.get("max_context_length"))
+    if explicit_context is not None and max_ctx is not None and explicit_context > max_ctx:
+        return _result(None, rejected=True)
+
+    current_context = _loaded_context(target_entry)
+    if current_context is not None:
+        return _result(current_context)
+
+    loaded_instances = target_entry.get("loaded_instances")
+    if not isinstance(loaded_instances, list) or loaded_instances:
+        return _result(None)
+
+    load_payload: dict[str, Any] = {"model": model, "echo_load_config": True}
+    if explicit_context is not None:
+        load_payload["context_length"] = explicit_context
+    body = json.dumps(load_payload).encode()
+    load_headers = dict(headers)
+    load_headers["Content-Type"] = "application/json"
+    try:
+        load_request = urllib.request.Request(
+            server_root + "/api/v1/models/load",
+            data=body,
+            headers=load_headers,
+            method="POST",
+        )
+        with _urlopen_model_catalog_request(load_request, timeout=timeout) as resp:
+            response_body = resp.read()
+    except Exception:
+        return _result(None, load_attempted=True)
+
+    try:
+        response_payload = json.loads(response_body.decode())
+    except Exception:
+        response_payload = None
+    load_config = response_payload.get("load_config") if isinstance(response_payload, dict) else None
+    applied_context = (
+        _positive_int(load_config.get("context_length"))
+        if isinstance(load_config, dict)
+        else None
+    )
+    if applied_context is not None:
+        return _result(applied_context, load_attempted=True)
+
+    try:
+        refreshed_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=10)
+    except Exception:
+        refreshed_models = None
+    if refreshed_models is None:
+        return _result(None, load_attempted=True)
+    refreshed_entry = _find_entry(refreshed_models)
+    refreshed_context = _loaded_context(refreshed_entry) if refreshed_entry is not None else None
+    return _result(refreshed_context, load_attempted=True)
+
+
+def lmstudio_model_reasoning_options(
+    model: str,
+    base_url: Optional[str],
+    api_key: Optional[str] = None,
+    timeout: float = 5.0,
+) -> list[str]:
+    """Return the reasoning ``allowed_options`` LM Studio publishes for ``model``.
+
+    Reads ``capabilities.reasoning.allowed_options`` from ``/api/v1/models``; returns ``[]``
+    when the model is unknown, the endpoint is unreachable, or no reasoning capability is
+    declared.
+    """
+    from hermes_cli.models import _lmstudio_fetch_raw_models
+    try:
+        raw_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=timeout)
+    except Exception:
+        raw_models = None
+    if not raw_models:
+        return []
+
+    for raw in raw_models:
+        if not isinstance(raw, dict):
+            continue
+        if raw.get("key") != model and raw.get("id") != model:
+            continue
+        caps = raw.get("capabilities")
+        reasoning = caps.get("reasoning") if isinstance(caps, dict) else None
+        opts = reasoning.get("allowed_options") if isinstance(reasoning, dict) else None
+        if isinstance(opts, list):
+            return [str(o).strip().lower() for o in opts if isinstance(o, str)]
+        return []
+    return []
+
+
+def ollama_model_supports_thinking(
+    model: str,
+    base_url: Optional[str],
+    api_key: Optional[str] = None,
+    timeout: float = 5.0,
+) -> Optional[bool]:
+    """Return True if an Ollama (Cloud or local) model advertises ``thinking``.
+
+    Probes native ``/api/show`` and checks ``capabilities`` — the authoritative source, since
+    the OpenAI-compat ``/v1/models`` endpoint omits it. Tri-state: True when ``thinking`` is
+    declared, False when the probe succeeded without it, None when the probe failed so the
+    caller picks the fallback (treated as "don't emit").
+    """
+    import httpx
+
+    server_url = (base_url or "").strip().rstrip("/")
+    if server_url.endswith("/v1"):
+        server_url = server_url[:-3]
+    if not server_url:
+        return None
+
+    bare_model = _strip_ollama_cloud_suffix((model or "").strip())
+    if not bare_model:
+        return None
+
+    token = str(api_key or "").strip()
+    headers = {"Authorization": f"Bearer {token}"} if token else {}
+
+    try:
+        with httpx.Client(timeout=timeout, headers=headers) as client:
+            resp = client.post(f"{server_url}/api/show", json={"name": bare_model})
+            if resp.status_code != 200:
+                return None
+            caps = resp.json().get("capabilities")
+            if isinstance(caps, list):
+                return "thinking" in caps
+    except Exception:
+        return None
+    return None
+
+
+_OLLAMA_CLOUD_CACHE_TTL = 3600  # 1 hour
+
+
+def _strip_ollama_cloud_suffix(model_id: str) -> str:
+    """Strip :cloud / -cloud suffixes that models.dev appends to Ollama Cloud IDs.
+
+    The live API uses clean IDs (e.g. 'kimi-k2.6') while models.dev sometimes returns them as
+    'kimi-k2.6:cloud'. Normalising before the dedup merge prevents duplicate entries in the merged
+    model list.
+    """
+    for suffix in (":cloud", "-cloud"):
+        if model_id.endswith(suffix):
+            return model_id[: -len(suffix)]
+    return model_id
+
+
+def _ollama_cloud_cache_path() -> Path:
+    """Return the path for the Ollama Cloud model cache."""
+    from hermes_constants import get_hermes_home
+    return get_hermes_home() / "ollama_cloud_models_cache.json"
+
+
+def _load_ollama_cloud_cache(*, ignore_ttl: bool = False) -> Optional[dict]:
+    """Load cached Ollama Cloud models from disk (None when missing, empty, or stale)."""
+    from hermes_cli.models import _read_json_cache
+
+    try:
+        data = _read_json_cache(_ollama_cloud_cache_path())
+        if data is None:
+            return None
+        models = data.get("models")
+        if not (isinstance(models, list) and models):
+            return None
+        if not ignore_ttl and (time.time() - data.get("cached_at", 0)) > _OLLAMA_CLOUD_CACHE_TTL:
+            return None  # stale
+        return data
+    except Exception:
+        return None
+
+
+def _save_ollama_cloud_cache(models: list[str]) -> None:
+    """Persist the merged Ollama Cloud model list to disk. Best-effort."""
+    from hermes_cli.models import _write_json_cache
+
+    try:
+        _write_json_cache(_ollama_cloud_cache_path(), {"models": models, "cached_at": time.time()}, indent=None)
+    except Exception:
+        pass
+
+
+def fetch_ollama_cloud_models(
+    api_key: Optional[str] = None,
+    base_url: Optional[str] = None,
+    *,
+    force_refresh: bool = False,
+) -> list[str]:
+    """Fetch Ollama Cloud models by merging live API + models.dev, with disk cache.
+
+    Resolution order: 1. Disk cache (if fresh, < 1 hour, and not force_refresh) 2. Live
+    ``/v1/models`` endpoint (primary — freshest source) 3. models.dev registry (secondary — fills
+    gaps for unlisted models) 4. Merge: live models first, then models.dev additions (deduped)
+
+    Returns a list of model IDs (never None — empty list on total failure).
+    """
+    from hermes_cli.models import fetch_api_models
+    # 1. Check disk cache
+    if not force_refresh:
+        cached = _load_ollama_cloud_cache()
+        if cached is not None:
+            return cached["models"]
+
+    # 2. Live API probe
+    if not api_key:
+        api_key = os.getenv("OLLAMA_API_KEY", "")
+    if not base_url:
+        base_url = os.getenv("OLLAMA_BASE_URL", "") or "https://ollama.com/v1"
+
+    live_models: list[str] = []
+    if api_key:
+        result = fetch_api_models(api_key, base_url, timeout=8.0)
+        if result:
+            live_models = result
+
+    # 3. models.dev registry
+    mdev_models: list[str] = []
+    try:
+        from agent.models_dev import list_agentic_models
+        mdev_models = list_agentic_models("ollama-cloud")
+    except Exception:
+        pass
+
+    # 4. Merge: live first, then models.dev additions (deduped, order-preserving)
+    if live_models or mdev_models:
+        seen: set[str] = set()
+        merged: list[str] = []
+        for m in live_models:
+            if m and m not in seen:
+                seen.add(m)
+                merged.append(m)
+        for m in mdev_models:
+            normalized = _strip_ollama_cloud_suffix(m)
+            if normalized and normalized not in seen:
+                seen.add(normalized)
+                merged.append(normalized)
+        if merged:
+            _save_ollama_cloud_cache(merged)
+            return merged
+
+    # Total failure — return stale cache if available (ignore TTL)
+    stale = _load_ollama_cloud_cache(ignore_ttl=True)
+    if stale is not None:
+        return stale["models"]
+
+    return []

hermes_cli/models_pricing.py

@@ -0,0 +1,615 @@
+"""Live model pricing.
+
+OpenRouter-compatible ``/v1/models`` pricing fetch with a per-endpoint/per-credential cache,
+Nous Portal sale chrome and org-policy filtering, and the Vercel AI Gateway / Novita / Fireworks /
+DeepInfra pricing adapters.
+
+Split out of ``hermes_cli.models``; every moved name is re-imported there, so
+``hermes_cli.models.<name>`` keeps resolving (and monkeypatching) as before.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import urllib.request
+from typing import Any, Optional
+from hermes_cli.models_reasoning_caps import _seed_reasoning_caps
+
+
+# Cache: maps model_id → {"prompt": str, "completion": str} per endpoint
+_pricing_cache: dict[str, dict[str, dict[str, str]]] = {}
+
+
+# A failed fetch caches its empty result too, so an unreachable endpoint isn't
+# re-dialed on every call — but only until this deadline. Cached forever, one
+# bad moment (a blip during startup, a key that hadn't been written yet) turns
+# into no live model discovery for the life of the process, and the processes
+# that read this most are the ones that run for weeks: the gateway, the desktop
+# backend. Every caller falls back to a curated list meanwhile, so the cost of
+# the stale entry is silent and invisible.
+_FAILED_CATALOG_TTL_SECONDS = 120.0
+
+
+_pricing_cache_retry_after: dict[str, float] = {}
+
+
+def _cached_catalog(cache_key: str) -> Optional[dict[str, dict[str, Any]]]:
+    """The cached catalog for *cache_key*, or None to go fetch it."""
+    from hermes_cli.models import _pricing_cache, _pricing_cache_retry_after
+    cached = _pricing_cache.get(cache_key)
+    if cached is None:
+        return None
+    retry_after = _pricing_cache_retry_after.get(cache_key)
+    if retry_after is not None and time.monotonic() >= retry_after:
+        _pricing_cache.pop(cache_key, None)
+        _pricing_cache_retry_after.pop(cache_key, None)
+        return None
+    return cached
+
+
+def _cache_catalog(
+    cache_key: str,
+    result: dict[str, dict[str, Any]],
+    ttl_seconds: Optional[float] = None,
+) -> dict[str, dict[str, Any]]:
+    """Cache a catalog result, giving an empty one an expiry.
+
+    *ttl_seconds* expires a non-empty result too. Only a catalog whose contents depend on server-
+    side state the client cannot observe needs it — an org's model policy can change while a long-
+    lived process holds the entry.
+    """
+    from hermes_cli.models import _pricing_cache, _pricing_cache_retry_after
+    _pricing_cache[cache_key] = result
+    if result:
+        if ttl_seconds:
+            _pricing_cache_retry_after[cache_key] = time.monotonic() + ttl_seconds
+        else:
+            _pricing_cache_retry_after.pop(cache_key, None)
+    else:
+        _pricing_cache_retry_after[cache_key] = (
+            time.monotonic() + _FAILED_CATALOG_TTL_SECONDS
+        )
+    return result
+
+
+# NUL cannot appear in a URL, so this cannot collide with a real base URL.
+_PRICING_AUTH_KEY_PREFIX = "\x00auth:"
+
+
+def _pricing_auth_fingerprint(api_key: str | None) -> str:
+    """Key suffix identifying the credential a catalog was read with.
+
+    A governed endpoint answers each token with the catalog its org may reach, so two credentials
+    cannot share an entry. blake2b for cache-key fingerprinting only, same rationale as
+    :func:`_custom_endpoint_fingerprint`.
+    """
+    if not api_key:
+        return ""
+    import hashlib
+
+    digest = hashlib.blake2b(api_key.encode("utf-8", errors="replace"), digest_size=8)
+    return _PRICING_AUTH_KEY_PREFIX + digest.hexdigest()
+
+
+def peek_cached_pricing(base_url: str) -> dict[str, dict[str, Any]]:
+    """Pricing already cached for *base_url*, or ``{}``. Never fetches.
+
+    Accepts a ``/v1``-suffixed URL as well as the pre-``/v1`` root the fetchers key on, and
+    prefers an authenticated catalog. Scans rather than rebuilding a key because callers hold no
+    credential — newest first, skipping expired entries, so a rotated credential does not keep
+    answering from the catalog its predecessor read.
+    """
+    from hermes_cli.models import _pricing_cache
+    root = (base_url or "").rstrip("/")
+    if root.endswith("/v1"):
+        root = root[:-3].rstrip("/")
+    authed_prefix = root + _PRICING_AUTH_KEY_PREFIX
+    for key in reversed(list(_pricing_cache)):
+        if key.startswith(authed_prefix):
+            cached = _cached_catalog(key)
+            if cached:
+                return cached
+    return _cached_catalog(root) or {}
+
+
+def _format_price_per_mtok(per_token_str: str) -> str:
+    """Convert a per-token price string to a human-friendly $/Mtok string.
+
+    Always uses 2 decimal places so that prices align vertically when right-justified in a column
+    (the decimal point stays in the same position).
+
+    Sub-cent prices (e.g. deep-discount cache-hit promos) extend precision instead of collapsing to
+    "$0.00": the smallest decimal place that makes the value non-zero is found, then one extra digit
+    is kept and trailing zeros trimmed.
+    """
+    try:
+        val = float(per_token_str)
+    except (TypeError, ValueError):
+        return "?"
+    if val == 0:
+        return "free"
+    per_m = val * 1_000_000
+    text = f"{per_m:.2f}"
+    if per_m < 0.01:
+        # Non-zero price below one cent per Mtok — widen precision until the
+        # value shows, keep one extra significant digit, trim trailing zeros.
+        prec = 3
+        while prec < 12 and round(per_m, prec) == 0:
+            prec += 1
+        text = f"{per_m:.{min(prec + 1, 12)}f}".rstrip("0").rstrip(".")
+    return f"${text}"
+
+
+def compute_sale_discount(
+    prompt: str,
+    completion: str,
+    original: Any,
+) -> tuple[int, str, str] | None:
+    """Derive sale chrome from gateway ``pricing.original`` when cheaper.
+
+    Nous Portal-only feature: callers gate on the provider; this helper only sees ``original``
+    because the Nous fetch path opted in via ``include_sale_original=True``.
+
+    Returns ``(discount_percent, was_prompt_raw, was_completion_raw)`` only when ``original`` is a
+    dict and the current prompt (fallback: completion) rate is strictly below the corresponding
+    original.
+    """
+    def _finite(raw: Any) -> float | None:
+        try:
+            n = float(raw)
+        except (TypeError, ValueError):
+            return None
+        return n if n > 0 and n == n else None  # n == n rejects NaN
+
+    def _nonneg(raw: Any) -> float | None:
+        try:
+            n = float(raw)
+        except (TypeError, ValueError):
+            return None
+        return n if n >= 0 and n == n else None
+
+    orig_dict = original if isinstance(original, dict) else {}
+    was_prompt = orig_dict.get("prompt")
+    was_completion = orig_dict.get("completion")
+
+    # Free / $0 models: flat 100% off, with "was" prices only when the
+    # gateway actually served an original (e.g. a :free sibling); a
+    # natively-free model (stealth/ox-alpha) gets bare "-100%" chrome.
+    cur_prompt_any = _nonneg(prompt) if prompt not in (None, "") else None
+    cur_comp_any = _nonneg(completion) if completion not in (None, "") else None
+    if cur_prompt_any == 0 and cur_comp_any in (0, None):
+        return (
+            100,
+            str(was_prompt) if was_prompt not in (None, "") else "",
+            str(was_completion) if was_completion not in (None, "") else "",
+        )
+
+    if not isinstance(original, dict):
+        return None
+
+    if was_prompt in (None, "") and was_completion in (None, ""):
+        return None
+
+    cur_prompt = _finite(prompt) if prompt not in (None, "") else None
+    orig_prompt = _finite(was_prompt) if was_prompt not in (None, "") else None
+    if cur_prompt is not None and orig_prompt is not None and cur_prompt < orig_prompt:
+        pct = int(round((1.0 - (cur_prompt / orig_prompt)) * 100))
+        if pct < 1:
+            return None
+        return (
+            pct,
+            str(was_prompt),
+            str(was_completion) if was_completion not in (None, "") else "",
+        )
+
+    cur_comp = _finite(completion) if completion not in (None, "") else None
+    orig_comp = _finite(was_completion) if was_completion not in (None, "") else None
+    if cur_comp is not None and orig_comp is not None and cur_comp < orig_comp:
+        pct = int(round((1.0 - (cur_comp / orig_comp)) * 100))
+        if pct < 1:
+            return None
+        return (
+            pct,
+            str(was_prompt) if was_prompt not in (None, "") else "",
+            str(was_completion),
+        )
+
+    return None
+
+
+def _get_json(url: str, headers: dict[str, str], timeout: float, opener=None) -> Optional[dict]:
+    """GET *url* as JSON via the origin's catalog opener (or *opener*); None on any failure."""
+    from hermes_cli.models import _urlopen_model_catalog_request
+
+    try:
+        req = urllib.request.Request(url, headers=headers)
+        with (opener or _urlopen_model_catalog_request)(req, timeout=timeout) as resp:
+            return json.loads(resp.read().decode())
+    except Exception:
+        return None
+
+
+def _pricing_entry(pricing: dict, prompt_key: str = "prompt", completion_key: str = "completion") -> dict[str, Any]:
+    """Picker-shape ``{prompt, completion[, input_cache_read, input_cache_write]}`` from a catalog
+    ``pricing`` block whose cache fields already use the hermes names."""
+    entry: dict[str, Any] = {
+        "prompt": str(pricing.get(prompt_key, "")),
+        "completion": str(pricing.get(completion_key, "")),
+    }
+    for key in ("input_cache_read", "input_cache_write"):
+        if pricing.get(key):
+            entry[key] = str(pricing[key])
+    return entry
+
+
+def fetch_models_with_pricing(
+    api_key: str | None = None,
+    base_url: str = "https://openrouter.ai/api",
+    timeout: float = 8.0,
+    *,
+    force_refresh: bool = False,
+    include_sale_original: bool = False,
+    cache_ttl_seconds: Optional[float] = None,
+) -> dict[str, dict[str, Any]]:
+    """Fetch ``/v1/models`` and return ``{model_id: {prompt, completion, ...}}``.
+
+    Results are cached per *base_url* and per credential, so repeated calls are free and one
+    caller's catalog never answers another's read. Works with any OpenRouter-compatible endpoint
+    (OpenRouter, Nous Portal).
+
+    When *include_sale_original* is true (Nous Portal only) and the gateway advertises a global
+    discount under ``pricing.original``, those pre-discount rates are copied through as a nested
+    ``original`` dict so pickers can show sale chrome.
+    """
+    from hermes_cli.models import _HERMES_USER_AGENT
+    url_root = (base_url or "").rstrip("/")
+    cache_key = url_root + _pricing_auth_fingerprint(api_key)
+    if not force_refresh:
+        cached = _cached_catalog(cache_key)
+        if cached is not None:
+            return cached
+
+    url = url_root + "/v1/models"
+    headers = {"Accept": "application/json", "User-Agent": _HERMES_USER_AGENT}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    payload = _get_json(url, headers, timeout)
+    if payload is None:
+        return _cache_catalog(cache_key, {})
+
+    # Same document the reasoning-capability fetch would pull, and every picker/pricing surface
+    # goes through here — mirror it so a later hot-path lookup (and the next process) has an
+    # answer without its own round-trip.
+    _seed_reasoning_caps(url, payload.get("data"))
+
+    result: dict[str, dict[str, Any]] = {}
+    for item in payload.get("data", []):
+        mid = item.get("id")
+        pricing = item.get("pricing")
+        if mid and isinstance(pricing, dict):
+            entry = _pricing_entry(pricing)
+            # Sale chrome is Nous Portal-only; never copy pricing.original for other catalogs.
+            original = pricing.get("original") if include_sale_original else None
+            if isinstance(original, dict):
+                orig_entry = {
+                    key: str(original[key])
+                    for key in ("prompt", "completion", "input_cache_read", "input_cache_write")
+                    if original.get(key) not in (None, "")
+                }
+                if orig_entry.get("prompt") or orig_entry.get("completion"):
+                    entry["original"] = orig_entry
+            result[mid] = entry
+
+    return _cache_catalog(cache_key, result, cache_ttl_seconds)
+
+
+def fetch_ai_gateway_pricing(
+    timeout: float = 8.0,
+    *,
+    force_refresh: bool = False,
+) -> dict[str, dict[str, str]]:
+    """Fetch Vercel AI Gateway /v1/models and return hermes-shaped pricing.
+
+    Vercel uses ``input`` / ``output`` field names; hermes's picker expects ``prompt`` /
+    ``completion``. This translates. Cache read/write field names already match.
+    """
+    from hermes_constants import AI_GATEWAY_BASE_URL
+
+    cache_key = AI_GATEWAY_BASE_URL.rstrip("/")
+    if not force_refresh:
+        cached = _cached_catalog(cache_key)
+        if cached is not None:
+            return cached
+
+    payload = _get_json(f"{cache_key}/models", {"Accept": "application/json"}, timeout, opener=urllib.request.urlopen)
+    if payload is None:
+        return _cache_catalog(cache_key, {})
+
+    result: dict[str, dict[str, str]] = {}
+    for item in payload.get("data", []):
+        if not isinstance(item, dict):
+            continue
+        mid = item.get("id")
+        pricing = item.get("pricing")
+        if mid and isinstance(pricing, dict):
+            result[mid] = _pricing_entry(pricing, "input", "output")
+    return _cache_catalog(cache_key, result)
+
+
+def _resolve_openrouter_api_key() -> str:
+    """Best-effort OpenRouter API key for pricing fetch."""
+    return os.getenv("OPENROUTER_API_KEY", "").strip()
+
+
+_DEFAULT_NOUS_INFERENCE_BASE = "https://inference-api.nousresearch.com"
+
+
+def _resolve_nous_pricing_credentials() -> tuple[str, str]:
+    """Return ``(api_key, base_url)`` for Nous Portal pricing.
+
+    Base URL precedence (mirrors runtime credential resolution): 1. ``NOUS_INFERENCE_BASE_URL`` env
+    override (staging / preview) 2. Resolved runtime credential ``base_url`` 3. Production default
+
+    Without (1), a staging profile's sale ``pricing.original`` never reaches the pickers — the
+    anonymous fallback would hit prod, which has no ``original`` field.
+    """
+    try:
+        from hermes_cli.auth import _nous_inference_env_override
+
+        env_base = _nous_inference_env_override()
+    except Exception:
+        env_base = None
+
+    api_key = ""
+    creds_base = ""
+    try:
+        from hermes_cli.auth import resolve_nous_runtime_credentials
+
+        creds = resolve_nous_runtime_credentials()
+        if creds:
+            api_key = creds.get("api_key", "") or ""
+            creds_base = (creds.get("base_url", "") or "").strip()
+    except Exception:
+        pass
+
+    base_url = (env_base or creds_base or _DEFAULT_NOUS_INFERENCE_BASE).rstrip("/")
+    # Credential bases arrive with or without the ``/v1`` suffix. Callers
+    # append their own path, so hand back the bare origin.
+    if base_url.endswith("/v1"):
+        base_url = base_url[:-3]
+    return (api_key, base_url)
+
+
+def nous_policy_allowed_ids(*, force_refresh: bool = False) -> Optional[set[str]]:
+    """The Nous model ids the caller's org may reach, or ``None`` to not filter.
+
+    The gateway omits policy-blocked rows from an authenticated ``GET /v1/models``, so that
+    response's keys are the reachable set.
+
+    ``None`` means "leave the caller's list alone", for the three states that cannot support
+    narrowing one: no policy (or a token too old to say), an anonymous read whose catalog is
+    unfiltered, and an empty read, which is a fetch failure rather than an org that may reach
+    nothing.
+    """
+    from hermes_cli.models import _resolve_nous_pricing_credentials, fetch_models_with_pricing
+    try:
+        from hermes_cli.nous_account import nous_policy_present
+
+        if nous_policy_present() is not True:
+            return None
+    except Exception:
+        return None
+
+    api_key, base_url = _resolve_nous_pricing_credentials()
+    if not api_key or not base_url:
+        return None
+
+    # Same arguments as get_pricing_for_provider's nous branch, so a caller
+    # asking for pricing too shares this entry instead of paying for a second
+    # request.
+    pricing = fetch_models_with_pricing(
+        api_key=api_key,
+        base_url=base_url,
+        force_refresh=force_refresh,
+        include_sale_original=True,
+        cache_ttl_seconds=_NOUS_CATALOG_TTL_SECONDS,
+    )
+    return set(pricing) or None
+
+
+# Past this size an allowed set reads as a whole catalog rather than an
+# allowlist, and is not worth showing in place of an empty picker.
+_NOUS_POLICY_APPEND_MAX = 64
+
+
+# How long a Nous catalog stays trusted. Its contents depend on the org's
+# policy, which an admin can change at any time and the client cannot observe,
+# so a long-lived process must re-ask instead of holding the first answer for
+# its whole life. Other providers' catalogs carry no such state and keep the
+# default no-expiry caching.
+_NOUS_CATALOG_TTL_SECONDS = 300.0
+
+
+def restrict_to_nous_policy(
+    model_ids: list[str],
+    allowed: Optional[set[str]],
+    *,
+    rescue_empty: bool = False,
+) -> list[str]:
+    """*model_ids* narrowed to *allowed*, preserving the caller's order.
+
+    A ``:free`` sibling is kept when its base model is reachable, mirroring the gateway, which
+    admits a row when any of its requestable ids passes. Prefer over-listing: that costs a 403 from
+    the authoritative gate, while hiding a row the gate would serve is unrecoverable from the
+    client.
+    """
+    if not allowed:
+        return list(model_ids)
+    kept = [
+        mid
+        for mid in model_ids
+        if mid in allowed or mid.split(":", 1)[0] in allowed
+    ]
+
+    # An allowlist can name only models the curated manifest lacks, leaving an
+    # empty picker — worse than no filter, since the models the org may use are
+    # the ones dropped. Opt-in per list: an already-empty list (a paid tier's
+    # gated models) means "nothing to gate", not "nothing survived".
+    if rescue_empty and not kept and len(allowed) <= _NOUS_POLICY_APPEND_MAX:
+        return sorted(allowed)
+    return kept
+
+
+def get_pricing_for_provider(provider: str, *, force_refresh: bool = False) -> dict[str, dict[str, str]]:
+    """Return live pricing for providers that support it (openrouter, nous, ai-gateway, novita)."""
+    from hermes_cli.models import _resolve_nous_pricing_credentials, fetch_models_with_pricing, normalize_provider
+    normalized = normalize_provider(provider)
+    if normalized == "openrouter":
+        return fetch_models_with_pricing(
+            api_key=_resolve_openrouter_api_key(),
+            base_url="https://openrouter.ai/api",
+            force_refresh=force_refresh,
+        )
+    if normalized == "ai-gateway":
+        return fetch_ai_gateway_pricing(force_refresh=force_refresh)
+    if normalized == "novita":
+        return _fetch_novita_pricing(force_refresh=force_refresh)
+    if normalized == "deepinfra":
+        return _fetch_deepinfra_pricing(force_refresh=force_refresh)
+    if normalized == "fireworks":
+        return _fireworks_pricing_from_models_dev(force_refresh=force_refresh)
+    if normalized == "nous":
+        api_key, base_url = _resolve_nous_pricing_credentials()
+        if base_url:
+            return fetch_models_with_pricing(
+                api_key=api_key,
+                base_url=base_url,
+                force_refresh=force_refresh,
+                # Sale chrome (pricing.original) is Nous Portal-only.
+                include_sale_original=True,
+                cache_ttl_seconds=_NOUS_CATALOG_TTL_SECONDS,
+            )
+    return {}
+
+
+def _fireworks_pricing_from_models_dev(
+    *,
+    force_refresh: bool = False,
+) -> dict[str, dict[str, str]]:
+    """Derive Fireworks picker pricing from the models.dev registry cache.
+
+    No dedicated network fetch: ``fetch_models_dev()`` already maintains an in-memory + disk cache
+    (1h TTL) that every picker surface shares, so this is a pure dict transform on the picker path —
+    no added latency and no per-render network call.
+    """
+    cache_key = "models.dev/fireworks"
+    if not force_refresh:
+        cached = _cached_catalog(cache_key)
+        if cached is not None:
+            return cached
+
+    result: dict[str, dict[str, str]] = {}
+    try:
+        from agent.models_dev import _get_provider_models
+
+        models = _get_provider_models("fireworks") or {}
+        for mid, entry in models.items():
+            if not isinstance(entry, dict):
+                continue
+            cost = entry.get("cost")
+            if not isinstance(cost, dict):
+                continue
+            inp = cost.get("input")
+            out = cost.get("output")
+            if inp is None and out is None:
+                continue
+            row: dict[str, str] = {
+                "prompt": str(float(inp or 0) / 1_000_000),
+                "completion": str(float(out or 0) / 1_000_000),
+            }
+            cache_read = cost.get("cache_read")
+            if cache_read:
+                row["input_cache_read"] = str(float(cache_read) / 1_000_000)
+            result[str(mid)] = row
+    except Exception:
+        result = {}
+
+    return _cache_catalog(cache_key, result)
+
+
+def _fetch_novita_pricing(
+    timeout: float = 8.0,
+    *,
+    force_refresh: bool = False,
+) -> dict[str, dict[str, str]]:
+    """Fetch pricing from NovitaAI /v1/models.
+
+    NovitaAI reports per-million-token prices in units of 0.0001 USD; they are converted to the
+    per-token strings the shared pricing formatter expects. Results are cached in
+    ``_pricing_cache`` keyed on the resolved base URL so menu renders don't re-hit the network.
+    """
+    from hermes_cli.models import _HERMES_USER_AGENT
+    api_key = os.getenv("NOVITA_API_KEY", "").strip()
+    if not api_key:
+        return {}
+
+    base_url = os.getenv("NOVITA_BASE_URL", "").strip() or "https://api.novita.ai/openai/v1"
+    cache_key = base_url.rstrip("/")
+    if not force_refresh:
+        cached = _cached_catalog(cache_key)
+        if cached is not None:
+            return cached
+
+    headers = {"Authorization": f"Bearer {api_key}", "Accept": "application/json", "User-Agent": _HERMES_USER_AGENT}
+    payload = _get_json(cache_key + "/models", headers, timeout)
+    if payload is None:
+        return _cache_catalog(cache_key, {})
+
+    result: dict[str, dict[str, str]] = {}
+    for item in payload.get("data", []):
+        if not isinstance(item, dict):
+            continue
+        mid = item.get("id")
+        if not mid:
+            continue
+        inp = item.get("input_token_price_per_m")
+        out = item.get("output_token_price_per_m")
+        if inp is None and out is None:
+            continue
+        result[str(mid)] = {
+            "prompt": str(float(inp or 0) / 10_000 / 1_000_000),
+            "completion": str(float(out or 0) / 10_000 / 1_000_000),
+        }
+
+    return _cache_catalog(cache_key, result)
+
+
+def _fetch_deepinfra_pricing(
+    timeout: float = 5.0,
+    *,
+    force_refresh: bool = False,
+) -> dict[str, dict[str, str]]:
+    """Return picker-shape pricing for DeepInfra chat models.
+
+    DeepInfra publishes ``input_tokens``/``output_tokens``/``cache_read_tokens`` in $/MTok; the
+    picker expects per-token strings under ``prompt``/``completion``/``input_cache_read``
+    (OpenRouter shape). Cached via the catalog helper so repeated picker renders are free.
+    """
+    from hermes_cli.models import _fetch_deepinfra_models_by_tag
+    items = _fetch_deepinfra_models_by_tag("chat", timeout=timeout, force_refresh=force_refresh)
+    result: dict[str, dict[str, str]] = {}
+    for item in items or []:
+        metadata = item.get("metadata") or {}
+        pricing = metadata.get("pricing") if isinstance(metadata, dict) else None
+        if not isinstance(pricing, dict):
+            continue
+        entry = {
+            ours: str(float(pricing[theirs]) / 1_000_000)
+            for theirs, ours in (("input_tokens", "prompt"), ("output_tokens", "completion"), ("cache_read_tokens", "input_cache_read"))
+            if pricing.get(theirs) is not None
+        }
+        if entry:
+            result[item["id"]] = entry
+    return result

hermes_cli/models_reasoning_caps.py

@@ -0,0 +1,310 @@
+"""Per-model reasoning capabilities from OpenRouter-schema ``/v1/models`` catalogs.
+
+Split out of ``hermes_cli.models``; every public/patched name is re-imported there. The
+OpenRouter and Nous Portal catalogs share one implementation parametrized by
+:class:`_CapsSource`; the per-source module globals (``_openrouter_reasoning_caps_cache``,
+``_nous_caps_disk_checked``, ...) stay defined on ``hermes_cli.models`` — tests reset them there —
+and are read/written by attribute name through the origin module.
+
+Tri-state contract for callers deciding whether to emit reasoning controls:
+- dict with ``supports_reasoning: True`` (+ ``supported_efforts``, ``mandatory``) — the route
+  advertises reasoning controls;
+- dict with ``supports_reasoning: False`` — the catalog knows the model and it does NOT accept
+  reasoning controls (definitive negative);
+- ``None`` — unknown: catalog not loaded, model not listed (private/custom route), malformed.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import threading
+import time
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+
+logger = logging.getLogger("hermes_cli.models")
+
+Caps = dict[str, Optional[dict[str, Any]]]
+
+
+def _origin():
+    from hermes_cli import models
+
+    return models
+
+
+def parse_openrouter_reasoning_capabilities(item: Any) -> Optional[dict[str, Any]]:
+    """Normalize one OpenRouter catalog entry's reasoning metadata.
+
+    ``supported_parameters`` contains ``"reasoning"`` when the route accepts reasoning controls at
+    all; a top-level ``reasoning`` object may add detail (``mandatory``, ``supported_efforts``).
+    A missing/malformed ``supported_parameters`` is "unknown" (None), mirroring the permissive
+    stance of ``_openrouter_model_supports_tools``.
+    """
+    if not isinstance(item, dict):
+        return None
+    params = item.get("supported_parameters")
+    if not isinstance(params, list):
+        return None
+    if "reasoning" not in params:
+        return {"supports_reasoning": False}
+    reasoning = item.get("reasoning")
+    mandatory = isinstance(reasoning, dict) and reasoning.get("mandatory") is True
+    efforts: Optional[list[str]] = None
+    if isinstance(reasoning, dict):
+        raw_efforts = reasoning.get("supported_efforts")
+        if isinstance(raw_efforts, list):
+            efforts = list(dict.fromkeys(
+                str(effort).strip().lower()
+                for effort in raw_efforts
+                if str(effort).strip()
+            ))
+    return {
+        "supports_reasoning": True,
+        "supported_efforts": efforts,
+        "mandatory": mandatory,
+    }
+
+
+# ── Disk mirror ────────────────────────────────────────────────────────
+#
+# The in-process caches are always cold in a short-lived process, and every consumer is on a hot
+# path that must never block on HTTP — so without a disk copy, `hermes -p`, a cron job, or a
+# freshly booted gateway answers "capability unknown" for its whole first turn and falls back to
+# the conservative wire shape. One file holds every catalog, keyed by the URL it came from:
+# OpenRouter and the Nous Portal list different models, and a staging Portal must not answer for
+# production.
+_REASONING_CAPS_DISK_TTL_SECONDS = 24 * 3600
+
+
+def _reasoning_caps_disk_path() -> Path:
+    from hermes_constants import get_hermes_home
+    return get_hermes_home() / "cache" / "reasoning_caps.json"
+
+
+def _read_reasoning_caps_disk() -> dict[str, Any]:
+    from hermes_cli.models import _read_json_cache
+
+    return _read_json_cache(_reasoning_caps_disk_path()) or {}
+
+
+def _load_reasoning_caps_disk(url: str) -> tuple[Optional[Caps], float]:
+    """Return ``(caps, age_seconds)`` for *url*, or ``(None, 0.0)``."""
+    entry = _origin()._read_reasoning_caps_disk().get(url)
+    if not isinstance(entry, dict):
+        return None, 0.0
+    caps = entry.get("caps")
+    if not isinstance(caps, dict) or not caps:
+        return None, 0.0
+    try:
+        age = max(0.0, time.time() - float(entry.get("ts") or 0))
+    except (TypeError, ValueError):
+        age = float(_REASONING_CAPS_DISK_TTL_SECONDS)
+    return {str(mid): model_caps for mid, model_caps in caps.items()}, age
+
+
+def _save_reasoning_caps_disk(url: str, caps: Caps) -> None:
+    """Merge *url*'s catalog into the shared disk mirror, atomically."""
+    from hermes_cli.models import _write_json_cache
+
+    try:
+        data = _origin()._read_reasoning_caps_disk()
+        data[url] = {"ts": time.time(), "caps": caps}
+        _write_json_cache(_reasoning_caps_disk_path(), data, indent=0, separators=(",", ":"))
+    except Exception as exc:
+        logger.debug("Failed to save reasoning-caps disk cache: %s", exc)
+
+
+def _warm_reasoning_caps_async(refresh) -> None:
+    """Run *refresh* in a background thread. Fire-and-forget.
+
+    Called from hot paths that found the cache cold or the disk copy stale, so the next call — or,
+    via the disk mirror, the next process — benefits without this turn ever blocking on HTTP.
+    Callers own the once-per-process guard; the fetch keeps its own failure TTL.
+    """
+    if os.environ.get("PYTEST_CURRENT_TEST"):
+        return
+    threading.Thread(target=refresh, name="reasoning-caps-warm", daemon=True).start()
+
+
+def _hydrate_reasoning_caps_from_disk(url: str, refresh) -> Optional[Caps]:
+    """The disk copy of *url*'s catalog, queueing *refresh* when it's stale.
+
+    A copy past its TTL is still returned — a stale verdict beats no verdict, and reasoning
+    capabilities change rarely — with a background refresh so the next run is current.
+    """
+    caps, age = _load_reasoning_caps_disk(url)
+    if caps is None:
+        return None
+    if age >= _REASONING_CAPS_DISK_TTL_SECONDS:
+        _warm_reasoning_caps_async(refresh)
+    return caps
+
+
+def _seed_reasoning_caps(url: str, items: Any) -> Optional[Caps]:
+    """Parse a ``/v1/models`` ``data`` array and mirror it for *url*.
+
+    Takes the payload rather than fetching it, so picker and pricing fetches (which pull the same
+    document) leave the mirror warm at no network cost. Returns None when the array has no usable
+    entries, which callers remember as a failure rather than caching as empty.
+    """
+    if not isinstance(items, list):
+        return None
+    caps_by_id: Caps = {}
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        mid = str(item.get("id") or "").strip()
+        if not mid:
+            continue
+        caps_by_id[mid] = parse_openrouter_reasoning_capabilities(item)
+    if not caps_by_id:
+        return None
+    _save_reasoning_caps_disk(url, caps_by_id)
+    return caps_by_id
+
+
+def _fetch_reasoning_caps_catalog(url: str, timeout: float) -> Optional[Caps]:
+    """Fetch one OpenRouter-shaped ``/v1/models`` catalog → per-model caps.
+
+    Returns None when the catalog is unreachable or has no usable entries, so callers remember the
+    failure and fall back rather than caching an empty result. Sends a User-Agent because the
+    Portal 403s anonymous catalog reads.
+    """
+    m = _origin()
+    headers = {"Accept": "application/json", "User-Agent": m._HERMES_USER_AGENT}
+    try:
+        req = urllib.request.Request(url, headers=headers)
+        with m._urlopen_model_catalog_request(req, timeout=timeout) as resp:
+            payload = json.loads(resp.read().decode())
+    except Exception:
+        return None
+    return _seed_reasoning_caps(url, payload.get("data"))
+
+
+# ── Per-source cache (OpenRouter, Nous Portal) ─────────────────────────
+
+@dataclass(frozen=True)
+class _CapsSource:
+    """One catalog's cache slots on ``hermes_cli.models`` plus how to name its URL.
+
+    ``cache``: model id → parsed caps, populated by one full-catalog fetch and kept for the process
+    lifetime (capabilities don't change). ``failed_at``: monotonic timestamp of the last FAILED
+    fetch; suppresses re-fetch storms from per-turn callers while the catalog is unreachable (60s,
+    mirrors the LM Studio/Ollama capability-probe caching). ``disk_checked`` / ``warm_started``:
+    once-per-process guards for the disk hydrate and the background warm.
+    """
+    cache: str
+    failed_at: str
+    disk_checked: str
+    warm_started: str
+    url: Callable[[], str]
+
+
+def _fetch_caps(src: _CapsSource, timeout: float = 6.0, *, force: bool = False) -> Optional[Caps]:
+    """Fetch + cache the source's per-model caps. None (without poisoning the cache) when
+    unreachable, so callers retry later and fall back meanwhile."""
+    m = _origin()
+    cached = getattr(m, src.cache)
+    if cached is not None and not force:
+        return cached
+    failed_at = getattr(m, src.failed_at)
+    if failed_at is not None and (time.monotonic() - failed_at) < 60:
+        return None
+    caps_by_id = _fetch_reasoning_caps_catalog(src.url(), timeout)
+    if caps_by_id is None:
+        setattr(m, src.failed_at, time.monotonic())
+        return None
+    setattr(m, src.cache, caps_by_id)
+    return caps_by_id
+
+
+def _caps_cached(src: _CapsSource) -> Optional[Caps]:
+    """Cache-only caps: memory, else the disk mirror. Never HTTP.
+
+    Guarded to one disk attempt per process: for the Portal, naming the catalog means resolving
+    credentials, which can itself reach the network to refresh a token — far too expensive for a
+    caller that runs every turn.
+    """
+    m = _origin()
+    if getattr(m, src.cache) is None and not getattr(m, src.disk_checked):
+        setattr(m, src.disk_checked, True)
+        setattr(m, src.cache, _hydrate_reasoning_caps_from_disk(src.url(), lambda: _fetch_caps(src, force=True)))
+    return getattr(m, src.cache)
+
+
+def _model_caps(src: _CapsSource, model_id: Optional[str], *, timeout: float, allow_fetch: bool) -> Optional[dict[str, Any]]:
+    model = str(model_id or "").strip()
+    if not model:
+        return None
+    caps_by_id = _caps_cached(src)
+    if caps_by_id is None and allow_fetch:
+        caps_by_id = _fetch_caps(src, timeout=timeout)
+    if caps_by_id is None:
+        return None
+    return caps_by_id.get(model)
+
+
+def _warm_caps_async(src: _CapsSource) -> None:
+    m = _origin()
+    if getattr(m, src.warm_started) or _caps_cached(src) is not None:
+        return
+    setattr(m, src.warm_started, True)
+    _warm_reasoning_caps_async(lambda: _fetch_caps(src, force=True))
+
+
+_OPENROUTER_CATALOG_URL = "https://openrouter.ai/api/v1/models"
+
+_OPENROUTER_CAPS = _CapsSource(
+    "_openrouter_reasoning_caps_cache", "_openrouter_reasoning_caps_failed_at",
+    "_openrouter_caps_disk_checked", "_openrouter_caps_warm_started",
+    lambda: _OPENROUTER_CATALOG_URL,
+)
+# Nous Portal serves OpenRouter's catalog schema, so the same parser and contract apply. Its own
+# cache because the two catalogs list different models (and different capabilities for shared ids).
+_NOUS_CAPS = _CapsSource(
+    "_nous_reasoning_caps_cache", "_nous_reasoning_caps_failed_at",
+    "_nous_caps_disk_checked", "_nous_caps_warm_started",
+    lambda: _origin().nous_catalog_url(),
+)
+
+
+def nous_catalog_url() -> str:
+    """The Portal ``/v1/models`` URL for the endpoint we actually talk to.
+
+    Resolved through the ladder ``NOUS_INFERENCE_BASE_URL`` → resolved credential base → prod
+    rather than pinned to production, so a staging profile reads staging's capabilities.
+    """
+    return f"{_origin()._resolve_nous_pricing_credentials()[1]}/v1/models"
+
+
+def openrouter_model_reasoning_capabilities(
+    model_id: Optional[str], *, timeout: float = 6.0, allow_fetch: bool = False,
+) -> Optional[dict[str, Any]]:
+    """Live-catalog reasoning capabilities for an OpenRouter model (tri-state, see module doc).
+
+    CACHE-ONLY by default — safe on per-request hot paths (never blocks on HTTP)."""
+    return _model_caps(_OPENROUTER_CAPS, model_id, timeout=timeout, allow_fetch=allow_fetch)
+
+
+def nous_model_reasoning_capabilities(
+    model_id: Optional[str], *, timeout: float = 6.0, allow_fetch: bool = False,
+) -> Optional[dict[str, Any]]:
+    """Nous Portal counterpart of :func:`openrouter_model_reasoning_capabilities`; warm the cache
+    with :func:`warm_nous_reasoning_caps_async` from hot paths."""
+    return _model_caps(_NOUS_CAPS, model_id, timeout=timeout, allow_fetch=allow_fetch)
+
+
+def warm_openrouter_reasoning_caps_async() -> None:
+    """Warm the OpenRouter reasoning-capability cache in the background."""
+    _warm_caps_async(_OPENROUTER_CAPS)
+
+
+def warm_nous_reasoning_caps_async() -> None:
+    """Nous Portal counterpart of :func:`warm_openrouter_reasoning_caps_async`."""
+    _warm_caps_async(_NOUS_CAPS)

hermes_cli/models_validate.py

@@ -0,0 +1,629 @@
+"""Validate a requested ``/model`` value against the active provider's catalog.
+
+Split out of ``hermes_cli.models``; :func:`validate_requested_model` is re-imported there, so
+``hermes_cli.models.validate_requested_model`` keeps resolving. Every catalog fetcher this module
+calls is looked up on ``hermes_cli.models`` at call time (``_m.<name>``), so existing
+``patch("hermes_cli.models.<name>")`` mocks keep intercepting.
+
+Every provider branch returns one of four verdict shapes (see :func:`_verdict`) or ``None`` to
+mean "not decided here — keep walking the ladder". The ladder ORDER is behavior: moa → whitespace
+→ OpenRouter preset parse → LM Studio → Ollama native → custom → codex/xai static → MiniMax →
+Anthropic native → Anthropic Messages → live listing → Bedrock → curated-catalog fallback.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from difflib import get_close_matches
+from typing import Any, Optional
+
+from utils import base_url_host_matches
+
+
+# ── Verdicts ─────────────────────────────────────────────────────────────
+
+def _verdict(accepted: bool, persist: bool, recognized: bool, message: Optional[str],
+             corrected_model: Optional[str] = None) -> dict[str, Any]:
+    """Build the verdict dict. ``corrected_model`` is only present when set (key order matters
+    to nobody, but keep it identical to the historical literals anyway)."""
+    out: dict[str, Any] = {"accepted": accepted, "persist": persist, "recognized": recognized}
+    if corrected_model is not None:
+        out["corrected_model"] = corrected_model
+    out["message"] = message
+    return out
+
+
+def _accept() -> dict[str, Any]:
+    return _verdict(True, True, True, None)
+
+
+def _reject(message: str) -> dict[str, Any]:
+    return _verdict(False, False, False, message)
+
+
+def _soft_accept(message: Optional[str]) -> dict[str, Any]:
+    """Accept + persist an unrecognized name, with a warning."""
+    return _verdict(True, True, False, message)
+
+
+def _corrected(requested: str, corrected: str) -> dict[str, Any]:
+    return _verdict(True, True, True, f"Auto-corrected `{requested}` → `{corrected}`",
+                    corrected_model=corrected)
+
+
+# ── Catalog matching ─────────────────────────────────────────────────────
+
+@dataclass
+class _Match:
+    exact: bool = False
+    corrected: Optional[str] = None
+    suggestion_text: str = ""
+
+
+def _match_in_catalog(
+    query: str,
+    candidates,
+    *,
+    case_insensitive: bool = False,
+    auto_correct: bool = True,
+    suggest_query: Optional[str] = None,
+    suggest_cutoff: float = 0.5,
+    suggest_label: str = "Similar models",
+) -> _Match:
+    """The shared ladder: exact membership → typo auto-correct (cutoff .9) → suggestion text.
+
+    ``case_insensitive`` matches on lower-cased ids and maps results back to the catalog's
+    spelling (MiniMax ships mixed-case ids). ``suggest_query`` overrides the string the
+    suggestion search uses (some branches search on the raw request, not the lookup form).
+    """
+    candidates = list(candidates)
+    if case_insensitive:
+        display = {c.lower(): c for c in candidates}
+        pool = list(display)
+        query = query.lower()
+        suggest_query = query if suggest_query is None else suggest_query.lower()
+    else:
+        display = None
+        pool = candidates
+        suggest_query = query if suggest_query is None else suggest_query
+
+    def _show(cid: str) -> str:
+        return display[cid] if display is not None else cid
+
+    if query in set(pool):
+        return _Match(exact=True)
+    if auto_correct:
+        auto = get_close_matches(query, pool, n=1, cutoff=0.9)
+        if auto:
+            return _Match(corrected=_show(auto[0]))
+    suggestions = get_close_matches(suggest_query, pool, n=3, cutoff=suggest_cutoff)
+    text = ""
+    if suggestions:
+        text = f"\n  {suggest_label}: " + ", ".join(f"`{_show(s)}`" for s in suggestions)
+    return _Match(suggestion_text=text)
+
+
+# ── Request context ──────────────────────────────────────────────────────
+
+@dataclass
+class _Request:
+    requested: str
+    lookup: str                 # id used for catalog membership (copilot-normalized / preset base)
+    provider: Optional[str]     # raw caller value (Ollama checks look at this, not ``normalized``)
+    normalized: str
+    api_key: Optional[str]
+    base_url: Optional[str]
+    api_mode: Optional[str]
+    headers: Optional[dict[str, str]]
+    preset_suffix: str = ""
+
+    def with_preset_suffix(self, model_id: str) -> str:
+        """Re-attach a preserved ``@preset/<slug>`` suffix after auto-correction."""
+        return f"{model_id}{self.preset_suffix}"
+
+
+# ── Provider branches (None = not decided here) ─────────────────────────
+
+def _validate_moa(requested: str) -> dict[str, Any]:
+    try:
+        from hermes_cli.config import load_config
+        from hermes_cli.moa_config import normalize_moa_config
+
+        cfg = normalize_moa_config(load_config().get("moa") or {})
+        if requested in cfg["presets"]:
+            return _accept()
+        return _reject(f"MoA preset `{requested}` was not found. Run `hermes moa list`.")
+    except Exception as exc:
+        return _reject(f"Could not read MoA presets: {exc}")
+
+
+def _parse_openrouter_preset(req: _Request) -> Optional[dict[str, Any]]:
+    """OpenRouter presets are account-scoped, so ``@preset/<slug>`` never appears in the public
+    /v1/models listing. A bare preset is accepted unverified; ``<model>@preset/<slug>`` validates
+    the base model and preserves the suffix through auto-correction. OpenRouter validates the slug
+    at request time."""
+    marker = "@preset/"
+    if marker not in req.requested:
+        return None
+    if req.requested.count(marker) != 1:
+        preset_slug, preset_base = "", req.requested
+    else:
+        preset_base, preset_slug = req.requested.split(marker, 1)
+    if re.fullmatch(r"[A-Za-z0-9._~-]+", preset_slug) is None:
+        return _reject(
+            "OpenRouter preset slugs must be non-empty URL-safe "
+            "identifiers using only letters, digits, '.', '_', "
+            "'~', or '-'."
+        )
+    req.preset_suffix = f"{marker}{preset_slug}"
+    if not preset_base:
+        return _soft_accept(None)
+    req.lookup = preset_base
+    return None
+
+
+def _validate_lmstudio(req: _Request) -> dict[str, Any]:
+    from hermes_cli import models as _m
+    from hermes_cli.auth import AuthError
+
+    # probe_lmstudio_models distinguishes None (unreachable / malformed) from [] (reachable,
+    # nothing chat-capable loaded); fetch_lmstudio_models collapses both to [].
+    try:
+        models = _m.probe_lmstudio_models(api_key=req.api_key, base_url=req.base_url)
+    except AuthError as exc:
+        return _reject(f"{exc} Set `LM_API_KEY` (or update it) to match the server's bearer token.")
+    if models is None:
+        return _reject(f"Could not reach LM Studio's `/api/v1/models` to validate `{req.requested}`.")
+    if not models:
+        return _reject(
+            f"LM Studio is reachable but no chat-capable models are loaded. "
+            f"Load `{req.requested}` in LM Studio (Developer tab → Load Model) and try again."
+        )
+    if req.lookup in set(models):
+        return _accept()
+    return _reject(f"Model `{req.requested}` was not found in LM Studio's model listing.")
+
+
+def _ollama_probe_headers(req: _Request) -> dict[str, str]:
+    """Headers for the Ollama native probe.
+
+    Configured ``providers.ollama.extra_headers`` are only applied when the probed endpoint is
+    the configured one (never leak them to a different host). Caller headers win; a caller
+    ``api_key`` becomes the Authorization header unless the caller already sent one.
+    """
+    from hermes_cli import models as _m
+    from hermes_cli.models_local import _configured_ollama_base_url
+
+    configured_base = _configured_ollama_base_url()
+    configured_allowed = not (
+        configured_base and not _m._same_ollama_native_root(req.base_url or "", configured_base)
+    )
+    if req.headers is None:
+        return _m._get_ollama_native_headers(req.base_url, api_key=req.api_key) if configured_allowed else {}
+    out: dict[str, str] = {}
+    if configured_allowed:
+        out.update(_m._get_ollama_native_headers(req.base_url, api_key=req.api_key))
+    for key in tuple(out):
+        if key.lower() == "authorization":
+            del out[key]
+    out.update(req.headers)
+    caller_has_authorization = any(key.lower() == "authorization" for key in req.headers)
+    if req.api_key and not caller_has_authorization:
+        for key in tuple(out):
+            if key.lower() == "authorization":
+                del out[key]
+        out["Authorization"] = f"Bearer {req.api_key}"
+    return out
+
+
+def _validate_ollama_native(req: _Request) -> Optional[dict[str, Any]]:
+    """Runs for EVERY provider: the native ``/api/tags`` catalog is used whenever the endpoint
+    looks like a local Ollama server. Also resolves ``base_url`` for the raw ``ollama`` provider,
+    which later branches (custom) rely on."""
+    from hermes_cli import models as _m
+
+    if str(req.provider or "").strip().lower() == "ollama" and not req.base_url:
+        req.base_url = _m._get_ollama_base_url()
+    headers = _ollama_probe_headers(req)
+    if not _m.should_use_ollama_native_catalog(req.provider, req.base_url, headers=headers):
+        return None
+    models = _m.probe_ollama_local_models(req.base_url, headers=headers)
+    if models is None:
+        # A failed native probe is not authoritative; fall back to the OpenAI-compatible
+        # catalog before accepting blindly.
+        models = _m.probe_api_models(
+            req.api_key,
+            _m._normalize_openai_base_url(req.base_url),
+            request_headers=headers,
+        ).get("models")
+    if models is None:
+        return _soft_accept(
+            f"Note: could not reach this Ollama endpoint's `/api/tags` model listing to validate `{req.requested}`. "
+            "Hermes will save the model name, but local Ollama model discovery could not verify it."
+        )
+    match = _match_in_catalog(req.lookup, models, auto_correct=False,
+                              suggest_label="Similar local Ollama models")
+    if match.exact:
+        return _accept()
+    empty_hint = " No models are currently listed by `/api/tags`." if not models else ""
+    return _soft_accept(
+        f"Note: `{req.requested}` was not found in this Ollama endpoint's `/api/tags` model listing."
+        f"{empty_hint} It may still work if the server supports hidden or aliased models."
+        f"{match.suggestion_text}"
+    )
+
+
+def _validate_custom(req: _Request) -> dict[str, Any]:
+    from hermes_cli import models as _m
+
+    # Probe with the auth shape the api_mode expects.
+    if req.api_mode == "anthropic_messages":
+        probe = _m.probe_api_models(req.api_key, req.base_url, api_mode=req.api_mode,
+                                    request_headers=req.headers)
+    else:
+        probe = _m.probe_api_models(req.api_key, req.base_url, request_headers=req.headers)
+    api_models = probe.get("models")
+    if api_models is not None:
+        match = _match_in_catalog(req.lookup, api_models, suggest_query=req.requested)
+        if match.exact:
+            return _accept()
+        if match.corrected:
+            return _corrected(req.requested, match.corrected)
+        message = (
+            f"Note: `{req.requested}` was not found in this custom endpoint's model listing "
+            f"({probe.get('probed_url')}). It may still work if the server supports hidden or aliased models."
+            f"{match.suggestion_text}"
+        )
+        if probe.get("used_fallback"):
+            message += (
+                f"\n  Endpoint verification succeeded after trying `{probe.get('resolved_base_url')}`. "
+                f"Consider saving that as your base URL."
+            )
+        return _soft_accept(message)
+
+    message = (
+        f"Note: could not reach this custom endpoint's model listing at `{probe.get('probed_url')}`. "
+        f"Hermes will still save `{req.requested}`, but the endpoint should expose `/models` for verification."
+    )
+    if req.api_mode == "anthropic_messages":
+        message += (
+            "\n  Many Anthropic-compatible proxies do not implement the Models API "
+            "(GET /v1/models).  The model name has been accepted without verification."
+        )
+    if probe.get("suggested_base_url"):
+        message += f"\n  If this server expects `/v1`, try base URL: `{probe.get('suggested_base_url')}`"
+    # Anthropic-style proxies routinely lack /v1/models, so only they are accepted unverified.
+    return _verdict(req.api_mode == "anthropic_messages", True, False, message)
+
+
+def _static_catalog(normalized: str) -> list[str]:
+    from hermes_cli import models as _m
+
+    try:
+        return _m.provider_model_ids(normalized)
+    except Exception:
+        return []
+
+
+_STATIC_FAMILY_PREFIXES = {
+    "openai-codex": ("gpt-", "codex-", "o1", "o3", "o4"),
+    "xai-oauth": ("grok-",),
+}
+_STATIC_LABELS = {"openai-codex": "OpenAI Codex", "xai-oauth": "xAI Grok OAuth (SuperGrok / Premium+)"}
+
+
+def _validate_static_catalog(req: _Request) -> Optional[dict[str, Any]]:
+    """openai-codex / xai-oauth: no /v1/models probing — validate against the curated catalog.
+    Returns None (fall through) when the catalog is empty."""
+    catalog = _static_catalog(req.normalized)
+    if req.normalized == "openai-codex":
+        from agent.model_metadata import CODEX_CONTEXT_VARIANT_SUFFIX, is_codex_context_variant
+
+        # Ineligible ``-900k`` aliases must be rejected BEFORE the hidden-slug soft-accept:
+        # the suffix is a Hermes picker convention, so an unknown `*-900k` can never be a real
+        # hidden provider slug — soft-accepting one silently runs at 272K on a different model.
+        if req.lookup.strip().lower().endswith(CODEX_CONTEXT_VARIANT_SUFFIX) and req.lookup not in set(catalog):
+            if is_codex_context_variant(req.lookup):
+                # Valid variant a stale catalog hasn't synthesized yet. Accept directly — the typo
+                # auto-corrector would otherwise "fix" it to the base slug and drop the opt-in.
+                return _accept()
+            base_guess = req.lookup[: -len(CODEX_CONTEXT_VARIANT_SUFFIX)]
+            return _reject(
+                f"`{req.requested}` is not a valid large-context variant — "
+                f"`{base_guess}` enforces the standard 272K window on "
+                f"Codex, so no `-900k` option exists for it. Pick the "
+                f"base model, or a verified variant from the `/model` "
+                f"picker (e.g. `gpt-5.6-sol-900k`)."
+            )
+    if not catalog:
+        return None
+    match = _match_in_catalog(req.lookup, catalog)
+    if match.exact:
+        return _accept()
+    if match.corrected:
+        return _corrected(req.requested, match.corrected)
+    label = _STATIC_LABELS[req.normalized]
+    # Plausibility gate: the soft-accept exists for entitlement-gated *hidden* slugs the curated
+    # listing hasn't caught up with — always the provider's own family (gpt-* / grok-*). An
+    # unrelated name (`qwen3.5-4b`) would turn an actionable "did you mean --provider <x>?" into
+    # a confusing success that 400s on the next turn, so reject it with guidance instead.
+    prefixes = _STATIC_FAMILY_PREFIXES.get(req.normalized, ())
+    lower = req.lookup.strip().lower()
+    if prefixes and not any(lower.startswith(p) for p in prefixes):
+        return _reject(
+            f"`{req.requested}` doesn't look like a {label} model "
+            f"and isn't in its listing, so it was not accepted. If it "
+            f"belongs to another configured provider, switch with "
+            f"`--provider <slug>` (or select it from the `/model` "
+            f"picker)."
+            f"{match.suggestion_text}"
+        )
+    return _soft_accept(
+        f"Note: `{req.requested}` was not found in the {label} model listing. "
+        "It may still work if your account has access to a newer or hidden model ID."
+        f"{match.suggestion_text}"
+    )
+
+
+def _validate_minimax(req: _Request) -> Optional[dict[str, Any]]:
+    """MiniMax has no /models endpoint — static catalog, case-insensitive (ids like MiniMax-M2.7).
+    Returns None when the catalog is empty."""
+    catalog = _static_catalog(req.normalized)
+    if not catalog:
+        return None
+    match = _match_in_catalog(req.lookup, catalog, case_insensitive=True)
+    if match.exact:
+        return _accept()
+    if match.corrected:
+        return _corrected(req.requested, match.corrected)
+    return _soft_accept(
+        f"Note: `{req.requested}` was not found in the MiniMax catalog."
+        f"{match.suggestion_text}"
+        "\n  MiniMax does not expose a /models endpoint, so Hermes cannot verify the model name."
+        "\n  The model may still work if it exists on the server."
+    )
+
+
+def _validate_anthropic(req: _Request) -> Optional[dict[str, Any]]:
+    """Native Anthropic: /v1/models needs x-api-key (or OAuth Bearer) + anthropic-version, so the
+    generic Bearer probe 401s — use the native fetcher. Returns None (fall through to the generic
+    ladder) when no token is resolvable or the network failed."""
+    from hermes_cli import models as _m
+
+    models = _m._fetch_anthropic_models(base_url=req.base_url or None, api_key=req.api_key or None)
+    if models is None:
+        return None
+    match = _match_in_catalog(req.lookup, models, suggest_query=req.requested)
+    if match.exact:
+        return _accept()
+    if match.corrected:
+        return _corrected(req.requested, match.corrected)
+    # Accept anyway — Anthropic gates newer/preview models (snapshot IDs, early access) behind
+    # accounts even though they aren't listed on /v1/models.
+    return _soft_accept(
+        f"Note: `{req.requested}` was not found in Anthropic's /v1/models listing. "
+        f"It may still work if you have early-access or snapshot IDs."
+        f"{match.suggestion_text}"
+    )
+
+
+def _validate_anthropic_messages(req: _Request) -> dict[str, Any]:
+    """Anthropic Messages transport: many proxies don't implement /v1/models — probe, and accept
+    with a warning when the probe fails or the model isn't listed."""
+    from hermes_cli import models as _m
+
+    models = _m.fetch_api_models(req.api_key, req.base_url, api_mode=req.api_mode)
+    if models is not None:
+        match = _match_in_catalog(req.lookup, models)
+        if match.exact:
+            return _accept()
+        if match.corrected:
+            return _corrected(req.requested, match.corrected)
+    return _soft_accept(
+        f"Note: could not verify `{req.requested}` against this endpoint's "
+        f"model listing.  Many Anthropic-compatible proxies do not "
+        f"implement GET /v1/models.  The model name has been accepted "
+        f"without verification."
+    )
+
+
+def _nous_portal_recommended_names() -> set[str]:
+    """Lower-cased ids from the Portal's live recommended-models feed (empty on any failure)."""
+    from hermes_cli import models as _m
+
+    try:
+        payload = _m.fetch_nous_recommended_models(_m._resolve_nous_portal_url())
+        return {
+            name.lower()
+            for tier in ("freeRecommendedModels", "paidRecommendedModels")
+            for entry in (payload.get(tier) or [])
+            if (name := _m._extract_model_name(entry))
+        }
+    except Exception:
+        return set()
+
+
+def _validate_live_listing(req: _Request) -> Optional[dict[str, Any]]:
+    """Generic live /v1/models probe. Returns None when the API was unreachable (the caller then
+    tries Bedrock discovery / the curated catalog)."""
+    from hermes_cli import models as _m
+
+    api_models = _m.fetch_api_models(req.api_key, req.base_url)
+    if api_models is None:
+        return None
+    if req.normalized == "gemini":
+        # Gemini's OpenAI-compat listing prefixes ids with "models/"; curated list and user
+        # input use the bare id, so strip before comparing.
+        api_models = [
+            m[len("models/"):] if isinstance(m, str) and m.startswith("models/") else m
+            for m in api_models
+        ]
+    match = _match_in_catalog(req.lookup, api_models)
+    if match.exact:
+        return _accept()
+    # OpenRouter routing variants (":nitro", ":floor", ...) are request-time modifiers, not
+    # catalog entries — validate the BASE but keep the suffixed id. Must run BEFORE fuzzy
+    # auto-correction, which would otherwise "correct" `model:nitro` → `model` and silently
+    # strip the routing opt-in.
+    variant_base = _m._openrouter_variant_base(req.lookup) if req.normalized == "openrouter" else None
+    if variant_base is not None and variant_base in set(api_models):
+        return _accept()
+    # Listed but not found: the account may reach models absent from the public listing
+    # (e.g. Z.AI Pro/Max plans use glm-5 on coding endpoints) — warn but allow where plausible.
+    if match.corrected:
+        corrected = req.with_preset_suffix(match.corrected)
+        return _corrected(req.requested, corrected)
+    # Curated-catalog soft-accept: providers omit valid models from live listings (stale cache,
+    # partial rollout, gated previews). EXCEPTION: official OpenAI hosts (canonical + data-
+    # residency regional) — their listing is access-scoped and authoritative, so an absent model
+    # is one this key CANNOT serve; a soft-accept would 400 at first use. Custom OpenAI-compatible
+    # proxies keep the fallback.
+    listing_authoritative = False
+    if req.normalized in ("openai", "openai-api"):
+        from hermes_cli.providers import is_official_openai_host
+
+        listing_authoritative = is_official_openai_host(req.base_url)
+    if not listing_authoritative and _m._model_in_provider_catalog(
+        (variant_base or req.lookup).lower(), _m._provider_keys(req.normalized)
+    ):
+        return _verdict(True, True, True,
+                        f"Note: `{req.requested}` was not found in the live /v1/models listing "
+                        f"but exists in the curated catalog — accepted.")
+    # Nous: the Portal's recommended-models feed can list a model before the curated list or the
+    # docs-hosted manifest catches up; `hermes chat` already accepts those at model-list build
+    # time, so mirror that source of truth for per-message /model validation.
+    if req.normalized == "nous" and req.lookup.lower() in _nous_portal_recommended_names():
+        return _verdict(True, True, True,
+                        f"Note: `{req.requested}` was not found in the live /v1/models "
+                        f"listing but is a current Nous Portal recommendation — accepted.")
+    return _reject(
+        f"Model `{req.requested}` was not found in this provider's model listing."
+        f"{match.suggestion_text}"
+    )
+
+
+def _validate_bedrock(req: _Request) -> Optional[dict[str, Any]]:
+    """Bedrock's runtime URL has no /models; discovery goes through the AWS control plane
+    (ListFoundationModels + ListInferenceProfiles). Any failure falls through (None)."""
+    try:
+        from agent.bedrock_adapter import discover_bedrock_models, resolve_bedrock_runtime_region
+
+        region = resolve_bedrock_runtime_region()
+        discovered_ids = {m["id"] for m in discover_bedrock_models(region)}
+        match = _match_in_catalog(req.requested, list(discovered_ids), auto_correct=False,
+                                  suggest_cutoff=0.4)
+        if match.exact:
+            return _accept()
+        # Still accept (custom inference profiles / cross-account access), but warn.
+        return _soft_accept(
+            f"Note: `{req.requested}` was not found in Bedrock model discovery for {region}. "
+            f"It may still work with custom inference profiles or cross-account access."
+            f"{match.suggestion_text}"
+        )
+    except Exception:
+        return None
+
+
+def _validate_catalog_fallback(req: _Request) -> dict[str, Any]:
+    """The /models probe was unreachable: validate against the curated ``provider_model_ids()``
+    list so gateway /model switches keep working while a provider's endpoint is down (otherwise
+    switch_model() would fail and the gateway never writes the session override). No catalog at
+    all → accept with a warning."""
+    from hermes_cli import models as _m
+
+    label = _m._PROVIDER_LABELS.get(req.normalized, req.normalized)
+    catalog = _static_catalog(req.normalized)
+    if not catalog:
+        return _soft_accept(
+            f"Note: could not reach the {label} API to validate `{req.requested}`. "
+            f"If the service isn't down, this model may not be valid."
+        )
+    match = _match_in_catalog(req.lookup, catalog, case_insensitive=True)
+    if match.exact:
+        return _accept()
+    # Same OpenRouter routing-variant rule as the live-listing path.
+    if req.normalized == "openrouter":
+        variant_base = _m._openrouter_variant_base(req.lookup)
+        if variant_base is not None and variant_base.lower() in {m.lower() for m in catalog}:
+            return _accept()
+    if match.corrected:
+        corrected = req.with_preset_suffix(match.corrected)
+        return _corrected(req.requested, corrected)
+    return _soft_accept(
+        f"Note: `{req.requested}` was not found in the {label} curated catalog "
+        f"and the /models endpoint was unreachable.{match.suggestion_text}"
+        f"\n  The model may still work if it exists on the provider."
+    )
+
+
+# ── Orchestrator ─────────────────────────────────────────────────────────
+
+def validate_requested_model(
+    model_name: str,
+    provider: Optional[str],
+    *,
+    api_key: Optional[str] = None,
+    base_url: Optional[str] = None,
+    api_mode: Optional[str] = None,
+    headers: Optional[dict[str, str]] = None,
+) -> dict[str, Any]:
+    """Validate a ``/model`` value for the active provider.
+
+    Returns a dict with: - accepted: whether the CLI should switch to the requested model now -
+    persist: whether it is safe to save to config - recognized: whether it matched a known provider
+    catalog - message: optional warning / guidance for the user (- corrected_model: when a typo
+    was auto-corrected).
+    """
+    from hermes_cli import models as _m
+
+    requested = (model_name or "").strip()
+    normalized = _m.normalize_provider(provider)
+    if normalized == "openrouter" and base_url and not base_url_host_matches(base_url, "openrouter.ai"):
+        normalized = "custom"
+    lookup = requested
+    if normalized == "copilot":
+        lookup = _m.normalize_copilot_model_id(requested, api_key=api_key) or requested
+
+    if not requested:
+        return _reject("Model name cannot be empty.")
+    if normalized == "moa":
+        return _validate_moa(requested)
+    if any(ch.isspace() for ch in requested):
+        return _reject("Model names cannot contain spaces.")
+
+    req = _Request(requested, lookup, provider, normalized, api_key, base_url, api_mode, headers)
+    if normalized == "openrouter":
+        verdict = _parse_openrouter_preset(req)
+        if verdict is not None:
+            return verdict
+    if normalized == "lmstudio":
+        return _validate_lmstudio(req)
+    verdict = _validate_ollama_native(req)
+    if verdict is not None:
+        return verdict
+    if normalized == "custom" or normalized.startswith("custom:"):
+        return _validate_custom(req)
+    if normalized in {"openai-codex", "xai-oauth"}:
+        verdict = _validate_static_catalog(req)
+        if verdict is not None:
+            return verdict
+    if normalized in {"minimax", "minimax-cn"}:
+        verdict = _validate_minimax(req)
+        if verdict is not None:
+            return verdict
+    if normalized == "anthropic":
+        verdict = _validate_anthropic(req)
+        if verdict is not None:
+            return verdict
+    if api_mode == "anthropic_messages":
+        return _validate_anthropic_messages(req)
+    verdict = _validate_live_listing(req)
+    if verdict is not None:
+        return verdict
+    # API unreachable — accept and persist, but warn so typos don't silently break things.
+    if normalized == "bedrock":
+        verdict = _validate_bedrock(req)
+        if verdict is not None:
+            return verdict
+    return _validate_catalog_fallback(req)