hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 31 Packages Projects Releases Wiki Activity

refactor(config): collapse thin _load_config copies onto the canonical readers

Browse Source 
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.
This commit is contained in:
teknium1
2026-09-12 20:28:17 -07:00
committed by Teknium
parent 93d0dba281
commit 469a87f7a5
9 changed files with 27 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
agent/onboarding.py
View File

@@ -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.<flag> = 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")

9
hermes_cli/credential_lifecycle.py
View File

@@ -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] = []

11
hermes_cli/doctor_live.py
View File

@@ -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):

11
hermes_cli/kanban_decompose.py
View File

@@ -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.<key>`` 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(

10
hermes_cli/web_routers/local_models.py
View File

@@ -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

16
tests/hermes_cli/test_doctor_live.py
View File

@@ -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 = {}

2
tests/hermes_cli/test_kanban_decompose.py
View File

@@ -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")

2
tests/tools/test_code_execution.py
View File

@@ -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, {})

6
tools/code_execution_tool.py
View File

@@ -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 {}