fix(config): detect file replacements in stat-based cache signatures

Config caches, the profile re-scan watcher, the MCP reconciler, managed-scope
reads, the completer memo, OAuth heal marks, and the skill-manifest snapshot
keyed change detection on (st_mtime_ns, st_size) only, so a replacement that
preserved both (cp -p, rsync -t, timestamp-pinning scripts, sync clients) was
treated as unchanged and stale values were served until restart.

Add st_ino (fresh inode on atomic replace) and st_ctime_ns (cannot be
backdated via os.utime) to every signature via a shared
hermes_cli.managed_scope.file_signature() helper.

Fixes #111105

agent/prompt_builder.py

@@ -1089,8 +1089,10 @@ def clear_skills_system_prompt_cache(*, clear_snapshot: bool = False) -> None:
 
 
 def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
-    """mtime/size manifest of every SKILL.md and DESCRIPTION.md; only the ACTIVE org mirror participates, and
+    """File-signature manifest of every SKILL.md and DESCRIPTION.md; only the ACTIVE org mirror participates, and
     the ``.active_org`` marker is included so switching/leaving an org invalidates the snapshot by itself."""
+    from hermes_cli.managed_scope import file_signature
+
     manifest: dict[str, list[int]] = {}
     skills_dir_str = str(skills_dir)
     prefix_len = len(os.path.join(skills_dir_str, ""))
@@ -1098,7 +1100,7 @@ def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
     org_root = os.path.join(skills_dir_str, ORG_MIRROR_DIR_NAME)
     try:
         st = os.stat(os.path.join(org_root, ORG_ACTIVE_MARKER))
-        manifest[ORG_MIRROR_DIR_NAME + "/" + ORG_ACTIVE_MARKER] = [int(st.st_mtime), int(st.st_size)]
+        manifest[ORG_MIRROR_DIR_NAME + "/" + ORG_ACTIVE_MARKER] = list(file_signature(st))
     except OSError:
         pass
     for root, dirs, files in os.walk(skills_dir_str, followlinks=True):
@@ -1113,7 +1115,7 @@ def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
             try:
                 if filename in files:
                     st = os.stat(path)
-                    manifest[path[prefix_len:]] = [st.st_mtime_ns, st.st_size]
+                    manifest[path[prefix_len:]] = list(file_signature(st))
             except OSError:
                 pass
     return manifest

gateway/run_profile_reconcile.py

@@ -22,6 +22,7 @@ from pathlib import Path
 from typing import Any, Dict, Optional
 
 from gateway.run_shutdown import _log_suppressed
+from hermes_cli.managed_scope import file_signature
 
 logger = logging.getLogger(__name__)
 
@@ -30,12 +31,12 @@ _PROFILE_SIGNATURE_FILES = ("config.yaml", ".env")
 
 
 def profile_serve_signature(home: "Path") -> tuple:
-    """Cheap change detector for a served profile's credentials/config: (mtime_ns, size) per file."""
+    """Cheap change detector for a served profile's credentials/config: file signature per file."""
     sig = []
     for name in _PROFILE_SIGNATURE_FILES:
         try:
             st = os.stat(Path(home) / name)
-            sig.append((st.st_mtime_ns, st.st_size))
+            sig.append(file_signature(st))
         except OSError:
             sig.append(None)
     return tuple(sig)
@@ -220,7 +221,7 @@ def _mcp_config_reconciler(runner=None):
     """Housekeeping chore keeping live MCP servers in step with ``mcp_servers`` on disk: an entry
     the user removed (or disabled) after boot must stop — a parked one otherwise self-probes every
     ``_PARKED_RETRY_INTERVAL`` for the life of the process. One ``stat`` per profile per tick; the
-    reconcile runs when ``config.yaml``'s (mtime, size) changed, and again on the next tick while a
+    reconcile runs when ``config.yaml``'s signature changed, and again on the next tick while a
     dropped server was still mid-connect (``pending``) and could not be torn down yet. Interactive
     OAuth is suppressed — this runs on a housekeeping thread nobody is watching."""
     from hermes_cli.config import get_config_path
@@ -230,9 +231,9 @@ def _mcp_config_reconciler(runner=None):
     def _sig(path) -> tuple:
         try:
             st = os.stat(path)
