From 469a87f7a549d67ab2323930cb1c05053856c7c9 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sat, 12 Sep 2026 20:28:17 -0700 Subject: [PATCH] refactor(config): collapse thin _load_config copies onto the canonical readers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit doctor_live and kanban_decompose carried byte-identical `try: load_config() or {}` wrappers; local_models wrapped load_config in _quiet; each is now a direct load_config_readonly() call (read-only callers; the canonical already fails open and returns a mapping). Tests that patched the local wrappers patch hermes_cli.config.load_config_readonly instead. tools/code_execution_tool._load_config read the RAW file, so a managed-pinned `code_execution.mode` and the DEFAULT_CONFIG keys were invisible at tool discovery — it now reads load_config_readonly() (behavior change: the managed overlay applies to execute_code's mode/timeout). onboarding.mark_seen and credential_lifecycle's config mirror scrub parsed config.yaml with a bare safe_load; both are read→mutate→write round-trips and use read_user_config_raw, the documented write-back primitive. --- agent/onboarding.py | 10 +++------- hermes_cli/credential_lifecycle.py | 9 ++++----- hermes_cli/doctor_live.py | 11 ++--------- hermes_cli/kanban_decompose.py | 11 ++--------- hermes_cli/web_routers/local_models.py | 10 +++------- tests/hermes_cli/test_doctor_live.py | 16 ++++++++-------- tests/hermes_cli/test_kanban_decompose.py | 2 +- tests/tools/test_code_execution.py | 2 +- tools/code_execution_tool.py | 6 +++--- 9 files changed, 27 insertions(+), 50 deletions(-) diff --git a/agent/onboarding.py b/agent/onboarding.py index e8bb0ded65..bb984d6bb3 100644 --- a/agent/onboarding.py +++ b/agent/onboarding.py @@ -153,16 +153,12 @@ def is_seen(config: Mapping[str, Any], flag: str) -> bool: def mark_seen(config_path: Path, flag: str) -> bool: """Persist ``onboarding.seen. = True`` atomically; False on any error (best-effort).""" try: - import yaml - from hermes_cli.config import atomic_config_write + from hermes_cli.config import atomic_config_write, read_user_config_raw except Exception as e: # pragma: no cover — dependency issue - logger.debug("onboarding: failed to import yaml/utils: %s", e) + logger.debug("onboarding: failed to import config helpers: %s", e) return False try: - cfg: dict = {} - if config_path.exists(): - with open(config_path, encoding="utf-8") as f: - cfg = yaml.safe_load(f) or {} + cfg: dict = read_user_config_raw(config_path) if not isinstance(cfg.get("onboarding"), dict): cfg["onboarding"] = {} seen = cfg["onboarding"].get("seen") diff --git a/hermes_cli/credential_lifecycle.py b/hermes_cli/credential_lifecycle.py index 9a9bfd2027..79079c97ed 100644 --- a/hermes_cli/credential_lifecycle.py +++ b/hermes_cli/credential_lifecycle.py @@ -87,19 +87,18 @@ def _scrub_config_yaml_mirrors(old_value: str, new_value: str | None) -> List[st """ if not old_value: return [] - from utils import atomic_yaml_write, fast_safe_load + from utils import atomic_yaml_write - from hermes_cli.config import get_config_path, require_readable_config_before_write + from hermes_cli.config import get_config_path, read_user_config_raw, require_readable_config_before_write config_path = get_config_path() if not config_path.exists(): return [] try: - with open(config_path, encoding="utf-8") as f: - user_config = fast_safe_load(f) or {} + user_config = read_user_config_raw(config_path) except Exception: return [] - if not isinstance(user_config, dict): + if not user_config: return [] touched: List[str] = [] diff --git a/hermes_cli/doctor_live.py b/hermes_cli/doctor_live.py index 08057cf1a7..b8762dc9ac 100644 --- a/hermes_cli/doctor_live.py +++ b/hermes_cli/doctor_live.py @@ -40,14 +40,6 @@ class ProbeResult: # ── Small seams (monkeypatchable in tests, and single points of control) ── -def _load_config() -> dict: - try: - from hermes_cli.config import load_config - return load_config() or {} - except Exception: - return {} - - def _http_get(url: str, headers: Optional[dict] = None, timeout: Optional[float] = None): """Single HTTP GET seam for all metadata probes.""" import httpx @@ -175,7 +167,8 @@ def _run_one(name: str, fn: Callable[[], ProbeResult], issues: List[str]) -> Pro def run_live_checks(issues: List[str]) -> List[ProbeResult]: """Run one bounded, read-only probe per configured tool backend — sequential by design (predictable output ordering). Appends a remediation line to ``issues`` per failed probe; skipped backends never append.""" - config = _load_config() + from hermes_cli.config import load_config_readonly + config = load_config_readonly() try: timeout = float((config.get("doctor") or {}).get("live_probe_timeout", DEFAULT_PROBE_TIMEOUT)) except (TypeError, ValueError): diff --git a/hermes_cli/kanban_decompose.py b/hermes_cli/kanban_decompose.py index 3bdb89927e..e0ea386a9d 100644 --- a/hermes_cli/kanban_decompose.py +++ b/hermes_cli/kanban_decompose.py @@ -126,14 +126,6 @@ def _profile_author() -> str: return _specify_author("decomposer") -def _load_config() -> dict: - try: - from hermes_cli.config import load_config - return load_config() or {} - except Exception: - return {} - - def _resolve_profile_from_cfg(cfg: dict, key: str) -> str: """``kanban.`` if it names an existing profile, else the active default profile — so a task is never stranded for lack of an owner. @@ -202,7 +194,8 @@ class _Routing: def _load_routing() -> _Routing: - cfg = _load_config() + from hermes_cli.config import load_config_readonly + cfg = load_config_readonly() kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} roster, valid_names = _build_roster() return _Routing( diff --git a/hermes_cli/web_routers/local_models.py b/hermes_cli/web_routers/local_models.py index 9ccbce01ba..a85a8d477d 100644 --- a/hermes_cli/web_routers/local_models.py +++ b/hermes_cli/web_routers/local_models.py @@ -190,12 +190,8 @@ def _router_request(endpoint: Dict[str, Any], path: str, *, timeout: float, payl return None if payload is not None else json.loads(r.read()) -def _load_config() -> dict: - return _quiet(config_mod.load_config, {}) - - def _runtime_section() -> dict: - return (_load_config() or {}).get("local_runtime") or {} + return (config_mod.load_config_readonly() or {}).get("local_runtime") or {} def _set_runtime_enabled(enabled: bool) -> dict: @@ -445,7 +441,7 @@ def _active_llamacpp_model_id() -> str | None: """The active main model when it is one of ours (config authority: the model.provider + model.default that /api/model/set writes).""" def read() -> str | None: - model_section = (_load_config() or {}).get("model") or {} + model_section = (config_mod.load_config_readonly() or {}).get("model") or {} if str(model_section.get("provider", "")).strip().lower() in _LLAMACPP_PROVIDERS: return str(model_section.get("default") or model_section.get("name") or "").strip() or None return None @@ -635,7 +631,7 @@ def _restart_on_new_tag(job: Dict[str, Any], tag: str, previous: list) -> bool: return False _step(job, "restarting", "Switching the running server to the new build") bootstrap.shutdown_local_runtime() - bootstrap.ensure_local_runtime(_load_config(), force=True) + bootstrap.ensure_local_runtime(config_mod.load_config_readonly(), force=True) return True diff --git a/tests/hermes_cli/test_doctor_live.py b/tests/hermes_cli/test_doctor_live.py index 48d4efba0e..49d0f44b0b 100644 --- a/tests/hermes_cli/test_doctor_live.py +++ b/tests/hermes_cli/test_doctor_live.py @@ -34,7 +34,7 @@ def _clean_env(monkeypatch): "ELEVENLABS_API_KEY", "GROQ_API_KEY"): monkeypatch.delenv(var, raising=False) # Default: empty config, no MCP servers, local tts/stt. - monkeypatch.setattr(doctor_live, "_load_config", lambda: {}) + monkeypatch.setattr("hermes_cli.config.load_config_readonly", lambda: {}) # Default: browser not installed. monkeypatch.setattr(doctor_live, "_browser_available", lambda: False) @@ -134,7 +134,7 @@ class TestConfiguredOnlySelection: def test_mcp_servers_probed_per_configured_server(self, monkeypatch): monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"mcp_servers": {"alpha": {"url": "https://x"}, "beta": {"command": "foo"}}}) probed = [] @@ -148,7 +148,7 @@ class TestConfiguredOnlySelection: def test_tts_local_provider_skipped(self, monkeypatch): monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"tts": {"provider": "edge"}}) results = {r.name: r for r in run_live_checks([])} assert results["TTS"].status == "skip" @@ -156,7 +156,7 @@ class TestConfiguredOnlySelection: def test_tts_openai_probed_with_key(self, monkeypatch): monkeypatch.setenv("OPENAI_API_KEY", "sk-test") monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"tts": {"provider": "openai"}}) monkeypatch.setattr( doctor_live, "_http_get", @@ -167,7 +167,7 @@ class TestConfiguredOnlySelection: def test_stt_groq_probed_with_key(self, monkeypatch): monkeypatch.setenv("GROQ_API_KEY", "gsk-test") monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"stt": {"provider": "groq"}}) monkeypatch.setattr( doctor_live, "_http_get", @@ -177,7 +177,7 @@ class TestConfiguredOnlySelection: def test_stt_provider_configured_but_key_missing_warns(self, monkeypatch): monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"stt": {"provider": "groq"}}) results = {r.name: r for r in run_live_checks([])} assert results["STT"].status == "warn" @@ -248,7 +248,7 @@ class TestFailureIsolation: def test_mcp_probe_failure_isolated_per_server(self, monkeypatch): monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"mcp_servers": {"bad": {"url": "https://x"}, "good": {"url": "https://y"}}}) @@ -278,7 +278,7 @@ class TestTimeoutHandling: def test_probe_timeout_bounded_and_configurable(self, monkeypatch): monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test") monkeypatch.setattr( - doctor_live, "_load_config", + "hermes_cli.config.load_config_readonly", lambda: {"doctor": {"live_probe_timeout": 3}}) seen = {} diff --git a/tests/hermes_cli/test_kanban_decompose.py b/tests/hermes_cli/test_kanban_decompose.py index 0b5f57489e..c37ee1aab3 100644 --- a/tests/hermes_cli/test_kanban_decompose.py +++ b/tests/hermes_cli/test_kanban_decompose.py @@ -131,7 +131,7 @@ def test_decompose_fanout_false_invalid_llm_assignee_uses_default(kanban_home): p.start() try: with _patch_aux_client(llm_payload), _patch_extra_body(), patch( - "hermes_cli.kanban_decompose._load_config", + "hermes_cli.config.load_config_readonly", return_value={"kanban": {"default_assignee": "fallback"}}, ): outcome = decomp.decompose_task(tid, author="me") diff --git a/tests/tools/test_code_execution.py b/tests/tools/test_code_execution.py index d8a899d515..da8b97e227 100644 --- a/tests/tools/test_code_execution.py +++ b/tests/tools/test_code_execution.py @@ -715,7 +715,7 @@ class TestLoadConfig(unittest.TestCase): mock_cli = MagicMock() mock_cli.CLI_CONFIG = {"code_execution": {"timeout": 999}} with patch.dict("sys.modules", {"cli": mock_cli}), \ - patch("hermes_cli.config.read_raw_config", return_value={}): + patch("hermes_cli.config.load_config_readonly", return_value={}): result = _load_config() self.assertEqual(result, {}) diff --git a/tools/code_execution_tool.py b/tools/code_execution_tool.py index 2d29ead869..4289081851 100644 --- a/tools/code_execution_tool.py +++ b/tools/code_execution_tool.py @@ -758,11 +758,11 @@ def _kill_process_group(proc, escalate: bool = False): def _load_config() -> dict: - """``code_execution`` config section via the lightweight raw reader — runs while the + """Effective ``code_execution`` section (defaults + user file + managed overlay) — runs while the module-level schema is built at tool discovery, so it must not import ``cli``.""" try: - from hermes_cli.config import read_raw_config - cfg = read_raw_config().get("code_execution", {}) + from hermes_cli.config import load_config_readonly + cfg = load_config_readonly().get("code_execution", {}) return cfg if isinstance(cfg, dict) else {} except Exception: return {}