"""Custom-provider resolution: ``providers:`` / ``custom_providers:`` lookup, identity recovery, custom credential pools, and the named-custom runtime builder. Extracted from :mod:`hermes_cli.runtime_provider`; origin-internal collaborators (``load_config``, ``_get_model_config``, ``load_pool``, ``has_usable_secret``, …) are looked up on the origin module AT CALL TIME via :func:`_rp` so ``monkeypatch.setattr(runtime_provider, name, …)`` keeps working for moved bodies.""" from __future__ import annotations import logging import os from typing import Any, Callable, Dict, Optional, Tuple from hermes_cli.providers import custom_provider_aliases, custom_provider_slug from agent.custom_entry_pool import custom_entry_for_endpoint, entry_declares_credential, entry_scoped_pool from agent.secret_scope import get_secret_str from utils import base_url_hostname logger = logging.getLogger("hermes_cli.runtime_provider") _LLAMACPP_ALIASES = ("llamacpp", "llama.cpp", "llama-cpp") def _rp(): """Origin module, late-bound so test patches on ``hermes_cli.runtime_provider.*`` apply.""" import hermes_cli.runtime_provider as origin return origin def _normalize_custom_provider_name(value: str) -> str: return value.strip().lower().replace(" ", "-") def _normalize_base_url_for_match(value) -> str: return str(value or "").strip().rstrip("/").lower() def _route_key(value) -> str: """An endpoint compared for identity: scheme and host fold case, the path does not.""" from hermes_cli.route_identity import normalize_route_base_url return normalize_route_base_url(_clean(value)) def _clean(value: Any) -> str: return str(value or "").strip() def _key_env_secret(entry: dict[str, Any], label: str) -> str: """The credential named by ``key_env`` / ``api_key_env`` on a config block, or "". A declared variable that resolves to nothing is logged: every custom rung substitutes ``no-key-required`` for an empty key (keyless local servers), so a misnamed or unexported variable otherwise surfaces only as the provider's 401/403 (#67453). A block with no key_env at all stays silent — that IS the keyless-server configuration. A ``providers:`` entry with a declared credential never falls back to an implicit env key, so the placeholder is what it gets. """ key_env = _clean(entry.get("key_env") or entry.get("api_key_env")) if not key_env: return "" value = get_secret_str(key_env, "").strip() if not value: logger.warning("%s: key_env %s is set but the variable is empty/unset — unless another credential " "is configured for it, the request carries the placeholder no-key-required and the " "endpoint will reject it", label, key_env) return value def _model_cfg_key_env_for(model_cfg: dict[str, Any], base_url: str) -> str: """``model.key_env`` for a bare ``provider: custom`` runtime, only when ``base_url`` IS the configured ``model.base_url`` — the key was declared for that endpoint, never for a direct alias or CUSTOM_BASE_URL pointing elsewhere.""" cfg_base_url = _clean(model_cfg.get("base_url")).rstrip("/") if not cfg_base_url or cfg_base_url != _clean(base_url).rstrip("/"): return "" return _key_env_secret(model_cfg, "model") def _entry_url(entry: dict[str, Any]) -> str: return entry.get("api") or entry.get("url") or entry.get("base_url") or "" # ── field lifting shared by ``providers:`` and legacy ``custom_providers:`` entries ──────── def _filter_capabilities(value: Any) -> dict[str, bool]: """Return the string-keyed boolean capabilities accepted at runtime.""" if not isinstance(value, dict): return {} return {k: v for k, v in value.items() if isinstance(k, str) and isinstance(v, bool)} def _lift_model_capabilities(entry: dict[str, Any], model: Optional[str], result: dict[str, Any]) -> None: """Copy explicit boolean per-model capabilities into the runtime.""" capabilities = _filter_capabilities(entry.get("capabilities")) models = entry.get("models") model_config = models.get(model) if isinstance(models, dict) and model else None if isinstance(model_config, dict): capabilities.update(_filter_capabilities(model_config)) if capabilities: result["capabilities"] = capabilities def _lift_extra_headers(entry: dict[str, Any], result: dict[str, Any]) -> None: """Copy a validated ``extra_headers`` dict. SECURITY: values carry credentials — never log.""" extra_headers = _rp().normalize_extra_headers(entry.get("extra_headers")) if extra_headers: result["extra_headers"] = extra_headers def _lift_common_custom_fields(entry: dict[str, Any], result: dict[str, Any], *, provider_key: str, key_env: str, api_mode: Optional[str]) -> None: """Copy the optional fields shared by ``providers:`` and legacy ``custom_providers:`` entries.""" if key_env: result["key_env"] = key_env if provider_key: result["provider_key"] = provider_key extra_body = entry.get("extra_body") if isinstance(extra_body, dict): result["extra_body"] = dict(extra_body) _lift_extra_headers(entry, result) if api_mode: result["api_mode"] = api_mode _lift_model_capabilities(entry, None, result) # ── config lookup ────────────────────────────────────────────────────────────────────────── def _shadowed_by_builtin(requested_norm: str) -> bool: """Raw names map to custom providers only when they are not canonical built-ins. Explicit ``custom:`` keys always target the saved entry, and bare ``custom`` is exempt: a user may literally name a ``providers:`` entry "custom" (returning None before the config scan made such cron jobs fail with ``auth_unavailable``). Defer to the built-in only when the raw name IS the canonical provider (``nous``); an entry matching merely an alias (``kimi`` → ``kimi-coding``) is the user's target.""" if requested_norm == "custom" or requested_norm.startswith("custom:"): return False rp = _rp() try: canonical = rp.auth_mod.resolve_provider(requested_norm) except rp.AuthError: return False return (canonical or "").strip().lower() == requested_norm def _match_new_style_provider( requested_norm: str, providers: dict[str, Any], *, resolve_credentials: bool = True, ) -> Optional[tuple[dict[str, Any], Any]]: """Scan ``providers:`` (new-style, keyed) for ``requested_norm``; ``(entry, config key)``.""" from hermes_cli.config import is_provider_enabled rp = _rp() for ep_name, entry in providers.items(): # ``providers..enabled: false`` entries stay in config but are invisible here. if not isinstance(entry, dict) or not is_provider_enabled(entry): continue if requested_norm not in custom_provider_aliases(str(entry.get("name", "") or ep_name), str(ep_name)): continue base_url = _entry_url(entry) if not base_url: continue # Resolve credentials only after identity and endpoint validation. Merely scanning an # unrelated entry must not read its profile-scoped secret. key_env = _clean(entry.get("key_env") or entry.get("api_key_env")) api_key = get_secret_str(key_env, "").strip() if key_env and resolve_credentials else "" result: dict[str, Any] = {"name": entry.get("name", ep_name), "base_url": base_url.strip(), "api_key": api_key or _clean(entry.get("api_key", "")), "model": entry.get("default_model", "")} # Command that PRINTS a short-lived credential; wrapped in a per-request token provider. key_cmd = _clean(entry.get("key_cmd", "")) if key_cmd: result["key_cmd"] = key_cmd # v12 migration writes ``transport``; hand-edited configs may still use ``api_mode``. # Accept both or migrated configs silently downgrade to chat_completions. _lift_common_custom_fields( entry, result, provider_key=_clean(ep_name), key_env=key_env, api_mode=rp._parse_api_mode(entry.get("api_mode") or entry.get("transport")), ) return result, ep_name return None def _match_legacy_custom_provider(requested_norm: str, custom_providers) -> Optional[tuple[dict[str, Any], dict]]: """Scan the legacy ``custom_providers:`` list for ``requested_norm``; ``(result, matched entry)``.""" for entry in custom_providers: name, base_url = (entry.get("name"), entry.get("base_url")) if isinstance(entry, dict) else (None, None) if not isinstance(name, str) or not isinstance(base_url, str): continue provider_key = _clean(entry.get("provider_key", "")) if requested_norm not in custom_provider_aliases(name, provider_key): continue result = {"name": name.strip(), "base_url": base_url.strip(), "api_key": _clean(entry.get("api_key", ""))} model_name = _clean(entry.get("model", "")) if model_name: result["model"] = model_name _lift_common_custom_fields(entry, result, provider_key=provider_key, key_env=_clean(entry.get("key_env", "")), api_mode=_rp()._parse_api_mode(entry.get("api_mode"))) return result, entry return None def _get_named_custom_provider( requested_provider: str, *, config: Optional[dict[str, Any]] = None, resolve_credentials: bool = True, ) -> Optional[dict[str, Any]]: """Resolve a named endpoint. Creation-time config copying supplies its own config snapshot and skips credential reads; ordinary runtime callers keep their existing scoped resolution.""" found = _select_named_custom_provider(requested_provider, config=config, resolve_credentials=resolve_credentials) return found[0] if found else None def named_custom_provider_config_path(requested_provider: str, config: dict[str, Any]) -> Optional[tuple]: """The config path (``("providers", key)`` / ``("custom_providers", index)``) of the ONE entry the runtime routes *requested_provider* through, else None. Reads no credential.""" found = _select_named_custom_provider(requested_provider, config=config, resolve_credentials=False) return found[1] if found else None def custom_provider_config_path_for_url(base_url: Any, config: dict[str, Any]) -> Optional[tuple]: """The config path of the entry a bare ``custom`` route at *base_url* takes its key from: the first compatible ``custom_providers`` / ``providers`` entry with that URL (trailing slash ignored), the same owner the credential pool seeds from. Reads no credential.""" url = str(base_url or "").strip().rstrip("/") if not url or isinstance(config.get("custom_providers"), dict): return None for entry in _rp().get_compatible_custom_providers(config) or (): if isinstance(entry, dict) and str(entry.get("base_url") or "").strip().rstrip("/") == url: return _compatible_entry_path(config, entry) return None def _select_named_custom_provider( requested_provider: str, *, config: Optional[dict[str, Any]], resolve_credentials: bool, ) -> Optional[tuple[dict[str, Any], Optional[tuple]]]: """``(resolved entry, its config path)`` for the first usable match: ``providers:`` first, then the legacy list view.""" requested_norm = _normalize_custom_provider_name(requested_provider or "") if not requested_norm or requested_norm == "auto" or _shadowed_by_builtin(requested_norm): return None rp = _rp() if config is None: config = rp.load_config() providers = config.get("providers") found = ( _match_new_style_provider(requested_norm, providers, resolve_credentials=resolve_credentials) if isinstance(providers, dict) else None ) if found: return found[0], ("providers", found[1]) if isinstance(config.get("custom_providers"), dict): logger.warning("custom_providers in config.yaml is a dict, not a list. " "Each entry must be prefixed with '-' in YAML. " "Run 'hermes doctor' for details.") return None custom_providers = rp.get_compatible_custom_providers(config) found = _match_legacy_custom_provider(requested_norm, custom_providers) if custom_providers else None return (found[0], _compatible_entry_path(config, found[1])) if found else None def _compatible_entry_path(config: dict[str, Any], entry: dict[str, Any]) -> Optional[tuple]: """Where a ``get_compatible_custom_providers`` entry lives in *config*. The view keeps the first of identical legacy entries, so the first raw entry normalizing to it is its source.""" from hermes_cli.config_providers import _normalize_custom_provider_entry providers = config.get("providers") key = entry.get("provider_key") if key and isinstance(providers, dict): return next((("providers", k) for k in providers if str(k).strip() == key), None) legacy = config.get("custom_providers") return next((("custom_providers", i) for i, raw in enumerate(legacy if isinstance(legacy, list) else ()) if _normalize_custom_provider_entry(raw) == entry), None) def has_named_custom_provider(requested_provider: str) -> bool: """True when config defines a ``providers:`` / ``custom_providers:`` entry matching the request (public wrapper so e.g. the cronjob tool need not reach into a private helper).""" try: return _rp()._get_named_custom_provider(requested_provider) is not None except Exception: return False def codex_model_provider_id(requested_provider: str) -> Optional[str]: """Codex ``[model_providers.]`` key for a configured named custom provider — its ``custom:`` identity without the prefix (the ``providers:`` config key; legacy ``custom_providers:`` entries use their normalized display name). None for bare ``custom``, aliases that resolve to custom (ollama, vllm, …) and unknown names: codex has no stable id to look up for those (#75186).""" if _normalize_custom_provider_name(requested_provider or "") in {"", "custom"}: return None try: entry = _rp()._get_named_custom_provider(requested_provider) except Exception: return None if not entry: return None return custom_provider_slug(str(entry.get("name") or ""), str(entry.get("provider_key") or "")).split(":", 1)[1] or None # ── identity recovery (bare "custom" -> durable ``custom:``) ───────────────────────── def _find_custom_identity(matches: Callable[[dict[str, Any]], bool], config: Optional[dict[str, Any]] = None, prefer: Optional[str] = None) -> Optional[str]: """First enabled entry in ``providers:`` then legacy ``custom_providers:`` where ``matches(entry)`` holds (*prefer* when it is one of them), as its canonical ``custom:`` slug. A legacy entry whose slug a ``providers:`` entry with an endpoint owns never matches: runtime resolution runs that one. *config* defaults to the active profile's.""" from hermes_cli.config import is_provider_enabled rp = _rp() try: config = rp.load_config() if config is None else config except Exception: return None first, owned = None, set() providers = config.get("providers") if isinstance(providers, dict): for ep_name, entry in providers.items(): if not (isinstance(entry, dict) and is_provider_enabled(entry)): continue slug = custom_provider_slug(str(ep_name), str(ep_name)) if _entry_url(entry): owned.add(slug) if matches(entry): if prefer is None or slug == prefer: return slug first = first or slug try: custom_providers = rp.get_compatible_custom_providers(config) except Exception: custom_providers = None for entry in custom_providers or []: name = entry.get("name") if isinstance(entry, dict) else None if isinstance(name, str) and name.strip() and matches(entry): slug = custom_provider_slug(name, str(entry.get("provider_key", "") or "")) if slug in owned and not entry.get("provider_key"): continue if prefer is None or slug == prefer: return slug first = first or slug return first def find_custom_provider_identity(base_url: str) -> Optional[str]: """Map an endpoint URL back to its canonical ``custom:`` menu key. Session persistence stores the agent's *resolved* provider, which for every named custom endpoint is the literal string ``"custom"`` — the entry name is lost, and the api_key is deliberately never persisted.""" target = _route_key(base_url) if not target: return None return _find_custom_identity(lambda entry: _route_key(_entry_url(entry)) == target) def _model_id_matches(value: Any, target: str) -> bool: return isinstance(value, str) and value.strip().lower() == target def find_custom_provider_identity_by_model(model: str) -> Optional[str]: """Map a model id back to the ``custom:`` entry that serves it — companion to :func:`find_custom_provider_identity` for persistence paths where no base_url survived the round-trip (the session row always stores the model name).""" target = str(model or "").strip().lower() return _find_custom_identity(lambda entry: _entry_serves_model(entry, target)) if target else None def _entry_serves_model(entry: dict[str, Any], target: str) -> bool: if any(_model_id_matches(entry.get(key), target) for key in ("model", "default_model")): return True models = entry.get("models") if isinstance(models, dict): return any(str(mid).strip().lower() == target for mid in models) if isinstance(models, list): return any(_model_id_matches(item.get("id") or item.get("name") if isinstance(item, dict) else item, target) for item in models) return False def routed_custom_identity(base_url: Optional[str], model: Optional[str], config: Optional[dict[str, Any]] = None, held: Optional[str] = None) -> Optional[str]: """The configured ``custom:`` entry a bare ``custom`` pick (the billing class every named endpoint reports) names: on *base_url*, the entry also serving *model* beats the first one there; without a URL the model alone names it. Among entries serving *model*, the *held* one (the profile's own) wins over config order.""" url = _route_key(base_url) target = str(model or "").strip().lower() def on_url(entry: dict[str, Any]) -> bool: return not url or _route_key(_entry_url(entry)) == url return ((target and _find_custom_identity(lambda e: on_url(e) and _entry_serves_model(e, target), config, held)) or (url and _find_custom_identity(on_url, config)) or None) def _endpoint_identity(base_url: str) -> Optional[str]: """The menu identity that owns ``base_url``: a configured entry, else the managed server.""" identity = find_custom_provider_identity(base_url) if identity: return identity # The managed server has no custom-provider config entry. Recover its menu key # from the ownership-checked endpoint, never from a model name or a fixed port. from hermes_cli.local_runtime.endpoint import _state_endpoint endpoint = _state_endpoint() if endpoint and _route_key(base_url) == _route_key(endpoint["base_url"]): return "llamacpp" return None def canonical_custom_identity(*, base_url: Optional[str] = None, config_provider: Optional[str] = None, model: Optional[str] = None, requested: Optional[str] = None) -> Optional[str]: """Recover the durable menu identity for a bare custom provider. Match a configured endpoint first, then the ownership-checked managed server, then a configured model or provider. Every session persistence/restore path shares this lookup.""" rp = _rp() # The entry the live agent resolved (*requested*) while it still serves base_url, else its owner. identity = session_provider_identity("custom", requested, base_url) if base_url else "custom" if identity != "custom": return identity identity = find_custom_provider_identity_by_model(model) if model else None if identity: return identity candidate = str(config_provider or "").strip() if not candidate: try: candidate = str(rp._get_model_config().get("provider") or "").strip() except Exception: candidate = "" if not candidate: candidate = os.environ.get("HERMES_INFERENCE_PROVIDER", "").strip() candidate_norm = _normalize_custom_provider_name(candidate) # A bare/non-routable candidate cannot heal a bare custom override. if not candidate_norm or candidate_norm in {"custom", "auto", "openrouter"}: return None # Only when it resolves to a configured entry — never invent a ``custom:`` resolution # can't honor. ``candidate`` may be the entry's DISPLAY NAME, not the durable identity of a # keyed ``providers:`` entry — re-resolve via its endpoint so every path returns the same # config-key slug. try: entry = rp._get_named_custom_provider(candidate) except Exception: return None if entry is None: return None try: identity = find_custom_provider_identity(str(entry.get("base_url") or "")) except Exception: return None return identity or custom_provider_slug(candidate_norm) def session_provider_identity(provider: Any, requested_provider: Any, base_url: Any, model: Any = None, *, config: Optional[dict[str, Any]] = None) -> str: """A session's identity: the requested entry while it still serves ``base_url``, else the endpoint's owner, else the adapter. A name for an entry that left ``base_url`` pairs its new key with the old endpoint on resume, so with an endpoint known the model and config fallbacks never apply.""" adapter = _clean(provider) if adapter.lower() != "custom": return adapter requested = _clean(requested_provider) url = _route_key(base_url) if url and requested.lower() not in {"", "custom", "auto"}: entry = _get_named_custom_provider(requested, config=config, resolve_credentials=False) if entry and _route_key(_entry_url(entry)) == url: return requested if url: return _endpoint_identity(_clean(base_url)) or adapter return canonical_custom_identity(model=_clean(model) or None) or adapter def is_routable_provider(provider: Optional[str]) -> bool: """Whether a provider name currently resolves to a routable route. Empty/None/``auto`` is vacuously routable (agent build falls back to the configured default). Bare ``custom`` is the resolved billing class shared by every named entry — not a routable identity; restore paths must heal it (:func:`canonical_custom_identity`) or fall back. Anything else is routable iff the full chain (built-in -> ``providers:`` -> ``custom_providers:`` -> models.dev) resolves it.""" name = str(provider or "").strip() if not name or name.lower() == "auto": return True if name.lower() == "custom": return False try: from hermes_cli.providers import resolve_provider_full rp = _rp() config = rp.load_config() return resolve_provider_full(name, config.get("providers"), rp.get_compatible_custom_providers(config)) is not None except Exception: return False # ── runtime builders ─────────────────────────────────────────────────────────────────────── def _owning_entry(rp, base_url: str, provider_name: Optional[str] = None) -> Optional[dict[str, Any]]: """The ``providers:`` entry that owns a custom endpoint, from the frozen route's config when the session froze one (a live edit then neither drops nor reassigns it).""" frozen = rp._FROZEN_RUNTIME_CONFIG.get() match = custom_entry_for_endpoint(base_url, provider_name, frozen) return match[1] if match else None def _owning_entry_credential(rp, base_url: str) -> tuple[bool, Any]: """``(declared, key)`` for a bare ``provider: custom`` route whose URL a ``providers:`` entry owns. ``declared`` is False when no entry owns the URL or it names no credential; the caller then keeps its implicit env fallbacks. Otherwise ``key`` is the entry's own (key_cmd first, as on the named path), or "" when it resolves to nothing: no other account's env key may stand in for it.""" entry = _owning_entry(rp, base_url) if entry is None or not entry_declares_credential(entry): return False, "" return True, _entry_credential(rp, entry, str(entry.get("name") or "custom")) def _entry_credential(rp, entry: dict[str, Any], name: str) -> Any: """A ``providers:`` entry's own credential, for the named and the URL-matched bare routes alike: ``key_cmd`` (minted per request) first, else the first usable of ``api_key`` / ``key_env``, else "". A placeholder ``api_key`` never hides a working ``key_env``.""" key_cmd = _clean(entry.get("key_cmd")) if key_cmd: from agent.command_token_source import build_command_token_provider if (token_provider := build_command_token_provider(key_cmd, name)) is not None: return token_provider candidates = (_clean(entry.get("api_key")), _key_env_secret(entry, f"custom provider '{name}'")) return next((c for c in candidates if rp.has_usable_secret(c)), "") def _try_resolve_from_custom_pool( base_url: str, provider_label: str, api_mode_override: Optional[str] = None, provider_name: Optional[str] = None, ) -> Optional[dict[str, Any]]: """Runtime dict from the first credential pool that owns this custom endpoint, else None. A pool the owning entry shares with a same-named built-in yields only the entry's rows (entry_scoped_pool).""" rp = _rp() try: # A frozen route's entry owns the pool, never the live one (a live URL edit would drop it). frozen = rp._FROZEN_RUNTIME_CONFIG.get() scope = {} if frozen is None else {"config": frozen} raw_keys = list(rp.custom_provider_pool_key_candidates(base_url, provider_name, **scope)) owner = _owning_entry(rp, base_url, provider_name) except Exception: raw_keys, owner = [], None # Order-preserving dedupe of normalized keys. candidates = list(dict.fromkeys(k for k in (str(key or "").strip().lower() for key in raw_keys) if k)) for pool_key in candidates: try: if (pool := entry_scoped_pool(rp.load_pool(pool_key), pool_key, owner)) is None: continue entry = pool.select() if pool.has_credentials() else None pool_api_key = rp._pool_entry_api_key(entry) if entry is not None else "" if not pool_api_key: continue if not rp.has_usable_secret(pool_api_key) and rp._loopback_hostname(base_url_hostname(base_url)): # Legacy configs used short placeholder keys ('123', 'm') for local no-auth # services; has_usable_secret's 4-char floor rejects them. Every other path # substitutes "no-key-required" for a loopback endpoint — this was the one gap. # Every OTHER resolution path in this file already substitutes "no-key-required" for a # loopback endpoint with no usable secret (the config-based custom_providers fallback a few # hundred lines below, and the "actual" provider's local-offline exemption further down) -- # this pool path was the one gap (issue #86864). pool_api_key = "no-key-required" return rp._runtime(provider_label, api_mode_override or rp._detect_api_mode_for_url(base_url) or "chat_completions", base_url, pool_api_key, source=f"pool:{pool_key}", credential_pool=pool) except Exception: continue return None def _custom_provider_request_overrides(custom_provider: dict[str, Any]) -> dict[str, Any]: extra_body = custom_provider.get("extra_body") if not isinstance(extra_body, dict) or not extra_body: return {} return {"extra_body": dict(extra_body)} def _apply_custom_provider_extras(custom_provider: dict[str, Any], target_model: Optional[str], result: dict[str, Any]) -> None: """Copy model / capabilities / extra_headers / request_overrides onto a resolved custom runtime. An explicit ``target_model`` wins over the provider's configured default (auxiliary slots / background-review resolve a concrete model and must not fall back to ``default_model``). ``extra_headers`` may carry credentials — NEVER log them.""" model_name = target_model or custom_provider.get("model") if model_name: result["model"] = model_name _lift_model_capabilities(custom_provider, model_name, result) if custom_provider.get("extra_headers"): result["extra_headers"] = dict(custom_provider["extra_headers"]) request_overrides = _custom_provider_request_overrides(custom_provider) if request_overrides: result["request_overrides"] = {**(result.get("request_overrides") or {}), **request_overrides} class LocalRuntimeUnavailableError(ValueError): """The selected model is the managed local runtime and its server is not serving. Typed so readiness probes can say "configured, but the local server is down" instead of "nothing configured": provider onboarding cannot start a local server. """ error_code = "local_runtime_unavailable" def __init__(self, message: str, *, retryable: bool = False) -> None: super().__init__(message) self.retryable = retryable def _resolve_llamacpp_runtime(requested_provider: str, explicit_api_key: Optional[str]) -> dict[str, Any]: """Managed llama.cpp runtime: the supervised (or detected external) server, or a typed error. No server => say so and stop; falling through to the generic custom path would surface "local server is off" as OpenRouter's baffling "401 Invalid API key". The switch's state picks the message (server off → point at the switch; else the setup pane).""" rp = _rp() try: from hermes_cli.local_runtime.endpoint import resolve_llamacpp_endpoint endpoint = resolve_llamacpp_endpoint(config=rp.load_config()) except Exception: endpoint = None if endpoint: return rp._runtime("custom", "chat_completions", endpoint["base_url"], (explicit_api_key or "").strip() or endpoint["api_key"] or "no-key-required", source="local-runtime", requested_provider=requested_provider) try: enabled = bool((rp.load_config().get("local_runtime") or {}).get("enabled")) except Exception: enabled = False if enabled: raise LocalRuntimeUnavailableError("The local model server isn't running. It may still be " "starting — try again in a moment, or switch to another " "model with /model.", retryable=True) # Every chat surface (messaging, TUI, Webapp) shows this sentence, so it names both places a # user can start the server and no Desktop-only pane; the Desktop client links its pane itself. raise LocalRuntimeUnavailableError("The local model server is turned off. Start it from the Hermes " "desktop app or `hermes model` → Local models, or switch to " "another model with /model.") def _custom_runtime(rp, base_url: str, api_key: Any, api_mode: Optional[str], **extra: Any) -> dict[str, Any]: """``custom`` runtime dict with URL-detected api_mode fallback and the no-auth placeholder.""" return rp._runtime("custom", api_mode or rp._detect_api_mode_for_url(base_url) or "chat_completions", base_url, api_key or "no-key-required", **extra) # Aliases for direct REST APIs not modeled in PROVIDER_REGISTRY, so ``provider: openai`` (aux slots, # background review, curator, MoA slots, the main model) resolves to a working ``custom`` endpoint # instead of "Unknown provider" and a silent fall-back to the main model (#116055). _DIRECT_API_BASE_URLS: dict[str, str] = {"openai": "https://api.openai.com/v1"} def expand_direct_api_alias(provider: Optional[str], existing_base: Optional[str]) -> tuple[Optional[str], Optional[str]]: """``provider: openai`` → custom + the user's OpenAI endpoint, api.openai.com/v1 only as the last resort. The ONE normalization both aux paths (``agent.auxiliary_client`` and ``resolve_runtime_provider``) apply, so the same ``auxiliary..provider`` value routes identically everywhere. A ``providers.openai`` entry keeps the provider name so the named-custom branch applies its base_url and key; otherwise ``OPENAI_BASE_URL`` (a proxy/gateway the OPENAI_API_KEY was issued for) wins over the public endpoint — sending the proxy key to api.openai.com 401s and then quarantines a valid key. """ if not provider: return provider, existing_base target_base = _DIRECT_API_BASE_URLS.get(provider.strip().lower()) if target_base is None or _rp()._get_named_custom_provider(provider) is not None: return provider, existing_base return "custom", (existing_base or "").strip() or get_secret_str("OPENAI_BASE_URL", "").strip().rstrip("/") or target_base def _resolve_direct_alias_runtime(requested_provider: str, explicit_api_key: Optional[str], explicit_base_url: str) -> dict[str, Any]: """Bare ``custom`` + explicit base_url (e.g. a ``model_aliases:`` direct alias).""" rp = _rp() base_url = explicit_base_url.strip().rstrip("/") # Pool first — mirrors the named-custom path so bare `provider: custom` with a configured # custom_providers entry gets its api_key from the pool instead of env fallbacks. An explicit # launch key still wins over the pool (R2-M1). explicit_key = (explicit_api_key or "").strip() pool_result = None if rp.has_usable_secret(explicit_key) else rp._try_resolve_from_custom_pool(base_url, "custom", None) if pool_result: pool_result["source"] = "direct-alias" return pool_result # OLLAMA_API_KEY gets its own gate here: without it a `model_aliases:` entry pointing at # Ollama Cloud resolved no key at all. # ``model.key_env`` only when this alias endpoint IS the configured model.base_url (#67453). # A ``providers:`` entry that owns the URL and declares a credential supplies it instead of # the host-gated env keys (on the entry's built-in's host those are another account's). declared, entry_key = _owning_entry_credential(rp, base_url) candidates = [(explicit_api_key or "").strip(), _model_cfg_key_env_for(rp._get_model_config(), base_url), *([] if declared else rp._host_gated_env_key_candidates(base_url, ollama=True))] api_key = next((c for c in candidates if rp.has_usable_secret(c)), "") or entry_key return _custom_runtime(rp, base_url, api_key, None, source="direct-alias", requested_provider=requested_provider) def _opencode_family_for_custom(requested_provider: str, base_url: str) -> Optional[str]: """OpenCode family by provider name, else by opencode.ai host (``/zen/go`` => opencode-go).""" # Custom providers in the OpenCode family (name extends opencode-go/zen, or base_url hosted on # opencode.ai) serve models behind different API surfaces per model — a static api_mode 503s for # /v1/responses-only models like grok-4.5 (#85589). Re-derive api_mode from the effective model and # normalize the /v1 suffix, exactly like the built-in opencode-zen/go paths do. from hermes_cli.models import opencode_provider_family family = opencode_provider_family(requested_provider) if family is not None: return family try: if base_url_hostname(base_url).lower() == "opencode.ai": return "opencode-go" if "/zen/go" in base_url.lower() else "opencode-zen" except Exception: pass return None def _resolve_named_custom_runtime(*, requested_provider: str, explicit_api_key: Optional[str] = None, explicit_base_url: Optional[str] = None, target_model: Optional[str] = None, config=None) -> Optional[dict[str, Any]]: """Runtime for a llamacpp alias, a bare-custom direct alias, or a configured custom entry. Aliases resolving to "custom" (ollama, vllm, llamacpp, …) are treated like bare ``custom``. A llamacpp alias with no explicit base_url resolves to the managed server first; an explicit base_url always wins.""" rp = _rp() # Bare `provider="custom"` with an explicit base_url (e.g. propagated from a `model_aliases:` # direct-alias resolution) — build a runtime directly so the alias's base_url actually takes effect. # GitHub #27132: provider aliases that resolve to "custom" at runtime (ollama, vllm, llamacpp, …) are # treated identically here, so a YAML `provider: ollama` with a LAN/WireGuard `base_url` doesn't # silently fall through to OpenRouter. requested_norm = (requested_provider or "").strip().lower() custom_provider = None if requested_norm in _LLAMACPP_ALIASES and not explicit_base_url: custom_provider = rp._get_named_custom_provider(requested_provider) if not custom_provider: return _resolve_llamacpp_runtime(requested_provider, explicit_api_key) if requested_norm and requested_norm != "custom" and rp._resolves_to_custom(requested_norm): requested_norm = "custom" if requested_norm == "custom" and explicit_base_url: return _resolve_direct_alias_runtime(requested_provider, explicit_api_key, explicit_base_url) if custom_provider is None: custom_provider = (rp._get_named_custom_provider(requested_provider) if config is None else _get_named_custom_provider(requested_provider, config=config)) if not custom_provider: return None base_url = ((explicit_base_url or "").strip() or custom_provider.get("base_url", "")).rstrip("/") if not base_url: return None explicit_key = (explicit_api_key or "").strip() # The credential pool is the profile's auth store (``hermes auth add``), not config: a frozen # config snapshot still owes a pooled ``providers:`` entry its keys, or the turn ships the # ``no-key-required`` placeholder to a host that wants one (C11 pool rows). An explicit launch # key is the operator's credential for this route and wins over the pool (R2-M1). An entry # that names its own credential never borrows a same-named built-in's env-seeded pool rows. pool_result = None if rp.has_usable_secret(explicit_key) else rp._try_resolve_from_custom_pool( base_url, "custom", custom_provider.get("api_mode"), provider_name=custom_provider.get("provider_key") or custom_provider.get("name"), ) if pool_result: # The pool doesn't know the custom_providers fields — propagate them here too. _apply_custom_provider_extras(custom_provider, target_model, pool_result) return pool_result # An explicit --api-key wins as the one-off recovery escape hatch. ``key_cmd`` credentials are # minted per request (short-lived bearers would go stale mid-session); both wire clients accept a # callable api_key (the Entra ID contract). A declared credential that resolves to nothing gets no # implicit host-gated env key: on the entry's built-in's host that is another account's key. api_key: Any = explicit_key if rp.has_usable_secret(explicit_key) else ( _entry_credential(rp, custom_provider, str(custom_provider.get("name", requested_provider) or "custom")) or next((c for c in ([] if entry_declares_credential(custom_provider) else rp._host_gated_env_key_candidates(base_url, ollama=False)) if rp.has_usable_secret(c)), "")) result = _custom_runtime(rp, base_url, api_key, custom_provider.get("api_mode"), source=f"custom_provider:{custom_provider.get('name', requested_provider)}", requested_provider=requested_provider) _apply_custom_provider_extras(custom_provider, target_model, result) # OpenCode-family custom providers (opencode-go/zen names, or opencode.ai hosts) serve models # on different API surfaces — a static api_mode 503s for /v1/responses-only models. Re-derive # api_mode from the model and normalize /v1 like the built-in paths. family = _opencode_family_for_custom(requested_provider, base_url) if family is not None and not custom_provider.get("api_mode"): from hermes_cli.models import normalize_opencode_base_url, opencode_model_api_mode effective_model = str(target_model or custom_provider.get("model") or rp._get_model_config().get("default") or "").strip() if effective_model: result["api_mode"] = opencode_model_api_mode(family, effective_model) result["base_url"] = normalize_opencode_base_url(family, result["api_mode"], result["base_url"]) return result