refactor(config): add cfg_get() helper; migrate 20 nested-get call sites (#17304)

The "cfg.get('X', {}).get('Y', default)" pattern appears 50+ times
across tools/, gateway/, and plugins/. Each call site manually handles
the same three gotchas:

  1. Missing intermediate key → empty dict → chain works
  2. Non-dict value at intermediate position → AttributeError
     (uncaught in most sites, so a misconfigured YAML crashes the tool)
  3. cfg is None → AttributeError

Introduces cfg_get(cfg, *keys, default=None) in hermes_cli/config.py
as the canonical helper. Handles all three uniformly, returns default
only when the final key is *absent* (matches dict.get semantics —
explicit None values are preserved, falsy values like 0 / False / ''
are preserved).

Named cfg_get rather than cfg_path to avoid shadowing the existing
'cfg_path = _hermes_home / "config.yaml"' local variable that appears
in gateway/run.py, cron/scheduler.py, hermes_cli/main.py, etc.

Migrated 20 call sites as the first-batch proof-of-value:

  gateway/run.py            10 sites (agent/display subtrees)
  tools/browser_tool.py      3 sites
  tools/vision_tools.py      2 sites
  tools/browser_camofox.py   1 site
  tools/approval.py          1 site
  tools/skills_tool.py       1 site
  tools/skill_manager_tool.py 1 site
  tools/credential_files.py  1 site
  tools/env_passthrough.py   1 site

The remaining ~30 sites across plugins/ and smaller tool files can be
migrated opportunistically — the helper is now available and the
pattern is established.

Fixed a latent bug along the way: tools/vision_tools.py had its
cfg_get usage at line 560 inside a function that locally re-imports
'from hermes_cli.config import load_config', but the AST-based
migration script wrote the top-level cfg_get import to a different
function scope, leaving line 560's cfg_get as a NameError silently
swallowed by the surrounding try/except. Test
test_vision_uses_configured_temperature_and_timeout caught it. Fixed
by including cfg_get in the function-local import.

Verified:
- 7880/7893 tests/tools/ + tests/gateway/ + tests/hermes_cli/test_config
  tests pass; all 13 failures pre-existing on main (MCP, delegate,
  session_split_brain — verified earlier in the sweep).
- All 20 migrated sites AST-verified to have cfg_get in scope (either
  module-level or function-local).
- Live 'hermes chat' smoke: 2 turns + /model switch + tool calls +
  /quit, zero errors. Agent correctly counted 20 cfg_get hits across
  8 tool files — matching the migration.

Semantic parity verified against the original pattern across 8 edge
cases (missing keys, None values, falsy values, empty strings, string
instead of dict, None cfg, nested levels).
This commit is contained in:
Teknium 2026-04-28 23:17:39 -07:00 committed by GitHub
parent 5e68503d2f
commit 2d137074a3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 77 additions and 24 deletions

View File

@ -38,6 +38,7 @@ from typing import Dict, Optional, Any, List
# gateway is a long-running daemon, so its boot cost matters less than
# preserving the established test-patch surface.
from agent.account_usage import fetch_account_usage, render_account_usage_lines
from hermes_cli.config import cfg_get
# --- Agent cache tuning ---------------------------------------------------
# Bounds the per-session AIAgent cache to prevent unbounded growth in
@ -1549,7 +1550,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
return (cfg.get("agent", {}).get("system_prompt", "") or "").strip()
return (cfg_get(cfg, "agent", "system_prompt", default="") or "").strip()
except Exception:
pass
return ""
@ -1570,7 +1571,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
effort = str(cfg.get("agent", {}).get("reasoning_effort", "") or "").strip()
effort = str(cfg_get(cfg, "agent", "reasoning_effort", default="") or "").strip()
except Exception:
pass
result = parse_reasoning_effort(effort)
@ -1653,7 +1654,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
raw = str(cfg.get("agent", {}).get("service_tier", "") or "").strip()
raw = str(cfg_get(cfg, "agent", "service_tier", default="") or "").strip()
except Exception:
pass
@ -1674,7 +1675,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
return bool(cfg.get("display", {}).get("show_reasoning", False))
return bool(cfg_get(cfg, "display", "show_reasoning", default=False))
except Exception:
pass
return False
@ -1690,7 +1691,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
mode = str(cfg.get("display", {}).get("busy_input_mode", "") or "").strip().lower()
mode = str(cfg_get(cfg, "display", "busy_input_mode", default="") or "").strip().lower()
except Exception:
pass
if mode == "queue":
@ -1710,7 +1711,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
raw = str(cfg.get("agent", {}).get("restart_drain_timeout", "") or "").strip()
raw = str(cfg_get(cfg, "agent", "restart_drain_timeout", default="") or "").strip()
except Exception:
pass
value = parse_restart_drain_timeout(raw)
@ -1743,7 +1744,7 @@ class GatewayRunner:
if cfg_path.exists():
with open(cfg_path, encoding="utf-8") as _f:
cfg = _y.safe_load(_f) or {}
raw = cfg.get("display", {}).get("background_process_notifications")
raw = cfg_get(cfg, "display", "background_process_notifications")
if raw is False:
mode = "off"
elif raw not in (None, ""):
@ -6449,7 +6450,7 @@ class GatewayRunner:
try:
config = _load_gateway_config()
personalities = config.get("agent", {}).get("personalities", {}) if config else {}
personalities = cfg_get(config, "agent", "personalities", default={})
except Exception:
config = {}
personalities = {}
@ -7450,7 +7451,7 @@ class GatewayRunner:
# --- check config gate ------------------------------------------------
try:
user_config = _load_gateway_config()
gate_enabled = user_config.get("display", {}).get("tool_progress_command", False)
gate_enabled = cfg_get(user_config, "display", "tool_progress_command", default=False)
except Exception:
gate_enabled = False
@ -10069,7 +10070,7 @@ class GatewayRunner:
tool_progress_hint_gateway,
)
_cfg = _load_gateway_config()
gate_on = bool(_cfg.get("display", {}).get("tool_progress_command", False))
gate_on = bool(cfg_get(_cfg, "display", "tool_progress_command", default=False))
if gate_on and not is_seen(_cfg, TOOL_PROGRESS_FLAG):
long_tool_hint_fired[0] = True
progress_queue.put(tool_progress_hint_gateway())

