[eric] backend: stat-validated caches for settings.json and tools dir, disk parse only on change

backend/apps/settings/store.py

@@ -66,10 +66,30 @@ def _preserve_corrupt_settings() -> None:
         pass
 
 
+# In-memory mirror of SETTINGS_FILE, revalidated by stat (mtime+size) on every load
+# so even a hand-edited file or an unexpected writer is picked up immediately. A stat
+# skips the open+parse+validate that Defender turns into 5-50ms on Windows. Copies on
+# both sides keep handler isolation: callers mutate their copy, never the cache.
+_cached_settings: AppSettings | None = None
+_cached_sig: tuple[int, int] | None = None
+
+
+def _settings_sig() -> tuple[int, int] | None:
+    try:
+        st = os.stat(SETTINGS_FILE)
+        return (st.st_mtime_ns, st.st_size)
+    except OSError:
+        return None
+
+
 def load_settings() -> AppSettings:
     """Load settings from JSON file, returning defaults if not found. Never raises
     on a corrupt or version-mismatched file: a single bad settings.json must not
     brick boot (it is read at startup, by the settings endpoint, and per dispatch)."""
+    global _cached_settings, _cached_sig
+    sig = _settings_sig()
+    if sig is not None and _cached_settings is not None and sig == _cached_sig:
+        return _cached_settings.model_copy(deep=True)
     if os.path.exists(SETTINGS_FILE):
         try:
             with open(SETTINGS_FILE) as f:
@@ -84,6 +104,8 @@ def load_settings() -> AppSettings:
         settings = _coerce_settings(_migrate_legacy_fields(raw))
         if settings.default_system_prompt is None:
             settings.default_system_prompt = DEFAULT_SYSTEM_PROMPT
+        _cached_settings = settings.model_copy(deep=True)
+        _cached_sig = sig
         return settings
     return AppSettings()
 
@@ -94,6 +116,7 @@ _settings_write_lock = threading.Lock()
 
 def _atomic_write_settings(payload: dict) -> None:
     """Atomic SETTINGS_FILE write; call via save_settings*, not directly."""
+    global _cached_settings, _cached_sig
     with _settings_write_lock:
         os.makedirs(DATA_DIR, exist_ok=True)
         fd, tmp = tempfile.mkstemp(prefix=".settings.", suffix=".tmp", dir=DATA_DIR)
@@ -104,6 +127,11 @@ def _atomic_write_settings(payload: dict) -> None:
             for attempt in range(2):
                 try:
                     os.replace(tmp, SETTINGS_FILE)
+                    # Refresh the cache inside the lock so cache order matches disk order.
+                    _cached_settings = _coerce_settings(_migrate_legacy_fields(dict(payload)))
+                    if _cached_settings.default_system_prompt is None:
+                        _cached_settings.default_system_prompt = DEFAULT_SYSTEM_PROMPT
+                    _cached_sig = _settings_sig()
                     return
                 except PermissionError:
                     if attempt == 1:

backend/apps/tools_lib/tools_lib.py

@@ -114,7 +114,33 @@ def _reclassify_existing_tools() -> None:
 GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
 
 
+# Tool JSONs total ~1.5MB and _load_all runs on every dispatch, prompt build, and
+# MCPSearch keystroke; the cache skips re-parsing, revalidated by a per-file stat
+# signature so any write (ours or external) invalidates instantly. Callers treat
+# the returned ToolDefinitions as immutable; mutate via _load(tool_id) + _save.
+_tools_cache: list[ToolDefinition] | None = None
+_tools_cache_sig: tuple | None = None
+
+
+def _tools_sig() -> tuple | None:
+    if not os.path.exists(DATA_DIR):
+        return ()
+    try:
+        entries = []
+        for fname in sorted(os.listdir(DATA_DIR)):
+            if fname.endswith(".json"):
+                st = os.stat(os.path.join(DATA_DIR, fname))
+                entries.append((fname, st.st_mtime_ns, st.st_size))
+        return tuple(entries)
+    except OSError:
+        return None
+
+
 def _load_all() -> list[ToolDefinition]:
+    global _tools_cache, _tools_cache_sig
+    sig = _tools_sig()
+    if sig is not None and _tools_cache is not None and sig == _tools_cache_sig:
+        return list(_tools_cache)
     result = []
     if not os.path.exists(DATA_DIR):
         return result
@@ -122,6 +148,9 @@ def _load_all() -> list[ToolDefinition]:
         if fname.endswith(".json"):
             with open(os.path.join(DATA_DIR, fname)) as f:
                 result.append(ToolDefinition(**json.load(f)))
