From 5d4b97939ed6642b230d2d1d52764362b3cddfec Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 14:38:32 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hclib):=20config=20=E2=80=94=20config/?= =?UTF-8?q?config=5Fmigrations/tools=5Fconfig/toolset=5F*=20dispatch=20tab?= =?UTF-8?q?les=20and=20dedupe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/config.py | 4554 +++++++++++------------------- hermes_cli/config_defaults.py | 7 +- hermes_cli/config_migrations.py | 523 ++-- hermes_cli/env_loader.py | 219 +- hermes_cli/fallback_config.py | 25 +- hermes_cli/mcp_config.py | 270 +- hermes_cli/moa_config.py | 356 +-- hermes_cli/setup_hidden_env.py | 20 +- hermes_cli/skills_config.py | 32 +- hermes_cli/tools_config.py | 4486 +++++++++++------------------ hermes_cli/toolset_scope.py | 6 +- hermes_cli/toolset_validation.py | 116 +- 12 files changed, 3841 insertions(+), 6773 deletions(-) diff --git a/hermes_cli/config.py b/hermes_cli/config.py index add9141382..3bfe8dcb39 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1,18 +1,4 @@ -""" -Configuration management for Hermes Agent. - -Config files are stored in ~/.hermes/ for easy access: -- ~/.hermes/config.yaml - All settings (model, toolsets, terminal, etc.) -- ~/.hermes/.env - API keys and secrets - -This module provides: -- hermes config - Show current configuration -- hermes config edit - Open config in editor -- hermes config get - Print a resolved configuration value -- hermes config set - Set a specific value -- hermes config unset - Remove a user configuration value -- hermes config wizard - Re-run setup wizard -""" +"""Configuration management for Hermes Agent.""" import copy from decimal import Decimal, InvalidOperation @@ -60,23 +46,12 @@ class InvalidUserConfigError(RuntimeError): def _backup_corrupt_config(config_path: Path) -> Optional[Path]: """Preserve a corrupted ``config.yaml`` by copying it to a timestamped ``.bak``. - When the YAML can't be parsed, ``load_config()`` silently falls back to - ``DEFAULT_CONFIG`` and the user's broken file stays on disk untouched. - That file is still the user's only copy of their intended overrides — if - they re-run the setup wizard or ``hermes config set`` (which rewrites - ``config.yaml``), the broken-but-recoverable content is gone for good. + When the YAML can't be parsed, ``load_config()`` silently falls back to ``DEFAULT_CONFIG`` and + the user's broken file stays on disk untouched. - This snapshots the corrupted file to ``config.yaml.corrupt..bak`` so - the user can diff/repair it. Unlike Gemini CLI's policy-file recovery - (which resets the live file to a clean state), we deliberately leave - ``config.yaml`` in place: hermes never silently mutates the user's config, - and leaving it means a hand-fixed file is re-read on the next load. The - backup is best-effort — any failure (permissions, symlink, disk full) is - swallowed so config loading is never blocked by backup problems. - - Returns the backup path on success, else ``None``. Symlinks are not - followed/copied (mirrors the Gemini #21541 lstat guard) to avoid - clobbering whatever a malicious/misconfigured symlink points at. + Returns the backup path on success, else ``None``. Symlinks are not followed/copied (mirrors the + Gemini #21541 lstat guard) to avoid clobbering whatever a malicious/misconfigured symlink points + at. """ try: if config_path.is_symlink(): @@ -116,25 +91,9 @@ def _warn_config_parse_failure( ) -> None: """Surface a config.yaml parse failure to user, log, and stderr. - A YAML parse error in ``~/.hermes/config.yaml`` causes ``load_config()`` - to silently fall back to ``DEFAULT_CONFIG``, which means every user - override (auxiliary providers, fallback chain, model overrides, etc.) - is dropped. Before this helper that was a one-line ``print(...)`` that - scrolled off-screen on the first invocation and was never seen again. - - Now: warn once per (path, mtime_ns, size) on stderr **and** in - ``agent.log`` / ``errors.log`` at WARNING level so ``hermes logs`` - surfaces it. Re-warns automatically if the file changes (different - mtime/size), so users editing the config see the next failure. On the - first warning for a given broken file we also snapshot it to a - timestamped ``.bak`` (best-effort) so the user's recoverable content - survives any later rewrite of ``config.yaml`` by the setup wizard or - ``hermes config set``. - - ``fallback`` selects the message wording: ``"defaults"`` (fresh process, - nothing else to serve) or ``"last-known-good"`` (in-process retention of - the previously loaded config — see the codex#31188 port in - ``_load_config_impl``). + A YAML parse error in ``~/.hermes/config.yaml`` causes ``load_config()`` to silently fall back + to ``DEFAULT_CONFIG``, which means every user override (auxiliary providers, fallback chain, + model overrides, etc.) is dropped. """ try: st = config_path.stat() @@ -184,18 +143,9 @@ def _warn_config_parse_failure( def get_active_config_parse_failure() -> Optional[str]: """Return the parse-error message if the ACTIVE config.yaml is corrupt. - Probes the failure record written by :func:`_warn_config_parse_failure` - for the current :func:`get_config_path` and re-stats the file NOW: the - recorded error is returned only while the file is byte-identical - (mtime_ns + size) to the one that failed to parse. The moment the user - fixes (or deletes) the file, this returns ``None`` without any cache - invalidation dance. - - Consumers: ``hermes_cli.auth.resolve_provider`` uses this to refuse - env-key/pool auto-adoption of a paid provider while the user's real - config — which may name a different provider entirely — is unreadable - (#81952). Returns ``None`` when no failure was ever recorded, when the - file changed since the failure, or on any stat error. + Probes the failure record written by :func:`_warn_config_parse_failure` for the current + :func:`get_config_path` and re-stats the file NOW: the recorded error is returned only while the + file is byte-identical (mtime_ns + size) to the one that failed to parse. """ try: path = get_config_path() @@ -214,44 +164,21 @@ def get_active_config_parse_failure() -> Optional[str]: _IS_WINDOWS = platform.system() == "Windows" _ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") -# Env var names that influence how the next subprocess executes — -# never writable through ``save_env_value``. Anything that controls -# the loader, interpreter, shell, or replacement editor counts: +# Env var names that influence how the next subprocess executes — never writable +# through ``save_env_value``: dynamic loader (LD_*/DYLD_*: attacker code loads +# before main()), interpreter init (PYTHON*, NODE_*: Hermes restarts through +# them), PATH (too broad; fix tool lookup with absolute paths in integration +# config instead), git rewrites (fire on every plugin install / update), +# implicitly-invoked commands (BROWSER/EDITOR/VISUAL/PAGER = RCE on next $EDITOR), +# SHELL (shell=True defense in depth), and Hermes runtime-location flags +# (config.yaml is the supported surface; .env writes would relocate state). # -# * ``LD_PRELOAD`` / ``LD_LIBRARY_PATH`` / ``LD_AUDIT`` — Linux dynamic -# loader. ``DYLD_*`` — macOS equivalent. Planting a path here means -# the next ``subprocess.run([...])`` Hermes makes loads attacker code -# before main(). -# * ``PYTHONPATH`` / ``PYTHONHOME`` / ``PYTHONSTARTUP`` / -# ``PYTHONUSERBASE`` — Python interpreter init. Hermes itself starts -# from one of these on every restart. -# * ``NODE_OPTIONS`` / ``NODE_PATH`` — Node interpreter; affects npm, -# ``hermes update``, the TUI build. -# * ``PATH`` — too broad to allow. The dashboard never needs to rewrite -# the operator's PATH; if a tool can't be found, the fix is to add an -# absolute path in the integration config, not to mutate PATH globally. -# * ``GIT_SSH_COMMAND`` / ``GIT_EXEC_PATH`` — git rewrites that fire -# on every plugin install / ``hermes update``. -# * ``BROWSER`` / ``EDITOR`` / ``VISUAL`` / ``PAGER`` — commands the -# shell or CLI invokes implicitly. Wrong values here = RCE on next -# ``$EDITOR``. -# * ``SHELL`` — what subprocess uses with ``shell=True`` (we try to -# avoid that, but defense in depth). -# * ``HERMES_HOME`` / ``HERMES_PROFILE`` / ``HERMES_CONFIG`` / -# ``HERMES_ENV`` — Hermes runtime location flags. Writing these into -# ``.env`` would relocate state in ways the user did not request from -# the dashboard. ``config.yaml`` is the supported surface for these. -# -# IMPORTANT: ``HERMES_*`` overall is NOT blocked. Many legitimate -# integration credentials follow that prefix (HERMES_LANGFUSE_PUBLIC_KEY, -# HERMES_SPOTIFY_CLIENT_ID, ...). The -# denylist is name-by-name on purpose so the gate stays narrow and -# doesn't accidentally break provider setup wizards. -# -# This is enforced on *write* only — values already in ``.env`` (set -# by the operator out-of-band, or pre-existing) keep working. The -# point is that the dashboard's writable surface cannot escalate by -# planting them. +# IMPORTANT: ``HERMES_*`` overall is NOT blocked — many integration credentials +# use that prefix (HERMES_LANGFUSE_PUBLIC_KEY, HERMES_SPOTIFY_CLIENT_ID, ...). +# The denylist is name-by-name so the gate stays narrow and cannot break provider +# setup wizards. Enforced on *write* only: pre-existing/out-of-band ``.env`` +# values keep working; the point is the dashboard's writable surface cannot +# escalate by planting them. _ENV_VAR_NAME_DENYLIST: frozenset[str] = frozenset({ # Loader / linker "LD_PRELOAD", "LD_LIBRARY_PATH", "LD_AUDIT", "LD_DEBUG", @@ -290,20 +217,16 @@ _ENV_VAR_NAME_DENYLIST: frozenset[str] = frozenset({ def _env_var_policy_name(key: str, *, is_windows: Optional[bool] = None) -> str: """Return the name used for environment policy comparisons. - Windows environment names are case-insensitive; POSIX names are not. The - explicit override keeps both semantics directly testable without pretending - the test interpreter is running on another host OS. + Windows environment names are case-insensitive; POSIX names are not. The explicit override keeps + both semantics directly testable without pretending the test interpreter is running on another + host OS. """ windows = _IS_WINDOWS if is_windows is None else is_windows return key.upper() if windows else key def _reject_denylisted_env_var(key: str) -> None: - """Raise if ``key`` is in :data:`_ENV_VAR_NAME_DENYLIST`. - - Centralised so both the regular and "secure" env writers share the - same gate, and so the message is consistent for callers. - """ + """Raise if ``key`` is in :data:`_ENV_VAR_NAME_DENYLIST`.""" if _env_var_policy_name(key) in _ENV_VAR_NAME_DENYLIST: raise ValueError( f"Environment variable {key!r} is on the writer denylist. " @@ -319,26 +242,23 @@ def _reject_denylisted_env_var(key: str) -> None: def validate_env_var_name_for_write(key: str) -> None: """Validate an environment name before a generic persistence write. - Exposed separately from :func:`save_env_value` so batch-style callers can - validate their complete request before writing the first value. + Exposed separately from :func:`save_env_value` so batch-style callers can validate their + complete request before writing the first value. """ if not _ENV_VAR_NAME_RE.match(key): raise ValueError(f"Invalid environment variable name: {key!r}") _reject_denylisted_env_var(key) _LAST_EXPANDED_CONFIG_BY_PATH: Dict[str, Any] = {} -# (path, mtime_ns, size) -> cached expanded config dict. -# load_config() returns a deepcopy of the cached value when the file -# hasn't changed since the last load, skipping yaml.safe_load + -# _deep_merge + _normalize_* + _expand_env_vars (~13 ms/call). -# save_config() + migrate_config() write via atomic_yaml_write which -# produces a fresh inode, so stat() sees a new mtime_ns and the next -# load repopulates automatically — no explicit invalidation hook. -# Cached tuple is (user_mtime_ns, user_size, managed_mtime_ns, managed_size, -# merged_value, env_ref_snapshot) — the managed-file signature is folded in so -# editing the managed-scope config.yaml invalidates the cache (see -# managed_scope), and the env snapshot invalidates it when a referenced ${VAR} -# changes value (late .env load, in-process rotation — #58514). +# path -> (user_mtime_ns, user_size, managed_mtime_ns, managed_size, merged_value, +# env_ref_snapshot). load_config() returns a deepcopy of the cached value while +# the signature matches, skipping safe_load + _deep_merge + _normalize_* + +# _expand_env_vars (~13 ms/call). save_config()/migrate_config() write via +# atomic_yaml_write (fresh inode → new mtime_ns), so no explicit invalidation +# hook is needed. The managed-file signature is folded in so editing the +# managed-scope config.yaml invalidates the cache, and the env snapshot +# invalidates it when a referenced ${VAR} changes value (late .env load, +# in-process rotation). _LOAD_CONFIG_CACHE: Dict[str, Tuple[int, int, int, int, Dict[str, Any], Dict[str, Optional[str]]]] = {} # (path, mtime_ns, size) -> cached raw yaml dict. Same pattern as # _LOAD_CONFIG_CACHE but for read_raw_config() — used when callers want @@ -355,70 +275,48 @@ _CONFIG_LOCK = threading.RLock() # Env var names written to .env that aren't in OPTIONAL_ENV_VARS # (managed by setup/provider flows directly). _EXTRA_ENV_KEYS = frozenset({ - "OPENAI_API_KEY", "OPENAI_BASE_URL", - "ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", + "OPENAI_API_KEY", "OPENAI_BASE_URL", "ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "DISCORD_HOME_CHANNEL", "DISCORD_HOME_CHANNEL_NAME", "TELEGRAM_HOME_CHANNEL", "TELEGRAM_HOME_CHANNEL_NAME", "SLACK_HOME_CHANNEL", "SLACK_HOME_CHANNEL_NAME", - "SIGNAL_ACCOUNT", "SIGNAL_HTTP_URL", - "SIGNAL_ALLOWED_USERS", "SIGNAL_GROUP_ALLOWED_USERS", - "SIGNAL_HOME_CHANNEL", "SIGNAL_HOME_CHANNEL_NAME", - "SMS_HOME_CHANNEL", "SMS_HOME_CHANNEL_NAME", - "DINGTALK_CLIENT_ID", "DINGTALK_CLIENT_SECRET", - "DINGTALK_HOME_CHANNEL", "DINGTALK_HOME_CHANNEL_NAME", + "SIGNAL_ACCOUNT", "SIGNAL_HTTP_URL", "SIGNAL_ALLOWED_USERS", "SIGNAL_GROUP_ALLOWED_USERS", + "SIGNAL_HOME_CHANNEL", "SIGNAL_HOME_CHANNEL_NAME", "SMS_HOME_CHANNEL", "SMS_HOME_CHANNEL_NAME", + "DINGTALK_CLIENT_ID", "DINGTALK_CLIENT_SECRET", "DINGTALK_HOME_CHANNEL", "DINGTALK_HOME_CHANNEL_NAME", "FEISHU_APP_ID", "FEISHU_APP_SECRET", "FEISHU_ENCRYPT_KEY", "FEISHU_VERIFICATION_TOKEN", - "FEISHU_HOME_CHANNEL", "FEISHU_HOME_CHANNEL_NAME", - "YUANBAO_HOME_CHANNEL", "YUANBAO_HOME_CHANNEL_NAME", - "WECOM_BOT_ID", "WECOM_SECRET", - "WECOM_CALLBACK_CORP_ID", "WECOM_CALLBACK_CORP_SECRET", "WECOM_CALLBACK_AGENT_ID", - "WECOM_CALLBACK_TOKEN", "WECOM_CALLBACK_ENCODING_AES_KEY", - "WECOM_CALLBACK_HOST", "WECOM_CALLBACK_PORT", - "WECOM_HOME_CHANNEL", "WECOM_HOME_CHANNEL_NAME", + "FEISHU_HOME_CHANNEL", "FEISHU_HOME_CHANNEL_NAME", "YUANBAO_HOME_CHANNEL", "YUANBAO_HOME_CHANNEL_NAME", + "WECOM_BOT_ID", "WECOM_SECRET", "WECOM_CALLBACK_CORP_ID", "WECOM_CALLBACK_CORP_SECRET", + "WECOM_CALLBACK_AGENT_ID", "WECOM_CALLBACK_TOKEN", "WECOM_CALLBACK_ENCODING_AES_KEY", + "WECOM_CALLBACK_HOST", "WECOM_CALLBACK_PORT", "WECOM_HOME_CHANNEL", "WECOM_HOME_CHANNEL_NAME", "WEIXIN_ACCOUNT_ID", "WEIXIN_TOKEN", "WEIXIN_BASE_URL", "WEIXIN_CDN_BASE_URL", "WEIXIN_HOME_CHANNEL", "WEIXIN_HOME_CHANNEL_NAME", "WEIXIN_DM_POLICY", "WEIXIN_GROUP_POLICY", "WEIXIN_ALLOWED_USERS", "WEIXIN_GROUP_ALLOWED_USERS", "WEIXIN_ALLOW_ALL_USERS", - "BLUEBUBBLES_SERVER_URL", "BLUEBUBBLES_PASSWORD", - "BLUEBUBBLES_HOME_CHANNEL", "BLUEBUBBLES_HOME_CHANNEL_NAME", + "BLUEBUBBLES_SERVER_URL", "BLUEBUBBLES_PASSWORD", "BLUEBUBBLES_HOME_CHANNEL", "BLUEBUBBLES_HOME_CHANNEL_NAME", "QQ_APP_ID", "QQ_CLIENT_SECRET", "QQBOT_HOME_CHANNEL", "QQBOT_HOME_CHANNEL_NAME", "QQ_HOME_CHANNEL", "QQ_HOME_CHANNEL_NAME", # legacy aliases (pre-rename, still read for back-compat) "QQ_ALLOWED_USERS", "QQ_GROUP_ALLOWED_USERS", "QQ_ALLOW_ALL_USERS", "QQ_MARKDOWN_SUPPORT", "QQ_STT_API_KEY", "QQ_STT_BASE_URL", "QQ_STT_MODEL", - "IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", - "IRC_USE_TLS", "IRC_SERVER_PASSWORD", "IRC_NICKSERV_PASSWORD", - "TERMINAL_ENV", "TERMINAL_SSH_KEY", "TERMINAL_SSH_PORT", - # HERMES_TOOL_PROGRESS_MODE is deprecated (replaced by display.tool_progress - # in config.yaml) but STILL READ at runtime by the gateway as a back-compat - # fallback, so it must stay known to reload/compat paths. The boolean - # HERMES_TOOL_PROGRESS variant is fully unsupported since the v12 config - # support floor retired its only consumer (the v3→4 migration): it is no - # longer listed here and doctor flags it as ignored. + "IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS", "IRC_SERVER_PASSWORD", + "IRC_NICKSERV_PASSWORD", "TERMINAL_ENV", "TERMINAL_SSH_KEY", "TERMINAL_SSH_PORT", + # HERMES_TOOL_PROGRESS_MODE is deprecated (replaced by display.tool_progress) but STILL READ + # by the gateway as a back-compat fallback, so it must stay known to reload/compat paths. The + # boolean HERMES_TOOL_PROGRESS variant is unsupported since the v12 support floor retired its + # only consumer (the v3→4 migration): not listed here; doctor flags it as ignored. "HERMES_TOOL_PROGRESS_MODE", "WHATSAPP_MODE", "WHATSAPP_ENABLED", "MATTERMOST_HOME_CHANNEL", "MATTERMOST_HOME_CHANNEL_NAME", "MATTERMOST_REPLY_MODE", "MATRIX_PASSWORD", "MATRIX_ENCRYPTION", "MATRIX_DEVICE_ID", "MATRIX_HOME_ROOM", "MATRIX_REQUIRE_MENTION", "MATRIX_FREE_RESPONSE_ROOMS", "MATRIX_AUTO_THREAD", "MATRIX_DM_AUTO_THREAD", "MATRIX_RECOVERY_KEY", - # Langfuse observability plugin — optional tuning keys + standard SDK vars. - # Activation is via plugins.enabled (opt-in through `hermes plugins enable - # observability/langfuse` or `hermes tools → Langfuse`); credentials gate - # the plugin at runtime. - "HERMES_LANGFUSE_ENV", - "HERMES_LANGFUSE_RELEASE", - "HERMES_LANGFUSE_SAMPLE_RATE", - "HERMES_LANGFUSE_MAX_CHARS", - "HERMES_LANGFUSE_CAPTURE", - "HERMES_LANGFUSE_DEBUG", - "LANGFUSE_PUBLIC_KEY", - "LANGFUSE_SECRET_KEY", - "LANGFUSE_BASE_URL", - # ACP (Agent Client Protocol) keys — profile-isolable so different - # profiles can use different ACP backends without cross-leak. - "HERMES_ACP_AUTH_METHOD", - "HERMES_ACP_AUTO_APPROVE", - "HERMES_COPILOT_ACP_COMMAND", - "HERMES_COPILOT_ACP_ARGS", - "COPILOT_CLI_PATH", - "COPILOT_ACP_BASE_URL", + # Langfuse observability plugin — optional tuning keys + standard SDK vars. Activation is via + # plugins.enabled (`hermes plugins enable observability/langfuse` / `hermes tools → Langfuse`); + # credentials gate the plugin at runtime. + "HERMES_LANGFUSE_ENV", "HERMES_LANGFUSE_RELEASE", "HERMES_LANGFUSE_SAMPLE_RATE", + "HERMES_LANGFUSE_MAX_CHARS", "HERMES_LANGFUSE_CAPTURE", "HERMES_LANGFUSE_DEBUG", + "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_BASE_URL", + # ACP (Agent Client Protocol) keys — profile-isolable so different profiles can use + # different ACP backends without cross-leak. + "HERMES_ACP_AUTH_METHOD", "HERMES_ACP_AUTO_APPROVE", "HERMES_COPILOT_ACP_COMMAND", + "HERMES_COPILOT_ACP_ARGS", "COPILOT_CLI_PATH", "COPILOT_ACP_BASE_URL", }) import yaml @@ -449,11 +347,8 @@ _IGNORED_MANAGED_VALUES = frozenset({"brew", "homebrew"}) def get_managed_system() -> Optional[str]: """Return the package manager owning this install, if any.""" - raw = os.getenv("HERMES_MANAGED", "").strip() - marker = None - if raw: - marker = raw.lower() - else: + marker = os.getenv("HERMES_MANAGED", "").strip().lower() or None + if marker is None: managed_marker = get_hermes_home() / ".managed" # An interactive shell reads the marker, because it does not see the # HERMES_MANAGED variable of the service. A marker with content @@ -464,24 +359,18 @@ def get_managed_system() -> Optional[str]: except OSError: marker = "" - if marker is None: + if marker is None or marker in _IGNORED_MANAGED_VALUES: return None - - if marker in _IGNORED_MANAGED_VALUES: - return None - if marker == "" or marker in _MANAGED_TRUE_VALUES: return _LEGACY_MANAGED_SYSTEM - return marker def is_managed() -> bool: """Check if Hermes is running in package-manager-managed mode. - Two signals: the HERMES_MANAGED env var (set by the systemd service), - or a .managed marker file in HERMES_HOME (set by the NixOS activation - script, so interactive shells also see it). + Two signals: the HERMES_MANAGED env var (set by the systemd service) or a .managed marker + file in HERMES_HOME (set by the NixOS activation script so interactive shells see it too). """ return get_managed_system() is not None @@ -506,61 +395,23 @@ def get_managed_update_command() -> Optional[str]: def _install_method_project_root(project_root: Optional[Path] = None) -> Path: """Resolve the directory that holds the *running code* (the install tree). - This is the parent of ``hermes_cli/`` — i.e. the git checkout for source - installs, ``/opt/hermes`` inside the published image. It is a property of - the running interpreter, NOT of ``$HERMES_HOME``, which is why a - code-scoped stamp here is immune to two installs sharing one data - directory. + This is the parent of ``hermes_cli/`` — i.e. the git checkout for source installs, + ``/opt/hermes`` inside the published image. It is a property of the running interpreter, NOT of + ``$HERMES_HOME``, which is why a code-scoped stamp here is immune to two installs sharing one + data directory. """ - if project_root is not None: - return project_root - return Path(__file__).parent.parent.resolve() + return project_root if project_root is not None else get_project_root() def detect_install_method(project_root: Optional[Path] = None) -> str: - """Detect how Hermes was installed: 'apt', 'docker', 'nix', 'nixos', - 'home-manager', 'git', or 'unknown'. + """Detect how Hermes was installed: apt/docker/nix/nixos/home-manager/git/unknown. - Resolution order: - 1. Code-scoped stamp ``/.install_method`` (next to the - running code) — the authoritative marker. - 2. Legacy home-scoped stamp ``$HERMES_HOME/.install_method`` — read for - backward compatibility, but a ``docker`` value is IGNORED when we are - not actually running inside a container (see below). - 3. HERMES_MANAGED env / .managed marker (NixOS managed mode) - 4. /nix/store/ path detection -> 'nix' (nix run / nix profile install) - 5. .git directory presence -> 'git' - 6. Fallback -> 'unknown' - - Why the stamp is code-scoped, not home-scoped (issue: shared ``~/.hermes``) - -------------------------------------------------------------------------- - The install method describes *the binary that is running*, but - ``$HERMES_HOME`` is a shared DATA directory — the Docker docs deliberately - bind-mount it (``~/.hermes:/opt/data``) so config/sessions/memory persist - and can be shared with a host-side Desktop/CLI install. When a - containerised gateway and a host install share one ``$HERMES_HOME``, a - home-scoped stamp is a single slot describing two different installs: - the container stamps ``docker`` on every boot, the host install then reads - ``docker`` and ``hermes update`` refuses to run ("doesn't apply inside the - Docker container") even though the host binary is a perfectly updatable - git/pip install. Scoping the stamp to the install tree gives each install - its own truthful marker. - - Self-healing for already-poisoned homes: a legacy ``docker`` value in the - home-scoped stamp is only honoured when we are genuinely in a container. - On a host install that read a contaminating ``docker`` stamp, we fall - through to managed/.git detection instead — so existing shared-home - setups recover without the user touching anything. - - Note: running inside a container is NOT treated as "docker" on its own. - The supported installs self-identify via the code-scoped stamp: - - the curl installer (scripts/install.sh, the README/website install - command) git-clones the repo and stamps ``git`` next to the code; - - the published ``nousresearch/hermes-agent`` image bakes a ``docker`` - stamp into ``/opt/hermes`` at build time. - An unsupported manual install dropped into a container (no stamp) falls - through to the ``.git`` checks and behaves like any off-path install. - See issue #34397. + Order: code-scoped ``/.install_method`` stamp (authoritative) -> legacy + ``$HERMES_HOME/.install_method`` -> managed marker -> /nix/store path -> .git dir -> unknown. + The stamp lives next to the code because HERMES_HOME is shared data: a container and a host + install can bind-mount the same home, so a home-scoped ``docker`` stamp would make the host + ``hermes update`` refuse to run. A legacy ``docker`` value is therefore ignored unless we are + really inside a container, and being in a container alone never implies 'docker'. """ root = _install_method_project_root(project_root) # "apt" is intentionally the Termux APT distribution identifier, not a @@ -572,29 +423,25 @@ def detect_install_method(project_root: Optional[Path] = None) -> str: # home-manager install gives "unknown". supported_methods = {"apt", "docker", "nix", "nixos", "home-manager", "git", "unknown"} + def _stamp(path: Path) -> Optional[str]: + try: + method = path.read_text(encoding="utf-8").strip().lower() + except OSError: + return None + return method if method in supported_methods else None + # 1. Code-scoped stamp — authoritative, immune to shared $HERMES_HOME. - try: - method = (root / ".install_method").read_text(encoding="utf-8").strip().lower() - if method in supported_methods: - return method - except OSError: - pass + method = _stamp(root / ".install_method") + if method: + return method # 2. Legacy home-scoped stamp — back-compat. Ignore a ``docker`` value # when we are not actually containerised: that is the signature of a # host install whose shared $HERMES_HOME was stamped by a co-located # container, and honouring it wrongly blocks ``hermes update``. - try: - method = ( - (get_hermes_home() / ".install_method") - .read_text(encoding="utf-8") - .strip() - .lower() - ) - if method in supported_methods and not (method == "docker" and not _running_in_container()): - return method - except OSError: - pass + method = _stamp(get_hermes_home() / ".install_method") + if method and not (method == "docker" and not _running_in_container()): + return method managed = get_managed_system() if managed: @@ -611,16 +458,14 @@ def detect_install_method(project_root: Optional[Path] = None) -> str: except OSError: pass - # detect git repo installs (normal installer, development env) + # detect git repo installs (normal installer, development env) — a .git + # directory, or a ``gitdir:`` pointer file for worktrees. git_path = root / ".git" if git_path.is_dir(): return "git" - - # detect git repo installs from worktrees if git_path.is_file(): try: - content = git_path.read_text(encoding="utf-8").strip() - if content.startswith("gitdir:"): + if git_path.read_text(encoding="utf-8").strip().startswith("gitdir:"): return "git" except OSError: pass @@ -637,33 +482,12 @@ def _running_in_container() -> bool: return False -def stamp_install_method(method: str, project_root: Optional[Path] = None) -> None: - """Write the install method next to the running code (code-scoped stamp). - - The stamp lives in the install tree (``/.install_method``), - not in ``$HERMES_HOME``, so that two installs sharing one data directory - do not overwrite each other's marker. See ``detect_install_method`` for - the full rationale. - - Best-effort: if the install tree is read-only (e.g. the immutable - ``/opt/hermes`` in the published image, which instead bakes the stamp at - build time) the write silently no-ops and detection falls back to its - other signals. - """ - root = _install_method_project_root(project_root) - try: - root.mkdir(parents=True, exist_ok=True) - (root / ".install_method").write_text(method + "\n", encoding="utf-8") - except OSError: - pass - - def is_nix_install_method(method: str) -> bool: """Return True for every install method that Nix owns. - The callers that branch on the install method must treat "nix", - "nixos" and "home-manager" the same way. One helper keeps the three - names in one place, so a new Nix shape cannot miss a call site. + The callers that branch on the install method must treat "nix", "nixos" and "home-manager" the + same way. One helper keeps the three names in one place, so a new Nix shape cannot miss a call + site. """ return method == "nix" or method in _NIX_MANAGED_SYSTEMS @@ -693,21 +517,12 @@ def recommended_update_command() -> str: return recommended_update_command_for_method(method) -# Long-form text for ``hermes update`` / ``--check`` when running inside the -# Docker image. Surfaced by ``cmd_update`` and ``_cmd_update_check`` in -# hermes_cli/main.py; lives here so the wording stays consistent and we -# don't grow two slightly-different copies. -# -# Why this matters: -# - The published image excludes ``.git`` (see .dockerignore), so the -# git-based update path can never succeed inside the container. -# - The pre-existing fallback message ("✗ Not a git repository. Please -# reinstall: curl ... install.sh") is actively misleading inside Docker -# — that script installs a *new* host-side Hermes, it doesn't update -# the running container. -# - The right action is ``docker pull`` + restart the container; this -# helper spells that out, with notes on tag pinning and config -# persistence so users don't get blindsided. +# Long-form text for ``hermes update`` / ``--check`` inside the Docker image, +# shared by ``cmd_update`` and ``_cmd_update_check`` (hermes_cli/main.py) so the +# wording never forks. The published image excludes ``.git`` (.dockerignore), so +# the git update path can never succeed there, and the generic "Not a git +# repository, reinstall via install.sh" fallback is misleading (that installs a +# NEW host-side Hermes). The right action is ``docker pull`` + restart. _DOCKER_UPDATE_MESSAGE = """\ ✗ ``hermes update`` doesn't apply inside the Docker container. @@ -738,9 +553,8 @@ Notes: def format_docker_update_message() -> str: """Return the user-facing message for ``hermes update`` inside Docker. - Centralised so ``cmd_update`` (the apply path) and ``_cmd_update_check`` - (the dry-run path) share the same wording. See ``_DOCKER_UPDATE_MESSAGE`` - above for the full rationale. + Centralised so ``cmd_update`` (the apply path) and ``_cmd_update_check`` (the dry-run path) + share the same wording. See ``_DOCKER_UPDATE_MESSAGE`` above for the full rationale. """ return _DOCKER_UPDATE_MESSAGE @@ -765,13 +579,9 @@ def managed_error(action: str = "modify configuration"): def get_container_exec_info() -> Optional[dict]: """Read container mode metadata from HERMES_HOME/.container-mode. - Returns a dict with keys: backend, container_name, exec_user, hermes_bin - or None if container mode is not active, we're already inside the - container, or HERMES_DEV=1 is set. - - The .container-mode file is written by the NixOS activation script when - container.enable = true. It tells the host CLI to exec into the container - instead of running locally. + Written by the NixOS activation script when container.enable = true; tells the host CLI to + exec into the container instead of running locally. Returns None when container mode is off, + when already inside the container, or when HERMES_DEV=1 is set. """ if os.environ.get("HERMES_DEV") == "1": return None @@ -794,16 +604,11 @@ def get_container_exec_info() -> Optional[dict]: return None # All other exceptions (PermissionError, malformed data, etc.) propagate - backend = info.get("backend", "docker") - container_name = info.get("container_name", "hermes-agent") - exec_user = info.get("exec_user", "hermes") - hermes_bin = info.get("hermes_bin", "/data/current-package/bin/hermes") - return { - "backend": backend, - "container_name": container_name, - "exec_user": exec_user, - "hermes_bin": hermes_bin, + "backend": info.get("backend", "docker"), + "container_name": info.get("container_name", "hermes-agent"), + "exec_user": info.get("exec_user", "hermes"), + "hermes_bin": info.get("hermes_bin", "/data/current-package/bin/hermes"), } @@ -812,7 +617,7 @@ def get_container_exec_info() -> Optional[dict]: # ============================================================================= # Re-export from hermes_constants — canonical definition lives there. -from hermes_constants import get_hermes_home, get_process_hermes_home # noqa: F811,E402 +from hermes_constants import get_hermes_home, get_process_hermes_home # noqa: E402,F401 from utils import atomic_replace, fast_safe_load def get_config_path() -> Path: @@ -823,13 +628,13 @@ def get_config_path() -> Path: def require_parseable_user_config(*, ignore_user_config: bool = False) -> None: """Reject an existing invalid config before a non-interactive agent run. - Interactive surfaces retain ``load_config()``'s recovery behavior so the - operator can repair their configuration. A one-shot or single-query run - has no such repair opportunity: allowing defaults there can silently pick - a hosted provider/model and spend against credentials loaded from ``.env``. + Interactive surfaces retain ``load_config()``'s recovery behavior so the operator can repair + their configuration. A one-shot or single-query run has no such repair opportunity: allowing + defaults there can silently pick a hosted provider/model and spend against credentials loaded + from ``.env``. - Missing and empty files remain valid first-run states. The explicit - ``--ignore-user-config``/safe-mode escape hatch also remains authoritative. + Missing and empty files remain valid first-run states. The explicit ``--ignore-user- + config``/safe-mode escape hatch also remains authoritative. """ if ignore_user_config or os.environ.get("HERMES_IGNORE_USER_CONFIG") == "1": return @@ -872,44 +677,27 @@ def get_project_root() -> Path: def _resolve_hermes_uid_gid() -> tuple[Optional[int], Optional[int]]: """Read the HERMES_UID / HERMES_GID env vars set by Docker deployments. - Docker containers running Hermes commonly set these to map the in-container - user to a host user so volume-mounted state files end up with the right - ownership. The entrypoint chowns the top-level HERMES_HOME once, but - subdirectories created at runtime by ``ensure_hermes_home()`` (especially - for profile namespaces under ``profiles//``) need the same chown - or they land as ``root:root`` and block subsequent uid-mapped workers - with ``PermissionError [Errno 13]``. See #34107. - - Returns ``(uid, gid)`` parsed from the env vars, or ``(None, None)`` - when either is missing/invalid. Returns ``(None, None)`` on Windows - too (where chown is a no-op anyway). + The entrypoint chowns the top-level HERMES_HOME once, but subdirectories created at runtime + (notably ``profiles//``) need the same chown or they land as root:root and block later + uid-mapped workers with PermissionError. Returns (None, None) if unset/invalid or on Windows. """ if sys.platform == "win32": return None, None - uid_str = os.environ.get("HERMES_UID", "").strip() - gid_str = os.environ.get("HERMES_GID", "").strip() - try: - uid = int(uid_str) if uid_str else None - except ValueError: - uid = None - try: - gid = int(gid_str) if gid_str else None - except ValueError: - gid = None - return uid, gid + def _env_int(name: str) -> Optional[int]: + raw = os.environ.get(name, "").strip() + try: + return int(raw) if raw else None + except ValueError: + return None + + return _env_int("HERMES_UID"), _env_int("HERMES_GID") def _chown_to_hermes_uid(path) -> None: """Chown ``path`` to ``HERMES_UID:HERMES_GID`` if those env vars are set. - No-op when: - - Either env var is unset/invalid - - The current process isn't root (chown will EPERM — silently ignored) - - On Windows (chown semantics don't apply) - - Used by :func:`_secure_dir` to keep ownership consistent across all - directories created by :func:`ensure_hermes_home` on Docker deployments. - See #34107. + No-op when: - Either env var is unset/invalid - The current process isn't root (chown will EPERM + — silently ignored) - On Windows (chown semantics don't apply) """ uid, gid = _resolve_hermes_uid_gid() if uid is None and gid is None: @@ -931,28 +719,24 @@ def _chown_to_hermes_uid(path) -> None: def _secure_dir(path): """Set directory to owner-only access (0700 by default). No-op on Windows. - Skipped in managed mode — the NixOS module sets group-readable - permissions (0750) so interactive users in the hermes group can - share state with the gateway service. + The mode can be overridden via the HERMES_HOME_MODE environment variable (e.g. + HERMES_HOME_MODE=0701) for deployments where a web server (nginx, caddy, etc.) needs to traverse + HERMES_HOME to reach a served subdirectory. The execute-only bit on a directory permits cd- + through without exposing directory listings. - The mode can be overridden via the HERMES_HOME_MODE environment variable - (e.g. HERMES_HOME_MODE=0701) for deployments where a web server (nginx, - caddy, etc.) needs to traverse HERMES_HOME to reach a served subdirectory. - The execute-only bit on a directory permits cd-through without exposing - directory listings. - - Also applies ``HERMES_UID``/``HERMES_GID``-based ownership when those env - vars are set (#34107 — Docker deployments need this so profile subdirs - created at runtime by kanban workers don't land as root:root and block - subsequent uid-mapped workers). + Also applies ``HERMES_UID``/``HERMES_GID``-based ownership when those env vars are set (#34107 — + Docker deployments need this so profile subdirs created at runtime by kanban workers don't land + as root:root and block subsequent uid-mapped workers). """ if is_managed(): return + mode = 0o700 try: mode_str = os.environ.get("HERMES_HOME_MODE", "").strip() - mode = int(mode_str, 8) if mode_str else 0o700 + if mode_str: + mode = int(mode_str, 8) except ValueError: - mode = 0o700 + pass try: os.chmod(path, mode) except (OSError, NotImplementedError): @@ -963,10 +747,8 @@ def _secure_dir(path): def _is_container() -> bool: """Detect if we're running inside a Docker/Podman/LXC container. - When Hermes runs in a container with volume-mounted config files, forcing - 0o600 permissions breaks multi-process setups where the gateway and - dashboard run as different UIDs or the volume mount requires broader - permissions. + Used to skip forcing 0o600 on volume-mounted config: in containers the gateway and dashboard + may run as different UIDs, or the mount itself needs broader permissions. """ # Explicit opt-out if os.environ.get("HERMES_CONTAINER") or os.environ.get("HERMES_SKIP_CHMOD"): @@ -988,11 +770,8 @@ def _is_container() -> bool: def _secure_file(path): """Set file to owner-only read/write (0600). No-op on Windows. - Skipped in managed mode — the NixOS activation script sets - group-readable permissions (0640) on config files. - - Skipped in containers — Docker/Podman volume mounts often need broader - permissions. Set HERMES_SKIP_CHMOD=1 to force-skip on other systems. + Skipped in managed mode (the NixOS activation script sets 0640 group-readable config) and in + containers (volume mounts often need broader permissions). HERMES_SKIP_CHMOD=1 forces a skip. """ if is_managed() or _is_container(): return @@ -1006,10 +785,9 @@ def _secure_file(path): def _ensure_default_soul_md(home: Path) -> None: """Seed a default SOUL.md into HERMES_HOME, upgrading legacy empty templates. - First run: write DEFAULT_SOUL_MD. Existing installs whose SOUL.md is still - the old comment-only scaffold (seeded by older install.sh / install.ps1 / - docker images, which shadowed the runtime default) get upgraded in place to - DEFAULT_SOUL_MD. A SOUL.md the user actually customized is never touched. + First run: write DEFAULT_SOUL_MD. Installs whose SOUL.md is still the old comment-only + scaffold (seeded by older installers/images, shadowing the runtime default) are upgraded in + place. A SOUL.md the user actually customized is never touched. """ soul_path = home / "SOUL.md" if soul_path.exists(): @@ -1033,17 +811,9 @@ _HERMES_HOME_ENSURED: set = set() def ensure_hermes_home(): """Ensure ~/.hermes directory structure exists with secure permissions. - In managed mode (NixOS), dirs are created by the activation script with - setgid + group-writable (2770). We skip mkdir and set umask(0o007) so - any files created (e.g. SOUL.md) are group-writable (0660). - - Memoized per home path: this runs on EVERY ``load_config()`` (inside the - config lock), and the ~14 mkdir/chmod syscalls per call made repeated - config loads the dominant cost of hot read paths like ``model.options``. - After the first successful pass for a given ``HERMES_HOME`` we only re-run - the full walk if the home directory itself has vanished (a deleted home is - recreated on the next load, as before). Profile switches change - ``get_hermes_home()`` and therefore re-run for the new path. + Memoized per home path: this runs on EVERY ``load_config()`` (inside the config lock), and the + ~14 mkdir/chmod syscalls per call made repeated config loads the dominant cost of hot read paths + like ``model.options``. """ home = get_hermes_home() key = str(home) @@ -1126,40 +896,23 @@ REQUIRED_ENV_VARS = {} def get_missing_env_vars(required_only: bool = False) -> List[Dict[str, Any]]: - """ - Check which environment variables are missing. - - Returns list of dicts with var info for missing variables. - """ - missing = [] - - # Check required vars - for var_name, info in REQUIRED_ENV_VARS.items(): - if not get_env_value(var_name): - missing.append({"name": var_name, **info, "is_required": True}) - - # Check optional vars (if not required_only) + """Check which environment variables are missing.""" + groups = [(REQUIRED_ENV_VARS, True)] if not required_only: - for var_name, info in OPTIONAL_ENV_VARS.items(): - if not get_env_value(var_name): - missing.append({"name": var_name, **info, "is_required": False}) - - return missing + groups.append((OPTIONAL_ENV_VARS, False)) + return [ + {"name": var_name, **info, "is_required": is_required} + for table, is_required in groups + for var_name, info in table.items() + if not get_env_value(var_name) + ] def _split_key_path(key: str) -> list[str]: """Split a dotted config-key path, honoring backslash-escaped dots. - ``hermes config set`` uses ``.`` as the nesting separator, so a key that - itself contains a literal dot (e.g. provider names like - ``qwen3.5-397b-wafer``) was silently split into bogus nested segments - (#84064). A backslash escapes a dot:: - - _split_key_path("providers.qwen3\\.5-397b.api_key") - -> ["providers", "qwen3.5-397b", "api_key"] - - Backslashes before any other character are preserved verbatim. Keys - without escapes behave exactly as ``key.split(".")``. + Backslashes before any other character are preserved verbatim. Keys without escapes behave + exactly as ``key.split(".")``. """ parts: list[str] = [] current: list[str] = [] @@ -1184,77 +937,38 @@ def _split_key_path(key: str) -> list[str]: def _greedy_literal_match(container: dict, parts: list) -> Optional[Tuple[str, int]]: """Return ``(literal_key, n_consumed)`` for the longest dotted literal match. - Dots in config key names are the norm, not the exception — model IDs - (``grok-4.6``, ``glm-5.3``), Matrix room IDs (``!room:chat.example.cc``), - and versioned provider names all embed dots. Users typing - ``providers.myprov.models.grok-4.6.context_length`` do not know the - escape syntax exists, so when navigating an EXISTING mapping we prefer an - existing literal key equal to the dot-join of the next N path segments - (longest match wins) over blindly splitting. See #84064 / #80006 / - #91095 / #91607 / #99124. - - Backward compatible: when no multi-segment literal key exists, the single - segment ``parts[0]`` is the only candidate, which is exactly the historic - plain-split behavior. Returns ``None`` when nothing matches. + Backward compatible: when no multi-segment literal key exists, the single segment ``parts[0]`` + is the only candidate, which is exactly the historic plain-split behavior. Returns ``None`` when + nothing matches. """ if not isinstance(container, dict) or not parts: return None - for n in range(len(parts), 0, -1): - candidate = ".".join(parts[:n]) - if candidate in container: - return candidate, n - return None + return next( + ((".".join(parts[:n]), n) for n in range(len(parts), 0, -1) if ".".join(parts[:n]) in container), + None, + ) def _phantom_sibling(container: dict, part: str) -> Optional[str]: """Return an existing sibling key that ``part`` would shadow, if any. - Called when a write is about to CREATE a new intermediate mapping named - ``part``. If the mapping already holds a literal dotted key that starts - with ``part + "."`` (e.g. creating ``grok-4`` beside an existing - ``grok-4.5``), the split almost certainly chopped a dotted leaf name and - the write would produce a phantom sibling the runtime never reads. - Fail loudly instead of silently corrupting (#84064 discussion). + Called before CREATING an intermediate mapping named ``part``. If the mapping already holds + a literal dotted key starting with ``part + '.'`` (creating ``grok-4`` beside ``grok-4.5``), + the split chopped a dotted leaf and the write would produce a phantom sibling the runtime + never reads -- fail loudly instead of silently corrupting. """ if not isinstance(container, dict): return None prefix = part + "." - for k in container: - if isinstance(k, str) and k.startswith(prefix): - return k - return None + return next((k for k in container if isinstance(k, str) and k.startswith(prefix)), None) def _set_nested(config, dotted_key: str, value): """Set a value at an arbitrarily nested dotted key path. - Supports both dict and list navigation: - _set_nested(c, "a.b.c", 1) → c["a"]["b"]["c"] = 1 - _set_nested(c, "a.0.b", 1) → c["a"][0]["b"] = 1 - _set_nested(c, "providers.1", "x") → c["providers"][1] = "x" - - Intermediate dicts are created on demand. List indices are parsed - from numeric path segments; the referenced index must already exist - (we do not grow lists — the user is navigating into structure they - wrote themselves). If a segment targets a non-container leaf - (scalar), the leaf is replaced with a fresh dict so the write can - proceed — this preserves the pre-existing behavior for bare scalar - overrides (e.g. setting ``a.b.c`` where ``a.b`` was previously a - string). - - Guards against #17876: before this fix the code unconditionally - replaced any non-dict value (including lists) with ``{}``, silently - destroying list-typed config like ``custom_providers`` whenever a - caller used an indexed path. - - Dotted key names (#84064 family): when navigating an existing mapping, - an existing literal key equal to the dot-join of the next N segments is - preferred over blind splitting (see ``_greedy_literal_match``), so - ``models.grok-4.6.supports_vision`` lands on the real ``grok-4.6`` entry. - And when a write WOULD create a new intermediate mapping that shadows an - existing dotted sibling (``grok-4`` beside ``grok-4.5``), it raises - ``ValueError`` instead of silently writing a phantom the runtime never - reads. + Intermediate dicts are created on demand. List indices are parsed from numeric path segments; + the referenced index must already exist (we do not grow lists — the user is navigating into + structure they wrote themselves). """ parts = _split_key_path(dotted_key) current = config @@ -1323,11 +1037,10 @@ def clear_model_endpoint_credentials( ) -> Dict[str, Any]: """Remove stale inline endpoint credentials from a model config. - ``model.api_key`` is valid only for explicit custom endpoint assignments. - Built-in providers resolve credentials from env vars, auth.json, or the - credential pool. When switching away from a custom endpoint, leaving these - fields behind keeps secrets in config.yaml and can contaminate later custom - resolution paths. + ``model.api_key`` is valid only for explicit custom endpoint assignments. Built-in providers + resolve credentials from env vars, auth.json, or the credential pool. When switching away from a + custom endpoint, leaving these fields behind keeps secrets in config.yaml and can contaminate + later custom resolution paths. """ if not isinstance(model_cfg, dict): return model_cfg @@ -1344,86 +1057,67 @@ def clear_model_endpoint_credentials( _MISSING = object() -def _get_nested(config, dotted_key: str): - """Return a dotted-path value from nested dict/list config data. +def _locate_nested(config, parts: list): + """Walk *parts* through nested dicts/lists (escape-aware, greedy-literal like ``_set_nested``). - Mirrors ``_set_nested``'s navigation: honors backslash-escaped dots and - prefers an existing literal dotted key over blind splitting, so - ``config get providers.p.models.grok-4.6.context_length`` reads the real - ``grok-4.6`` entry instead of reporting the key unset (#84064). + Returns ``(parents, container, key)`` where ``container[key]`` is the addressed leaf and + ``parents`` lists the ``(container, key)`` hops above it, or ``None`` when any hop is missing, + a list index is non-numeric/out of range, or a scalar is hit before the path is consumed. """ - parts = _split_key_path(dotted_key) + parents = [] current = config i = 0 - while i < len(parts): + while True: remaining = parts[i:] if isinstance(current, list): try: - current = current[int(remaining[0])] + key = int(remaining[0]) + current[key] except (TypeError, ValueError, IndexError): - return _MISSING - i += 1 + return None + consumed = 1 elif isinstance(current, dict): match = _greedy_literal_match(current, remaining) if match is None: - return _MISSING + return None key, consumed = match - current = current[key] - i += consumed else: - return _MISSING - return current + return None + i += consumed + if i == len(parts): + return parents, current, key + parents.append((current, key)) + current = current[key] + + +def _get_nested(config, dotted_key: str): + """Return a dotted-path value from nested dict/list config data. + + Mirrors ``_set_nested`` navigation: honors backslash-escaped dots and prefers an existing + literal dotted key over blind splitting, so ``models.grok-4.6.context_length`` reads the real + ``grok-4.6`` entry instead of reporting the key unset. + """ + loc = _locate_nested(config, _split_key_path(dotted_key)) + if loc is None: + return _MISSING + _, container, key = loc + return container[key] def _unset_nested(config, dotted_key: str) -> bool: """Remove a dotted-path value from nested dict/list config data. - Same escape-aware, greedy-literal navigation as ``_set_nested`` / - ``_get_nested`` (#84064): unsetting an unescaped dotted key removes the - real literal entry rather than a phantom sibling. + Same escape-aware, greedy-literal navigation as ``_set_nested`` / ``_get_nested``: unsetting + an unescaped dotted key removes the real literal entry rather than a phantom sibling. """ - parts = _split_key_path(dotted_key) - if not parts: - return False - - parents = [] - current = config - removed = False - i = 0 - while i < len(parts): - remaining = parts[i:] - if isinstance(current, list): - part = remaining[0] - if len(remaining) == 1: - try: - current.pop(int(part)) - removed = True - break - except (TypeError, ValueError, IndexError): - return False - parents.append((current, part)) - try: - current = current[int(part)] - except (TypeError, ValueError, IndexError): - return False - i += 1 - elif isinstance(current, dict): - match = _greedy_literal_match(current, remaining) - if match is None: - return False - key, consumed = match - if i + consumed == len(parts): - del current[key] - removed = True - break - parents.append((current, key)) - current = current[key] - i += consumed - else: - return False - - if not removed: + loc = _locate_nested(config, _split_key_path(dotted_key)) + if loc is None: return False + parents, current, key = loc + if isinstance(current, list): + current.pop(key) + else: + del current[key] # Drop empty dict containers left behind by the deletion while preserving # user-authored empty lists and non-empty sibling branches. @@ -1431,12 +1125,8 @@ def _unset_nested(config, dotted_key: str) -> bool: if current != {}: break if isinstance(parent, list): - try: - idx = int(part) - except (TypeError, ValueError): - break - if 0 <= idx < len(parent) and parent[idx] == {}: - parent.pop(idx) + if 0 <= part < len(parent) and parent[part] == {}: + parent.pop(part) current = parent continue elif isinstance(parent, dict) and parent.get(part) == {}: @@ -1445,7 +1135,7 @@ def _unset_nested(config, dotted_key: str) -> bool: continue break - return removed + return True def _is_env_config_key(key: str) -> bool: @@ -1474,7 +1164,6 @@ def _is_env_config_key(key: str) -> bool: def _format_config_get_value(value, *, as_json: bool) -> str: """Format a config value for command-line output.""" if as_json: - import json return json.dumps(value, ensure_ascii=False) if isinstance(value, bool): return "true" if value else "false" @@ -1486,12 +1175,7 @@ def _format_config_get_value(value, *, as_json: bool) -> str: def get_missing_config_fields() -> List[Dict[str, Any]]: - """ - Check which config fields are missing or outdated (recursive). - - Walks the DEFAULT_CONFIG tree at arbitrary depth and reports any keys - present in defaults but absent from the user's loaded config. - """ + """Check which config fields are missing or outdated (recursive).""" config = load_config() missing = [] @@ -1514,12 +1198,7 @@ def get_missing_config_fields() -> List[Dict[str, Any]]: def get_missing_skill_config_vars() -> List[Dict[str, Any]]: - """Return skill-declared config vars that are missing or empty in config.yaml. - - Scans all enabled skills for ``metadata.hermes.config`` entries, then checks - which ones are absent or empty under ``skills.config.`` in the user's - config.yaml. Returns a list of dicts suitable for prompting. - """ + """Return skill-declared config vars that are missing or empty in config.yaml.""" try: from agent.skill_utils import discover_all_skill_config_vars, SKILL_CONFIG_PREFIX except Exception: @@ -1531,10 +1210,7 @@ def get_missing_skill_config_vars() -> List[Dict[str, Any]]: # A malformed SKILL.md, unreadable external skill dir, or similar # should never break `hermes update`. Skill-config prompting is a # post-migration nicety, not a blocker. - import logging - logging.getLogger(__name__).debug( - "discover_all_skill_config_vars failed: %s", e - ) + logger.debug("discover_all_skill_config_vars failed: %s", e) return [] if not all_vars: return [] @@ -1542,19 +1218,9 @@ def get_missing_skill_config_vars() -> List[Dict[str, Any]]: config = load_config() missing: List[Dict[str, Any]] = [] for var in all_vars: - # Skill config is stored under skills.config. - storage_key = f"{SKILL_CONFIG_PREFIX}.{var['key']}" - parts = storage_key.split(".") - current = config - value = None - for part in parts: - if isinstance(current, dict) and part in current: - current = current[part] - value = current - else: - value = None - break - # Missing = key doesn't exist or is empty string + # Skill config is stored under skills.config.; + # missing = key doesn't exist or is an empty string. + value = cfg_get(config, *f"{SKILL_CONFIG_PREFIX}.{var['key']}".split(".")) if value is None or (isinstance(value, str) and not value.strip()): missing.append(var) return missing @@ -1607,21 +1273,16 @@ _API_MODE_ALIASES = { 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. + 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. - - PyYAML loads unquoted scalars like ``provider: 2070`` / ``2070:`` as int, - and later ``.strip()`` / ``.lower()`` on that value 500s the Model tab. - """ + """Provider identity fields are strings.""" if value is None: return "" return str(value).strip() @@ -1630,9 +1291,8 @@ def coerce_provider_id(value: Any) -> str: 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; picker code then calls ``ep_name.lower()`` - and CRUD looks up ``"2070"`` and misses. + 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 {} @@ -1647,8 +1307,7 @@ def stringify_provider_map(providers: Any) -> dict: def find_provider_entry(providers: Any, key: Any) -> Tuple[Any, Optional[Dict[str, Any]]]: """Return ``(stored_key, entry)`` matching *key* by string identity. - Needed because PyYAML may have stored the key as ``2070`` (int) while - Desktop looks up ``"2070"``. Prefer an exact string hit, then scan. + Prefer an exact string hit, then scan. """ if not isinstance(providers, dict): return None, None @@ -1664,6 +1323,93 @@ def find_provider_entry(providers: Any, key: Any) -> Tuple[Any, Optional[Dict[st 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, *, @@ -1673,47 +1419,18 @@ def _normalize_custom_provider_entry( if not isinstance(entry, dict): return None - # Shallow-copy before the alias normalization below writes into the - # entry: callers (get_compatible_custom_providers, - # providers_dict_to_custom_providers) pass live sub-dicts from - # load_config_readonly()'s shared cache, and mutating those both - # violates the cache's no-mutation contract and leaks duplicated - # alias keys back into config.yaml through any later - # save_config(load_config()) round-trip. + # 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) - # Accept camelCase aliases commonly used in hand-written 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", - } # 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"] - _KNOWN_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 those (self-written) - # configs don't warn on every load. - "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", - } for camel, snake in _CAMEL_ALIASES.items(): if camel in entry and snake not in entry: _warn_once_per_provider( @@ -1723,7 +1440,7 @@ def _normalize_custom_provider_entry( provider_key or "?", camel, snake, ) entry[snake] = entry[camel] - unknown = set(entry.keys()) - _KNOWN_KEYS - set(_CAMEL_ALIASES.keys()) + unknown = set(entry.keys()) - _KNOWN_PROVIDER_KEYS - set(_CAMEL_ALIASES.keys()) if unknown: _warn_once_per_provider( provider_key, "unknown:" + ",".join(sorted(unknown)), @@ -1731,31 +1448,7 @@ def _normalize_custom_provider_entry( provider_key or "?", ", ".join(sorted(unknown)), ) - from urllib.parse import urlparse - - base_url = "" - for url_key in ("base_url", "url", "api"): - raw_url = entry.get(url_key) - if isinstance(raw_url, str) and raw_url.strip(): - candidate = raw_url.strip() - # Accept URLs containing unresolved placeholder tokens — both - # ``${ENV_VAR}`` env-refs and bare ``{region}``-style templates — - # without URL validation. They are expanded at runtime, so a - # caller reaching this normalizer with raw (un-expanded) config - # would otherwise see the provider silently dropped (#14457). - if re.search(r"\{[^}]+\}", candidate): - base_url = candidate - break - parsed = urlparse(candidate) - if parsed.scheme and parsed.netloc: - base_url = candidate - break - else: - logger.warning( - "providers.%s: '%s' value '%s' is not a valid URL " - "(no scheme or host) — skipped", - provider_key or "?", url_key, candidate, - ) + base_url = _pick_provider_base_url(entry, provider_key) if not base_url: return None @@ -1772,68 +1465,37 @@ def _normalize_custom_provider_entry( if provider_key: normalized["provider_key"] = provider_key - api_key = entry.get("api_key") - if isinstance(api_key, str) and api_key.strip(): - normalized["api_key"] = api_key.strip() + 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 "" - 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 _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.strip() + normalized["api_key_env"] = key_env - api_mode = entry.get("api_mode") or entry.get("transport") - if isinstance(api_mode, str) and api_mode.strip(): + api_mode = _stripped("api_mode", "transport") + if api_mode: normalized["api_mode"] = _canonical_api_mode(api_mode) - model_name = entry.get("model") or entry.get("default_model") - if isinstance(model_name, str) and model_name.strip(): - normalized["model"] = model_name.strip() + 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. - # Older Hermes versions wrote an in-mapping ``__discovered_model_catalog__`` - # sentinel instead; accept it on read and strip it from the models - # mapping so sentinel keys never surface as model IDs. - models_discovered = entry.get("models_discovered") is True - - models = entry.get("models") - if isinstance(models, dict) and models: - # Shallow-copy: `entry` may alias a cached config sub-dict, and the - # normalized entry escapes into long-lived runtime state - # (agent._custom_providers) — don't share the cached models mapping. - models_copy = dict(models) - if models_copy.pop("__discovered_model_catalog__", None) is True: - models_discovered = True - models_copy.pop("__explicit_model_allowlist__", None) - if models_copy: - normalized["models"] = models_copy - elif isinstance(models, list) and models: - # Hand-edited configs (and older Hermes versions) may write - # ``models`` as a plain list of ids or as ``[{id: ...}]`` rows. - # Preserve both by converting to the dict shape downstream code - # expects; otherwise normalize silently drops the list and /model - # shows the provider with (0) 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 - model_meta = { - k: v for k, v in item.items() if k not in {"id", "name"} - } - normalized_models[model_id.strip()] = model_meta - if normalized_models: - normalized["models"] = normalized_models - - if models_discovered: + # 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") @@ -1854,13 +1516,11 @@ def _normalize_custom_provider_entry( if isinstance(rate_limit_delay, (int, float)) and rate_limit_delay >= 0: normalized["rate_limit_delay"] = rate_limit_delay - discover_models = entry.get("discover_models") - if isinstance(discover_models, bool): - normalized["discover_models"] = discover_models + if isinstance(entry.get("discover_models"), bool): + normalized["discover_models"] = entry["discover_models"] - extra_body = entry.get("extra_body") - if isinstance(extra_body, dict): - normalized["extra_body"] = dict(extra_body) + 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 @@ -1898,18 +1558,9 @@ def _custom_provider_entry_to_provider_config( 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", + "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] @@ -1945,10 +1596,9 @@ def get_compatible_custom_providers( ) -> 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. + ``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() @@ -1990,6 +1640,43 @@ def get_compatible_custom_providers( 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 @@ -2010,21 +1697,7 @@ def get_custom_provider_tls_settings( config: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Return TLS settings from a matching ``custom_providers`` / ``providers`` entry.""" - 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) - for entry in custom_providers: - if not isinstance(entry, dict): - continue - entry_url = normalize_route_base_url(entry.get("base_url")) - if not entry_url or entry_url != target_url: - continue + 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(): @@ -2053,14 +1726,8 @@ def apply_custom_provider_tls_to_client_kwargs( def normalize_extra_headers(extra_headers: Any) -> Dict[str, str]: """Normalize a raw ``extra_headers`` value into a ``dict[str, str]``. - Stringifies keys and values and drops entries whose value is ``None``. - Returns ``{}`` for non-dict or empty inputs. This is the single shared - normalizer for per-provider ``extra_headers`` across config normalization, - runtime resolution, client construction, and live ``/models`` discovery. - - SECURITY: header values routinely carry credentials (Cloudflare Access - service tokens, proxy auth, custom bearer schemes). Callers must never - log the returned values. + 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 {} @@ -2074,29 +1741,14 @@ def get_custom_provider_extra_headers( ) -> 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. + 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. + SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy + auth, custom bearer schemes). Callers must never log the returned values. """ - 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) - for entry in custom_providers: - if not isinstance(entry, dict): - continue - entry_url = normalize_route_base_url(entry.get("base_url")) - if not entry_url or entry_url != target_url: - continue + for entry in _entries_for_route(base_url, custom_providers, config): headers = normalize_extra_headers(entry.get("extra_headers")) if headers: return headers @@ -2111,12 +1763,9 @@ def apply_custom_provider_extra_headers_to_client_kwargs( ) -> None: """Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers``. - Provider-specific headers win over provider/SDK defaults already present in - ``client_kwargs`` — they are the most specific configuration level. No-op - when the base_url matches no ``providers`` / ``custom_providers`` entry or - the entry declares no headers. - - SECURITY: values may carry credentials — never log them. + 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: @@ -2134,21 +1783,9 @@ def get_custom_provider_context_length( ) -> 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. - - This is the single source of truth for custom-provider context overrides, - used by: - * ``AIAgent.__init__`` (startup resolution) - * ``AIAgent.switch_model`` (mid-session ``/model`` switch) - * ``hermes_cli.model_switch.resolve_display_context_length`` (``/model`` confirmation display) - * ``gateway.run._format_session_info`` (``/info`` display) - * ``agent.model_metadata.get_model_context_length`` (when custom_providers is threaded through) - - Before this helper existed, the lookup was duplicated in ``run_agent.py``'s - startup path only; every other path (notably ``/model`` switch) fell back - to the 128K default. See #15779. + 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 @@ -2160,24 +1797,10 @@ def get_custom_provider_context_length( return None raw = config.get("custom_providers") custom_providers = raw if isinstance(raw, list) else [] - if not isinstance(custom_providers, list): - return None - target_url = normalize_route_base_url(base_url) - if not target_url: - return None - - for entry in custom_providers: - if not isinstance(entry, dict): - continue - entry_url = normalize_route_base_url(entry.get("base_url")) - if not entry_url or entry_url != target_url: - continue - models = entry.get("models") - if not isinstance(models, dict): - continue - model_cfg = models.get(model) - if not isinstance(model_cfg, dict): + 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: @@ -2200,9 +1823,9 @@ def get_custom_provider_model_capability( ) -> 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``. + 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 @@ -2217,24 +1840,10 @@ def get_custom_provider_model_capability( custom_providers = get_compatible_custom_providers(config) except Exception: return None - if not isinstance(custom_providers, list): - return None - target_url = normalize_route_base_url(base_url) - if not target_url: - return None - - for entry in custom_providers: - if not isinstance(entry, dict): - continue - entry_url = normalize_route_base_url(entry.get("base_url")) - if not entry_url or entry_url != target_url: - continue - models = entry.get("models") - if not isinstance(models, dict): - continue - model_cfg = models.get(model) - if not isinstance(model_cfg, dict): + 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): @@ -2256,33 +1865,26 @@ def _coerce_config_version(value: Any) -> int: def _raw_config_has_explicit_version() -> bool: """True when config.yaml exists, parses, and carries a ``_config_version`` key. - Distinguishes an ANCIENT config (explicit old version → refused by the - v12 support floor) from a fresh minimal/hand-written/cloned config with - no version key at all (→ migrated + stamped normally). Missing or - unparseable files return False so they never trip the floor gate. + Distinguishes an ANCIENT config (explicit old version → refused by the v12 support floor) from a + fresh minimal/hand-written/cloned config with no version key at all (→ migrated + stamped + normally). Missing or unparseable files return False so they never trip the floor gate. """ config_path = get_config_path() if not config_path.exists(): return False try: - with open(config_path, encoding="utf-8") as f: - raw = fast_safe_load(f) or {} + raw = read_user_config_raw(config_path) except Exception: return False - return isinstance(raw, dict) and "_config_version" in raw + return "_config_version" in raw def check_config_version() -> Tuple[int, int]: - """ - Check the raw on-disk config schema version. + """Check the raw on-disk config schema version; returns (current_version, latest_version). - ``load_config()`` deliberately starts from ``DEFAULT_CONFIG`` and deep-merges - the user's file, which is correct for runtime reads but wrong for deciding - whether the user's persisted schema has been migrated. A config file with no - raw ``_config_version`` must remain visible as legacy instead of inheriting - the latest default version in memory. - - Returns (current_version, latest_version). + Reads the raw file rather than ``load_config()``, which deep-merges over ``DEFAULT_CONFIG`` + and would make a file lacking ``_config_version`` inherit the latest version in memory, + hiding that the persisted schema was never migrated. """ latest = _coerce_config_version(DEFAULT_CONFIG.get("_config_version", 1)) or 1 config_path = get_config_path() @@ -2367,13 +1969,28 @@ class ConfigIssue: hint: str +def _require_fields( + issues: List["ConfigIssue"], + entry: Dict[str, Any], + label: str, + fields: Tuple[Tuple[str, str], ...], + suffix: str = "", +) -> None: + """Append a warning for every falsy ``field`` of *entry* (message: ``