data/narration.json

@@ -0,0 +1,3 @@
+{
+  "discovery": "You discovered a hidden cave in the {location}."
+}

src/timmy/sovereignty/sovereignty_loop.py

@@ -22,16 +22,59 @@ Refs: #953 (The Sovereignty Loop), #955, #956, #961
 from __future__ import annotations
 
 import functools
+import json
 import logging
 from collections.abc import Callable
+from pathlib import Path
 from typing import Any, TypeVar
 
+from timmy.sovereignty.auto_crystallizer import (
+    crystallize_reasoning,
+    get_rule_store,
+)
 from timmy.sovereignty.metrics import emit_sovereignty_event, get_metrics_store
 
 logger = logging.getLogger(__name__)
 
 T = TypeVar("T")
 
+# ── Module-level narration cache ─────────────────────────────────────────────
+
+_narration_cache: dict[str, str] | None = None
+_narration_cache_mtime: float = 0.0
+
+
+def _load_narration_store() -> dict[str, str]:
+    """Load narration templates from disk, with mtime-based caching."""
+    global _narration_cache, _narration_cache_mtime
+
+    from config import settings
+
+    narration_path = Path(settings.repo_root) / "data" / "narration.json"
+    if not narration_path.exists():
+        _narration_cache = {}
+        return _narration_cache
+
+    try:
+        mtime = narration_path.stat().st_mtime
+    except OSError:
+        if _narration_cache is not None:
+            return _narration_cache
+        return {}
+
+    if _narration_cache is not None and mtime == _narration_cache_mtime:
+        return _narration_cache
+
+    try:
+        with narration_path.open() as f:
+            _narration_cache = json.load(f)
+        _narration_cache_mtime = mtime
+    except Exception:
+        if _narration_cache is None:
+            _narration_cache = {}
+
+    return _narration_cache
+
 
 # ── Perception Layer ──────────────────────────────────────────────────────────
 
@@ -81,10 +124,7 @@ async def sovereign_perceive(
     raw = await vlm.analyze(screenshot)
 
     # Step 3: parse
-    if parse_fn is not None:
-        state = parse_fn(raw)
-    else:
-        state = raw
+    state = parse_fn(raw) if parse_fn is not None else raw
 
     # Step 4: crystallize
     if crystallize_fn is not None:
@@ -140,11 +180,6 @@ async def sovereign_decide(
     dict[str, Any]
         The decision result, with at least an ``"action"`` key.
     """
-    from timmy.sovereignty.auto_crystallizer import (
-        crystallize_reasoning,
-        get_rule_store,
-    )
-
     store = rule_store if rule_store is not None else get_rule_store()
 
     # Step 1: check rules
@@ -207,29 +242,16 @@ async def sovereign_narrate(
     template_store:
         Optional narration template store (dict-like mapping event types
         to template strings with ``{variable}`` slots).  If ``None``,
-        tries to load from ``data/narration.json``.
+        uses mtime-cached templates from ``data/narration.json``.
 
     Returns
     -------
     str
         The narration text.
     """
-    import json
-    from pathlib import Path
-
-    from config import settings
-
-    # Load template store
+    # Load templates from cache instead of disk every time
     if template_store is None:
-        narration_path = Path(settings.repo_root) / "data" / "narration.json"
-        if narration_path.exists():
-            try:
-                with narration_path.open() as f:
-                    template_store = json.load(f)
-            except Exception:
-                template_store = {}
-        else:
-            template_store = {}
+        template_store = _load_narration_store()
 
     event_type = event.get("type", "unknown")
 
@@ -270,8 +292,7 @@ def _crystallize_narration_template(
     Replaces concrete values in the narration with format placeholders
     based on event keys, then saves to ``data/narration.json``.
     """
-    import json
-    from pathlib import Path
+    global _narration_cache, _narration_cache_mtime
 
     from config import settings
 
@@ -289,6 +310,9 @@ def _crystallize_narration_template(
         narration_path.parent.mkdir(parents=True, exist_ok=True)
         with narration_path.open("w") as f:
             json.dump(template_store, f, indent=2)
+        # Update cache so next read skips disk
+        _narration_cache = template_store
+        _narration_cache_mtime = narration_path.stat().st_mtime
         logger.info("Crystallized narration template for event type '%s'", event_type)
     except Exception as exc:
         logger.warning("Failed to persist narration template: %s", exc)
@@ -347,17 +371,18 @@ def sovereignty_enforced(
     def decorator(fn: Callable) -> Callable:
         @functools.wraps(fn)
         async def wrapper(*args: Any, **kwargs: Any) -> Any:
+            session_id = kwargs.get("session_id", "")
+            store = get_metrics_store()
+
             # Check cache
             if cache_check is not None:
                 cached = cache_check(args, kwargs)
                 if cached is not None:
-                    store = get_metrics_store()
-                    store.record(sovereign_event, session_id=kwargs.get("session_id", ""))
+                    store.record(sovereign_event, session_id=session_id)
                     return cached
 
             # Cache miss — run the model
-            store = get_metrics_store()
-            store.record(miss_event, session_id=kwargs.get("session_id", ""))
+            store.record(miss_event, session_id=session_id)
             result = await fn(*args, **kwargs)
 
             # Crystallize
@@ -367,7 +392,7 @@ def sovereignty_enforced(
                     store.record(
                         "skill_crystallized",
                         metadata={"layer": layer},
-                        session_id=kwargs.get("session_id", ""),
+                        session_id=session_id,
                     )
                 except Exception as exc:
                     logger.warning("Crystallization failed for %s: %s", layer, exc)