+    if sig is not None:
+        _tools_cache = list(result)
+        _tools_cache_sig = sig
     return result
 
 

backend/tests/test_disk_caches.py

@@ -0,0 +1,110 @@
+import json
+import os
+import time
+
+import pytest
+
+from backend.apps.settings import store
+from backend.apps.settings.models import AppSettings
+from backend.apps.tools_lib import tools_lib
+from backend.apps.tools_lib.models import ToolDefinition
+
+
+@pytest.fixture
+def settings_tmp(tmp_path, monkeypatch):
+    f = tmp_path / "settings.json"
+    monkeypatch.setattr(store, "DATA_DIR", str(tmp_path))
+    monkeypatch.setattr(store, "SETTINGS_FILE", str(f))
+    monkeypatch.setattr(store, "_cached_settings", None)
+    monkeypatch.setattr(store, "_cached_sig", None)
+    return f
+
+
+@pytest.fixture
+def tools_tmp(tmp_path, monkeypatch):
+    d = tmp_path / "tools"
+    d.mkdir()
+    monkeypatch.setattr(tools_lib, "DATA_DIR", str(d))
+    monkeypatch.setattr(tools_lib, "_tools_cache", None)
+    monkeypatch.setattr(tools_lib, "_tools_cache_sig", None)
+    return d
+
+
+def _bump_mtime(path):
+    # FAT32-style coarse clocks could hide a same-size rewrite; force a distinct mtime.
+    st = os.stat(path)
+    os.utime(path, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000))
+
+
+def test_settings_write_through_is_fresh(settings_tmp):
+    s = store.load_settings()
+    s.theme = "light"
+    store.save_settings(s)
+    assert store.load_settings().theme == "light"
+    s2 = store.load_settings()
+    s2.theme = "dark"
+    store.save_settings(s2)
+    assert store.load_settings().theme == "dark"
+
+
+def test_settings_external_edit_detected(settings_tmp):
+    store.save_settings(AppSettings(theme="dark"))
+    assert store.load_settings().theme == "dark"
+    raw = json.loads(settings_tmp.read_text())
+    raw["theme"] = "light"
+    settings_tmp.write_text(json.dumps(raw))
+    _bump_mtime(settings_tmp)
+    assert store.load_settings().theme == "light"
+
+
+def test_settings_cache_returns_isolated_copies(settings_tmp):
+    store.save_settings(AppSettings(theme="dark"))
+    a = store.load_settings()
+    a.theme = "light"
+    assert store.load_settings().theme == "dark"
+
+
+def test_settings_file_deleted_falls_back_to_defaults(settings_tmp):
+    store.save_settings(AppSettings(theme="light"))
+    os.remove(settings_tmp)
+    assert store.load_settings().theme == AppSettings().theme
+
+
+def test_tools_write_then_list_is_fresh(tools_tmp):
+    assert tools_lib._load_all() == []
+    t = ToolDefinition(name="Alpha", description="a")
+    tools_lib._save(t)
+    _bump_mtime(tools_tmp / f"{t.id}.json")
+    names = [x.name for x in tools_lib._load_all()]
+    assert names == ["Alpha"]
+
+    t2 = ToolDefinition(name="Beta", description="b")
+    tools_lib._save(t2)
+    assert sorted(x.name for x in tools_lib._load_all()) == ["Alpha", "Beta"]
+
+
+def test_tools_delete_detected(tools_tmp):
+    t = ToolDefinition(name="Gone", description="g")
+    tools_lib._save(t)
+    assert [x.name for x in tools_lib._load_all()] == ["Gone"]
+    os.remove(tools_tmp / f"{t.id}.json")
+    assert tools_lib._load_all() == []
+
+
+def test_tools_in_place_rewrite_detected(tools_tmp):
+    t = ToolDefinition(name="Old", description="x")
+    tools_lib._save(t)
+    assert [x.name for x in tools_lib._load_all()] == ["Old"]
+    t.name = "New"
+    tools_lib._save(t)
+    _bump_mtime(tools_tmp / f"{t.id}.json")
+    assert [x.name for x in tools_lib._load_all()] == ["New"]
+
+
+def test_tools_cached_hit_skips_reparse(tools_tmp, monkeypatch):
+    tools_lib._save(ToolDefinition(name="Once", description="o"))
+    tools_lib._load_all()
+    def boom(*a, **k):
+        raise AssertionError("disk re-parse on unchanged dir")
+    monkeypatch.setattr(json, "load", boom)
+    assert [x.name for x in tools_lib._load_all()] == ["Once"]