hermes-agent/hermes_cli/runtime_provider_custom.py

"""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:<name>`` 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.<name>.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.<id>]`` 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:<name>``) ─────────────────────────


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:<name>`` 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:<name>`` 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:<name>`` 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:<name>`` 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:<x>`` 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.<task>.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