View File

@ -3477,6 +3477,52 @@ def _normalize_max_turns_config(config: Dict[str, Any]) -> Dict[str, Any]:
return config
def cfg_get(cfg: Optional[Dict[str, Any]], *keys: str, default: Any = None) -> Any:
"""Traverse nested dict keys safely, returning ``default`` on any miss.
Canonical helper for the ``cfg.get("X", {}).get("Y", default)`` pattern
that appears 50+ times across the codebase. Handles three common gotchas
in one place:
1. Missing intermediate keys (returns ``default``, no KeyError).
2. An intermediate value that's not a dict (e.g. a user wrote a string
where a section was expected). Returns ``default`` instead of
AttributeError on ``.get()``.
3. ``cfg is None`` (callers sometimes pass ``load_config() or None``).
Named ``cfg_get`` rather than ``cfg_path`` to avoid shadowing the
ubiquitous ``cfg_path = _hermes_home / "config.yaml"`` local variable
that appears in gateway/run.py, cron/scheduler.py, main.py, etc.
Explicit ``None`` values are returned as-is (matches ``dict.get(key,
default)`` semantics ``default`` is only returned when the key is
*absent*, not when it's present but set to ``None``).
Examples:
>>> cfg_get({"agent": {"reasoning_effort": "high"}}, "agent", "reasoning_effort")
'high'
>>> cfg_get({}, "agent", "reasoning_effort", default="medium")
'medium'
>>> cfg_get({"agent": "oops_a_string"}, "agent", "reasoning_effort", default="low")
'low'
>>> cfg_get(None, "anything", default=42)
42
>>> cfg_get({"a": {"b": None}}, "a", "b", default="def") # explicit None preserved
>>> cfg_get({"a": {"b": False}}, "a", "b", default=True) # falsy values preserved
False
"""
if not isinstance(cfg, dict):
return default
node: Any = cfg
for key in keys:
if not isinstance(node, dict):
return default
if key not in node:
return default
node = node[key]
return node
def read_raw_config() -> Dict[str, Any]:
"""Read ~/.hermes/config.yaml as-is, without merging defaults or migrating.

View File

@ -17,6 +17,7 @@ import threading
import time
import unicodedata
from typing import Optional
from hermes_cli.config import cfg_get
logger = logging.getLogger(__name__)
@ -711,7 +712,7 @@ def _get_cron_approval_mode() -> str:
try:
from hermes_cli.config import load_config
config = load_config()
mode = str(config.get("approvals", {}).get("cron_mode", "deny")).lower().strip()
mode = str(cfg_get(config, "approvals", "cron_mode", default="deny")).lower().strip()
if mode in ("approve", "off", "allow", "yes"):
return "approve"
return "deny"

View File

@ -32,7 +32,7 @@ from typing import Any, Dict, Optional
import requests
from hermes_cli.config import load_config
from hermes_cli.config import cfg_get, load_config
from tools.browser_camofox_state import get_camofox_identity
from tools.registry import tool_error
@ -544,7 +544,7 @@ def camofox_vision(question: str, annotate: bool = False,
try:
_cfg = load_config()
_vision_cfg = _cfg.get("auxiliary", {}).get("vision", {})
_vision_cfg = cfg_get(_cfg, "auxiliary", "vision", default={})
_vision_timeout = float(_vision_cfg.get("timeout", 120))
_vision_temperature = float(_vision_cfg.get("temperature", 0.1))
except Exception:

View File

@ -68,6 +68,7 @@ from pathlib import Path
from agent.auxiliary_client import call_llm
from hermes_constants import get_hermes_home
from utils import is_truthy_value
from hermes_cli.config import cfg_get
try:
from tools.website_policy import check_website_access
@ -192,7 +193,7 @@ def _get_command_timeout() -> int:
try:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
val = cfg.get("browser", {}).get("command_timeout")
val = cfg_get(cfg, "browser", "command_timeout")
if val is not None:
result = max(int(val), 5) # Floor at 5s to avoid instant kills
except Exception as e:
@ -2245,7 +2246,7 @@ def _maybe_start_recording(task_id: str):
from hermes_cli.config import read_raw_config
hermes_home = get_hermes_home()
cfg = read_raw_config()
record_enabled = cfg.get("browser", {}).get("record_sessions", False)
record_enabled = cfg_get(cfg, "browser", "record_sessions", default=False)
if not record_enabled:
return
@ -2448,7 +2449,7 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
try:
from hermes_cli.config import load_config
_cfg = load_config()
_vision_cfg = _cfg.get("auxiliary", {}).get("vision", {})
_vision_cfg = cfg_get(_cfg, "auxiliary", "vision", default={})
_vt = _vision_cfg.get("timeout")
if _vt is not None:
vision_timeout = float(_vt)

View File

@ -25,6 +25,7 @@ import os
from contextvars import ContextVar
from pathlib import Path
from typing import Dict, List
from hermes_cli.config import cfg_get
logger = logging.getLogger(__name__)
@ -138,7 +139,7 @@ def _load_config_files() -> List[Dict[str, str]]:
from hermes_cli.config import read_raw_config
hermes_home = _resolve_hermes_home()
cfg = read_raw_config()
cred_files = cfg.get("terminal", {}).get("credential_files")
cred_files = cfg_get(cfg, "terminal", "credential_files")
if isinstance(cred_files, list):
from tools.path_security import validate_within_dir

View File

@ -22,6 +22,7 @@ from __future__ import annotations
import logging
from contextvars import ContextVar
from typing import Iterable
from hermes_cli.config import cfg_get
logger = logging.getLogger(__name__)
@ -109,7 +110,7 @@ def _load_config_passthrough() -> frozenset[str]:
try:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
passthrough = cfg.get("terminal", {}).get("env_passthrough")
passthrough = cfg_get(cfg, "terminal", "env_passthrough")
if isinstance(passthrough, list):
for item in passthrough:
if isinstance(item, str) and item.strip():

View File

@ -43,6 +43,7 @@ from hermes_constants import get_hermes_home, display_hermes_home
from typing import Dict, Any, Optional, Tuple
from utils import atomic_replace
from hermes_cli.config import cfg_get
logger = logging.getLogger(__name__)
@ -66,7 +67,7 @@ def _guard_agent_created_enabled() -> bool:
try:
from hermes_cli.config import load_config
cfg = load_config()
return bool(cfg.get("skills", {}).get("guard_agent_created", False))
return bool(cfg_get(cfg, "skills", "guard_agent_created", default=False))
except Exception:
return False

View File

@ -77,6 +77,7 @@ from pathlib import Path
from typing import Dict, Any, List, Optional, Set, Tuple
from tools.registry import registry, tool_error
from hermes_cli.config import cfg_get
logger = logging.getLogger(__name__)
@ -535,7 +536,7 @@ def _is_skill_disabled(name: str, platform: str = None) -> bool:
skills_cfg = config.get("skills", {})
resolved_platform = platform or os.getenv("HERMES_PLATFORM") or _get_session_platform()
if resolved_platform:
platform_disabled = skills_cfg.get("platform_disabled", {}).get(resolved_platform)
platform_disabled = cfg_get(skills_cfg, "platform_disabled", resolved_platform)
if platform_disabled is not None:
return name in platform_disabled
return name in skills_cfg.get("disabled", [])

View File

@ -56,9 +56,9 @@ def _resolve_download_timeout() -> float:
except ValueError:
pass
try:
from hermes_cli.config import load_config
from hermes_cli.config import cfg_get, load_config
cfg = load_config()
val = cfg.get("auxiliary", {}).get("vision", {}).get("download_timeout")
val = cfg_get(cfg, "auxiliary", "vision", "download_timeout")
if val is not None:
return float(val)
except Exception:
@ -555,9 +555,9 @@ async def vision_analyze_tool(
vision_timeout = 120.0
vision_temperature = 0.1
try:
from hermes_cli.config import load_config
from hermes_cli.config import cfg_get, load_config
_cfg = load_config()
_vision_cfg = _cfg.get("auxiliary", {}).get("vision", {})
_vision_cfg = cfg_get(_cfg, "auxiliary", "vision", default={})
_vt = _vision_cfg.get("timeout")
if _vt is not None:
vision_timeout = float(_vt)