-            return (st.st_mtime_ns, st.st_size)
+            return file_signature(st)
         except OSError:
-            return (None, None)
+            return (None, None, None, None)
 
     def _reconcile_current(label: str) -> None:
         from tools.mcp_oauth import suppress_interactive_oauth

hermes_cli/auth_oauth_grants.py

@@ -12,6 +12,7 @@ import os
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 from hermes_cli.auth_constants import _decode_jwt_claims
+from hermes_cli.managed_scope import file_signature
 
 # Log-record parity with the origin module (caplog tests pin "hermes_cli.auth").
 logger = logging.getLogger("hermes_cli.auth")
@@ -382,14 +383,14 @@ def _heal_forked_provider_block(
     return adopted
 
 
-def _stat_sig(p: Optional[Path]) -> Optional[Tuple[int, int]]:
-    """``(mtime_ns, size)`` from ONE stat, or None when absent — a torn pair from two stats could
+def _stat_sig(p: Optional[Path]) -> Optional[Tuple[int, int, int, int]]:
+    """File signature from ONE stat, or None when absent — a torn pair from two stats could
     leave a persisted clean mark matching a store rewritten between them."""
     try:
         st = p.stat() if p is not None else None
     except OSError:
         return None
-    return (st.st_mtime_ns, st.st_size) if st is not None else None
+    return file_signature(st) if st is not None else None
 
 
 def _pool_rows(store: Dict[str, Any], provider_id: str) -> Tuple[Any, List[Any]]:
@@ -573,9 +574,9 @@ def _heal_forked_single_use_oauth_grants(provider_id: str) -> Optional[Dict[str,
     # this heal consolidates root -> profile, so a new forked grant appearing in ROOT must
     # invalidate the mark. The in-memory mark omitted it and got away with it because it died
     # with the process; a persisted mark would otherwise keep skipping a heal that has become
-    # necessary. Sizes ride along with the mtimes for the same reason: a metadata-preserving
-    # rewrite (``rsync -t``, ``tar -p``, a restore) would otherwise leave a stale mark looking
-    # current indefinitely rather than for one process.
+    # necessary. Sizes, inodes and ctimes ride along with the mtimes for the same reason:
+    # a metadata-preserving rewrite (``rsync -t``, ``tar -p``, a restore) would otherwise leave
+    # a stale mark looking current indefinitely rather than for one process.
     fingerprint = (
         str(profile_path), _stat_sig(profile_path), _stat_sig(profile_singleton),
         str(root_path), _stat_sig(root_path), _stat_sig(root_singleton),

hermes_cli/commands_completion.py

@@ -25,19 +25,20 @@ _personalities_memo: Optional[
 
 
 def _personalities_from_cli_config() -> Dict[str, Any]:
-    """``available_personalities(load_cli_config())`` memoised on config path+mtime+size:
+    """``available_personalities(load_cli_config())`` memoised on config path+signature:
     load_cli_config() is a full YAML parse + deep merge and the completer runs per keystroke.
     Falls back to a fresh load when the file cannot be stat'ed."""
     global _personalities_memo
     from cli import load_cli_config
+    from hermes_cli.managed_scope import file_signature
     from hermes_cli.personality import available_personalities
     try:
         from hermes_cli.config import get_config_path
         cfg_path = get_config_path()
         st = cfg_path.stat()
-        sig = (str(cfg_path), st.st_mtime_ns, st.st_size)
+        sig = (str(cfg_path), *file_signature(st))
     except Exception:
-        sig = (None, None, None)
+        sig = (None, None, None, None, None)
     if _personalities_memo is None or _personalities_memo[0] != sig:
         _personalities_memo = (sig, available_personalities(load_cli_config()))
     return _personalities_memo[1]

hermes_cli/config.py

@@ -93,10 +93,11 @@ def _warn_config_parse_failure(
     """
     try:
         st = config_path.stat()
-        key = (str(config_path), st.st_mtime_ns, st.st_size)
-        _CONFIG_PARSE_FAILURES[str(config_path)] = (st.st_mtime_ns, st.st_size, str(exc))
+        sig = managed_scope.file_signature(st)
+        key = (str(config_path), *sig)
+        _CONFIG_PARSE_FAILURES[str(config_path)] = (*sig, str(exc))
     except OSError:
-        key = (str(config_path), 0, 0)
+        key = (str(config_path), 0, 0, 0, 0)
     if key in _CONFIG_PARSE_WARNED:
         return
     _CONFIG_PARSE_WARNED.add(key)
@@ -115,12 +116,11 @@ def _warn_config_parse_failure(
 
 def get_active_config_parse_failure() -> Optional[str]:
     """Return the recorded parse error while the ACTIVE config.yaml is still byte-identical
-    (mtime_ns + size) to the file that failed to parse; else None."""
+    (mtime_ns + size + ino + ctime_ns) to the file that failed to parse; else None."""
     try:
-        path = get_config_path()
-        mtime_ns, size, err = _CONFIG_PARSE_FAILURES[str(path)]
+        record = _CONFIG_PARSE_FAILURES[str(path := get_config_path())]
         st = path.stat()
-        return err if (st.st_mtime_ns, st.st_size) == (mtime_ns, size) else None
+        return record[4] if managed_scope.file_signature(st) == record[:4] else None
     except Exception:
         return None
 
@@ -201,11 +201,11 @@ _LAST_EXPANDED_CONFIG_BY_PATH: Dict[str, Any] = {}
 # (path, mtime_ns, size) -> cached expanded config dict. load_config() returns a deepcopy of the cached
 # value when the file hasn't changed since the last load, skipping yaml.safe_load + _deep_merge +
 # _normalize_* + _expand_env_vars (~13 ms/call). save_config() + migrate_config() write via
-# atomic_yaml_write which produces a fresh inode, so stat() sees a new mtime_ns and the next load
+# atomic_yaml_write which produces a fresh inode, so stat() sees a new signature and the next load
 # repopulates automatically — no explicit invalidation hook. See #58514.
-_LOAD_CONFIG_CACHE: Dict[str, Tuple[int, int, int, int, Dict[str, Any], Dict[str, Optional[str]]]] = {}
+_LOAD_CONFIG_CACHE: Dict[str, Tuple[int, ...]] = {}
 # path -> (mtime_ns, size, raw yaml dict) for read_raw_config() (no defaults merged in).
-_RAW_CONFIG_CACHE: Dict[str, Tuple[int, int, Dict[str, Any]]] = {}
+_RAW_CONFIG_CACHE: Dict[str, Tuple[int, ...]] = {}
 
 # Env var names written to .env that aren't in OPTIONAL_ENV_VARS (managed by setup/provider
 # flows directly). Also the set reload_env() may remove from os.environ.
@@ -1901,14 +1901,14 @@ def _read_raw_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
         try:
             config_path = get_config_path()
             st = config_path.stat()
-            cache_key = (st.st_mtime_ns, st.st_size)
+            cache_key = managed_scope.file_signature(st)
         except (FileNotFoundError, OSError):
             return {}
 
         path_key = str(config_path)
         cached = _RAW_CONFIG_CACHE.get(path_key)
-        if cached is not None and cached[:2] == cache_key:
-            return copy.deepcopy(cached[2]) if want_deepcopy else cached[2]
+        if cached is not None and cached[:len(cache_key)] == cache_key:
+            return copy.deepcopy(cached[len(cache_key)]) if want_deepcopy else cached[len(cache_key)]
 
         try:
             with open(config_path, encoding="utf-8") as f:
@@ -1922,13 +1922,13 @@ def _read_raw_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
         # The cache stores its own deepcopy. The readonly path returns THAT object (identity
         # invariant: later cache hits return the same dict); the mutable path returns the parse.
         cached_copy = copy.deepcopy(data)
-        _RAW_CONFIG_CACHE[path_key] = (cache_key[0], cache_key[1], cached_copy)
+        _RAW_CONFIG_CACHE[path_key] = (*cache_key, cached_copy)
         return data if want_deepcopy else cached_copy
 
 
 def read_raw_config() -> Dict[str, Any]:
     """Read config.yaml as-is (no defaults merged, no migration); ``{}`` if missing/unparseable.
-    Cached on (mtime_ns, size); returns a deepcopy since callers mutate before ``save_config()``."""
+    Cached on the file signature (mtime_ns, size, ino, ctime_ns); returns a deepcopy since callers mutate before ``save_config()``."""
     return _read_raw_config_impl(want_deepcopy=True)
 
 
@@ -2139,24 +2139,24 @@ def apply_terminal_config_to_env(
     return target
 
 
-def _load_config_cache_sig(config_path: Path) -> Tuple[Optional[Tuple[int, int]], Optional[Tuple[int, int, int, int]]]:
+def _load_config_cache_sig(config_path: Path) -> Tuple[Optional[Tuple[int, int, int, int]], Optional[Tuple[int, ...]]]:
     """Return ``(user_sig, cache_sig)`` for ``_LOAD_CONFIG_CACHE``.
-    The managed config file's (mtime, size) is folded in ((0, 0) = none) so editing it invalidates
+    The managed config file's signature is folded in ((0, 0, 0, 0) = none) so editing it invalidates
     the merged result. ``cache_sig`` is None only when neither file exists (nothing to cache on)."""
     try:
         st = config_path.stat()
-        user_sig: Optional[Tuple[int, int]] = (st.st_mtime_ns, st.st_size)
+        user_sig: Optional[Tuple[int, int, int, int]] = managed_scope.file_signature(st)
     except FileNotFoundError:
         user_sig = None
     managed_dir = managed_scope.get_managed_dir()
     try:
         mst = (managed_dir / "config.yaml").stat() if managed_dir else None
-        managed_sig = (mst.st_mtime_ns, mst.st_size) if mst else (0, 0)
+        managed_sig = managed_scope.file_signature(mst) if mst else (0, 0, 0, 0)
     except OSError:
-        managed_sig = (0, 0)
-    if user_sig is None and managed_sig == (0, 0):
+        managed_sig = (0, 0, 0, 0)
+    if user_sig is None and managed_sig == (0, 0, 0, 0):
         return None, None
-    return user_sig, (*(user_sig or (0, 0)), *managed_sig)
+    return user_sig, (*(user_sig or (0, 0, 0, 0)), *managed_sig)
 
 
 def _last_known_good_fallback(config_path: Path, path_key: str, cache_sig, exc: Exception) -> Optional[Dict[str, Any]]:
@@ -2223,15 +2223,15 @@ def _load_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
         user_sig, cache_sig = _load_config_cache_sig(config_path)
 
         cached = _LOAD_CONFIG_CACHE.get(path_key)
-        if cached is not None and cache_sig is not None and cached[:4] == cache_sig:
+        if cached is not None and cache_sig is not None and cached[:8] == cache_sig:
             # Signatures match, but the cached expansion is only valid if every ${VAR} it was
             # expanded against still has the same value — otherwise a load before
             # load_hermes_dotenv() pins unexpanded literals for the process lifetime.
             # Without this, a load_config() that ran before load_hermes_dotenv() pins unexpanded literals
             # (e.g. auxiliary.<task>.api_key) for the life of the process (#58514).
-            env_snapshot = cached[5] if len(cached) > 5 else {}
+            env_snapshot = cached[9] if len(cached) > 9 else {}
             if all(_env_ref_lookup(k) == v for k, v in env_snapshot.items()):
-                return copy.deepcopy(cached[4]) if want_deepcopy else cached[4]
+                return copy.deepcopy(cached[8]) if want_deepcopy else cached[8]
 
         config = copy.deepcopy(DEFAULT_CONFIG)
 

hermes_cli/managed_scope.py

@@ -13,10 +13,21 @@ import logging
 import os
 import threading
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Tuple
 
 import yaml
 
+
+def file_signature(st: "os.stat_result") -> Tuple[int, int, int, int]:
+    """Stat signature for cache invalidation: ``(st_mtime_ns, st_size, st_ino, st_ctime_ns)``.
+
+    ``mtime_ns`` + ``size`` alone miss replacements that preserve both (``cp -p``,
+    ``rsync -t``, timestamp-pinning scripts, sync clients). ``st_ino`` changes on an
+    atomic replace (fresh inode); ``st_ctime_ns`` cannot be backdated via ``os.utime``,
+    catching writers that pin mtime. macOS and Linux both expose these fields.
+    """
+    return (st.st_mtime_ns, st.st_size, st.st_ino, st.st_ctime_ns)
+
 logger = logging.getLogger(__name__)
 
 # POSIX default. Other-platform locations belong ONLY inside get_managed_dir().
@@ -59,7 +70,7 @@ def invalidate_managed_cache() -> None:
 
 
 def _cached_read(path: Path, cache: Dict[str, tuple], parse):
-    """Shared (mtime_ns, size)-keyed read; returns a deepcopy of the parsed value.
+    """Shared stat-signature-keyed read; returns a deepcopy of the parsed value.
 
     ``None`` when the file is absent or fails to parse (fail-open). A parse failure is logged
     LOUDLY — the admin needs to know their policy isn't applied — but never raises, so a malformed
@@ -69,12 +80,12 @@ def _cached_read(path: Path, cache: Dict[str, tuple], parse):
         st = path.stat()
     except OSError:
         return None  # absent
-    key = (st.st_mtime_ns, st.st_size)
+    key = file_signature(st)
     path_key = str(path)
     with _CACHE_LOCK:
         hit = cache.get(path_key)
-        if hit is not None and hit[:2] == key:
-            return copy.deepcopy(hit[2])
+        if hit is not None and hit[:len(key)] == key:
+            return copy.deepcopy(hit[len(key)])
     try:
         parsed = parse(path)
     except Exception as exc:  # noqa: BLE001 — fail-open, but LOUD

tests/gateway/test_profile_serve_signature.py

@@ -0,0 +1,27 @@
+"""Regression: the profile re-scan watcher must detect file *replacements*.
+
+``profile_serve_signature`` keyed files on ``(st_mtime_ns, st_size)`` only, so a
+replacement preserving both (``cp -p``, ``rsync -t``, timestamp-pinning) never
+triggered a profile re-scan and adapters were not rebuilt until restart.
+
+See #111105.
+"""
+import os
+import shutil
+
+
+def test_profile_serve_signature_detects_replacement_with_pinned_mtime(tmp_path):
+    from gateway.run_profile_reconcile import profile_serve_signature
+
+    cfg = tmp_path / "config.yaml"
+    cfg.write_text("x" * 64, encoding="utf-8")
+    before = profile_serve_signature(tmp_path)
+    st = cfg.stat()
+    other = tmp_path / "other"
+    other.write_text("y" * 64, encoding="utf-8")
+    shutil.copy2(other, cfg)
+    os.utime(cfg, ns=(st.st_atime_ns, st.st_mtime_ns))
+    after = cfg.stat()
+    assert after.st_mtime_ns == st.st_mtime_ns
+    assert after.st_size == st.st_size
+    assert profile_serve_signature(tmp_path) != before

tests/hermes_cli/test_config_cache_signature.py

@@ -0,0 +1,39 @@
+"""Regression: config cache must detect file *replacements*.
+
+A replacement that preserves ``st_mtime_ns`` and ``st_size`` (``cp -p``,
+``rsync -t``, timestamp-pinning scripts, sync clients) was treated as
+unchanged, so long-lived processes kept serving stale config until restart.
+The signature now also covers ``st_ino`` (fresh inode on atomic replace) and
+``st_ctime_ns`` (cannot be backdated via ``os.utime``).
+
+See #111105.
+"""
+import os
+import shutil
+from unittest.mock import patch
+
+from hermes_cli import config as config_mod
+from hermes_cli.config import load_config
+
+A = "model:\n  provider: opencode-go\n  default: aaaa-route\n"
+B = "model:\n  provider: opencode-go\n  default: bbbb-route\n"
+
+
+def test_load_config_detects_replacement_with_pinned_mtime(tmp_path):
+    with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}):
+        config_mod._LOAD_CONFIG_CACHE.clear()
+        config_mod._RAW_CONFIG_CACHE.clear()
+        cfg = tmp_path / "config.yaml"
+        cfg.write_text(A, encoding="utf-8")
+        assert load_config()["model"]["default"] == "aaaa-route"
+        # Replace content, keep st_mtime_ns + st_size identical.
+        before = cfg.stat()
+        other = tmp_path / "other.yaml"
+        other.write_text(B, encoding="utf-8")
+        shutil.copy2(other, cfg)
+        os.utime(cfg, ns=(before.st_atime_ns, before.st_mtime_ns))
+        after = cfg.stat()
+        assert after.st_mtime_ns == before.st_mtime_ns
+        assert after.st_size == before.st_size
+        assert cfg.read_text(encoding="utf-8").split("default: ")[1].strip() == "bbbb-route"
+        assert load_config()["model"]["default"] == "bbbb-route"