"""Small process-local cache for static character and prompt files only.""" from collections import OrderedDict from copy import deepcopy import json from pathlib import Path from threading import RLock _MAX_ENTRIES = 64 _CACHE = OrderedDict() _LOCK = RLock() def _signature(path): stat = path.stat() return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns) def _load(path, as_json): path = Path(path).resolve() key = (path, as_json) with _LOCK: # Always stat: changed or deleted files must never silently use stale data. signature = _signature(path) cached = _CACHE.get(key) if cached is not None and cached[0] == signature: _CACHE.move_to_end(key) value = cached[1] else: _CACHE.pop(key, None) text = path.read_text(encoding='utf-8') value = json.loads(text) if as_json else text # A concurrent file edit must not be cached under its older signature. if _signature(path) == signature: _CACHE[key] = (signature, value) while len(_CACHE) > _MAX_ENTRIES: _CACHE.popitem(last=False) return deepcopy(value) if as_json else value def load_static_json(path): """Return a private copy of parsed JSON, refreshing when file metadata changes.""" return _load(path, as_json=True) def load_static_text(path): """Return cached UTF-8 text without altering whitespace.""" return _load(path, as_json=False)