"""Persistent symbol-tree cache — eliminates re-parsing on every CLI invocation. Architecture ------------ Every call to ``symbols_for_snapshot`` currently re-parses all semantic files from their raw bytes: ~1,300 ms for a 400-file Python codebase. The parse result is fully determined by the file content — the same bytes always produce the same ``SymbolTree``. ``SymbolCache`` exploits this by persisting the parse results keyed by the SHA-256 of the file bytes (``object_id`` from the content-addressed object store, or a freshly computed SHA-256 for working-tree reads). On a warm cache hit the parse step is skipped entirely; the full snapshot loads in ~22 ms instead of ~1,300 ms — a 60× speedup. Cache key --------- The key is the 64-character lowercase SHA-256 hex digest of the **raw file bytes**, which is: * For committed files — the ``object_id`` from the snapshot manifest (content-addressed, so it is already the SHA-256 of the bytes). * For working-tree files — ``hashlib.sha256(raw_bytes).hexdigest()``, computed from the bytes already read from disk. Because the key is content-addressed, every hit is guaranteed correct. A file edit produces a new SHA-256 → cache miss → fresh parse → new entry. Old entries for changed files become dead weight but never produce wrong results. Storage ------- ``.muse/symbol_cache.msgpack`` — a single msgpack document:: { "version": 1, "entries": { "": { "": { "kind": "function", "name": "run", ... } } } } msgpack is used instead of JSON because: * 2.6× faster serialisation, 1.4× faster deserialisation. * Binary-safe (no base64 for blob fields). * Already a first-class dependency (used by the MWP wire protocol). Writes are atomic: data is flushed to a ``.tmp`` sibling then renamed over the target file so a crash mid-write never corrupts the cache. Pruning ------- Entries accumulate as new file versions are committed. The cache is naturally bounded by the object store: once ``muse gc`` collects an object, its symbol cache entry is dead weight. A future ``muse gc`` integration can prune the cache via :meth:`SymbolCache.prune`. Until then, the cache grows at the rate of unique file-content changes — typically kilobytes per commit. """ from __future__ import annotations import hashlib import logging import pathlib from typing import TypeGuard, get_args import msgpack from muse.core.store import MsgpackValue, read_msgpack_file from muse.plugins.code.ast_parser import SymbolKind, SymbolRecord, SymbolTree type _SymbolTreeMap = dict[str, "SymbolTree"] logger = logging.getLogger(__name__) _CACHE_VERSION = 1 _CACHE_FILENAME = "symbol_cache.msgpack" # All string fields in SymbolRecord, used for structural validation on load. _STR_FIELDS: frozenset[str] = frozenset({ "kind", "name", "qualified_name", "content_id", "body_hash", "signature_id", "metadata_id", "canonical_key", }) _INT_FIELDS: frozenset[str] = frozenset({"lineno", "end_lineno"}) _VALID_KINDS: frozenset[str] = frozenset(get_args(SymbolKind)) def _object_id_of(raw: bytes) -> str: """Return the SHA-256 hex digest of *raw* bytes — the content-addressed key.""" return hashlib.sha256(raw).hexdigest() def _is_symbol_record(obj: MsgpackValue) -> TypeGuard[SymbolRecord]: """Return ``True`` if *obj* is structurally a valid ``SymbolRecord``. Used during cache deserialization to validate entries from msgpack without constructing a new dict — the validated dict IS the SymbolRecord at runtime. """ if not isinstance(obj, dict): return False for field in _STR_FIELDS: if not isinstance(obj.get(field), str): return False for field in _INT_FIELDS: if not isinstance(obj.get(field), int): return False return obj["kind"] in _VALID_KINDS class SymbolCache: """Persistent msgpack cache mapping object_id → SymbolTree. Typical lifecycle inside ``symbols_for_snapshot``:: cache = SymbolCache.load(root / ".muse") for file_path, object_id in manifest.items(): tree = cache.get(object_id) if tree is None: raw = read_object(root, object_id) tree = parse_symbols(raw, file_path) cache.put(object_id, tree) cache.save() """ def __init__( self, muse_dir: pathlib.Path | None, entries: _SymbolTreeMap, ) -> None: self._muse_dir = muse_dir self._entries = entries self._dirty = False # ------------------------------------------------------------------ # Construction # ------------------------------------------------------------------ @classmethod def load(cls, muse_dir: pathlib.Path) -> SymbolCache: """Load the cache from *muse_dir*/symbol_cache.msgpack. Returns a fresh empty cache if the file is absent, unreadable, or version-mismatched — never raises. Invalid entries are skipped so a partially corrupt file does not poison the entire cache. """ cache_file = muse_dir / _CACHE_FILENAME if not cache_file.is_file(): return cls(muse_dir, {}) try: doc = read_msgpack_file(cache_file) if not isinstance(doc, dict) or doc.get("version") != _CACHE_VERSION: logger.debug("⚠️ symbol_cache version mismatch — starting fresh") return cls(muse_dir, {}) raw_entries = doc.get("entries") if not isinstance(raw_entries, dict): return cls(muse_dir, {}) entries: _SymbolTreeMap = {} for obj_id, tree_raw in raw_entries.items(): if not isinstance(obj_id, str) or not isinstance(tree_raw, dict): continue tree: SymbolTree = {} valid = True for addr, rec_raw in tree_raw.items(): # _is_symbol_record validates structure and narrows to SymbolRecord. if not _is_symbol_record(rec_raw): valid = False break tree[addr] = rec_raw if valid: entries[obj_id] = tree return cls(muse_dir, entries) except Exception as exc: # noqa: BLE001 logger.debug("⚠️ symbol_cache unreadable (%s) — starting fresh", exc) return cls(muse_dir, {}) @classmethod def empty(cls) -> SymbolCache: """Return a no-op cache for contexts without a ``.muse`` directory.""" return cls(None, {}) # ------------------------------------------------------------------ # Data access # ------------------------------------------------------------------ def get(self, object_id: str) -> SymbolTree | None: """Return the cached ``SymbolTree`` for *object_id*, or ``None`` on miss.""" return self._entries.get(object_id) def put(self, object_id: str, tree: SymbolTree) -> None: """Store *tree* under *object_id* and mark the cache dirty.""" self._entries[object_id] = tree self._dirty = True def prune(self, live_ids: set[str]) -> None: """Remove entries whose object IDs are not in *live_ids*. Call this from ``muse gc`` after identifying all reachable object IDs to keep the cache from growing unboundedly. """ stale = set(self._entries) - live_ids if stale: for k in stale: del self._entries[k] self._dirty = True logger.debug("🗑️ symbol_cache pruned %d stale entries", len(stale)) @property def size(self) -> int: """Number of cached object IDs.""" return len(self._entries) # ------------------------------------------------------------------ # Persistence # ------------------------------------------------------------------ def save(self) -> None: """Atomically persist the cache to disk if it has changed. Uses a temp-file-then-rename pattern so a crash mid-write never leaves a corrupt cache file. Silently skips when there is no ``.muse`` directory (e.g. in-memory unit tests). """ if not self._dirty or self._muse_dir is None: return doc = {"version": _CACHE_VERSION, "entries": self._entries} cache_file = self._muse_dir / _CACHE_FILENAME tmp = self._muse_dir / (_CACHE_FILENAME + ".tmp") try: tmp.write_bytes(msgpack.packb(doc, use_bin_type=True)) tmp.replace(cache_file) self._dirty = False logger.debug("✅ symbol_cache saved (%d entries)", len(self._entries)) except OSError as exc: logger.warning("⚠️ symbol_cache save failed: %s", exc) def load_symbol_cache(root: pathlib.Path) -> SymbolCache: """Convenience loader: return a ``SymbolCache`` for a repository root. Returns ``SymbolCache.empty()`` when *root* has no ``.muse`` directory so callers never need to guard against a missing repo. """ muse_dir = root / ".muse" if muse_dir.is_dir(): return SymbolCache.load(muse_dir) return SymbolCache.empty()