fix: credential pool fallback for custom providers (#4247)

Custom/aliased providers with keys in the credential pool (added via
hermes auth add) were reported unconfigured in the WebUI because provider
detection only checked env vars. Add _has_explicit_pool_credentials() that
checks the pool while filtering the ambient gh_cli entry (so Copilot doesn't
falsely show configured on any machine with gh installed), cache-routed to
avoid the cold per-provider load_pool latency, and resolve provider aliases
before load_pool() at every retrieval site so detection and retrieval read
the same pool (glm->zai, github->copilot, etc.). /api/models/live wrapped in
profile scope; getattr-safe runtime_api_key for OAuth-only entries.

Co-authored-by: akrhin <akrhin@users.noreply.github.com>
This commit is contained in:
nesquena-hermesandakrhin committed 2026-06-16 17:44:10 +00:00
1 parent 8fff266aaa
commit b81a92ecd5
4 files changed
+476 -7

No files matched your search

+61 -6
View File
@@ -4037,6 +4037,44 @@ def _static_models_catalog_without_live_probes() -> dict:
# only changes when the user adds/removes credentials, which is rare; a 24h TTL
# is plenty safe and ensures get_available_models() cold paths are fast.
_CREDENTIAL_POOL_CACHE: dict[str, tuple[float, "CredentialPool"]] = {} # noqa: F821 forward-ref string annotation, resolved at runtime # pid -> (ts, pool)
def _has_explicit_pool_credentials(provider_id: str) -> bool:
"""Return True when the credential pool has at least one non-ambient entry
for *provider_id* (i.e. not a gh-cli / GITHUB_TOKEN auto-detect).
Reuses ``_CREDENTIAL_POOL_CACHE`` so that callers on hot paths (provider
detection, model listing, live-model fetch) don't pay the ~10s load_pool
cost more than once per TTL window.
"""
try:
from agent.credential_pool import load_pool as _load_pool
_pid = _resolve_provider_alias(provider_id)
_cached = _CREDENTIAL_POOL_CACHE.get(_pid)
if _cached is not None:
_cp_ts, _cp_pool = _cached
if (time.time() - _cp_ts) < 86400.0:
_all_entries = _cp_pool.entries()
else:
_cp_pool = _load_pool(_pid)
_CREDENTIAL_POOL_CACHE[_pid] = (time.time(), _cp_pool)
_all_entries = _cp_pool.entries()
else:
_cp_pool = _load_pool(_pid)
_CREDENTIAL_POOL_CACHE[_pid] = (time.time(), _cp_pool)
_all_entries = _cp_pool.entries()
return any(
not _is_ambient_gh_cli_entry(
str(getattr(e, "source", "") or ""),
str(getattr(e, "label", "") or ""),
str(getattr(e, "key_source", "") or ""),
)
for e in _all_entries
)
except ImportError:
return False
_provider_models_invalidated_ts: dict[str, float] = {} # provider_id -> timestamp of last invalidation
# Disk-backed in-memory cache for get_available_models().
@@ -5426,13 +5464,30 @@ def get_available_models(*, prefer_cache: bool = False) -> dict:
_named_custom_groups[_slug] = (_cp_name, [])
_cp_base_url = str(_cp.get("base_url") or "").strip()
if _slug and _cp_base_url:
_cp_api_key = str(_cp.get("api_key") or "").strip()
if not _cp_api_key:
_cp_key_env = str(_cp.get("key_env") or "").strip()
if _cp_key_env:
_cp_api_key = str(os.getenv(_cp_key_env) or "").strip()
_cp_api_key = str(_cp.get("api_key") or "").strip()
if not _cp_api_key:
_cp_key_env = str(_cp.get("key_env") or "").strip()
if _cp_key_env:
_cp_api_key = str(os.getenv(_cp_key_env) or "").strip()
# Fallback: check credential pool for both api_key and base_url
if (not _cp_api_key or not _cp_base_url) and _slug:
try:
from api.config import _has_explicit_pool_credentials
if _has_explicit_pool_credentials(_slug):
from agent.credential_pool import load_pool
_resolved = _resolve_provider_alias(_slug)
_pool = load_pool(_resolved)
if _pool:
_entry = _pool.select()
if _entry:
if not _cp_api_key:
_cp_api_key = _entry.runtime_api_key
if not _cp_base_url:
_cp_base_url = str(getattr(_entry, "base_url", "") or "").strip()
except ImportError:
pass
if _slug and _cp_base_url:
# Check if user has configured models in config.yaml —
# configured models take priority over live /v1/models
# discovery (same as hermes-agent model_switch.py Section 4
+38
View File
@@ -991,6 +991,19 @@ def _provider_has_key(provider_id: str) -> bool:
return True
if _provider_value_counts_as_api_key(provider_id, os.getenv(alias)):
return True
# Check credential pool — covers custom providers registered via
# `hermes auth add` which store keys in auth.json (not config.yaml).
# Must be outside the `if env_var:` block above: custom providers
# (custom:bothub, etc.) have no env var, so that block is skipped.
# Uses the cached _has_explicit_pool_credentials helper which also
# filters gh-cli / GITHUB_TOKEN ambient entries so copilot doesn't
# appear just because `gh` is installed.
try:
from api.config import _has_explicit_pool_credentials
if _has_explicit_pool_credentials(provider_id):
return True
except ImportError:
pass
cfg = get_config()
# Check model.api_key — only match if this provider is the active one.
@@ -1068,6 +1081,23 @@ def _get_provider_api_key(provider_id: str) -> str | None:
return os.getenv(cp_key[2:-1], "").strip() or None
if _provider_value_counts_as_api_key(provider_id, cp_key):
return cp_key
# Fallback: try credential pool (e.g. bothub key stored via auth.json)
try:
from api.config import _has_explicit_pool_credentials, _resolve_provider_alias
if _has_explicit_pool_credentials(provider_id):
from agent.credential_pool import load_pool
# Must resolve alias here too: _has_explicit_pool_credentials does it
# internally, but load_pool sees the original unresolved provider_id.
_resolved = _resolve_provider_alias(provider_id)
pool = load_pool(_resolved)
if pool:
entry = pool.select()
if entry:
key = getattr(entry, "runtime_api_key", "") or getattr(entry, "access_token", "")
if key:
return key
except ImportError:
pass
return None
@@ -2343,6 +2373,14 @@ def get_providers() -> dict[str, Any]:
if cp_api_key.startswith("${") and cp_api_key.endswith("}"):
env_var = cp_api_key[2:-1]
cp_has_key = bool(os.getenv(env_var, "").strip())
# Fallback: check credential pool (key added via hermes auth add)
if not cp_has_key:
try:
from api.config import _has_explicit_pool_credentials
if _has_explicit_pool_credentials(cp_id):
cp_has_key = True
except ImportError:
pass
providers.append({
"id": cp_id,
"display_name": cp_name,
+20 -1
View File
@@ -6737,7 +6737,9 @@ def handle_get(handler, parsed) -> bool:
return j(handler, get_available_models())
if parsed.path == "/api/models/live":
return _handle_live_models(handler, parsed)
from api.profiles import profile_env_for_active_request
with profile_env_for_active_request("/api/models/live", logger_override=logger):
return _handle_live_models(handler, parsed)
# ── Auxiliary models (GET/POST) ──
if parsed.path == "/api/model/auxiliary":
@@ -12427,6 +12429,23 @@ def _handle_live_models(handler, parsed):
_model_cfg = cfg.get("model", {})
_base_url = _model_cfg.get("base_url")
_api_key = _model_cfg.get("api_key")
# Fallback: try credential pool for base_url + api_key
if (not _base_url or not _api_key) and provider.startswith("custom:"):
try:
from api.config import _has_explicit_pool_credentials, _resolve_provider_alias
if _has_explicit_pool_credentials(provider):
from agent.credential_pool import load_pool as _lpool
_resolved = _resolve_provider_alias(provider)
_lm_pool = _lpool(_resolved)
if _lm_pool:
_lm_entry = _lm_pool.select()
if _lm_entry:
if not _api_key:
_api_key = _lm_entry.runtime_api_key
if not _base_url:
_base_url = str(getattr(_lm_entry, "base_url", "") or "").strip()
except ImportError:
pass
if _base_url and _api_key:
try:
import urllib.request
+357
View File
@@ -49,6 +49,9 @@ def _install_fake_hermes_cli(monkeypatch, *, with_load_pool: bool = False, pool_
self.label = d.get("label", "")
self.key_source = d.get("key_source", "")
self.id = d.get("id", "")
self.runtime_api_key = (d.get("runtime_api_key") or d.get("access_token") or "")
self.access_token = (d.get("access_token") or "")
self.base_url = d.get("base_url", "")
class _FakePool:
def __init__(self, entries_list):
@@ -57,6 +60,9 @@ def _install_fake_hermes_cli(monkeypatch, *, with_load_pool: bool = False, pool_
def entries(self):
return self._entries
def select(self):
return self._entries[0] if self._entries else None
def _fake_load_pool(pid):
# Return ALL entries without filtering — mirrors the real load_pool()
# which does NOT suppress ambient gh-cli tokens on its own.
@@ -593,3 +599,354 @@ def test_fallback_path_resolves_alias_when_load_pool_unavailable(monkeypatch, tm
assert "Google" not in groups, (
f"Raw alias name must not leak when fallback path runs; got {list(groups)}"
)
# ── New tests for credential-pool changes (code-review #4247 follow-ups) ──
# --- Regression: ambient gh_cli MUST NOT mark copilot as configured ---
def test_ambient_gh_cli_not_detectable_by_provider_has_key(monkeypatch, tmp_path):
"""Regression: _provider_has_key('copilot') must return False when only
ambient gh auth token (gh_cli) is in the pool."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"copilot": [
{"id": "ambient-1", "label": "gh auth token", "source": "gh_cli", "auth_type": "api_key"},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
from api.providers import _provider_has_key
assert _provider_has_key("copilot") is False, (
"copilot must NOT be marked as configured when only ambient gh_cli token exists"
)
def test_ambient_gh_env_not_detectable_by_provider_has_key(monkeypatch, tmp_path):
"""Regression: _provider_has_key('copilot') must return False when only
GITHUB_TOKEN env-variable entry is in the pool."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"copilot": [
{"id": "env-1", "label": "env-token", "source": "env:github_token", "auth_type": "api_key"},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
from api.providers import _provider_has_key
assert _provider_has_key("copilot") is False, (
"copilot must NOT be marked as configured when only GITHUB_TOKEN env entry exists"
)
def test_ambient_key_source_not_detectable_by_provider_has_key(monkeypatch, tmp_path):
"""Regression: _provider_has_key('copilot') must return False when only
key_source='gh auth token' entries exist."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"copilot": [
{
"id": "key-src-1", "label": "copilot-pat",
"source": "manual", "key_source": "gh auth token",
"auth_type": "api_key",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
from api.providers import _provider_has_key
assert _provider_has_key("copilot") is False, (
"copilot must NOT be marked as configured when only key_source=gh auth token entry exists"
)
# --- Positive: custom:* providers with explicit credentials are detected ---
def test_custom_provider_explicit_credential_detected_by_provider_has_key(monkeypatch, tmp_path):
"""Positive: custom:bothub with explicit manual entry must show _provider_has_key=True."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"custom:bothub": [
{
"id": "bothub-1",
"label": "bothub-key",
"source": "manual",
"auth_type": "api_key",
"runtime_api_key": "sk-bh...test",
"base_url": "https://bothub.chat/v1",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
from api.providers import _provider_has_key
assert _provider_has_key("custom:bothub") is True, (
"custom:bothub with explicit pool entry must be detected as configured"
)
def test_custom_provider_detected_by_get_providers(monkeypatch, tmp_path):
"""Positive: custom:bothub must appear in get_providers() with has_key=True."""
config._CREDENTIAL_POOL_CACHE.clear()
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"custom:bothub": [
{
"id": "bothub-prov-1",
"label": "bothub-key",
"source": "manual",
"auth_type": "api_key",
"runtime_api_key": "sk-bh...test",
"base_url": "https://bothub.chat/v1",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
# Custom providers need to be in config.yaml for get_providers() to list them
old_cfg = dict(config.cfg)
old_mtime = config._cfg_mtime
config.cfg.clear()
config.cfg["model"] = {"provider": "custom:bothub"}
config.cfg["custom_providers"] = [
{
"name": "bothub",
"api_key": "sk-bh...test",
"base_url": "https://bothub.chat/v1",
"display_name": "Bothub",
},
]
try:
config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
except Exception:
config._cfg_mtime = 0.0
config.invalidate_models_cache()
try:
from api.providers import get_providers
providers = get_providers()
bothub = None
for p in providers["providers"]:
if p["id"] == "custom:bothub":
bothub = p
break
assert bothub is not None, (
f"custom:bothub must appear in get_providers(); "
f"got {[p['id'] for p in providers['providers']]}"
)
assert bothub["has_key"] is True, "custom:bothub must show has_key=True"
finally:
config.invalidate_models_cache()
config.cfg.clear()
config.cfg.update(old_cfg)
config._cfg_mtime = old_mtime
def test_custom_provider_detected_by_get_available_models(monkeypatch, tmp_path):
"""Positive: custom provider with explicit pool credentials must appear as a group."""
config._CREDENTIAL_POOL_CACHE.clear()
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"custom:bothub": [
{
"id": "bothub-mod-1",
"label": "bothub-key",
"source": "manual",
"auth_type": "api_key",
"runtime_api_key": "sk-bh...test",
"base_url": "https://bothub.chat/v1",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
# Need both auth.json (for pool) and custom_providers config (for model enumeration)
(tmp_path / "auth.json").write_text(json.dumps({
"version": 1,
"providers": {},
"active_provider": "openai-codex",
}), encoding="utf-8")
old_cfg = dict(config.cfg)
old_mtime = config._cfg_mtime
config.cfg.clear()
config.cfg["model"] = {}
config.cfg["custom_providers"] = [
{
"name": "bothub",
"api_key": "sk-bh...test",
"base_url": "https://bothub.chat/v1",
"display_name": "Bothub",
},
]
try:
config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
except Exception:
config._cfg_mtime = 0.0
config.invalidate_models_cache()
try:
result = config.get_available_models()
finally:
config.invalidate_models_cache()
config.cfg.clear()
config.cfg.update(old_cfg)
config._cfg_mtime = old_mtime
groups = _group_by_provider(result)
assert "bothub" in groups, (
f"Bothub must appear as a group in get_available_models; got {list(groups)}"
)
# Models may be empty since there's no real endpoint in test — the key is
# that the group appears at all (credential pool detection works).
# --- OAuth token order: runtime_api_key must be preferred over access_token ---
def test_get_provider_api_key_prefers_runtime_api_key_over_access_token(monkeypatch, tmp_path):
"""_get_provider_api_key must return runtime_api_key when both access_token
and runtime_api_key are present (runtime_api_key has priority)."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"custom:bothub": [
{
"id": "tok-order-1",
"label": "bothub-key",
"source": "manual",
"auth_type": "api_key",
"access_token": "sk-old-access-token",
"runtime_api_key": "sk-preferred-runtime-key",
"base_url": "https://bothub.chat/v1",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
from api.providers import _get_provider_api_key
key = _get_provider_api_key("custom:bothub")
assert key == "sk-preferred-runtime-key", (
f"runtime_api_key must take priority; got {key!r}"
)
def test_get_provider_api_key_falls_back_to_access_token(monkeypatch, tmp_path):
"""_get_provider_api_key must fall back to access_token when runtime_api_key is absent."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"custom:bothub": [
{
"id": "tok-fallback-1",
"label": "bothub-key",
"source": "manual",
"auth_type": "api_key",
"access_token": "sk-fallback-access-token",
"base_url": "https://bothub.chat/v1",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
from api.providers import _get_provider_api_key
key = _get_provider_api_key("custom:bothub")
assert key == "sk-fallback-access-token", (
f"access_token must be used when runtime_api_key absent; got {key!r}"
)
# --- _has_explicit_pool_credentials unit tests ---
def test_has_explicit_pool_credentials_ambient_only_is_false(monkeypatch, tmp_path):
"""_has_explicit_pool_credentials must return False when only ambient entries exist."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"copilot": [
{"id": "u-amb-1", "label": "gh auth token", "source": "gh_cli", "auth_type": "api_key"},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
config._CREDENTIAL_POOL_CACHE.clear()
from api.config import _has_explicit_pool_credentials
assert _has_explicit_pool_credentials("copilot") is False
def test_has_explicit_pool_credentials_explicit_is_true(monkeypatch, tmp_path):
"""_has_explicit_pool_credentials must return True when at least one explicit entry exists."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"custom:bothub": [
{
"id": "u-exp-1", "label": "bothub-key", "source": "manual",
"auth_type": "api_key", "runtime_api_key": "sk-test",
"base_url": "https://bothub.chat/v1",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
config._CREDENTIAL_POOL_CACHE.clear()
from api.config import _has_explicit_pool_credentials
assert _has_explicit_pool_credentials("custom:bothub") is True
def test_has_explicit_pool_credentials_mixed_is_true(monkeypatch, tmp_path):
"""_has_explicit_pool_credentials must return True when both ambient and explicit entries exist."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
"copilot": [
{"id": "u-mix-amb", "label": "gh auth token", "source": "gh_cli", "auth_type": "api_key"},
{
"id": "u-mix-exp", "label": "explicit-pat", "source": "manual",
"auth_type": "api_key", "runtime_api_key": "sk-pat",
"base_url": "https://api.githubcopilot.com",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
config._CREDENTIAL_POOL_CACHE.clear()
from api.config import _has_explicit_pool_credentials
assert _has_explicit_pool_credentials("copilot") is True
def test_has_explicit_pool_credentials_resolves_alias(monkeypatch, tmp_path):
"""_has_explicit_pool_credentials must resolve provider aliases (google -> gemini)."""
_install_fake_hermes_cli(monkeypatch, with_load_pool=True, pool_data={
# Pool data is stored under canonical ID 'gemini' after alias resolution
"gemini": [
{
"id": "u-alias-1", "label": "gemini-key", "source": "manual",
"auth_type": "api_key", "runtime_api_key": "sk-gemini",
"base_url": "https://generativelanguage.googleapis.com",
},
],
})
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
config._CREDENTIAL_POOL_CACHE.clear()
# Should work with both alias 'google' and canonical 'gemini'
from api.config import _has_explicit_pool_credentials
assert _has_explicit_pool_credentials("google") is True, (
"alias 'google' should resolve to canon 'gemini' and find pool entry"
)
assert _has_explicit_pool_credentials("gemini") is True, (
"canon 'gemini' should find pool entry directly"
)
def test_has_explicit_pool_credentials_import_error_is_false(monkeypatch, tmp_path):
"""_has_explicit_pool_credentials must return False when load_pool not available."""
_install_fake_hermes_cli(monkeypatch) # no with_load_pool — agent.credential_pool is removed
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
config._CREDENTIAL_POOL_CACHE.clear()
from api.config import _has_explicit_pool_credentials
assert _has_explicit_pool_credentials("copilot") is False