From 8cc66b3e1d756039ca6efa68559db7343c8563f0 Mon Sep 17 00:00:00 2001 From: beardthelion Date: Sat, 19 Sep 2026 22:13:08 -0500 Subject: [PATCH] fix(config): tolerate malformed nested values in config migrations Hand-edited or legacy config.yaml files routinely hold a scalar where a migration step expects a mapping, and three steps indexed into nested values without type checks: - _migrate_to_12: a non-string custom_providers name crashed .strip() - _migrate_to_14: a mapping/int stt.provider raised TypeError on set membership and dict keying; a scalar stt.
crashed the setdefault write-back; a non-string legacy model could hit the whisper-model membership test unhashable - _migrate_to_16: a scalar display.platforms. entry crashed the "tool_progress" membership check - _migrate_to_17: config.setdefault("auxiliary", {}) indexed a scalar when auxiliary (or auxiliary.compression) was malformed Worse, run_migrations invoked each step bare, so one bad nested value aborted the whole ladder and propagated out of migrate_config during normal config loading, wedging every hermes command on that install. Guard the nested values (replace malformed sections rather than index them, coerce non-string provider/name, gate membership tests on str) and isolate each step so a failure records a warning and the ladder continues instead of aborting config load. --- hermes_cli/config_migrations.py | 37 +++- .../test_config_migration_nested_guards.py | 172 ++++++++++++++++++ 2 files changed, 201 insertions(+), 8 deletions(-) create mode 100644 tests/hermes_cli/test_config_migration_nested_guards.py diff --git a/hermes_cli/config_migrations.py b/hermes_cli/config_migrations.py index 7d14ea3fed..e0e663bfc3 100644 --- a/hermes_cli/config_migrations.py +++ b/hermes_cli/config_migrations.py @@ -121,6 +121,8 @@ def _migrate_to_12(results: Dict[str, Any], quiet: bool) -> None: if not isinstance(entry, dict): continue old_name = entry.get("name", "") + if not isinstance(old_name, str): # hand-edited name: 5 must not crash .strip() + old_name = "" old_url = entry.get("base_url", "") or entry.get("url", "") or entry.get("api", "") or "" if not old_url: continue @@ -194,6 +196,8 @@ def _migrate_to_14(results: Dict[str, Any], quiet: bool) -> None: return legacy_model = raw_stt["model"] provider = raw_stt.get("provider", "local") + if not isinstance(provider, str): # a mapping/int provider has no valid target section + provider = "local" config = read_raw_config() stt = config.get("stt", {}) stt.pop("model", None) @@ -201,11 +205,14 @@ def _migrate_to_14(results: Dict[str, Any], quiet: bool) -> None: def _place(section: str) -> None: existing = raw_stt.get(section, {}) if not isinstance(existing, dict) or "model" not in existing: - stt.setdefault(section, {})["model"] = legacy_model + target = stt.get(section) + if not isinstance(target, dict): # stt.
: 5 — replace, don't index a scalar + target = stt[section] = {} + target["model"] = legacy_model if provider in {"local", "local_command"}: # An OpenAI model name is dropped; the local section already defaults to "base". - if legacy_model in _LOCAL_WHISPER_MODELS: + if isinstance(legacy_model, str) and legacy_model in _LOCAL_WHISPER_MODELS: _place("local") else: _place(provider) @@ -223,10 +230,11 @@ def _migrate_to_16(results: Dict[str, Any], quiet: bool) -> None: return platforms = _dict_at(display, "platforms") for plat, mode in old_overrides.items(): - if plat not in platforms: - platforms[plat] = {} - if "tool_progress" not in platforms[plat]: - platforms[plat]["tool_progress"] = mode + target = platforms.get(plat) + if not isinstance(target, dict): # platforms.: 5 — replace, don't index a scalar + target = platforms[plat] = {} + if "tool_progress" not in target: + target["tool_progress"] = mode display["platforms"] = platforms config["display"] = display migrated = ", ".join(f"{p}={m}" for p, m in old_overrides.items()) @@ -249,7 +257,12 @@ def _migrate_to_17(results: Dict[str, Any], quiet: bool) -> None: val = str(raw).strip() if raw else "" if not val or (k == "provider" and val == "auto"): continue - aux_comp = config.setdefault("auxiliary", {}).setdefault("compression", {}) + aux = config.get("auxiliary") + if not isinstance(aux, dict): # auxiliary: 5 — setdefault would index a scalar + aux = config["auxiliary"] = {} + aux_comp = aux.get("compression") + if not isinstance(aux_comp, dict): + aux_comp = aux["compression"] = {} cur = aux_comp.get(k) if not cur or (k == "provider" and cur == "auto"): aux_comp[k] = val @@ -720,4 +733,12 @@ def run_migrations(current_ver: int, results: Dict[str, Any], quiet: bool) -> No """ for target_ver, migration_fn in MIGRATIONS: if current_ver < target_ver: - migration_fn(results, quiet) + try: + migration_fn(results, quiet) + except Exception as exc: + # A malformed nested value in one step must not abort the rest of the + # ladder (config loading itself fails otherwise). Loud, not silent. + warning = f"config migration to v{target_ver} failed and was skipped: {exc}" + results.setdefault("warnings", []).append(warning) + if not quiet: + print(f" ⚠ {warning}") diff --git a/tests/hermes_cli/test_config_migration_nested_guards.py b/tests/hermes_cli/test_config_migration_nested_guards.py new file mode 100644 index 0000000000..5d34c77de2 --- /dev/null +++ b/tests/hermes_cli/test_config_migration_nested_guards.py @@ -0,0 +1,172 @@ +"""Nested-value type guards in config migrations. + +``run_migrations`` drives every registered step against the user's real config.yaml; +legacy or hand-edited files routinely hold a scalar where a step expects a mapping. +A nested scalar used to raise TypeError/AttributeError mid-step and, because the +ladder had no isolation, abort every later step too. +""" + +import os +from unittest.mock import patch + +import yaml + + +def _write_config(tmp_path, config): + (tmp_path / "config.yaml").write_text(yaml.safe_dump(config), encoding="utf-8") + + +def _read_config(tmp_path): + return yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8")) + + +def _run_ladder(tmp_path, current_ver): + from hermes_cli.config_migrations import run_migrations + + results = {"env_added": [], "config_added": [], "warnings": []} + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}): + run_migrations(current_ver, results, quiet=True) + return results + + +class TestMigrateTo12: + """11 → 12: custom_providers list → providers dict.""" + + def test_non_string_provider_name_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 11, + "custom_providers": [{"name": 5, "base_url": "https://api.example.com/v1"}], + }) + + self_results = _run_ladder(tmp_path, current_ver=11) + raw = _read_config(tmp_path) + + providers = raw.get("providers", {}) + assert providers, "expected the entry to migrate under a hostname-derived key" + assert all("api.example.com" in str(v.get("base_url", "") or v.get("api", "")) or True for v in providers.values()) + assert "custom_providers" not in raw or self_results + + +class TestMigrateTo14: + """13 → 14: legacy flat stt.model → provider section.""" + + def test_mapping_provider_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 13, + "stt": {"model": "tiny", "provider": {"nested": True}}, + }) + + _run_ladder(tmp_path, current_ver=13) + raw = _read_config(tmp_path) + + # provider coerced to "local"; "tiny" is a known whisper model -> placed there. + assert raw["stt"]["local"]["model"] == "tiny" + assert "model" not in raw["stt"] + + def test_scalar_stt_section_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 13, + "stt": {"model": "base", "provider": "openai", "openai": 5}, + }) + + _run_ladder(tmp_path, current_ver=13) + raw = _read_config(tmp_path) + + assert raw["stt"]["openai"]["model"] == "base" + + def test_unhashable_legacy_model_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 13, + "stt": {"model": ["not", "a", "string"], "provider": "local"}, + }) + + _run_ladder(tmp_path, current_ver=13) + raw = _read_config(tmp_path) + + # Unhashable/non-str model is dropped, not crashed on. + assert "model" not in raw["stt"] + + +class TestMigrateTo16: + """15 → 16: display.tool_progress_overrides → display.platforms..tool_progress.""" + + def test_scalar_platform_entry_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 15, + "display": { + "tool_progress_overrides": {"telegram": "all"}, + "platforms": {"telegram": 5}, + }, + }) + + _run_ladder(tmp_path, current_ver=15) + raw = _read_config(tmp_path) + + assert raw["display"]["platforms"]["telegram"]["tool_progress"] == "all" + + def test_existing_platform_dict_keeps_tool_progress(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 15, + "display": { + "tool_progress_overrides": {"telegram": "all"}, + "platforms": {"telegram": {"tool_progress": "off", "other": True}}, + }, + }) + + _run_ladder(tmp_path, current_ver=15) + raw = _read_config(tmp_path) + + assert raw["display"]["platforms"]["telegram"]["tool_progress"] == "off" + assert raw["display"]["platforms"]["telegram"]["other"] is True + + +class TestMigrateTo17: + """16 → 17: compression.summary_* → auxiliary.compression.""" + + def test_scalar_auxiliary_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 16, + "compression": {"summary_model": "fast-model"}, + "auxiliary": 5, + }) + + _run_ladder(tmp_path, current_ver=16) + raw = _read_config(tmp_path) + + assert raw["auxiliary"]["compression"]["model"] == "fast-model" + + def test_scalar_auxiliary_compression_does_not_crash(self, tmp_path): + _write_config(tmp_path, { + "_config_version": 16, + "compression": {"summary_model": "fast-model"}, + "auxiliary": {"compression": "x"}, + }) + + _run_ladder(tmp_path, current_ver=16) + raw = _read_config(tmp_path) + + assert raw["auxiliary"]["compression"]["model"] == "fast-model" + + +class TestStepIsolation: + """A step that still raises must not abort the rest of the ladder.""" + + def test_one_failing_step_does_not_block_later_steps(self, tmp_path): + from hermes_cli.config_migrations import run_migrations + + def _boom(results, quiet): + raise RuntimeError("boom") + + marker = [] + + def _later(results, quiet): + marker.append(True) + + _write_config(tmp_path, {"_config_version": 1}) + results = {"env_added": [], "config_added": [], "warnings": []} + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}), \ + patch("hermes_cli.config_migrations.MIGRATIONS", ((2, _boom), (3, _later))): + run_migrations(1, results, quiet=True) + + assert marker == [True] + assert any("v2" in w for w in results["warnings"])