refactor(hermes_cli): group B — docstring compaction, provider scalar-field table, config mirror _items helper, regex kebab collapse

This commit is contained in:
Teknium
2026-09-02 21:03:13 -07:00
parent 6121fefb78
commit 8b74aa07a4
4 changed files with 56 additions and 114 deletions
+2 -5
View File
@@ -116,11 +116,8 @@ def _migration_lines(config: dict) -> list[str]:
def apply(
config: dict, new_value: Optional[str], *, persist_callback=None) -> CodexRuntimeStatus:
"""Top-level entry point used by both CLI and gateway handlers.
``config`` is mutated in place when ``new_value`` is set (None means show current state only).
``persist_callback`` receives the mutated dict to write it to disk; skipped when None (tests).
"""
"""Entry point for CLI and gateway. ``config`` is mutated in place when ``new_value`` is set
(None = show current state); ``persist_callback(config)`` writes it, skipped when None."""
current = get_current_runtime(config)
# Cached per apply() call: the enable path would otherwise spawn `codex --version` up to 3x.
+6 -13
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import copy
import functools
import re
from typing import Any, Callable, Dict, List, Optional, Tuple
from urllib.parse import urlparse
@@ -79,11 +80,8 @@ def _rewrite_key(
message: str,
extra_guard: Callable[[Dict[str, Any]], bool] = lambda _m: True,
create_section: bool = False) -> None:
"""Rewrite ``<section>.<key>`` to *new* when ``match(current_value)`` holds.
``new=None`` deletes the key. A missing/non-mapping section is skipped unless
*create_section* (then ``match(None)`` decides).
"""
"""Rewrite ``<section>.<key>`` to *new* (None = delete) when ``match(current)`` holds; a
missing section is skipped unless *create_section*."""
config = read_raw_config()
raw = config.get(section)
if not isinstance(raw, dict):
@@ -100,11 +98,8 @@ def _rewrite_key(
def _rewrite_stale_default(*, old: Any, **kw: Any) -> Callable[[Dict[str, Any], bool], None]:
"""Step that rewrites a key only while it still equals the OLD default.
Never clobbers a value the user deliberately customized; unset keys inherit the new default at
read time.
"""
"""Step rewriting a key only while it still equals the OLD default — never clobbers a value
the user customized; unset keys inherit the new default at read time."""
return functools.partial(_rewrite_key, match=lambda cur: cur == old, **kw)
@@ -132,9 +127,7 @@ def _migrate_to_12(results: Dict[str, Any], quiet: bool) -> None:
# kebab-case key from the display name; fall back to the URL hostname.
key = old_name.strip().lower().replace(" ", "-").replace("(", "").replace(")", "")
while "--" in key:
key = key.replace("--", "-")
key = key.strip("-")
key = re.sub(r"-{2,}", "-", key).strip("-")
if not key:
try:
key = (urlparse(old_url).hostname or "endpoint").replace(".", "-")
+26 -59
View File
@@ -70,11 +70,7 @@ def coerce_provider_id(value: Any) -> str:
def stringify_provider_map(providers: Any) -> dict:
"""Copy a ``providers:`` mapping so keys are strings.
An unquoted YAML key ``2070:`` loads as int, so picker code calling ``ep_name.lower()`` crashes
and CRUD lookups of ``"2070"`` miss.
"""
"""Copy a ``providers:`` mapping so keys are strings (unquoted YAML ``2070:`` loads as int)."""
if not isinstance(providers, dict):
return {}
out: Dict[str, Any] = {}
@@ -126,9 +122,8 @@ _KNOWN_PROVIDER_KEYS = {
def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str:
"""First usable URL among ``base_url``/``url``/``api``, or "".
URLs with unresolved placeholders — ``${ENV_VAR}`` env-refs and ``{region}`` templates — are
accepted without validation: they expand at runtime, and rejecting them here would silently
drop the provider.
URLs with unresolved ``${ENV_VAR}`` / ``{region}`` placeholders are accepted unvalidated: they
expand at runtime, and rejecting them here would silently drop the provider.
"""
for url_key in ("base_url", "url", "api"):
raw_url = entry.get(url_key)
@@ -150,9 +145,8 @@ def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str:
def _normalize_provider_models(models: Any) -> Tuple[Dict[str, Any], bool]:
"""Normalize an entry's ``models`` to ``(models_dict, discovered_flag)``.
Older Hermes versions wrote an in-mapping ``__discovered_model_catalog__`` sentinel (accepted,
stripped so it never surfaces as a model id). Hand-edited configs may write a plain list of ids
or ``[{id: ...}]`` rows; both are converted so /model doesn't show the provider with (0) models.
The legacy in-mapping ``__discovered_model_catalog__`` sentinel is accepted and stripped; a
plain list of ids or ``[{id: ...}]`` rows is converted so /model doesn't show (0) models.
"""
discovered = False
if isinstance(models, dict) and models:
@@ -187,13 +181,11 @@ def _normalize_custom_provider_entry(
"""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; mutating those violates its 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 (azure-foundry guide).
if "api_key_env" in entry and "key_env" not in entry:
entry["key_env"] = entry["api_key_env"]
@@ -213,15 +205,10 @@ def _normalize_custom_provider_entry(
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:
if not base_url or not name:
return None
normalized: Dict[str, Any] = {"name": name, "base_url": base_url}
if provider_key:
normalized["provider_key"] = provider_key
@@ -258,17 +245,13 @@ def _normalize_custom_provider_entry(
key: value for key, value in capabilities.items()
if isinstance(key, str) and isinstance(value, bool)})
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"]
for field, ok in (
("context_length", lambda v: isinstance(v, int) and v > 0),
("rate_limit_delay", lambda v: isinstance(v, (int, float)) and v >= 0),
("discover_models", lambda v: isinstance(v, bool)),
):
if ok(entry.get(field)):
normalized[field] = entry[field]
if isinstance(entry.get("extra_body"), dict):
normalized["extra_body"] = dict(entry["extra_body"])
@@ -322,10 +305,9 @@ def providers_dict_to_custom_providers(providers_dict: Any) -> List[Dict[str, An
def get_compatible_custom_providers(
config: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
"""Deduplicated list-shaped view over legacy ``custom_providers`` and v12+ ``providers``.
"""Deduplicated list view over legacy ``custom_providers`` and v12+ ``providers``.
Runtime and picker flows need one list; the compatibility layer is never materialised back
into config.yaml because it duplicates entries in UIs.
Never materialised back into config.yaml (it would duplicate entries in UIs).
"""
from hermes_cli.config import load_config
if config is None:
@@ -366,10 +348,9 @@ 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*.
"""Yield 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.
None *custom_providers* → ``get_compatible_custom_providers(config)`` (failure → none).
"""
from hermes_cli.config import get_compatible_custom_providers
if custom_providers is None:
@@ -418,10 +399,7 @@ def _coerce_ssl_verify(value: Any) -> Optional[bool]:
return value
if isinstance(value, str):
lowered = value.strip().lower()
if lowered in _FALSE_WORDS:
return False
if lowered in _TRUE_WORDS:
return True
return False if lowered in _FALSE_WORDS else True if lowered in _TRUE_WORDS else None
return None
@@ -470,10 +448,8 @@ 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`` of the first route-matching entry that declares any, else ``{}``.
SECURITY: values may carry credentials — callers must never log them.
"""
"""``extra_headers`` of the first route-matching entry declaring any, else ``{}``.
SECURITY: values may carry credentials — callers must never log them."""
for entry in _entries_for_route(base_url, custom_providers, config):
headers = normalize_extra_headers(entry.get("extra_headers"))
if headers:
@@ -486,11 +462,8 @@ def apply_custom_provider_extra_headers_to_client_kwargs(
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). SECURITY: values may carry credentials -- never log them.
"""
"""Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers`` (provider wins
over SDK defaults, most specific level). SECURITY: values may carry credentials; never log."""
extra_headers = get_custom_provider_extra_headers(base_url, custom_providers, config)
if not extra_headers:
return
@@ -536,11 +509,8 @@ def get_custom_provider_model_capability(
capability: str,
custom_providers: Optional[List[Dict[str, Any]]] = None,
config: Optional[Dict[str, Any]] = None) -> Optional[bool]:
"""Explicit boolean capability for one custom-provider model, or ``None``.
Matching is scoped to the normalized route and exact runtime model id so aliases can declare
capabilities without changing the id sent upstream.
"""
"""Explicit boolean capability for one custom-provider model, or ``None``. Scoped to the
normalized route + exact runtime model id so aliases can declare capabilities."""
from hermes_cli.config import get_compatible_custom_providers, load_config_readonly
if not model or not base_url or not capability:
return None
@@ -562,11 +532,8 @@ def get_custom_provider_model_capability(
def is_provider_enabled(provider_cfg: Optional[Dict[str, Any]]) -> bool:
"""Whether a ``providers.<name>`` block is enabled (default True; only explicit false hides it).
An explicit ``enabled: false`` hides the provider from the model picker, ``/models``, the
runtime resolver and doctor/status output.
"""
"""Whether a ``providers.<name>`` block is enabled: default True; only an explicit
``enabled: false`` hides it from the picker, ``/models``, runtime resolver and doctor."""
if not isinstance(provider_cfg, dict):
return True
flag = provider_cfg.get("enabled", True)
+22 -37
View File
@@ -49,9 +49,8 @@ def _for_each_provider(providers: List[str], import_path: str, *args: Any) -> No
def _prune_env_pool_entries(env_var: str) -> List[str]:
"""Drop ``credential_pool`` entries seeded from ``env:<env_var>``; return providers pruned.
Operates across ALL providers: the source names the env var unambiguously and shared vars like
GITHUB_TOKEN may seed more than one provider. Entries with any other source (OAuth,
device-code, manual, borrowed-CLI) and ``providers.<id>`` blocks are preserved verbatim.
Spans ALL providers (shared vars like GITHUB_TOKEN seed several). Entries with any other
source (OAuth, device-code, manual, borrowed-CLI) are preserved verbatim.
"""
from hermes_cli.auth import _auth_store_lock, _load_auth_store, _save_auth_store
@@ -80,12 +79,11 @@ def _prune_env_pool_entries(env_var: str) -> List[str]:
def _scrub_config_yaml_mirrors(old_value: str, new_value: str | None) -> List[str]:
"""Reconcile config.yaml api_key mirrors holding ``old_value``; return the dotted paths touched.
"""Reconcile config.yaml api_key mirrors holding ``old_value``; return dotted paths touched.
Value-matched on purpose: only an entry that provably holds the SAME credential that just
changed in ``.env`` is touched — an independent key for a different endpoint is left alone.
``new_value=None`` removes the field; a string replaces it. Operates on the RAW user config so
the write doesn't bake defaults into the user's file. Paths only — never values.
Value-matched on purpose: only an entry holding the SAME credential that just changed in
``.env`` is touched. ``new_value=None`` removes the field. Operates on the RAW user config
so defaults are never baked into the user's file.
"""
if not old_value:
return []
@@ -121,29 +119,23 @@ def _scrub_config_yaml_mirrors(old_value: str, new_value: str | None) -> List[st
section.pop(field, None)
touched.append(f"{key_path}.{field}")
def _items(value: Any, allow_list: bool):
if isinstance(value, dict):
return value.items()
return enumerate(value) if allow_list and isinstance(value, list) else ()
_fix(user_config.get("model"), "model")
aux = user_config.get("auxiliary")
if isinstance(aux, dict):
for task, slot_cfg in aux.items():
_fix(slot_cfg, f"auxiliary.{task}")
custom = user_config.get("custom_providers")
if isinstance(custom, list):
for idx, entry in enumerate(custom):
_fix(entry, f"custom_providers.{idx}")
elif isinstance(custom, dict):
for name, entry in custom.items():
_fix(entry, f"custom_providers.{name}")
for task, slot_cfg in _items(user_config.get("auxiliary"), False):
_fix(slot_cfg, f"auxiliary.{task}")
for name, entry in _items(user_config.get("custom_providers"), True):
_fix(entry, f"custom_providers.{name}")
# ``providers.<id>.api_key`` (v12+) is where dashboard/desktop write custom-endpoint
# credentials. It is a real inline secret with higher precedence than the env var, so a stale
# copy shadows a rotation (persistent 401 with a key the UI no longer shows) and survives a
# removal that promised to clear EVERY store.
keyed_providers = user_config.get("providers")
if isinstance(keyed_providers, dict):
for provider_id, entry in keyed_providers.items():
_fix(entry, f"providers.{provider_id}", fields=("api_key",))
for provider_id, entry in _items(user_config.get("providers"), False):
_fix(entry, f"providers.{provider_id}", fields=("api_key",))
if touched:
require_readable_config_before_write(config_path)
@@ -173,12 +165,9 @@ def purge_env_credential_references(
def save_provider_env_credential(env_var: str, value: str) -> Dict[str, Any]:
"""Save/update a credential in ``.env`` and reconcile every mirror.
Any config.yaml mirror that held the PREVIOUS value (``model.api_key`` etc.) is updated so a
stale higher-precedence copy cannot shadow the rotation. The save also forces an immediate
``load_pool()`` for every provider registered against this env var so the env-seeded
``credential_pool`` entry is materialized to ``auth.json`` now — the live runtime reads from
the pool, and a ``.env``-only write left env-backed providers 401'ing until the user ran
``hermes auth add`` separately.
config.yaml mirrors of the PREVIOUS value are updated so a stale higher-precedence copy cannot
shadow the rotation, and ``load_pool()`` runs now so the env-seeded ``credential_pool`` entry
lands in ``auth.json`` (a ``.env``-only write left env-backed providers 401'ing).
"""
from hermes_cli.config import load_env, save_env_value
@@ -201,12 +190,8 @@ def save_provider_env_credential(env_var: str, value: str) -> Dict[str, Any]:
def remove_provider_env_credential(env_var: str) -> Dict[str, Any]:
"""Remove a credential from EVERY store it lives in.
Clears the ``.env`` entry (and process env), prunes env-seeded ``credential_pool`` entries,
drops the affected providers' model-cache rows, and removes any config.yaml mirror holding the
same value. OAuth/device-code/manual credentials are preserved.
"""
"""Remove a credential from EVERY store: ``.env`` (and process env), env-seeded
``credential_pool`` entries, model-cache rows, config.yaml mirrors of the same value."""
from hermes_cli.config import load_env, remove_env_value
old_value = load_env().get(env_var)