mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(config): persist last-known-good configuration

Browse source
This commit is contained in:
teknium1 2026-07-09 18:56:15 -07:00
parent a7f65e3bcd
commit 56875f4078
No known key found for this signature in database
3 changed files with 251 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

111
hermes_cli/config.py
View file

@ -38,6 +38,58 @@ logger = logging.getLogger(__name__)
# every time. Cleared automatically when the file changes (different mtime).
_CONFIG_PARSE_WARNED: set = set()
# Profile-local, machine-maintained recovery copy of the last config that both
# parsed and passed structural validation. It intentionally contains the same
# YAML values as config.yaml (including ${VAR} templates), never expanded env
# values, so maintaining it cannot persist secrets that config.yaml did not.
_VALIDATED_CONFIG_SNAPSHOT = "config.validated.yaml"
def _validated_snapshot_path(config_path: Path) -> Path:
return config_path.with_name(_VALIDATED_CONFIG_SNAPSHOT)
def _config_has_validation_errors(config: Dict[str, Any]) -> bool:
return any(issue.severity == "error" for issue in validate_config_structure(config))
def _write_validated_config_snapshot(config_path: Path, config: Dict[str, Any]) -> bool:
"""Atomically refresh the profile-local LKG when *config* validates."""
if not isinstance(config, dict) or _config_has_validation_errors(config):
return False
try:
from utils import atomic_yaml_write
snapshot_path = _validated_snapshot_path(config_path)
atomic_yaml_write(snapshot_path, config)
_secure_file(snapshot_path)
return True
except Exception as exc:
logger.warning("Could not update validated config snapshot for %s: %s", config_path, exc)
return False
def _read_validated_config_snapshot(config_path: Path) -> Optional[Dict[str, Any]]:
"""Read a durable LKG, rejecting missing, corrupt, or invalid snapshots."""
snapshot_path = _validated_snapshot_path(config_path)
try:
with snapshot_path.open(encoding="utf-8") as f:
snapshot = fast_safe_load(f) or {}
if not isinstance(snapshot, dict) or _config_has_validation_errors(snapshot):
raise ValueError("snapshot failed config structure validation")
return snapshot
except FileNotFoundError:
return None
except Exception as exc:
msg = f"last-known-good snapshot is unusable ({snapshot_path}: {exc})"
logger.warning(msg)
try:
sys.stderr.write(f"⚠️ hermes config: {msg}\n")
sys.stderr.flush()
except Exception:
pass
return None
def _backup_corrupt_config(config_path: Path) -> Optional[Path]:
"""Preserve a corrupted ``config.yaml`` by copying it to a timestamped ``.bak``.
@ -135,12 +187,18 @@ def _warn_config_parse_failure(
f"Keeping the previously loaded config for this process — "
f"edits to config.yaml are being IGNORED until the YAML is fixed."
)
elif fallback == "durable-last-known-good":
msg = (
f"Failed to parse {config_path}: {exc}. "
f"USING DURABLE LAST-KNOWN-GOOD CONFIG; the invalid config.yaml is untouched "
f"and ignored until fixed. Existing security policy remains enforced."
)
else:
msg = (
f"Failed to parse {config_path}: {exc}. "
f"Falling back to default config — every user override "
f"(auxiliary providers, fallback chain, model settings) is being IGNORED. "
f"Fix the YAML and restart."
f"NO VALID LAST-KNOWN-GOOD CONFIG EXISTS. Falling back to defaults; "
f"user overrides and security policy may be unavailable. The invalid "
f"config.yaml is untouched. Fix the YAML immediately."
)
if backup_path is not None:
msg += f" A copy of the corrupted file was saved to {backup_path}."
@ -6733,6 +6791,8 @@ def atomic_config_write(config_path: Path, data: Any, **kwargs: Any) -> None:
require_readable_config_before_write(config_path)
atomic_yaml_write(config_path, data, **kwargs)
if isinstance(data, dict):
_write_validated_config_snapshot(config_path, copy.deepcopy(data))
def load_config() -> Dict[str, Any]:
@ -6957,6 +7017,10 @@ def _load_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
user_config.pop("max_turns", None)
config = _deep_merge(config, user_config)
validated_user_config = _normalize_root_model_keys(
_normalize_max_turns_config(copy.deepcopy(user_config))
)
_write_validated_config_snapshot(config_path, validated_user_config)
except Exception as e:
# Last-known-good fallback (port of openai/codex#31188's
# invariant: a parse failure in a policy/config file must not
@ -6966,25 +7030,39 @@ def _load_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
# rules, which are supposed to block commands even under yolo.
# A long-running gateway whose user mid-edits config.yaml into
# broken YAML would silently lose those rules on the next load.
# Within a running process we still have the last successfully
# loaded config — keep serving it until the file is fixed.
# Fresh processes with no last-known-good keep the existing
# DEFAULT_CONFIG fallback.
# Prefer the in-process copy, then the profile-local durable
# snapshot. Durable content is raw/normalized config (env-ref
# templates intact), so merge defaults before expansion.
lkg = _LAST_EXPANDED_CONFIG_BY_PATH.get(path_key)
_warn_config_parse_failure(
config_path,
e,
fallback="last-known-good" if lkg is not None else "defaults",
)
fallback_kind = "last-known-good"
if lkg is None:
durable = _read_validated_config_snapshot(config_path)
if durable is not None:
durable_normalized = _normalize_root_model_keys(
_normalize_max_turns_config(durable)
)
lkg = _deep_merge(copy.deepcopy(DEFAULT_CONFIG), durable_normalized)
fallback_kind = "durable-last-known-good"
else:
fallback_kind = "defaults"
_warn_config_parse_failure(config_path, e, fallback=fallback_kind)
if lkg is not None:
# save_config() stores the pre-expansion normalized dict
# (env-ref templates preserved); the load path stores the
# expanded one. Expand defensively — idempotent when the
# stored value is already expanded.
# Env refs remain templates on disk and are expanded only
# for the returned runtime config.
from typing import cast as _cast
lkg_copy: Dict[str, Any] = _cast(
Dict[str, Any], _expand_env_vars(copy.deepcopy(lkg))
)
if fallback_kind == "durable-last-known-good":
# Managed policy is not duplicated into the profile
# snapshot. Reapply the live managed layer exactly as a
# normal load does so recovery cannot weaken an
# administrator-pinned security setting.
recovery_managed = managed_scope.load_managed_config()
if recovery_managed:
lkg_copy = _deep_merge(
lkg_copy, _expand_env_vars(recovery_managed)
)
if cache_sig is not None:
# Cache under the corrupt file's signature (empty env
# snapshot: always valid) so repeated loads don't
@ -7213,6 +7291,7 @@ def save_config(
extra_content="".join(parts) if parts else None,
)
_secure_file(config_path)
_write_validated_config_snapshot(config_path, copy.deepcopy(normalized))
_LAST_EXPANDED_CONFIG_BY_PATH[str(config_path)] = copy.deepcopy(current_normalized)

2
tests/hermes_cli/test_config.py
View file

@ -160,7 +160,7 @@ class TestLoadConfigParseFailure:
# WARNING-level log was emitted with file path + reason
assert any(
str(tmp_path / "config.yaml") in rec.message
and "Falling back to default config" in rec.message
and "NO VALID LAST-KNOWN-GOOD CONFIG EXISTS" in rec.message
for rec in caplog.records
), f"expected WARNING log, got: {[r.message for r in caplog.records]}"

155
tests/hermes_cli/test_config_lkg.py Normal file
View file

@ -0,0 +1,155 @@
"""Durable last-known-good recovery for profile-local config.yaml."""
import json
import os
import stat
import subprocess
import sys
from pathlib import Path
import yaml
from hermes_cli.config import atomic_config_write, load_config, save_config
LKG_NAME = "config.validated.yaml"
def _run_load(repo: Path, home: Path) -> subprocess.CompletedProcess[str]:
env = os.environ.copy()
env["HERMES_HOME"] = str(home)
env["PYTHONPATH"] = str(repo)
return subprocess.run(
[
sys.executable,
"-c",
"import json; from hermes_cli.config import load_config; "
"print(json.dumps(load_config()))",
],
cwd=repo,
env=env,
text=True,
capture_output=True,
check=True,
)
def test_successful_load_atomically_maintains_restrictive_profile_snapshot(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text(
"model:\n default: test/secure\napprovals:\n deny:\n - 'curl*evil*'\n",
encoding="utf-8",
)
loaded = load_config()
snapshot = tmp_path / LKG_NAME
assert loaded["approvals"]["deny"] == ["curl*evil*"]
assert yaml.safe_load(snapshot.read_text(encoding="utf-8"))["approvals"]["deny"] == [
"curl*evil*"
]
if os.name == "posix":
assert stat.S_IMODE(snapshot.stat().st_mode) == 0o600
assert not list(tmp_path.glob(f".{snapshot.stem}_*.tmp"))
def test_fresh_process_uses_durable_snapshot_and_leaves_broken_config_untouched(tmp_path):
repo = Path(__file__).resolve().parents[2]
config_path = tmp_path / "config.yaml"
good = "model:\n default: test/secure\napprovals:\n deny:\n - 'curl*evil*'\n"
config_path.write_text(good, encoding="utf-8")
first = _run_load(repo, tmp_path)
assert json.loads(first.stdout)["approvals"]["deny"] == ["curl*evil*"]
broken = "approvals:\n deny: [unclosed\n"
config_path.write_text(broken, encoding="utf-8")
second = _run_load(repo, tmp_path)
recovered = json.loads(second.stdout)
assert recovered["model"]["default"] == "test/secure"
assert recovered["approvals"]["deny"] == ["curl*evil*"]
assert config_path.read_text(encoding="utf-8") == broken
assert "DURABLE LAST-KNOWN-GOOD" in second.stderr
assert "security policy" in second.stderr
def test_absent_or_corrupt_snapshot_falls_back_loudly_without_touching_config(tmp_path):
repo = Path(__file__).resolve().parents[2]
broken = "model: [unclosed\n"
(tmp_path / "config.yaml").write_text(broken, encoding="utf-8")
absent = _run_load(repo, tmp_path)
assert "model" in json.loads(absent.stdout)
assert "NO VALID LAST-KNOWN-GOOD" in absent.stderr
assert (tmp_path / "config.yaml").read_text(encoding="utf-8") == broken
(tmp_path / LKG_NAME).write_text("also: [broken\n", encoding="utf-8")
corrupt = _run_load(repo, tmp_path)
assert "model" in json.loads(corrupt.stdout)
assert "last-known-good snapshot is unusable" in corrupt.stderr
assert (tmp_path / LKG_NAME).read_text(encoding="utf-8") == "also: [broken\n"
def test_save_and_shared_atomic_writer_refresh_snapshot(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
save_config({"model": {"default": "test/saved"}, "approvals": {"deny": ["rm *"]}})
snapshot = yaml.safe_load((tmp_path / LKG_NAME).read_text(encoding="utf-8"))
assert snapshot["model"]["default"] == "test/saved"
assert snapshot["approvals"]["deny"] == ["rm *"]
atomic_config_write(
tmp_path / "config.yaml",
{"model": {"default": "test/atomic"}, "approvals": {"deny": ["sudo *"]}},
)
snapshot = yaml.safe_load((tmp_path / LKG_NAME).read_text(encoding="utf-8"))
assert snapshot["model"]["default"] == "test/atomic"
assert snapshot["approvals"]["deny"] == ["sudo *"]
def test_snapshot_is_profile_local(tmp_path, monkeypatch):
one = tmp_path / "profiles" / "one"
two = tmp_path / "profiles" / "two"
one.mkdir(parents=True)
two.mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(one))
save_config({"model": {"default": "test/one"}})
monkeypatch.setenv("HERMES_HOME", str(two))
save_config({"model": {"default": "test/two"}})
assert yaml.safe_load((one / LKG_NAME).read_text())["model"]["default"] == "test/one"
assert yaml.safe_load((two / LKG_NAME).read_text())["model"]["default"] == "test/two"
def test_snapshot_preserves_env_reference_instead_of_expanded_secret(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("PRIVATE_TOKEN", "expanded-secret-must-not-be-persisted")
(tmp_path / "config.yaml").write_text(
"custom_providers:\n"
" - name: private\n"
" base_url: https://example.invalid/v1\n"
" api_key: ${PRIVATE_TOKEN}\n",
encoding="utf-8",
)
loaded = load_config()
assert loaded["custom_providers"][0]["api_key"] == "expanded-secret-must-not-be-persisted"
snapshot_text = (tmp_path / LKG_NAME).read_text(encoding="utf-8")
assert "${PRIVATE_TOKEN}" in snapshot_text
assert "expanded-secret-must-not-be-persisted" not in snapshot_text
def test_structurally_invalid_config_does_not_replace_validated_snapshot(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
config_path = tmp_path / "config.yaml"
config_path.write_text("model:\n default: test/good\n", encoding="utf-8")
load_config()
before = (tmp_path / LKG_NAME).read_text(encoding="utf-8")
# Parses as YAML, but validation classifies this shape as an error.
config_path.write_text("custom_providers:\n name: broken-shape\n", encoding="utf-8")
load_config()
assert (tmp_path / LKG_NAME).read_text(encoding="utf-8") == before