From 55f47d8e9589837802998939a4153ebd811ca754 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:31:26 -0700 Subject: [PATCH] refactor(config): extract custom-provider resolution to hermes_cli/config_providers.py --- hermes_cli/config.py | 674 ++------------------------------- hermes_cli/config_providers.py | 668 ++++++++++++++++++++++++++++++++ 2 files changed, 696 insertions(+), 646 deletions(-) create mode 100644 hermes_cli/config_providers.py diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 3bfe8dcb39..8ba7824c07 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -20,7 +20,6 @@ from dataclasses import dataclass from pathlib import Path from typing import Dict, Any, Optional, List, Tuple, Set -from hermes_cli.route_identity import normalize_route_base_url from hermes_cli.secret_prompt import masked_secret_prompt logger = logging.getLogger(__name__) @@ -869,6 +868,34 @@ def _ensure_hermes_home_managed(home: Path): # ============================================================================= from hermes_cli.config_defaults import DEFAULT_CONFIG, OPTIONAL_ENV_VARS # noqa: F401 +from hermes_cli.config_providers import ( # noqa: F401 (re-exported; callers/tests use hermes_cli.config.) + _API_MODE_ALIASES, + _CAMEL_ALIASES, + _KNOWN_PROVIDER_KEYS, + _PROVIDER_NORMALIZE_WARNED, + _canonical_api_mode, + _coerce_ssl_verify, + _custom_provider_entry_to_provider_config, + _entries_for_route, + _normalize_custom_provider_entry, + _normalize_provider_models, + _pick_provider_base_url, + _route_model_cfg, + _warn_once_per_provider, + apply_custom_provider_extra_headers_to_client_kwargs, + apply_custom_provider_tls_to_client_kwargs, + coerce_provider_id, + find_provider_entry, + get_compatible_custom_providers, + get_custom_provider_context_length, + get_custom_provider_extra_headers, + get_custom_provider_model_capability, + get_custom_provider_tls_settings, + is_provider_enabled, + normalize_extra_headers, + providers_dict_to_custom_providers, + stringify_provider_map, +) # ============================================================================= # Config Migration System @@ -1226,631 +1253,6 @@ def get_missing_skill_config_vars() -> List[Dict[str, Any]]: return missing -# ``_normalize_custom_provider_entry`` runs on every ``load_picker_context()`` -# call (i.e. per interactive picker/inventory request), so any warning it emits -# fires repeatedly for the same static config. Deduplicate per (provider, -# signature): on Windows a repeated-warning storm contends on -# ``concurrent-log-handler``'s cross-process rotation lock and can peg a core / -# stall the gateway/serve event loop. The cache lives for the process lifetime. -_PROVIDER_NORMALIZE_WARNED: set = set() - - -def _warn_once_per_provider( - provider_key: str, signature: str, msg: str, *args: Any -) -> None: - """Emit ``logger.warning(msg, *args)`` at most once per (provider, signature).""" - dedup_key = (provider_key or "?", signature) - if dedup_key in _PROVIDER_NORMALIZE_WARNED: - return - _PROVIDER_NORMALIZE_WARNED.add(dedup_key) - logger.warning(msg, *args) - - -_API_MODE_ALIASES = { - # Values accepted by earlier releases (and natural spellings) mapped to - # the canonical transport names consumed by agent_init. Before this map - # existed, an unrecognized api_mode was silently ignored and the - # transport fell through to hostname-based guessing, so a config that - # said ``api_mode: openai`` (valid on older releases) could flip to - # ``codex_responses`` after an update and break the provider (#66543 - # discussion; observed live against api.actual.inc). - "openai": "chat_completions", - "openai_chat": "chat_completions", - "openai-chat": "chat_completions", - "chat-completions": "chat_completions", - "chatcompletions": "chat_completions", - "responses": "codex_responses", - "openai_responses": "codex_responses", - "openai-responses": "codex_responses", - "anthropic": "anthropic_messages", - "anthropic-messages": "anthropic_messages", - "messages": "anthropic_messages", - "bedrock": "bedrock_converse", - "bedrock-converse": "bedrock_converse", -} - - -def _canonical_api_mode(api_mode: str) -> str: - """Map legacy/alias ``api_mode`` spellings to canonical transport names. - - Unknown values pass through unchanged (callers keep their existing fall-through behavior); known - aliases are rewritten so downstream consumers (``agent_init``'s accepted-set check, runtime - resolution) see a canonical name instead of silently discarding the user's intent. - """ - cleaned = api_mode.strip() - return _API_MODE_ALIASES.get(cleaned.lower(), cleaned) - - -def coerce_provider_id(value: Any) -> str: - """Provider identity fields are strings.""" - if value is None: - return "" - return str(value).strip() - - -def stringify_provider_map(providers: Any) -> dict: - """Copy a ``providers:`` mapping so keys are strings. - - Desktop Custom Endpoints store the name as the dict key; an unquoted YAML key ``2070:`` loads - as int, so picker code calling ``ep_name.lower()`` crashes and CRUD lookups of ``"2070"`` miss. - """ - if not isinstance(providers, dict): - return {} - out: Dict[str, Any] = {} - for stored, value in providers.items(): - key = coerce_provider_id(stored) - if key: - out[key] = value - return out - - -def find_provider_entry(providers: Any, key: Any) -> Tuple[Any, Optional[Dict[str, Any]]]: - """Return ``(stored_key, entry)`` matching *key* by string identity. - - Prefer an exact string hit, then scan. - """ - if not isinstance(providers, dict): - return None, None - want = coerce_provider_id(key) - if not want: - return None, None - exact = providers.get(want) - if isinstance(exact, dict): - return want, exact - for stored, entry in providers.items(): - if coerce_provider_id(stored) == want and isinstance(entry, dict): - return stored, entry - return None, None - - -# camelCase aliases commonly used in hand-written provider configs. -_CAMEL_ALIASES: Dict[str, str] = { - "apiKey": "api_key", - "baseUrl": "base_url", - "apiMode": "api_mode", - "keyEnv": "key_env", - "apiKeyEnv": "key_env", # alias — OpenClaw-compatible + docs variant - "defaultModel": "default_model", - "contextLength": "context_length", - "rateLimitDelay": "rate_limit_delay", -} -_KNOWN_PROVIDER_KEYS = { - # ``provider`` duplicates the ``providers.`` mapping key and is unused - # here, but Hermes' own config writer has historically emitted it into - # provider entries. Accept it silently so self-written configs don't warn. - "provider", - "name", "api", "url", "base_url", "api_key", "key_env", "api_key_env", "key_cmd", - "api_mode", "transport", "model", "default_model", "models", "models_discovered", - "context_length", "rate_limit_delay", "request_timeout_seconds", "stale_timeout_seconds", - "discover_models", "extra_body", "extra_headers", "capabilities", "ssl_ca_cert", "ssl_verify", -} - - -def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str: - """First usable URL among ``base_url``/``url``/``api``, or "". - - URLs containing unresolved placeholder tokens — ``${ENV_VAR}`` env-refs and bare ``{region}`` - templates — are accepted without validation: they are expanded at runtime, and a caller reaching - this normalizer with raw config would otherwise see the provider silently dropped. - """ - from urllib.parse import urlparse - - for url_key in ("base_url", "url", "api"): - raw_url = entry.get(url_key) - if not (isinstance(raw_url, str) and raw_url.strip()): - continue - candidate = raw_url.strip() - if re.search(r"\{[^}]+\}", candidate): - return candidate - parsed = urlparse(candidate) - if parsed.scheme and parsed.netloc: - return candidate - logger.warning( - "providers.%s: '%s' value '%s' is not a valid URL " - "(no scheme or host) — skipped", - provider_key or "?", url_key, candidate, - ) - return "" - - -def _normalize_provider_models(models: Any) -> Tuple[Dict[str, Any], bool]: - """Normalize an entry's ``models`` to the dict shape downstream expects. - - Returns ``(models_dict, discovered_flag)``. Older Hermes versions wrote an in-mapping - ``__discovered_model_catalog__`` sentinel (accepted on read, stripped so sentinel keys never - surface as model IDs). Hand-edited/older configs may write a plain list of ids or ``[{id: ...}]`` - rows; both are converted so /model doesn't show the provider with (0) models. - """ - discovered = False - if isinstance(models, dict) and models: - # Shallow-copy: `models` may alias a cached config sub-dict, and the - # normalized entry escapes into long-lived runtime state. - models_copy = dict(models) - if models_copy.pop("__discovered_model_catalog__", None) is True: - discovered = True - models_copy.pop("__explicit_model_allowlist__", None) - return models_copy, discovered - if isinstance(models, list) and models: - normalized_models: Dict[str, Any] = {} - for item in models: - if isinstance(item, str) and item.strip(): - normalized_models[item.strip()] = {} - continue - if not isinstance(item, dict): - continue - model_id = item.get("id") - if not isinstance(model_id, str) or not model_id.strip(): - model_id = item.get("name") - if not isinstance(model_id, str) or not model_id.strip(): - continue - normalized_models[model_id.strip()] = { - k: v for k, v in item.items() if k not in {"id", "name"} - } - return normalized_models, discovered - return {}, discovered - - -def _normalize_custom_provider_entry( - entry: Any, - *, - provider_key: str = "", -) -> Optional[Dict[str, Any]]: - """Return a runtime-compatible custom provider entry or ``None``.""" - if not isinstance(entry, dict): - return None - - # Shallow-copy before alias normalization writes into the entry: callers - # pass live sub-dicts from load_config_readonly()'s shared cache, and - # mutating those violates the cache's no-mutation contract and leaks alias - # keys back into config.yaml on a later save_config(load_config()). - entry = dict(entry) - provider_key = coerce_provider_id(provider_key) - - # api_key_env is a documented snake_case alias for key_env (see - # website/docs/guides/azure-foundry.md). Normalize it up front so the - # rest of the normalizer treats it as the canonical field. - if "api_key_env" in entry and "key_env" not in entry: - entry["key_env"] = entry["api_key_env"] - for camel, snake in _CAMEL_ALIASES.items(): - if camel in entry and snake not in entry: - _warn_once_per_provider( - provider_key, f"camel:{camel}", - "providers.%s: camelCase key '%s' auto-mapped to '%s' " - "(use snake_case to avoid this warning)", - provider_key or "?", camel, snake, - ) - entry[snake] = entry[camel] - unknown = set(entry.keys()) - _KNOWN_PROVIDER_KEYS - set(_CAMEL_ALIASES.keys()) - if unknown: - _warn_once_per_provider( - provider_key, "unknown:" + ",".join(sorted(unknown)), - "providers.%s: unknown config keys ignored: %s", - provider_key or "?", ", ".join(sorted(unknown)), - ) - - base_url = _pick_provider_base_url(entry, provider_key) - if not base_url: - return None - - name = coerce_provider_id(entry.get("name")) or provider_key - if not name: - return None - - normalized: Dict[str, Any] = { - "name": name, - "base_url": base_url, - } - - provider_key = provider_key.strip() - if provider_key: - normalized["provider_key"] = provider_key - - def _stripped(*keys: str) -> str: - val = None - for k in keys: - val = entry.get(k) - if val: - break - return val.strip() if isinstance(val, str) else "" - - if _stripped("api_key"): - normalized["api_key"] = _stripped("api_key") - - key_env = _stripped("key_env", "api_key_env") - if key_env: - normalized["key_env"] = key_env - if entry.get("api_key_env") and not entry.get("key_env"): - normalized["api_key_env"] = key_env - - api_mode = _stripped("api_mode", "transport") - if api_mode: - normalized["api_mode"] = _canonical_api_mode(api_mode) - - model_name = _stripped("model", "default_model") - if model_name: - normalized["model"] = model_name - - # Entry-level marker: the ``models`` mapping was auto-discovered by Hermes - # (``_save_discovered_models_to_config``), not hand-curated. - models_dict, discovered = _normalize_provider_models(entry.get("models")) - if models_dict: - normalized["models"] = models_dict - if entry.get("models_discovered") is True or discovered: - normalized["models_discovered"] = True - - capabilities = entry.get("capabilities") - if isinstance(capabilities, dict): - normalized_capabilities = { - key: value - for key, value in capabilities.items() - if isinstance(key, str) and isinstance(value, bool) - } - if normalized_capabilities: - normalized["capabilities"] = normalized_capabilities - - context_length = entry.get("context_length") - if isinstance(context_length, int) and context_length > 0: - normalized["context_length"] = context_length - - rate_limit_delay = entry.get("rate_limit_delay") - if isinstance(rate_limit_delay, (int, float)) and rate_limit_delay >= 0: - normalized["rate_limit_delay"] = rate_limit_delay - - if isinstance(entry.get("discover_models"), bool): - normalized["discover_models"] = entry["discover_models"] - - if isinstance(entry.get("extra_body"), dict): - normalized["extra_body"] = dict(entry["extra_body"]) - - # Per-provider extra HTTP headers (proxies, gateways, custom auth). - # Values may carry credentials (e.g. CF-Access-Client-Secret) — never - # log them anywhere downstream. - normalized_headers = normalize_extra_headers(entry.get("extra_headers")) - if normalized_headers: - normalized["extra_headers"] = normalized_headers - - ssl_ca_cert = entry.get("ssl_ca_cert") - if isinstance(ssl_ca_cert, str) and ssl_ca_cert.strip(): - normalized["ssl_ca_cert"] = ssl_ca_cert.strip() - - ssl_verify = entry.get("ssl_verify") - if isinstance(ssl_verify, bool): - normalized["ssl_verify"] = ssl_verify - elif isinstance(ssl_verify, str) and ssl_verify.strip(): - normalized["ssl_verify"] = ssl_verify.strip() - - return normalized - - -def _custom_provider_entry_to_provider_config( - entry: Any, - *, - provider_key: str = "", -) -> Optional[Dict[str, Any]]: - """Translate a legacy custom provider entry to the v12 providers shape.""" - normalized = _normalize_custom_provider_entry( - dict(entry) if isinstance(entry, dict) else entry, - provider_key=provider_key, - ) - if normalized is None: - return None - - provider_entry: Dict[str, Any] = {"api": normalized["base_url"]} - - for field in ( - "name", "api_key", "key_env", "models", "models_discovered", "context_length", - "rate_limit_delay", "discover_models", "extra_body", "extra_headers", - "ssl_ca_cert", "ssl_verify", - ): - if field in normalized: - provider_entry[field] = normalized[field] - - if "model" in normalized: - provider_entry["default_model"] = normalized["model"] - if "api_mode" in normalized: - provider_entry["transport"] = normalized["api_mode"] - - return provider_entry - - -def providers_dict_to_custom_providers(providers_dict: Any) -> List[Dict[str, Any]]: - """Normalize ``providers`` config entries into the legacy custom-provider shape.""" - if not isinstance(providers_dict, dict): - return [] - - custom_providers: List[Dict[str, Any]] = [] - for key, entry in providers_dict.items(): - if isinstance(entry, dict) and not is_provider_enabled(entry): - continue - normalized = _normalize_custom_provider_entry( - entry, provider_key=coerce_provider_id(key) - ) - if normalized is not None: - custom_providers.append(normalized) - - return custom_providers - - -def get_compatible_custom_providers( - config: Optional[Dict[str, Any]] = None, -) -> List[Dict[str, Any]]: - """Return a deduplicated custom-provider view across legacy and v12+ config. - - ``custom_providers`` remains the on-disk legacy format, while ``providers`` is the newer keyed - schema. Runtime and picker flows still need a single list-shaped view, but we should not - materialise that compatibility layer back into config.yaml because it duplicates entries in UIs. - """ - if config is None: - config = load_config() - - compatible: List[Dict[str, Any]] = [] - seen_provider_keys: set = set() - seen_name_url_pairs: set = set() - - def _append_if_new(entry: Optional[Dict[str, Any]]) -> None: - if entry is None: - return - provider_key = str(entry.get("provider_key", "") or "").strip().lower() - name = str(entry.get("name", "") or "").strip().lower() - base_url = str(entry.get("base_url", "") or "").strip().rstrip("/").lower() - model = str(entry.get("model", "") or "").strip().lower() - pair = (name, base_url, model) - - if provider_key and provider_key in seen_provider_keys: - return - if name and base_url and pair in seen_name_url_pairs: - return - - compatible.append(entry) - if provider_key: - seen_provider_keys.add(provider_key) - if name and base_url: - seen_name_url_pairs.add(pair) - - custom_providers = config.get("custom_providers") - if custom_providers is not None: - if not isinstance(custom_providers, list): - return [] - for entry in custom_providers: - _append_if_new(_normalize_custom_provider_entry(entry)) - - for entry in providers_dict_to_custom_providers(config.get("providers")): - _append_if_new(entry) - - return compatible - - -def _entries_for_route( - base_url: str, - custom_providers: Optional[List[Dict[str, Any]]], - config: Optional[Dict[str, Any]], -): - """Yield custom-provider entries whose normalized route identity equals *base_url*. - - Loads ``get_compatible_custom_providers(config)`` when *custom_providers* is None (failures → - no entries). Yields nothing for an empty *base_url* or non-list input. - """ - if custom_providers is None: - try: - custom_providers = get_compatible_custom_providers(config) - except Exception: - custom_providers = [] - if not base_url or not isinstance(custom_providers, list): - return - target_url = normalize_route_base_url(base_url) - if not target_url: - return - for entry in custom_providers: - if not isinstance(entry, dict): - continue - entry_url = normalize_route_base_url(entry.get("base_url")) - if entry_url and entry_url == target_url: - yield entry - - -def _route_model_cfg(entry: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]: - """Return ``entry.models[model]`` when both are mappings, else None.""" - models = entry.get("models") - if not isinstance(models, dict): - return None - model_cfg = models.get(model) - return model_cfg if isinstance(model_cfg, dict) else None - - -def _coerce_ssl_verify(value: Any) -> Optional[bool]: - if value is None: - return None - if isinstance(value, bool): - return value - if isinstance(value, str): - lowered = value.strip().lower() - if lowered in {"false", "0", "no", "off"}: - return False - if lowered in {"true", "1", "yes", "on"}: - return True - return None - - -def get_custom_provider_tls_settings( - base_url: str, - custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Dict[str, Any]: - """Return TLS settings from a matching ``custom_providers`` / ``providers`` entry.""" - for entry in _entries_for_route(base_url, custom_providers, config): - out: Dict[str, Any] = {} - ca = entry.get("ssl_ca_cert") - if isinstance(ca, str) and ca.strip(): - out["ssl_ca_cert"] = ca.strip() - verify = _coerce_ssl_verify(entry.get("ssl_verify")) - if verify is not None: - out["ssl_verify"] = verify - return out - return {} - - -def apply_custom_provider_tls_to_client_kwargs( - client_kwargs: Dict[str, Any], - base_url: str, - custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> None: - """Attach per-provider TLS knobs to OpenAI client kwargs when matched.""" - tls = get_custom_provider_tls_settings(base_url, custom_providers, config) - if tls.get("ssl_ca_cert"): - client_kwargs["ssl_ca_cert"] = tls["ssl_ca_cert"] - if "ssl_verify" in tls: - client_kwargs["ssl_verify"] = tls["ssl_verify"] - - -def normalize_extra_headers(extra_headers: Any) -> Dict[str, str]: - """Normalize a raw ``extra_headers`` value into a ``dict[str, str]``. - - SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy - auth, custom bearer schemes). Callers must never log the returned values. - """ - if not isinstance(extra_headers, dict) or not extra_headers: - return {} - return {str(k): str(v) for k, v in extra_headers.items() if v is not None} - - -def get_custom_provider_extra_headers( - base_url: str, - custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Dict[str, str]: - """Return ``extra_headers`` from a matching ``providers`` / ``custom_providers`` entry. - - Matches the entry whose normalized route identity equals *base_url*, mirroring - :func:`get_custom_provider_tls_settings`, and returns its ``extra_headers`` dict, or ``{}`` when - no entry matches or declares none. - - SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy - auth, custom bearer schemes). Callers must never log the returned values. - """ - for entry in _entries_for_route(base_url, custom_providers, config): - headers = normalize_extra_headers(entry.get("extra_headers")) - if headers: - return headers - return {} - - -def apply_custom_provider_extra_headers_to_client_kwargs( - client_kwargs: Dict[str, Any], - base_url: str, - custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> None: - """Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers``. - - Provider-specific headers win over SDK/provider defaults already in ``client_kwargs`` (they - are the most specific level). No-op when base_url matches no provider entry or none declares - headers. SECURITY: values may carry credentials -- never log them. - """ - extra_headers = get_custom_provider_extra_headers(base_url, custom_providers, config) - if not extra_headers: - return - merged = dict(client_kwargs.get("default_headers") or {}) - merged.update(extra_headers) - client_kwargs["default_headers"] = merged - - -def get_custom_provider_context_length( - model: str, - base_url: str, - custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Optional[int]: - """Look up a per-model ``context_length`` override from ``custom_providers``. - - Matches any entry whose normalized route identity equals ``base_url`` and returns - ``custom_providers[i].models..context_length`` if present and valid. Returns ``None`` - when no override applies. - """ - if not model or not base_url: - return None - if custom_providers is None: - try: - custom_providers = get_compatible_custom_providers(config) - except Exception: - if config is None: - return None - raw = config.get("custom_providers") - custom_providers = raw if isinstance(raw, list) else [] - - for entry in _entries_for_route(base_url, custom_providers, config): - model_cfg = _route_model_cfg(entry, model) - if model_cfg is None: - continue - raw_ctx = model_cfg.get("context_length") - if raw_ctx is None: - continue - try: - ctx = int(raw_ctx) - except (TypeError, ValueError): - continue - if ctx > 0: - return ctx - return None - - -def get_custom_provider_model_capability( - model: str, - base_url: str, - capability: str, - custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Optional[bool]: - """Return an explicit boolean capability for one custom-provider model. - - Matching is scoped to the normalized route and exact runtime model id so aliases can declare - capabilities without changing the id sent upstream. Missing or non-boolean declarations return - ``None``. - """ - if not model or not base_url or not capability: - return None - if custom_providers is None: - try: - if config is None: - # Read-only path: this helper never mutates the entries it - # scans, and get_compatible_custom_providers shallow-copies - # each entry before normalizing, so the no-deepcopy cache is - # safe here (~135us saved per call on the blank-stub paths). - config = load_config_readonly() - custom_providers = get_compatible_custom_providers(config) - except Exception: - return None - - for entry in _entries_for_route(base_url, custom_providers, config): - model_cfg = _route_model_cfg(entry, model) - if model_cfg is None: - continue - value = model_cfg.get(capability) - if isinstance(value, bool): - return value - return None - - def _coerce_config_version(value: Any) -> int: """Return a safe integer config version, treating invalid values as legacy.""" if isinstance(value, bool): @@ -2900,26 +2302,6 @@ def _normalize_max_turns_config(config: Dict[str, Any]) -> Dict[str, Any]: return config -def is_provider_enabled(provider_cfg: Optional[Dict[str, Any]]) -> bool: - """Return whether a ``providers.`` config block is enabled. - - A provider is enabled by default. Only an explicit ``enabled: false`` in the block hides it from - the model picker, ``/models`` listings, the runtime resolver and the doctor / status output. - - Backward-compat: configs without the ``enabled`` key keep working as before — the default is - ``True``. - """ - if not isinstance(provider_cfg, dict): - return True - flag = provider_cfg.get("enabled", True) - if isinstance(flag, bool): - return flag - # YAML can produce strings for "true"/"false" depending on quoting. - if isinstance(flag, str): - return flag.strip().lower() not in {"false", "0", "no", "off"} - return bool(flag) - - # Sentinel for an unlimited turn budget. ``sys.maxsize`` survives the str→int # round-trip through the HERMES_MAX_ITERATIONS env bridge (gateway/run.py), works # in every ``<``/``>=``/``max - used`` comparison in agent/iteration_budget.py and diff --git a/hermes_cli/config_providers.py b/hermes_cli/config_providers.py new file mode 100644 index 0000000000..ad657c5d68 --- /dev/null +++ b/hermes_cli/config_providers.py @@ -0,0 +1,668 @@ +"""Custom-provider entry normalization and per-route lookups (TLS, headers, context length, capabilities). + +Split out of ``hermes_cli/config.py``; every name is re-imported there, so +``hermes_cli.config.`` keeps resolving (and monkeypatching) as before. +""" + +import logging +import re +from typing import Any, Dict, List, Optional, Tuple +from hermes_cli.route_identity import normalize_route_base_url + +# Log-record parity with the origin module. +logger = logging.getLogger("hermes_cli.config") + + +# ``_normalize_custom_provider_entry`` runs on every ``load_picker_context()`` +# call (i.e. per interactive picker/inventory request), so any warning it emits +# fires repeatedly for the same static config. Deduplicate per (provider, +# signature): on Windows a repeated-warning storm contends on +# ``concurrent-log-handler``'s cross-process rotation lock and can peg a core / +# stall the gateway/serve event loop. The cache lives for the process lifetime. +_PROVIDER_NORMALIZE_WARNED: set = set() + + +def _warn_once_per_provider( + provider_key: str, signature: str, msg: str, *args: Any +) -> None: + """Emit ``logger.warning(msg, *args)`` at most once per (provider, signature).""" + dedup_key = (provider_key or "?", signature) + if dedup_key in _PROVIDER_NORMALIZE_WARNED: + return + _PROVIDER_NORMALIZE_WARNED.add(dedup_key) + logger.warning(msg, *args) + + +_API_MODE_ALIASES = { + # Values accepted by earlier releases (and natural spellings) mapped to + # the canonical transport names consumed by agent_init. Before this map + # existed, an unrecognized api_mode was silently ignored and the + # transport fell through to hostname-based guessing, so a config that + # said ``api_mode: openai`` (valid on older releases) could flip to + # ``codex_responses`` after an update and break the provider (#66543 + # discussion; observed live against api.actual.inc). + "openai": "chat_completions", + "openai_chat": "chat_completions", + "openai-chat": "chat_completions", + "chat-completions": "chat_completions", + "chatcompletions": "chat_completions", + "responses": "codex_responses", + "openai_responses": "codex_responses", + "openai-responses": "codex_responses", + "anthropic": "anthropic_messages", + "anthropic-messages": "anthropic_messages", + "messages": "anthropic_messages", + "bedrock": "bedrock_converse", + "bedrock-converse": "bedrock_converse", +} + + +def _canonical_api_mode(api_mode: str) -> str: + """Map legacy/alias ``api_mode`` spellings to canonical transport names. + + Unknown values pass through unchanged (callers keep their existing fall-through behavior); known + aliases are rewritten so downstream consumers (``agent_init``'s accepted-set check, runtime + resolution) see a canonical name instead of silently discarding the user's intent. + """ + cleaned = api_mode.strip() + return _API_MODE_ALIASES.get(cleaned.lower(), cleaned) + + +def coerce_provider_id(value: Any) -> str: + """Provider identity fields are strings.""" + if value is None: + return "" + return str(value).strip() + + +def stringify_provider_map(providers: Any) -> dict: + """Copy a ``providers:`` mapping so keys are strings. + + Desktop Custom Endpoints store the name as the dict key; an unquoted YAML key ``2070:`` loads + as int, so picker code calling ``ep_name.lower()`` crashes and CRUD lookups of ``"2070"`` miss. + """ + if not isinstance(providers, dict): + return {} + out: Dict[str, Any] = {} + for stored, value in providers.items(): + key = coerce_provider_id(stored) + if key: + out[key] = value + return out + + +def find_provider_entry(providers: Any, key: Any) -> Tuple[Any, Optional[Dict[str, Any]]]: + """Return ``(stored_key, entry)`` matching *key* by string identity. + + Prefer an exact string hit, then scan. + """ + if not isinstance(providers, dict): + return None, None + want = coerce_provider_id(key) + if not want: + return None, None + exact = providers.get(want) + if isinstance(exact, dict): + return want, exact + for stored, entry in providers.items(): + if coerce_provider_id(stored) == want and isinstance(entry, dict): + return stored, entry + return None, None + + +# camelCase aliases commonly used in hand-written provider configs. +_CAMEL_ALIASES: Dict[str, str] = { + "apiKey": "api_key", + "baseUrl": "base_url", + "apiMode": "api_mode", + "keyEnv": "key_env", + "apiKeyEnv": "key_env", # alias — OpenClaw-compatible + docs variant + "defaultModel": "default_model", + "contextLength": "context_length", + "rateLimitDelay": "rate_limit_delay", +} + + +_KNOWN_PROVIDER_KEYS = { + # ``provider`` duplicates the ``providers.`` mapping key and is unused + # here, but Hermes' own config writer has historically emitted it into + # provider entries. Accept it silently so self-written configs don't warn. + "provider", + "name", "api", "url", "base_url", "api_key", "key_env", "api_key_env", "key_cmd", + "api_mode", "transport", "model", "default_model", "models", "models_discovered", + "context_length", "rate_limit_delay", "request_timeout_seconds", "stale_timeout_seconds", + "discover_models", "extra_body", "extra_headers", "capabilities", "ssl_ca_cert", "ssl_verify", +} + + +def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str: + """First usable URL among ``base_url``/``url``/``api``, or "". + + URLs containing unresolved placeholder tokens — ``${ENV_VAR}`` env-refs and bare ``{region}`` + templates — are accepted without validation: they are expanded at runtime, and a caller reaching + this normalizer with raw config would otherwise see the provider silently dropped. + """ + from urllib.parse import urlparse + + for url_key in ("base_url", "url", "api"): + raw_url = entry.get(url_key) + if not (isinstance(raw_url, str) and raw_url.strip()): + continue + candidate = raw_url.strip() + if re.search(r"\{[^}]+\}", candidate): + return candidate + parsed = urlparse(candidate) + if parsed.scheme and parsed.netloc: + return candidate + logger.warning( + "providers.%s: '%s' value '%s' is not a valid URL " + "(no scheme or host) — skipped", + provider_key or "?", url_key, candidate, + ) + return "" + + +def _normalize_provider_models(models: Any) -> Tuple[Dict[str, Any], bool]: + """Normalize an entry's ``models`` to the dict shape downstream expects. + + Returns ``(models_dict, discovered_flag)``. Older Hermes versions wrote an in-mapping + ``__discovered_model_catalog__`` sentinel (accepted on read, stripped so sentinel keys never + surface as model IDs). Hand-edited/older configs may write a plain list of ids or ``[{id: ...}]`` + rows; both are converted so /model doesn't show the provider with (0) models. + """ + discovered = False + if isinstance(models, dict) and models: + # Shallow-copy: `models` may alias a cached config sub-dict, and the + # normalized entry escapes into long-lived runtime state. + models_copy = dict(models) + if models_copy.pop("__discovered_model_catalog__", None) is True: + discovered = True + models_copy.pop("__explicit_model_allowlist__", None) + return models_copy, discovered + if isinstance(models, list) and models: + normalized_models: Dict[str, Any] = {} + for item in models: + if isinstance(item, str) and item.strip(): + normalized_models[item.strip()] = {} + continue + if not isinstance(item, dict): + continue + model_id = item.get("id") + if not isinstance(model_id, str) or not model_id.strip(): + model_id = item.get("name") + if not isinstance(model_id, str) or not model_id.strip(): + continue + normalized_models[model_id.strip()] = { + k: v for k, v in item.items() if k not in {"id", "name"} + } + return normalized_models, discovered + return {}, discovered + + +def _normalize_custom_provider_entry( + entry: Any, + *, + provider_key: str = "", +) -> Optional[Dict[str, Any]]: + """Return a runtime-compatible custom provider entry or ``None``.""" + from hermes_cli.config import normalize_extra_headers + if not isinstance(entry, dict): + return None + + # Shallow-copy before alias normalization writes into the entry: callers + # pass live sub-dicts from load_config_readonly()'s shared cache, and + # mutating those violates the cache's no-mutation contract and leaks alias + # keys back into config.yaml on a later save_config(load_config()). + entry = dict(entry) + provider_key = coerce_provider_id(provider_key) + + # api_key_env is a documented snake_case alias for key_env (see + # website/docs/guides/azure-foundry.md). Normalize it up front so the + # rest of the normalizer treats it as the canonical field. + if "api_key_env" in entry and "key_env" not in entry: + entry["key_env"] = entry["api_key_env"] + for camel, snake in _CAMEL_ALIASES.items(): + if camel in entry and snake not in entry: + _warn_once_per_provider( + provider_key, f"camel:{camel}", + "providers.%s: camelCase key '%s' auto-mapped to '%s' " + "(use snake_case to avoid this warning)", + provider_key or "?", camel, snake, + ) + entry[snake] = entry[camel] + unknown = set(entry.keys()) - _KNOWN_PROVIDER_KEYS - set(_CAMEL_ALIASES.keys()) + if unknown: + _warn_once_per_provider( + provider_key, "unknown:" + ",".join(sorted(unknown)), + "providers.%s: unknown config keys ignored: %s", + provider_key or "?", ", ".join(sorted(unknown)), + ) + + base_url = _pick_provider_base_url(entry, provider_key) + if not base_url: + return None + + name = coerce_provider_id(entry.get("name")) or provider_key + if not name: + return None + + normalized: Dict[str, Any] = { + "name": name, + "base_url": base_url, + } + + provider_key = provider_key.strip() + if provider_key: + normalized["provider_key"] = provider_key + + def _stripped(*keys: str) -> str: + val = None + for k in keys: + val = entry.get(k) + if val: + break + return val.strip() if isinstance(val, str) else "" + + if _stripped("api_key"): + normalized["api_key"] = _stripped("api_key") + + key_env = _stripped("key_env", "api_key_env") + if key_env: + normalized["key_env"] = key_env + if entry.get("api_key_env") and not entry.get("key_env"): + normalized["api_key_env"] = key_env + + api_mode = _stripped("api_mode", "transport") + if api_mode: + normalized["api_mode"] = _canonical_api_mode(api_mode) + + model_name = _stripped("model", "default_model") + if model_name: + normalized["model"] = model_name + + # Entry-level marker: the ``models`` mapping was auto-discovered by Hermes + # (``_save_discovered_models_to_config``), not hand-curated. + models_dict, discovered = _normalize_provider_models(entry.get("models")) + if models_dict: + normalized["models"] = models_dict + if entry.get("models_discovered") is True or discovered: + normalized["models_discovered"] = True + + capabilities = entry.get("capabilities") + if isinstance(capabilities, dict): + normalized_capabilities = { + key: value + for key, value in capabilities.items() + if isinstance(key, str) and isinstance(value, bool) + } + if normalized_capabilities: + normalized["capabilities"] = normalized_capabilities + + context_length = entry.get("context_length") + if isinstance(context_length, int) and context_length > 0: + normalized["context_length"] = context_length + + rate_limit_delay = entry.get("rate_limit_delay") + if isinstance(rate_limit_delay, (int, float)) and rate_limit_delay >= 0: + normalized["rate_limit_delay"] = rate_limit_delay + + if isinstance(entry.get("discover_models"), bool): + normalized["discover_models"] = entry["discover_models"] + + if isinstance(entry.get("extra_body"), dict): + normalized["extra_body"] = dict(entry["extra_body"]) + + # Per-provider extra HTTP headers (proxies, gateways, custom auth). + # Values may carry credentials (e.g. CF-Access-Client-Secret) — never + # log them anywhere downstream. + normalized_headers = normalize_extra_headers(entry.get("extra_headers")) + if normalized_headers: + normalized["extra_headers"] = normalized_headers + + ssl_ca_cert = entry.get("ssl_ca_cert") + if isinstance(ssl_ca_cert, str) and ssl_ca_cert.strip(): + normalized["ssl_ca_cert"] = ssl_ca_cert.strip() + + ssl_verify = entry.get("ssl_verify") + if isinstance(ssl_verify, bool): + normalized["ssl_verify"] = ssl_verify + elif isinstance(ssl_verify, str) and ssl_verify.strip(): + normalized["ssl_verify"] = ssl_verify.strip() + + return normalized + + +def _custom_provider_entry_to_provider_config( + entry: Any, + *, + provider_key: str = "", +) -> Optional[Dict[str, Any]]: + """Translate a legacy custom provider entry to the v12 providers shape.""" + from hermes_cli.config import _normalize_custom_provider_entry + normalized = _normalize_custom_provider_entry( + dict(entry) if isinstance(entry, dict) else entry, + provider_key=provider_key, + ) + if normalized is None: + return None + + provider_entry: Dict[str, Any] = {"api": normalized["base_url"]} + + for field in ( + "name", "api_key", "key_env", "models", "models_discovered", "context_length", + "rate_limit_delay", "discover_models", "extra_body", "extra_headers", + "ssl_ca_cert", "ssl_verify", + ): + if field in normalized: + provider_entry[field] = normalized[field] + + if "model" in normalized: + provider_entry["default_model"] = normalized["model"] + if "api_mode" in normalized: + provider_entry["transport"] = normalized["api_mode"] + + return provider_entry + + +def providers_dict_to_custom_providers(providers_dict: Any) -> List[Dict[str, Any]]: + """Normalize ``providers`` config entries into the legacy custom-provider shape.""" + from hermes_cli.config import _normalize_custom_provider_entry + if not isinstance(providers_dict, dict): + return [] + + custom_providers: List[Dict[str, Any]] = [] + for key, entry in providers_dict.items(): + if isinstance(entry, dict) and not is_provider_enabled(entry): + continue + normalized = _normalize_custom_provider_entry( + entry, provider_key=coerce_provider_id(key) + ) + if normalized is not None: + custom_providers.append(normalized) + + return custom_providers + + +def get_compatible_custom_providers( + config: Optional[Dict[str, Any]] = None, +) -> List[Dict[str, Any]]: + """Return a deduplicated custom-provider view across legacy and v12+ config. + + ``custom_providers`` remains the on-disk legacy format, while ``providers`` is the newer keyed + schema. Runtime and picker flows still need a single list-shaped view, but we should not + materialise that compatibility layer back into config.yaml because it duplicates entries in UIs. + """ + from hermes_cli.config import _normalize_custom_provider_entry, load_config + if config is None: + config = load_config() + + compatible: List[Dict[str, Any]] = [] + seen_provider_keys: set = set() + seen_name_url_pairs: set = set() + + def _append_if_new(entry: Optional[Dict[str, Any]]) -> None: + if entry is None: + return + provider_key = str(entry.get("provider_key", "") or "").strip().lower() + name = str(entry.get("name", "") or "").strip().lower() + base_url = str(entry.get("base_url", "") or "").strip().rstrip("/").lower() + model = str(entry.get("model", "") or "").strip().lower() + pair = (name, base_url, model) + + if provider_key and provider_key in seen_provider_keys: + return + if name and base_url and pair in seen_name_url_pairs: + return + + compatible.append(entry) + if provider_key: + seen_provider_keys.add(provider_key) + if name and base_url: + seen_name_url_pairs.add(pair) + + custom_providers = config.get("custom_providers") + if custom_providers is not None: + if not isinstance(custom_providers, list): + return [] + for entry in custom_providers: + _append_if_new(_normalize_custom_provider_entry(entry)) + + for entry in providers_dict_to_custom_providers(config.get("providers")): + _append_if_new(entry) + + return compatible + + +def _entries_for_route( + base_url: str, + custom_providers: Optional[List[Dict[str, Any]]], + config: Optional[Dict[str, Any]], +): + """Yield custom-provider entries whose normalized route identity equals *base_url*. + + Loads ``get_compatible_custom_providers(config)`` when *custom_providers* is None (failures → + no entries). Yields nothing for an empty *base_url* or non-list input. + """ + from hermes_cli.config import get_compatible_custom_providers + if custom_providers is None: + try: + custom_providers = get_compatible_custom_providers(config) + except Exception: + custom_providers = [] + if not base_url or not isinstance(custom_providers, list): + return + target_url = normalize_route_base_url(base_url) + if not target_url: + return + for entry in custom_providers: + if not isinstance(entry, dict): + continue + entry_url = normalize_route_base_url(entry.get("base_url")) + if entry_url and entry_url == target_url: + yield entry + + +def _route_model_cfg(entry: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]: + """Return ``entry.models[model]`` when both are mappings, else None.""" + models = entry.get("models") + if not isinstance(models, dict): + return None + model_cfg = models.get(model) + return model_cfg if isinstance(model_cfg, dict) else None + + +def _coerce_ssl_verify(value: Any) -> Optional[bool]: + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"false", "0", "no", "off"}: + return False + if lowered in {"true", "1", "yes", "on"}: + return True + return None + + +def get_custom_provider_tls_settings( + base_url: str, + custom_providers: Optional[List[Dict[str, Any]]] = None, + config: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """Return TLS settings from a matching ``custom_providers`` / ``providers`` entry.""" + for entry in _entries_for_route(base_url, custom_providers, config): + out: Dict[str, Any] = {} + ca = entry.get("ssl_ca_cert") + if isinstance(ca, str) and ca.strip(): + out["ssl_ca_cert"] = ca.strip() + verify = _coerce_ssl_verify(entry.get("ssl_verify")) + if verify is not None: + out["ssl_verify"] = verify + return out + return {} + + +def apply_custom_provider_tls_to_client_kwargs( + client_kwargs: Dict[str, Any], + base_url: str, + custom_providers: Optional[List[Dict[str, Any]]] = None, + config: Optional[Dict[str, Any]] = None, +) -> None: + """Attach per-provider TLS knobs to OpenAI client kwargs when matched.""" + tls = get_custom_provider_tls_settings(base_url, custom_providers, config) + if tls.get("ssl_ca_cert"): + client_kwargs["ssl_ca_cert"] = tls["ssl_ca_cert"] + if "ssl_verify" in tls: + client_kwargs["ssl_verify"] = tls["ssl_verify"] + + +def normalize_extra_headers(extra_headers: Any) -> Dict[str, str]: + """Normalize a raw ``extra_headers`` value into a ``dict[str, str]``. + + SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy + auth, custom bearer schemes). Callers must never log the returned values. + """ + if not isinstance(extra_headers, dict) or not extra_headers: + return {} + return {str(k): str(v) for k, v in extra_headers.items() if v is not None} + + +def get_custom_provider_extra_headers( + base_url: str, + custom_providers: Optional[List[Dict[str, Any]]] = None, + config: Optional[Dict[str, Any]] = None, +) -> Dict[str, str]: + """Return ``extra_headers`` from a matching ``providers`` / ``custom_providers`` entry. + + Matches the entry whose normalized route identity equals *base_url*, mirroring + :func:`get_custom_provider_tls_settings`, and returns its ``extra_headers`` dict, or ``{}`` when + no entry matches or declares none. + + SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy + auth, custom bearer schemes). Callers must never log the returned values. + """ + from hermes_cli.config import normalize_extra_headers + for entry in _entries_for_route(base_url, custom_providers, config): + headers = normalize_extra_headers(entry.get("extra_headers")) + if headers: + return headers + return {} + + +def apply_custom_provider_extra_headers_to_client_kwargs( + client_kwargs: Dict[str, Any], + base_url: str, + custom_providers: Optional[List[Dict[str, Any]]] = None, + config: Optional[Dict[str, Any]] = None, +) -> None: + """Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers``. + + Provider-specific headers win over SDK/provider defaults already in ``client_kwargs`` (they + are the most specific level). No-op when base_url matches no provider entry or none declares + headers. SECURITY: values may carry credentials -- never log them. + """ + extra_headers = get_custom_provider_extra_headers(base_url, custom_providers, config) + if not extra_headers: + return + merged = dict(client_kwargs.get("default_headers") or {}) + merged.update(extra_headers) + client_kwargs["default_headers"] = merged + + +def get_custom_provider_context_length( + model: str, + base_url: str, + custom_providers: Optional[List[Dict[str, Any]]] = None, + config: Optional[Dict[str, Any]] = None, +) -> Optional[int]: + """Look up a per-model ``context_length`` override from ``custom_providers``. + + Matches any entry whose normalized route identity equals ``base_url`` and returns + ``custom_providers[i].models..context_length`` if present and valid. Returns ``None`` + when no override applies. + """ + from hermes_cli.config import get_compatible_custom_providers + if not model or not base_url: + return None + if custom_providers is None: + try: + custom_providers = get_compatible_custom_providers(config) + except Exception: + if config is None: + return None + raw = config.get("custom_providers") + custom_providers = raw if isinstance(raw, list) else [] + + for entry in _entries_for_route(base_url, custom_providers, config): + model_cfg = _route_model_cfg(entry, model) + if model_cfg is None: + continue + raw_ctx = model_cfg.get("context_length") + if raw_ctx is None: + continue + try: + ctx = int(raw_ctx) + except (TypeError, ValueError): + continue + if ctx > 0: + return ctx + return None + + +def get_custom_provider_model_capability( + model: str, + base_url: str, + capability: str, + custom_providers: Optional[List[Dict[str, Any]]] = None, + config: Optional[Dict[str, Any]] = None, +) -> Optional[bool]: + """Return an explicit boolean capability for one custom-provider model. + + Matching is scoped to the normalized route and exact runtime model id so aliases can declare + capabilities without changing the id sent upstream. Missing or non-boolean declarations return + ``None``. + """ + from hermes_cli.config import get_compatible_custom_providers, load_config_readonly + if not model or not base_url or not capability: + return None + if custom_providers is None: + try: + if config is None: + # Read-only path: this helper never mutates the entries it + # scans, and get_compatible_custom_providers shallow-copies + # each entry before normalizing, so the no-deepcopy cache is + # safe here (~135us saved per call on the blank-stub paths). + config = load_config_readonly() + custom_providers = get_compatible_custom_providers(config) + except Exception: + return None + + for entry in _entries_for_route(base_url, custom_providers, config): + model_cfg = _route_model_cfg(entry, model) + if model_cfg is None: + continue + value = model_cfg.get(capability) + if isinstance(value, bool): + return value + return None + + +def is_provider_enabled(provider_cfg: Optional[Dict[str, Any]]) -> bool: + """Return whether a ``providers.`` config block is enabled. + + A provider is enabled by default. Only an explicit ``enabled: false`` in the block hides it from + the model picker, ``/models`` listings, the runtime resolver and the doctor / status output. + + Backward-compat: configs without the ``enabled`` key keep working as before — the default is + ``True``. + """ + if not isinstance(provider_cfg, dict): + return True + flag = provider_cfg.get("enabled", True) + if isinstance(flag, bool): + return flag + # YAML can produce strings for "true"/"false" depending on quoting. + if isinstance(flag, str): + return flag.strip().lower() not in {"false", "0", "no", "off"} + return bool(flag)