feat(models): propagate native discovery across selectors

Co-authored-by: Taneli Mielikäinen <taneli.mielikainen@iki.fi>
This commit is contained in:
Vadelma
2026-08-08 23:42:42 -07:00
committed by Teknium
parent 638b72f5ef
commit fa1bb88e3e
9 changed files with 1122 additions and 80 deletions
+4 -1
View File
@@ -1442,9 +1442,11 @@ def _normalize_custom_provider_entry(
if isinstance(api_key, str) and api_key.strip():
normalized["api_key"] = api_key.strip()
key_env = entry.get("key_env")
key_env = entry.get("key_env") or entry.get("api_key_env")
if isinstance(key_env, str) and key_env.strip():
normalized["key_env"] = key_env.strip()
if entry.get("api_key_env") and not entry.get("key_env"):
normalized["api_key_env"] = key_env.strip()
api_mode = entry.get("api_mode") or entry.get("transport")
if isinstance(api_mode, str) and api_mode.strip():
@@ -1483,6 +1485,7 @@ def _normalize_custom_provider_entry(
}
normalized_models[model_id.strip()] = model_meta
if normalized_models:
normalized_models["__explicit_model_allowlist__"] = True
normalized["models"] = normalized_models
context_length = entry.get("context_length")
+2 -1
View File
@@ -3664,9 +3664,10 @@ def select_provider_and_model(args=None):
"name": name,
"base_url": base_url,
"api_key": entry.get("api_key", ""),
"key_env": entry.get("key_env", ""),
"key_env": entry.get("key_env") or entry.get("api_key_env", ""),
"model": entry.get("model", ""),
"models": entry.get("models", {}),
"extra_headers": entry.get("extra_headers", {}),
"discover_models": entry.get("discover_models", True),
"api_mode": entry.get("api_mode", ""),
"provider_key": provider_key,
+90 -15
View File
@@ -1505,14 +1505,22 @@ def _model_flow_azure_foundry(config, current_model=""):
def _model_flow_named_custom(config, provider_info):
"""Handle a named custom provider from config.yaml custom_providers list.
Always probes the endpoint's /models API to let the user pick a model.
Probes the endpoint's model catalog to let the user pick a model, using
native ``/api/tags`` for endpoints conservatively identified as Ollama.
If a model was previously saved, it is pre-selected in the menu.
Falls back to the saved model if probing fails.
"""
from hermes_cli.main import _custom_provider_api_key_config_value, _custom_provider_base_url_config_value, _save_custom_provider
from hermes_cli.auth import _save_model_choice, deactivate_provider
from hermes_cli.config import load_config, save_config
from hermes_cli.models import fetch_api_models
from hermes_cli.config import load_config, normalize_extra_headers, save_config
from hermes_cli.model_switch import _models_config_is_allowlist
from hermes_cli.models import (
fetch_api_models,
fetch_ollama_local_models,
_get_ollama_native_headers,
_normalize_openai_base_url,
should_use_ollama_native_catalog,
)
name = provider_info["name"]
base_url = provider_info["base_url"]
@@ -1538,13 +1546,24 @@ def _model_flow_named_custom(config, provider_info):
if isinstance(discover, str):
discover = discover.lower() not in {"false", "no", "0"}
configured_models: list[str] = []
native_catalog_empty = False
cfg_models = provider_info.get("models", {})
explicit_catalog = _models_config_is_allowlist(cfg_models)
if isinstance(cfg_models, dict):
configured_models = [str(m) for m in cfg_models if str(m).strip()]
elif isinstance(cfg_models, list):
configured_models = [
str(m) for m in cfg_models if isinstance(m, str) and m.strip()
str(m)
for m in cfg_models
if m != "__explicit_model_allowlist__" and str(m).strip()
]
elif isinstance(cfg_models, list):
configured_models = []
for model_entry in cfg_models:
if isinstance(model_entry, dict):
model_id = str(model_entry.get("id") or model_entry.get("model") or "").strip()
else:
model_id = str(model_entry).strip() if isinstance(model_entry, str) else ""
if model_id:
configured_models.append(model_id)
print(f" Provider: {name}")
print(f" URL: {base_url}")
@@ -1552,19 +1571,75 @@ def _model_flow_named_custom(config, provider_info):
print(f" Current: {saved_model}")
print()
if not discover and configured_models:
# Discovery disabled with an explicit list — use it verbatim, no probe.
print(f"Using configured models (discover_models: false): {len(configured_models)}")
models = configured_models
if not discover:
# Discovery disabled: never probe, even when only the singular active
# model is configured. The active model is useful as the sole picker
# choice, but it is not an endpoint catalog.
models = configured_models or ([saved_model] if saved_model else [])
print(
"Using configured models (discover_models: false): "
f"{len(models)}"
)
else:
print("Fetching available models...")
fetch_kwargs = {"timeout": 8.0}
if api_mode:
fetch_kwargs["api_mode"] = api_mode
live_models = fetch_api_models(api_key, base_url, **fetch_kwargs)
# If the probe came back empty but the operator configured an explicit
# list, fall back to it rather than forcing manual entry.
models = live_models or configured_models
native_catalog_provider = (
"ollama"
if provider_key.lower() == "ollama" or name.strip().lower() == "ollama"
else "custom"
)
extra_headers = normalize_extra_headers(provider_info.get("extra_headers")) or {}
candidate_headers = _get_ollama_native_headers(base_url, api_key=api_key)
for key in tuple(candidate_headers):
if any(key.lower() == existing.lower() for existing in extra_headers):
del candidate_headers[key]
candidate_headers.update(extra_headers)
caller_has_authorization = any(
key.lower() == "authorization" for key in extra_headers
)
if api_key and not caller_has_authorization:
for key in tuple(candidate_headers):
if key.lower() == "authorization":
del candidate_headers[key]
candidate_headers["Authorization"] = f"Bearer {api_key}"
use_native = should_use_ollama_native_catalog(
native_catalog_provider, base_url, headers=candidate_headers or None
)
native_headers_arg = candidate_headers or None if use_native else (extra_headers or None)
explicit_allowlist = explicit_catalog
if use_native:
if explicit_catalog and configured_models:
live_models = configured_models
native_catalog_empty = False
else:
live_models = fetch_ollama_local_models(
base_url,
timeout=8.0,
headers=native_headers_arg,
)
native_catalog_empty = live_models == []
if live_models is None:
live_models = fetch_api_models(
api_key,
_normalize_openai_base_url(base_url),
headers=native_headers_arg,
**fetch_kwargs,
)
native_catalog_empty = False
else:
live_models = fetch_api_models(
api_key, base_url, headers=native_headers_arg, **fetch_kwargs
)
native_catalog_empty = False
models = (
configured_models
if explicit_allowlist
else []
if native_catalog_empty
else (live_models or configured_models)
)
# Persist the live catalog back to the custom_providers entry so that
# no-probe surfaces (dashboard, desktop, ACP) show the full model list
# instead of collapsing to the single ``model:`` default. Mirrors the
@@ -1623,7 +1698,7 @@ def _model_flow_named_custom(config, provider_info):
except (ValueError, KeyboardInterrupt, EOFError):
print("\nCancelled.")
return
elif saved_model:
elif saved_model and not native_catalog_empty:
print("Could not fetch models from endpoint.")
try:
model_name = input(f"Model name [{saved_model}]: ").strip() or saved_model
+311 -60
View File
@@ -20,6 +20,7 @@ OpenRouter variant suffixes (``:free``, ``:extended``, ``:fast``).
from __future__ import annotations
import http.client
import logging
import os
import re
@@ -87,6 +88,8 @@ def _declared_model_ids(value: Any) -> list[str]:
if isinstance(value, dict):
for model_id in value:
if model_id == "__explicit_model_allowlist__":
continue
_add(model_id)
return ids
@@ -123,7 +126,7 @@ def _models_config_is_allowlist(value: Any) -> bool:
if isinstance(value, str):
return bool(value.strip())
if isinstance(value, dict):
return False
return bool(value.get("__explicit_model_allowlist__"))
if isinstance(value, (list, tuple)):
return bool(_declared_model_ids(value))
return False
@@ -202,6 +205,89 @@ def _bare_custom_provider_def(current_base_url: str) -> Optional[ProviderDef]:
)
_MODEL_DISCOVERY_ERRORS = (
ImportError,
OSError,
RuntimeError,
TimeoutError,
TypeError,
ValueError,
http.client.HTTPException,
)
class _NativePickerModelList(list[str]):
"""A successful native catalog, including an authoritative empty one."""
def _fetch_picker_live_models(
api_key: str,
api_url: str,
native_catalog_provider: str,
preserve_native_models: bool,
headers: dict[str, str] | None = None,
timeout: float = 5.0,
api_mode: str | None = None,
) -> list[str] | None:
"""Fetch picker models with native Ollama and cached generic discovery."""
from hermes_cli.models import (
_get_ollama_native_headers,
_normalize_openai_base_url,
cached_fetch_api_models,
fetch_ollama_local_models,
should_use_ollama_native_catalog,
)
candidate_headers = _get_ollama_native_headers(api_url, api_key=api_key)
caller_has_authorization = any(
key.lower() == "authorization" for key in (headers or {})
)
if caller_has_authorization:
for key in tuple(candidate_headers):
if key.lower() == "authorization":
del candidate_headers[key]
if headers:
for key in tuple(candidate_headers):
if any(key.lower() == existing.lower() for existing in headers):
del candidate_headers[key]
candidate_headers.update(headers)
if api_key and not caller_has_authorization:
for key in tuple(candidate_headers):
if key.lower() == "authorization":
del candidate_headers[key]
candidate_headers["Authorization"] = f"Bearer {api_key}"
use_native = should_use_ollama_native_catalog(
native_catalog_provider, api_url, headers=candidate_headers or None
)
resolved_headers = candidate_headers or None if use_native else headers
if use_native:
if preserve_native_models:
return None
native_models = fetch_ollama_local_models(
api_url, timeout=timeout, headers=resolved_headers
)
if native_models is not None:
return _NativePickerModelList(native_models)
# A failed native probe is not authoritative: retry the cached generic
# OpenAI-compatible catalog before reporting no models.
return cached_fetch_api_models(
api_key,
_normalize_openai_base_url(api_url),
timeout=timeout,
headers=resolved_headers,
api_mode=api_mode,
)
generic_models = cached_fetch_api_models(
api_key,
api_url,
timeout=timeout,
headers=resolved_headers,
api_mode=api_mode,
)
return generic_models if generic_models else None
# ---------------------------------------------------------------------------
# Non-agentic model warning
# ---------------------------------------------------------------------------
@@ -1340,6 +1426,9 @@ def switch_model(
detect_provider_for_model,
validate_requested_model,
opencode_model_api_mode,
_get_ollama_request_headers,
_get_provider_config_dict,
_same_ollama_native_root,
)
from hermes_cli.runtime_provider import resolve_runtime_provider
@@ -1689,9 +1778,11 @@ def switch_model(
api_key = current_api_key
base_url = current_base_url
api_mode = ""
ollama_headers: dict[str, str] = {}
validation_headers: dict[str, str] = {}
suppress_ollama_headers = False
if provider_changed or explicit_provider:
import os
# User-config providers (providers.<name> in config.yaml) carry their
# own base_url + transport + key reference. resolve_runtime_provider()
# resolves by provider NAME and doesn't know user-config slugs (e.g. a
@@ -1715,9 +1806,12 @@ def switch_model(
# os.getenv when multiplexing is off, fail-closed otherwise).
_ukey = _scoped_key_env(_ukey[2:-1])
if not _ukey:
_kenv = str(_ucfg.get("key_env", "") or "").strip()
_kenv = str(
_ucfg.get("key_env") or _ucfg.get("api_key_env") or ""
).strip()
if _kenv:
_ukey = _scoped_key_env(_kenv)
validation_headers = _extra_headers_from_config(_ucfg)
try:
runtime = resolve_runtime_provider(
requested=target_provider,
@@ -1728,6 +1822,7 @@ def switch_model(
api_key = runtime.get("api_key", "") or _ukey
base_url = runtime.get("base_url", "") or _user_pdef.base_url
api_mode = runtime.get("api_mode", "")
validation_headers = runtime.get("extra_headers") or validation_headers
except Exception:
api_key = _ukey
base_url = _user_pdef.base_url
@@ -1745,6 +1840,7 @@ def switch_model(
api_key = runtime.get("api_key", "")
base_url = runtime.get("base_url", "")
api_mode = runtime.get("api_mode", "")
validation_headers = runtime.get("extra_headers") or validation_headers
except Exception as e:
return ModelSwitchResult(
success=False,
@@ -1757,20 +1853,56 @@ def switch_model(
),
)
else:
try:
runtime = resolve_runtime_provider(
requested=current_provider,
target_model=new_model,
)
# If resolution fell through to "custom" (e.g. named custom provider like
# "ollama-launch" that resolve_runtime_provider doesn't know), keep existing
# credentials. Otherwise use the resolved values (picks up credential rotation,
# base_url adjustments for OpenCode, etc.).
api_key = runtime.get("api_key", "")
base_url = runtime.get("base_url", "")
api_mode = runtime.get("api_mode", "")
except Exception:
pass
keep_current_ollama_endpoint = False
if current_provider == "custom" and current_base_url:
try:
from hermes_cli.models import should_use_ollama_native_catalog
ollama_headers = _get_ollama_request_headers()
ollama_config = _get_provider_config_dict("ollama")
configured_ollama_base = str(
ollama_config.get("base_url")
or ollama_config.get("api")
or ollama_config.get("url")
or ""
).strip()
if configured_ollama_base and not _same_ollama_native_root(
current_base_url, configured_ollama_base
):
ollama_headers = {}
suppress_ollama_headers = True
elif not configured_ollama_base:
# Without an explicit configured root there is no safe
# origin to associate provider-level Ollama headers with.
ollama_headers = {}
suppress_ollama_headers = True
keep_current_ollama_endpoint = should_use_ollama_native_catalog(
current_provider,
current_base_url,
headers=ollama_headers,
)
except (ImportError, OSError, RuntimeError, TypeError, ValueError):
keep_current_ollama_endpoint = False
if keep_current_ollama_endpoint:
# Mid-session `/model <name>` on a local Ollama-compatible endpoint
# must keep the endpoint the session is already using. Re-resolving
# bare `custom` from config can fall through to an unrelated default
# provider, causing validation to probe the wrong model-list URL.
api_key = current_api_key or "no-key-required"
base_url = current_base_url
api_mode = determine_api_mode(current_provider, base_url)
validation_headers = ollama_headers
else:
try:
runtime = resolve_runtime_provider(
requested=current_provider,
target_model=new_model,
)
api_key = runtime.get("api_key", "")
base_url = runtime.get("base_url", "")
api_mode = runtime.get("api_mode", "")
validation_headers = runtime.get("extra_headers") or validation_headers
except (OSError, RuntimeError, TypeError, ValueError):
pass
# --- Direct alias override: use exact base_url from the alias if set ---
if resolved_alias:
@@ -1779,6 +1911,40 @@ def switch_model(
if _da is not None and _da.base_url:
base_url = _da.base_url
api_mode = "" # clear so determine_api_mode re-detects from URL
if target_provider.strip().lower() == "ollama":
_ollama_cfg = _get_provider_config_dict("ollama")
_ollama_cfg_base = str(
_ollama_cfg.get("base_url")
or _ollama_cfg.get("api")
or _ollama_cfg.get("url")
or ""
).strip()
if _ollama_cfg_base and _same_ollama_native_root(
base_url, _ollama_cfg_base
):
configured_key = str(_ollama_cfg.get("api_key") or "").strip()
if configured_key.startswith("${") and configured_key.endswith("}"):
configured_key = os.environ.get(configured_key[2:-1], "").strip()
if not configured_key:
key_env = str(_ollama_cfg.get("key_env") or "").strip()
if key_env:
configured_key = os.environ.get(key_env, "").strip()
if configured_key:
api_key = configured_key
if _ollama_cfg_base and not _same_ollama_native_root(
base_url, _ollama_cfg_base
):
# Do not carry providers.ollama credentials to an alias
# endpoint with a different origin.
validation_headers = {}
suppress_ollama_headers = True
api_key = "no-key-required"
elif not _ollama_cfg_base:
# Without an explicit configured root there is no safe
# origin to associate the provider-level headers with.
validation_headers = {}
suppress_ollama_headers = True
api_key = "no-key-required"
if not api_key:
api_key = "no-key-required"
@@ -1811,6 +1977,22 @@ def switch_model(
api_key=api_key,
base_url=base_url,
api_mode=api_mode or None,
headers=(
(
{}
if suppress_ollama_headers
else (validation_headers or _get_ollama_request_headers())
)
if target_provider.strip().lower() == "ollama"
else (
validation_headers
or (
_extra_headers_from_config(user_providers.get(target_provider))
if user_providers and target_provider in user_providers
else None
)
)
),
)
except Exception as e:
validation = {
@@ -2973,13 +3155,15 @@ def list_authenticated_providers(
or ep_cfg.get("url", "")
or ""
)
key_env = str(ep_cfg.get("key_env", "") or "").strip()
key_env = str(
ep_cfg.get("key_env") or ep_cfg.get("api_key_env") or ""
).strip()
inline_api_key = str(ep_cfg.get("api_key", "") or "").strip()
api_mode = str(
ep_cfg.get("api_mode")
or ep_cfg.get("transport")
or ""
).strip().lower()
).strip().lower() or None
credential_identity = (
inline_api_key
if inline_api_key
@@ -3100,8 +3284,10 @@ def list_authenticated_providers(
# local endpoints still show their full model catalog.
api_key = str(ep_cfg.get("api_key", "") or "").strip()
if not api_key:
key_env = str(ep_cfg.get("key_env", "") or "").strip()
api_key = _scoped_key_env(key_env)
key_env = str(
ep_cfg.get("key_env") or ep_cfg.get("api_key_env") or ""
).strip()
api_key = _scoped_key_env(key_env) if key_env else ""
discover = ep_cfg.get("discover_models", True)
if isinstance(discover, str):
discover = discover.lower() not in {"false", "no", "0"}
@@ -3136,20 +3322,49 @@ def list_authenticated_providers(
and (bool(api_key) or not has_explicit_models)
and _can_probe_custom_provider(row_is_current=_ep_is_current)
)
if _discovery_allowed:
native_catalog_empty = False
if _probe_live:
try:
from hermes_cli.models import cached_fetch_api_models
live_models = cached_fetch_api_models(
native_catalog_provider = (
ep_name
if str(ep_name).strip().lower()
in {"ollama", "custom:ollama"}
else "custom"
)
live_models = _fetch_picker_live_models(
api_key,
api_url,
timeout=1.5 if for_picker else 5.0, # picker: fail fast so a slow custom endpoint doesn't block /model
api_mode=grp.get("api_mode") or None,
native_catalog_provider,
has_explicit_models,
headers=_extra_headers_from_config(ep_cfg) or None,
cache_only=not _probe_live,
timeout=(1.5 if for_picker else 5.0),
api_mode=ep_cfg.get("api_mode"),
)
if live_models:
if isinstance(live_models, _NativePickerModelList):
native_catalog_empty = not live_models
if live_models is not None and (
live_models
or not has_explicit_models
or isinstance(live_models, _NativePickerModelList)
):
models_list = live_models
except Exception:
except _MODEL_DISCOVERY_ERRORS:
pass
elif _discovery_allowed:
try:
from hermes_cli.models import cached_fetch_api_models
cached_models = cached_fetch_api_models(
api_key,
api_url,
cache_only=True,
timeout=(1.5 if for_picker else 5.0),
headers=_extra_headers_from_config(ep_cfg) or None,
api_mode=ep_cfg.get("api_mode"),
)
if cached_models:
models_list = cached_models
except _MODEL_DISCOVERY_ERRORS:
pass
results.append({
@@ -3161,6 +3376,7 @@ def list_authenticated_providers(
"total_models": len(models_list) if models_list else 0,
"source": "user-config",
"api_url": api_url,
"native_catalog_empty": native_catalog_empty,
})
seen_slugs.add(ep_name.lower())
seen_slugs.update(_ep_aliases)
@@ -3208,21 +3424,34 @@ def list_authenticated_providers(
)
):
_models = [current_model] if current_model else []
# As in sections 3 and 4: with live probing suppressed, fall back to
# the cached catalog rather than to the single active model.
# With live probing suppressed, use the shared stale/cache path;
# otherwise probe through the native-aware picker helper.
native_catalog_empty = False
_probe_live = bool(refresh or probe_current_custom_provider)
try:
from hermes_cli.models import cached_fetch_api_models
if _probe_live:
_live_models = _fetch_picker_live_models(
"",
str(current_base_url).strip().rstrip("/"),
"custom",
False,
timeout=(1.5 if for_picker else 5.0),
)
else:
from hermes_cli.models import cached_fetch_api_models
_live_models = cached_fetch_api_models(
"",
str(current_base_url).strip().rstrip("/"),
timeout=1.5 if for_picker else 5.0, # picker: fail fast on a slow current endpoint
cache_only=not _probe_live,
)
if _live_models:
_live_models = cached_fetch_api_models(
"",
str(current_base_url).strip().rstrip("/"),
cache_only=True,
timeout=(1.5 if for_picker else 5.0),
)
if _live_models is not None:
native_catalog_empty = isinstance(
_live_models, _NativePickerModelList
) and not _live_models
_models = _live_models
except Exception:
except _MODEL_DISCOVERY_ERRORS:
pass
results.append({
"slug": "custom",
@@ -3233,6 +3462,7 @@ def list_authenticated_providers(
"total_models": len(_models),
"source": "model-config",
"api_url": str(current_base_url).strip().rstrip("/"),
"native_catalog_empty": native_catalog_empty,
})
seen_slugs.add("custom")
@@ -3280,7 +3510,7 @@ def list_authenticated_providers(
entry.get("api_mode")
or entry.get("transport")
or ""
).strip().lower()
).strip().lower() or None
credential_identity = (
inline_api_key
if inline_api_key
@@ -3326,6 +3556,7 @@ def list_authenticated_providers(
"models": [],
"has_explicit_models": False,
"discover_models": discover,
"api_mode": api_mode,
"extra_headers": entry_extra_headers,
# Part of group_key, so constant across the group. Needed
# in the render loop to key the model cache — api_mode
@@ -3478,31 +3709,50 @@ def list_authenticated_providers(
and (bool(api_key) or not grp.get("has_explicit_models"))
and _can_probe_custom_provider(row_is_current=_grp_is_current)
)
if _discovery_allowed:
native_catalog_empty = False
if _probe_live:
try:
native_catalog_provider = (
"ollama"
if str(slug).strip().lower() == "ollama"
or str(grp.get("name") or "").strip().lower() == "ollama"
else "custom"
)
live_models = _fetch_picker_live_models(
api_key,
api_url,
slug,
bool(grp.get("has_explicit_models")),
headers=grp.get("extra_headers") or None,
timeout=(1.5 if for_picker else 5.0),
api_mode=grp.get("api_mode"),
)
if live_models is not None and (
live_models or not bool(grp.get("has_explicit_models"))
):
if isinstance(live_models, _NativePickerModelList):
native_catalog_empty = not live_models
grp["models"] = live_models
grp["total_models"] = len(live_models)
_save_discovered_models_to_config(api_url, live_models)
except _MODEL_DISCOVERY_ERRORS:
pass
elif _discovery_allowed:
try:
from hermes_cli.models import cached_fetch_api_models
live_models = cached_fetch_api_models(
cached_models = cached_fetch_api_models(
api_key,
api_url,
timeout=1.5 if for_picker else 5.0, # picker: fail fast so a slow custom endpoint doesn't block /model
api_mode=grp.get("api_mode") or None,
cache_only=True,
timeout=(1.5 if for_picker else 5.0),
headers=grp.get("extra_headers") or None,
cache_only=not _probe_live,
api_mode=grp.get("api_mode"),
)
if live_models:
grp["models"] = live_models
grp["total_models"] = len(live_models)
# Auto-save discovered models back to config so
# ``discover_models: false`` has a populated cache
# on the next read. A failed save is non-fatal.
# Only after a real probe: a cache hit is already the
# product of an earlier probe that saved it.
if _probe_live:
_save_discovered_models_to_config(
api_url, live_models
)
except Exception:
if cached_models:
grp["models"] = cached_models
grp["total_models"] = len(cached_models)
except _MODEL_DISCOVERY_ERRORS:
pass
results.append({
"slug": slug,
@@ -3513,6 +3763,7 @@ def list_authenticated_providers(
"total_models": len(grp["models"]),
"source": "user-config",
"api_url": grp["api_url"],
"native_catalog_empty": native_catalog_empty,
})
seen_slugs.add(slug.lower())
_section4_emitted_slugs.add(slug.lower())
@@ -3549,7 +3800,7 @@ def list_authenticated_providers(
# which branch emitted the row.
if current_model:
for _row in results:
if not _row.get("is_current"):
if not _row.get("is_current") or _row.get("native_catalog_empty"):
continue
_models = _row.get("models") or []
if current_model not in _models:
+1 -1
View File
@@ -734,7 +734,7 @@ def resolve_user_provider(name: str, user_config: Dict[str, Any]) -> Optional[Pr
# Extract fields
display_name = entry.get("name", "") or name
api_url = entry.get("api", "") or entry.get("url", "") or entry.get("base_url", "") or ""
key_env = entry.get("key_env", "") or ""
key_env = entry.get("key_env") or entry.get("api_key_env") or ""
transport = entry.get("transport", "openai_chat") or "openai_chat"
env_vars: List[str] = []
+3 -1
View File
@@ -757,7 +757,9 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
if not is_provider_enabled(entry):
continue
# Resolve the API key from the env var name stored in key_env
key_env = str(entry.get("key_env", "") or "").strip()
key_env = str(
entry.get("key_env") or entry.get("api_key_env") or ""
).strip()
resolved_api_key = _getenv(key_env, "").strip() if key_env else ""
# Fall back to inline api_key when key_env is absent or unresolvable
if not resolved_api_key:
@@ -154,6 +154,7 @@ class TestCustomProviderModelSwitch:
mock_fetch.assert_called_once_with(
"sk-live-example-provider",
"https://api.example-provider.test/v1",
headers=None,
timeout=8.0,
)
config = yaml.safe_load(config_path.read_text()) or {}
@@ -445,6 +446,28 @@ class TestCustomProviderDiscoverModels:
instead of the endpoint's full live catalog."""
def test_discover_false_with_only_singular_model_skips_probe(self, config_home):
"""An active singular model is not an implicit discovery catalog."""
from hermes_cli.main import _model_flow_named_custom
provider_info = {
"name": "Headered Ollama",
"base_url": "http://127.0.0.1:11434",
"api_key": "no-key-required",
"discover_models": False,
"model": "qwen3:8b",
}
with patch("hermes_cli.models.fetch_api_models") as mock_fetch, \
patch("hermes_cli.models.fetch_ollama_local_models") as mock_ollama, \
patch("hermes_cli.curses_ui.curses_radiolist", side_effect=ImportError), \
patch("builtins.input", return_value="1"), \
patch("builtins.print"):
_model_flow_named_custom({}, provider_info)
mock_fetch.assert_not_called()
mock_ollama.assert_not_called()
def test_discover_false_saves_choice_from_configured_list(self, config_home):
"""User picks the 2nd configured model; it persists, list-driven."""
import yaml
@@ -93,6 +93,46 @@ def test_passthrough_kwargs_to_base(monkeypatch):
def test_current_custom_endpoint_passthrough_marks_current_row(monkeypatch):
"""Interactive picker should preserve current custom endpoint semantics."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr("agent.models_dev.PROVIDER_TO_MODELS_DEV", {})
monkeypatch.setattr("hermes_cli.providers.HERMES_OVERLAYS", {})
monkeypatch.setattr("hermes_cli.models.fetch_openrouter_models",
lambda *a, **kw: [])
result = model_switch.list_picker_providers(
current_provider="custom:ollama",
current_base_url="http://localhost:11434/v1",
current_model="glm-5.1",
user_providers={},
custom_providers=[
{
"name": "Ollama — GLM 5.1",
"base_url": "http://localhost:11434/v1",
"api_key": "ollama",
"model": "glm-5.1",
"discover_models": False,
},
{
"name": "Ollama — Qwen3",
"base_url": "http://localhost:11434/v1",
"api_key": "ollama",
"model": "qwen3",
"discover_models": False,
},
],
max_models=50,
)
custom_rows = [p for p in result if p.get("is_user_defined")]
assert len(custom_rows) == 1
row = custom_rows[0]
assert row["slug"] == "custom:ollama"
assert row["is_current"] is True
assert row["models"] == ["glm-5.1", "qwen3"]
# ---------------------------------------------------------------------------
# list_authenticated_providers: alias/canonical de-dup for Kimi (#49439)
@@ -3,6 +3,11 @@
The terminal `hermes model` flow already exposes `custom_providers`, but the
shared slash-command pipeline (`/model` in CLI/gateway/Telegram) historically
only looked at `providers:`.
Fixtures that set ``discover_models=False`` deliberately opt out of live
endpoint probing; they test saved/configured-provider behavior rather than a
local model server. The live-probe cases set it separately when discovery is
the behavior under test.
"""
import time
@@ -11,6 +16,7 @@ import hermes_cli.providers as providers_mod
import pytest
import yaml
from hermes_cli.model_switch import (
_fetch_picker_live_models,
_save_discovered_models_to_config,
list_authenticated_providers,
switch_model,
@@ -36,10 +42,53 @@ def _disable_live_custom_provider_model_probe(monkeypatch):
monkeypatch.setattr(
"hermes_cli.models.provider_model_ids", lambda *_a, **_kw: []
)
monkeypatch.setattr(
"hermes_cli.models.fetch_ollama_local_models", lambda *_a, **_kw: None
)
def test_picker_native_probe_failure_falls_back_to_openai_catalog(monkeypatch):
monkeypatch.setattr(
"hermes_cli.models.should_use_ollama_native_catalog", lambda *a, **k: True
)
monkeypatch.setattr(
"hermes_cli.models._get_ollama_native_headers", lambda *a, **k: {}
)
monkeypatch.setattr(
"hermes_cli.models.fetch_ollama_local_models", lambda *a, **k: None
)
monkeypatch.setattr(
"hermes_cli.models.fetch_api_models", lambda *a, **k: ["fallback-model"]
)
assert _fetch_picker_live_models(
"key", "http://127.0.0.1:11434/v1", "ollama", False
) == ["fallback-model"]
def test_picker_generic_discovery_preserves_api_mode(monkeypatch):
calls = []
def cached(*args, **kwargs):
calls.append((args, kwargs))
return ["model-a"]
monkeypatch.setattr(
"hermes_cli.models.should_use_ollama_native_catalog", lambda *a, **k: False
)
monkeypatch.setattr("hermes_cli.models.cached_fetch_api_models", cached)
assert _fetch_picker_live_models(
"key",
"https://proxy.example/anthropic/v1",
"custom",
False,
api_mode="anthropic_messages",
) == ["model-a"]
assert calls[0][1]["api_mode"] == "anthropic_messages"
def test_list_authenticated_providers_includes_custom_providers(monkeypatch):
"""No-args /model menus should include saved custom_providers entries."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
monkeypatch.setattr("hermes_cli.models.fetch_api_models", lambda *a, **k: [])
@@ -67,6 +116,55 @@ def test_list_authenticated_providers_includes_custom_providers(monkeypatch):
def test_providers_singular_model_does_not_suppress_ollama_native_discovery(monkeypatch):
"""A saved selection in ``providers:`` is not an explicit catalog."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
monkeypatch.setattr(
"hermes_cli.models.fetch_ollama_local_models",
lambda *a, **k: ["qwen3:latest", "llama3.2:latest"],
)
providers = list_authenticated_providers(
current_provider="openai-codex",
user_providers={
"ollama": {
"base_url": "http://localhost:11434/v1",
"model": "qwen3:latest",
}
},
custom_providers=[],
max_models=50,
)
ollama = next(p for p in providers if p["slug"] == "ollama")
assert ollama["models"] == ["qwen3:latest", "llama3.2:latest"]
def test_list_authenticated_providers_can_skip_custom_provider_live_probe(monkeypatch):
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
fetch = lambda *a, **k: (_ for _ in ()).throw(AssertionError("unexpected probe"))
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fetch)
providers = list_authenticated_providers(
user_providers={},
custom_providers=[
{
"name": "Slow Local",
"base_url": "http://127.0.0.1:8080/v1",
"api_key": "sk-local",
"model": "local-model",
}
],
probe_custom_providers=False,
)
row = next(p for p in providers if p["slug"] == "custom:slow-local")
assert row["models"] == ["local-model"]
assert row["total_models"] == 1
@@ -245,6 +343,64 @@ def test_is_aggregator_leaves_unknown_provider_non_aggregator():
assert providers_mod.is_aggregator("not-a-provider") is False
def test_switch_model_does_not_send_ollama_headers_to_unrelated_custom_endpoint(monkeypatch):
"""A custom endpoint must not inherit headers from configured Ollama."""
seen_headers = []
validation_headers = []
def fake_native_detection(provider, base_url, headers=None):
seen_headers.append(headers)
return True
def fake_validation(*args, **kwargs):
validation_headers.append(kwargs.get("headers"))
return _MOCK_VALIDATION
monkeypatch.setattr(
"hermes_cli.models.should_use_ollama_native_catalog",
fake_native_detection,
)
monkeypatch.setattr(
"hermes_cli.models._get_ollama_request_headers",
lambda: {"Authorization": "Bearer configured-ollama-secret"},
)
monkeypatch.setattr(
"hermes_cli.models._get_provider_config_dict",
lambda provider: (
{"base_url": "https://trusted-ollama.example:11434"}
if provider == "ollama"
else {}
),
)
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda **kwargs: {
"api_key": "custom-key",
"base_url": "https://attacker.example:11434/v1",
"api_mode": "chat_completions",
},
)
monkeypatch.setattr("hermes_cli.models.validate_requested_model", fake_validation)
monkeypatch.setattr("hermes_cli.model_switch.get_model_info", lambda *a, **k: None)
monkeypatch.setattr("hermes_cli.model_switch.get_model_capabilities", lambda *a, **k: None)
result = switch_model(
raw_input="new-model",
current_provider="custom",
current_model="old-model",
current_base_url="https://attacker.example:11434/v1",
current_api_key="custom-key",
explicit_provider="",
user_providers={},
custom_providers=[],
)
assert result.success is True
assert seen_headers == [{}]
assert validation_headers == [None]
def test_is_routing_aggregator_excludes_flat_namespace_resellers():
"""opencode-go / opencode-zen stay ``is_aggregator=True`` (model-switch
relies on it to search their flat bare-name catalog), but they are NOT
@@ -308,6 +464,395 @@ def test_picker_selection_resolves_named_custom_provider_model_id(monkeypatch):
def test_list_groups_same_name_custom_providers_into_one_row(monkeypatch):
"""Multiple custom_providers entries sharing a name should produce one row
with all models collected, not N duplicate rows."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
monkeypatch.setattr("hermes_cli.models.fetch_api_models", lambda *a, **k: [])
providers = list_authenticated_providers(
current_provider="openrouter",
user_providers={},
custom_providers=[
{"name": "Ollama Cloud", "base_url": "https://ollama.com/v1", "model": "qwen3-coder:480b-cloud"},
{"name": "Ollama Cloud", "base_url": "https://ollama.com/v1", "model": "glm-5.1:cloud"},
{"name": "Ollama Cloud", "base_url": "https://ollama.com/v1", "model": "kimi-k2.5"},
{"name": "Ollama Cloud", "base_url": "https://ollama.com/v1", "model": "minimax-m2.7:cloud"},
{"name": "Moonshot", "base_url": "https://api.moonshot.ai/v1", "model": "kimi-k2-thinking"},
],
max_models=50,
)
ollama_rows = [p for p in providers if p["name"] == "Ollama Cloud"]
assert len(ollama_rows) == 1, f"Expected 1 Ollama Cloud row, got {len(ollama_rows)}"
assert ollama_rows[0]["models"] == [
"qwen3-coder:480b-cloud", "glm-5.1:cloud", "kimi-k2.5", "minimax-m2.7:cloud"
]
assert ollama_rows[0]["total_models"] == 4
moonshot_rows = [p for p in providers if p["name"] == "Moonshot"]
assert len(moonshot_rows) == 1
assert moonshot_rows[0]["models"] == ["kimi-k2-thinking"]
def test_list_deduplicates_same_model_in_group(monkeypatch):
"""Duplicate model entries under the same provider name should not produce
duplicate entries in the models list."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
monkeypatch.setattr("hermes_cli.models.fetch_api_models", lambda *a, **k: [])
providers = list_authenticated_providers(
current_provider="openrouter",
user_providers={},
custom_providers=[
{"name": "MyProvider", "base_url": "http://localhost:11434/v1", "model": "llama3", "discover_models": False},
{"name": "MyProvider", "base_url": "http://localhost:11434/v1", "model": "llama3", "discover_models": False},
{"name": "MyProvider", "base_url": "http://localhost:11434/v1", "model": "mistral", "discover_models": False},
],
max_models=50,
)
my_rows = [p for p in providers if p["name"] == "MyProvider"]
assert len(my_rows) == 1
assert my_rows[0]["models"] == ["llama3", "mistral"]
assert my_rows[0]["total_models"] == 2
def test_custom_provider_no_key_singular_model_still_probes_live_models(monkeypatch):
"""A singular ``model:`` is the active selection, not an explicit catalog.
No-key local OpenAI-compatible endpoints such as llama.cpp should still be
probed so /model matches the terminal ``hermes model`` flow. Ollama-native
discovery is covered separately with a fake ``/api/tags`` server.
"""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
calls = []
def fake_fetch_api_models(api_key, base_url, **kwargs):
calls.append((api_key, base_url, kwargs))
return ["llama3", "mistral", "qwen3-coder"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fake_fetch_api_models)
providers = list_authenticated_providers(
current_provider="openai-codex",
user_providers={},
custom_providers=[
{
# Keep this generic and off Ollama's default :11434: this case
# covers /v1/models probing, not native /api/tags discovery.
"name": "Local llama.cpp",
"base_url": "http://localhost:8080/v1",
"model": "llama3",
}
],
max_models=50,
)
assert calls == [
("", "http://localhost:8080/v1", {
"timeout": 5.0,
"api_mode": None,
"headers": None,
})
]
row = next(p for p in providers if p["name"] == "Local llama.cpp")
assert row["models"] == ["llama3", "mistral", "qwen3-coder"]
assert row["total_models"] == 3
def test_custom_provider_model_metadata_dict_still_probes(monkeypatch):
"""Dict-shaped ``models:`` is metadata, not an explicit allowlist."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
calls = []
def fetch(*args, **kwargs):
calls.append((args, kwargs))
return ["unexpected-live-model"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fetch)
providers = list_authenticated_providers(
current_provider="custom:local-ollama",
user_providers={},
custom_providers=[
{
"name": "Local Ollama",
"base_url": "http://localhost:11434/v1",
"model": "llama3",
"models": {"llama3": {}},
}
],
)
row = next(p for p in providers if p["name"] == "Local Ollama")
assert calls == [
(
("", "http://localhost:11434/v1"),
{"timeout": 5.0, "api_mode": None, "headers": None},
)
]
assert row["models"] == ["unexpected-live-model"]
def test_custom_provider_group_explicit_duplicate_skips_probe(monkeypatch):
"""A later grouped entry can explicitly narrow to an existing model."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
calls = []
def fetch(*args, **kwargs):
calls.append((args, kwargs))
return ["unexpected-live-model"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fetch)
providers = list_authenticated_providers(
current_provider="custom:local-ollama",
user_providers={},
custom_providers=[
{
"name": "Local Ollama",
"base_url": "http://localhost:11434/v1",
"model": "llama3",
},
{
"name": "Local Ollama",
"base_url": "http://localhost:11434/v1",
"models": ["llama3"],
},
],
)
row = next(p for p in providers if p["name"] == "Local Ollama")
assert calls == []
assert row["models"] == ["llama3"]
def test_custom_provider_current_only_probe_respects_explicit_catalog(monkeypatch):
"""Normal GUI opens probe only the active singular-only provider."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
calls = []
def fetch(api_key, base_url, **kwargs):
calls.append((api_key, base_url, kwargs))
return ["live-a", "live-b"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fetch)
providers = list_authenticated_providers(
current_provider="custom:active",
current_base_url="http://active.local/v1",
user_providers={},
custom_providers=[
{
"name": "Active",
"base_url": "http://active.local/v1",
"model": "seed",
},
{
"name": "Offline",
"base_url": "http://offline.local/v1",
"model": "offline-seed",
},
{
"name": "Static",
"base_url": "http://static.local/v1",
"model": "only",
"models": ["only"],
},
],
probe_custom_providers=False,
probe_current_custom_provider=True,
)
assert calls == [
(
"",
"http://active.local/v1",
{"timeout": 5.0, "api_mode": None, "headers": None},
)
]
rows = {row["name"]: row for row in providers if row.get("is_user_defined")}
assert rows["Active"]["models"] == ["live-a", "live-b"]
assert rows["Offline"]["models"] == ["offline-seed"]
assert rows["Static"]["models"] == ["only"]
def test_custom_provider_current_explicit_catalog_skips_probe(monkeypatch):
"""Current-only GUI probing must still honor an explicit catalog."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
calls = []
def fetch(*args, **kwargs):
calls.append((args, kwargs))
return ["unexpected-live-model"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fetch)
providers = list_authenticated_providers(
current_provider="custom:static",
current_base_url="http://static.local/v1",
user_providers={},
custom_providers=[
{
"name": "Static",
"base_url": "http://static.local/v1",
"model": "only",
"models": ["only"],
}
],
probe_custom_providers=False,
probe_current_custom_provider=True,
)
assert calls == []
row = next(p for p in providers if p["name"] == "Static")
assert row["is_current"] is True
assert row["models"] == ["only"]
def test_custom_provider_empty_explicit_list_allows_probe(monkeypatch):
"""An empty ``models:`` declaration is not an explicit catalog."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
calls = []
def fetch(api_key, base_url, **kwargs):
calls.append((api_key, base_url, kwargs))
return ["live-a", "live-b"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fetch)
providers = list_authenticated_providers(
current_provider="custom:local",
user_providers={},
custom_providers=[
{
"name": "Local",
"base_url": "http://local.test/v1",
"model": "seed",
"models": [],
}
],
)
assert calls == [
(
"",
"http://local.test/v1",
{"timeout": 5.0, "api_mode": None, "headers": None},
)
]
row = next(p for p in providers if p["name"] == "Local")
assert row["models"] == ["live-a", "live-b"]
def test_list_enumerates_dict_format_models_alongside_default(monkeypatch):
"""custom_providers entry with dict-format ``models:`` plus singular
``model:`` should surface the default and every dict key.
Regression: Hermes's own writer stores configured models as a dict
keyed by model id, but the /model picker previously only honored the
singular ``model:`` field, so multi-model custom providers appeared
to have only the active model.
"""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
providers = list_authenticated_providers(
current_provider="openai-codex",
user_providers={},
custom_providers=[
{
"name": "DeepSeek",
"base_url": "https://api.deepseek.com",
"api_mode": "chat_completions",
"model": "deepseek-chat",
"models": {
"deepseek-chat": {"context_length": 128000},
"deepseek-reasoner": {"context_length": 128000},
},
}
],
max_models=50,
)
ds_rows = [p for p in providers if p["name"] == "DeepSeek"]
assert len(ds_rows) == 1
assert ds_rows[0]["models"] == ["deepseek-chat", "deepseek-reasoner"]
assert ds_rows[0]["total_models"] == 2
def test_list_enumerates_dict_format_models_without_singular_model(monkeypatch):
"""Dict-format ``models:`` with no singular ``model:`` should still
enumerate every dict key (previously the picker reported 0 models)."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
providers = list_authenticated_providers(
current_provider="openai-codex",
user_providers={},
custom_providers=[
{
"name": "Thor",
"base_url": "http://thor.lab:8337/v1",
"models": {
"gemma-4-26B-A4B-it-MXFP4_MOE": {"context_length": 262144},
"Qwen3.5-35B-A3B-MXFP4_MOE": {"context_length": 262144},
"gemma-4-31B-it-Q4_K_M": {"context_length": 262144},
},
}
],
max_models=50,
)
thor_rows = [p for p in providers if p["name"] == "Thor"]
assert len(thor_rows) == 1
assert set(thor_rows[0]["models"]) == {
"gemma-4-26B-A4B-it-MXFP4_MOE",
"Qwen3.5-35B-A3B-MXFP4_MOE",
"gemma-4-31B-it-Q4_K_M",
}
assert thor_rows[0]["total_models"] == 3
def test_list_dedupes_dict_model_matching_singular_default(monkeypatch):
"""When the singular ``model:`` is also a key in the ``models:`` dict,
it must appear exactly once in the picker."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
providers = list_authenticated_providers(
current_provider="openai-codex",
user_providers={},
custom_providers=[
{
"name": "DeepSeek",
"base_url": "https://api.deepseek.com",
"model": "deepseek-chat",
"models": {
"deepseek-chat": {"context_length": 128000},
"deepseek-reasoner": {"context_length": 128000},
},
}
],
max_models=50,
)
ds_rows = [p for p in providers if p["name"] == "DeepSeek"]
assert ds_rows[0]["models"].count("deepseek-chat") == 1
assert ds_rows[0]["models"] == ["deepseek-chat", "deepseek-reasoner"]
# ─────────────────────────────────────────────────────────────────────────────
@@ -315,6 +860,94 @@ def test_picker_selection_resolves_named_custom_provider_model_id(monkeypatch):
# ─────────────────────────────────────────────────────────────────────────────
def test_list_authenticated_providers_groups_same_endpoint(monkeypatch):
"""Multiple custom_providers entries sharing a base_url+api_key must be
returned as a single picker row with all their models merged."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
providers = list_authenticated_providers(
current_provider="custom",
current_base_url="http://localhost:11434/v1",
user_providers={},
custom_providers=[
{"name": "Ollama — MiniMax M2.7", "base_url": "http://localhost:11434/v1",
"api_key": "ollama", "model": "minimax-m2.7", "discover_models": False},
{"name": "Ollama — GLM 5.1", "base_url": "http://localhost:11434/v1",
"api_key": "ollama", "model": "glm-5.1", "discover_models": False},
{"name": "Ollama — Qwen3-coder", "base_url": "http://localhost:11434/v1",
"api_key": "ollama", "model": "qwen3-coder", "discover_models": False},
],
max_models=50,
probe_custom_providers=False,
)
custom_groups = [p for p in providers if p.get("is_user_defined")]
assert len(custom_groups) == 1, (
"Expected 1 group for shared endpoint, got "
f"{[p['slug'] for p in custom_groups]}"
)
group = custom_groups[0]
assert set(group["models"]) == {"minimax-m2.7", "glm-5.1", "qwen3-coder"}
assert group["total_models"] == 3
# Per-model suffix stripped from display name
assert group["name"] == "Ollama"
def test_list_authenticated_providers_current_endpoint_uses_current_slug(monkeypatch):
"""When current_base_url matches the grouped endpoint, the slug must
equal current_provider so picker selection routes through the live
credential pipeline — provided current_provider is a real slug, not
the corrupt bare "custom" (see #17478)."""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr(providers_mod, "HERMES_OVERLAYS", {})
providers = list_authenticated_providers(
current_provider="custom:ollama",
current_base_url="http://localhost:11434/v1",
user_providers={},
custom_providers=[
{"name": "Ollama — GLM 5.1", "base_url": "http://localhost:11434/v1",
"api_key": "ollama", "model": "glm-5.1"},
],
max_models=50,
)
matches = [p for p in providers if p.get("is_user_defined")]
assert len(matches) == 1
group = matches[0]
assert group["slug"] == "custom:ollama"
assert group["is_current"] is True
def test_picker_endpoint_authorization_overrides_inferred_bearer(monkeypatch):
from hermes_cli.model_switch import _fetch_picker_live_models
captured: dict[str, str] = {}
def fake_native(_url, *, timeout, headers):
captured.update(headers or {})
return ["model-a"]
monkeypatch.setattr("hermes_cli.models.should_use_ollama_native_catalog", lambda *a, **k: True)
monkeypatch.setattr("hermes_cli.models.fetch_ollama_local_models", fake_native)
result = _fetch_picker_live_models(
"endpoint-key",
"http://127.0.0.1:11434/v1",
"ollama",
False,
headers={"authorization": "Api-Key explicit"},
)
assert result == ["model-a"]
auth_headers = {
key: value
for key, value in captured.items()
if key.lower() == "authorization"
}
assert auth_headers == {"authorization": "Api-Key explicit"}
def test_list_authenticated_providers_bare_custom_slug_recovers(monkeypatch):
"""Regression for #17478: when a prior failed switch left the bare
literal "custom" in model.provider, the picker must NOT propagate
@@ -596,6 +1229,20 @@ def test_lmstudio_picker_skips_probe_when_not_configured(monkeypatch):
assert "base_url" not in captured
def test_custom_providers_uses_live_models_for_multi_model_endpoint(monkeypatch):
"""Custom providers with api_key + base_url should prefer live /models.