gabriel / muse public
indices.py python
336 lines 12.4 KB
Raw
sha256:057be43e401106b2b4d877e8028ae67d16a966562d7429707b71a5eaacd5027b docs(mwp-master): tick all ACs green; mark MWP-7 complete w… Sonnet 4.6 20 days ago
1 """Optional local index layer for Muse repositories.
2
3 Indexes live under ``.muse/indices/`` and are derived, versioned, and fully
4 rebuildable from the commit history. **No index is required for repository
5 correctness** — all commands work without them; indexes only accelerate
6 repeated queries.
7
8 Available indexes
9 -----------------
10
11 ``symbol_history``
12 Maps symbol addresses to their event timeline across all commits.
13 Enables O(1) ``muse symbol-log``, ``muse lineage``, and ``muse query-history``
14 instead of O(commits × files) scans.
15
16 ``hash_occurrence``
17 Maps ``body_hash`` values to the list of symbol addresses that share them.
18 Enables O(1) ``muse clones`` and ``muse find-symbol hash=``.
19
20 Storage format
21 --------------
22
23 Index files are stored as **plain JSON** (``*.json``) under ``.muse/indices/``.
24 Writes are **atomic**: data is flushed to a ``.tmp`` sibling and then renamed
25 over the target file so a crash mid-write never corrupts an index.
26
27 Rebuild
28 -------
29
30 Indexes are rebuilt by ``muse code index rebuild``. They can also be built
31 incrementally: the ``update_*_index`` functions accept an existing index
32 dict and patch it rather than rebuilding from scratch.
33 """
34
35 import logging
36 import pathlib
37 from typing import TypedDict
38
39 import json as _json
40
41 from muse.core.paths import indices_dir as _indices_dir
42 from muse._version import __version__ as _SCHEMA_VERSION
43 from muse.core.types import Metadata, MsgpackDict, now_utc_iso
44 from muse.core.types import MsgpackValue
45
46 logger = logging.getLogger(__name__)
47
48 type _EntryList = list[Metadata]
49 type _EntryMap = dict[str, _EntryList]
50 type _StringListMap = dict[str, list[str]]
51
52 # ---------------------------------------------------------------------------
53 # Typed index entry shapes
54 # ---------------------------------------------------------------------------
55
56 class SymbolHistoryEntry:
57 """One event in a symbol's history timeline."""
58
59 __slots__ = (
60 "commit_id", "committed_at", "op",
61 "content_id", "body_hash", "signature_id",
62 )
63
64 def __init__(
65 self,
66 commit_id: str,
67 committed_at: str,
68 op: str,
69 content_id: str,
70 body_hash: str,
71 signature_id: str,
72 ) -> None:
73 self.commit_id = commit_id
74 self.committed_at = committed_at
75 self.op = op
76 self.content_id = content_id
77 self.body_hash = body_hash
78 self.signature_id = signature_id
79
80 def to_dict(self) -> Metadata:
81 """Serialise to a plain string-keyed, string-valued mapping."""
82 return {
83 "commit_id": self.commit_id,
84 "committed_at": self.committed_at,
85 "op": self.op,
86 "content_id": self.content_id,
87 "body_hash": self.body_hash,
88 "signature_id": self.signature_id,
89 }
90
91 @classmethod
92 def from_dict(cls, d: "Metadata | MsgpackDict") -> "SymbolHistoryEntry":
93 """Deserialise from any string-keyed mapping (all fields coerced to ``str``)."""
94 def _s(key: str) -> str:
95 v = d.get(key, "")
96 return v if isinstance(v, str) else ""
97 return cls(
98 commit_id=_s("commit_id"),
99 committed_at=_s("committed_at"),
100 op=_s("op"),
101 content_id=_s("content_id"),
102 body_hash=_s("body_hash"),
103 signature_id=_s("signature_id"),
104 )
105
106 # ---------------------------------------------------------------------------
107 # On-disk document shapes (one TypedDict per index type)
108 # ---------------------------------------------------------------------------
109
110 class _SymbolHistoryDoc(TypedDict):
111 """Msgpack document layout for the ``symbol_history`` index file."""
112
113 schema_version: str
114 index: str
115 updated_at: str
116 entries: _EntryMap
117
118 class _HashOccurrenceDoc(TypedDict):
119 """Msgpack document layout for the ``hash_occurrence`` index file."""
120
121 schema_version: str
122 index: str
123 updated_at: str
124 entries: _StringListMap
125
126 # ---------------------------------------------------------------------------
127 # Index I/O helpers
128 # ---------------------------------------------------------------------------
129
130 def _index_path(root: pathlib.Path, name: str) -> pathlib.Path:
131 """Return the JSON file path for a named index."""
132 return _indices_dir(root) / f"{name}.json"
133
134 def _ensure_dir(root: pathlib.Path) -> None:
135 _indices_dir(root).mkdir(parents=True, exist_ok=True)
136
137 def _write_symbol_history_doc(path: pathlib.Path, doc: _SymbolHistoryDoc) -> None:
138 """Atomically serialise a :class:`_SymbolHistoryDoc` to *path*."""
139 _write_bytes_atomic(path, _json.dumps(doc, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
140
141 def _write_hash_occurrence_doc(path: pathlib.Path, doc: _HashOccurrenceDoc) -> None:
142 """Atomically serialise a :class:`_HashOccurrenceDoc` to *path*."""
143 _write_bytes_atomic(path, _json.dumps(doc, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
144
145 # ---------------------------------------------------------------------------
146 # Read safety constants (index-specific)
147 # ---------------------------------------------------------------------------
148
149 # Index files grow with repository size and can reach hundreds of MiB for
150 # large repos. The cap is more generous than the commit/snapshot store while
151 # still bounding worst-case memory allocation before read_bytes() is called.
152 _MAX_INDEX_BYTES: int = 512 * 1024 * 1024 # 512 MiB
153
154
155 def _write_bytes_atomic(path: pathlib.Path, packed: bytes) -> None:
156 """Replace *path* with *packed* using a ``.tmp`` rename to guarantee atomicity."""
157 tmp = path.with_suffix(".tmp")
158 try:
159 tmp.write_bytes(packed)
160 tmp.replace(path)
161 except OSError:
162 tmp.unlink(missing_ok=True)
163 raise
164
165 def _read_json(path: pathlib.Path) -> MsgpackValue:
166 """Read and parse an index JSON file, enforcing size limits.
167
168 Index files can grow into the hundreds of MiB for large repositories, so the
169 cap is more generous than the commit/snapshot store. The stat check fires
170 before ``read_bytes()`` so a gigantic file never triggers an OOM.
171
172 Raises :exc:`OSError` if the file exceeds ``_MAX_INDEX_BYTES``.
173 Raises :exc:`ValueError` if the file contains old binary msgpack content.
174 """
175 size = path.stat().st_size
176 if size > _MAX_INDEX_BYTES:
177 raise OSError(
178 f"Index file {path.name!r} is {size:,} bytes, exceeding the "
179 f"{_MAX_INDEX_BYTES // (1024 * 1024)} MiB read limit."
180 )
181 raw = path.read_bytes()
182 if raw and raw[0] > 0x7F:
183 raise ValueError(f"Index file {path.name!r} contains old binary format")
184 return _json.loads(raw.decode("utf-8"))
185
186 # ---------------------------------------------------------------------------
187 # Symbol history index
188 # ---------------------------------------------------------------------------
189
190 SymbolHistoryIndex = dict[str, list[SymbolHistoryEntry]]
191
192 def load_symbol_history(root: pathlib.Path) -> SymbolHistoryIndex:
193 """Load the symbol history index, returning an empty dict if absent."""
194 path = _index_path(root, "symbol_history")
195 if not path.exists():
196 return {}
197 try:
198 raw = _read_json(path)
199 if not isinstance(raw, dict):
200 raise ValueError("top-level document is not a dict")
201 entries_raw = raw.get("entries")
202 if not isinstance(entries_raw, dict):
203 return {}
204 result: SymbolHistoryIndex = {}
205 for address, entry_list in entries_raw.items():
206 if not isinstance(entry_list, list):
207 continue
208 entries: list[SymbolHistoryEntry] = []
209 for item in entry_list:
210 if isinstance(item, dict):
211 entries.append(SymbolHistoryEntry.from_dict(item))
212 result[address] = entries
213 return result
214 except Exception as exc:
215 logger.warning("⚠️ Corrupt symbol_history index: %s — returning empty", exc)
216 return {}
217
218 def save_symbol_history(root: pathlib.Path, index: SymbolHistoryIndex) -> None:
219 """Persist the symbol history index as JSON (atomic write)."""
220 _ensure_dir(root)
221 path = _index_path(root, "symbol_history")
222 doc = _SymbolHistoryDoc(
223 schema_version=_SCHEMA_VERSION,
224 index="symbol_history",
225 updated_at=now_utc_iso(),
226 entries={
227 addr: [e.to_dict() for e in evts]
228 for addr, evts in sorted(index.items())
229 },
230 )
231 _write_symbol_history_doc(path, doc)
232 logger.debug("✅ Saved symbol_history index (%d addresses)", len(index))
233
234 # ---------------------------------------------------------------------------
235 # Hash occurrence index
236 # ---------------------------------------------------------------------------
237
238 HashOccurrenceIndex = dict[str, list[str]]
239
240 def load_hash_occurrence(root: pathlib.Path) -> HashOccurrenceIndex:
241 """Load the hash occurrence index, returning an empty dict if absent."""
242 path = _index_path(root, "hash_occurrence")
243 if not path.exists():
244 return {}
245 try:
246 raw = _read_json(path)
247 if not isinstance(raw, dict):
248 raise ValueError("top-level document is not a dict")
249 entries_raw = raw.get("entries")
250 if not isinstance(entries_raw, dict):
251 return {}
252 result: HashOccurrenceIndex = {}
253 for body_hash, addresses in entries_raw.items():
254 if isinstance(addresses, list):
255 result[body_hash] = [a for a in addresses if isinstance(a, str)]
256 return result
257 except Exception as exc:
258 logger.warning("⚠️ Corrupt hash_occurrence index: %s — returning empty", exc)
259 return {}
260
261 def save_hash_occurrence(root: pathlib.Path, index: HashOccurrenceIndex) -> None:
262 """Persist the hash occurrence index as JSON (atomic write)."""
263 _ensure_dir(root)
264 path = _index_path(root, "hash_occurrence")
265 doc = _HashOccurrenceDoc(
266 schema_version=_SCHEMA_VERSION,
267 index="hash_occurrence",
268 updated_at=now_utc_iso(),
269 entries={h: sorted(addrs) for h, addrs in sorted(index.items())},
270 )
271 _write_hash_occurrence_doc(path, doc)
272 logger.debug("✅ Saved hash_occurrence index (%d hashes)", len(index))
273
274 # ---------------------------------------------------------------------------
275 # Index metadata
276 # ---------------------------------------------------------------------------
277
278 #: All index names known to this module.
279 KNOWN_INDEX_NAMES: tuple[str, ...] = ("symbol_history", "hash_occurrence")
280
281 class IndexInfoEntry(TypedDict):
282 """Status information for one local index — returned by :func:`index_info`."""
283
284 name: str
285 status: str # "present" | "absent" | "corrupt"
286 entries: int
287 updated_at: str | None
288
289 def index_info(root: pathlib.Path) -> list[IndexInfoEntry]:
290 """Return status information about all known indexes.
291
292 ``entries`` is always an ``int``. Callers must not convert it.
293 """
294 result: list[IndexInfoEntry] = []
295 for name in KNOWN_INDEX_NAMES:
296 path = _index_path(root, name)
297 if path.exists():
298 try:
299 raw = _read_json(path)
300 if not isinstance(raw, dict):
301 raise ValueError("not a dict")
302 upd = raw.get("updated_at")
303 updated_at: str | None = upd if isinstance(upd, str) else None
304 entries_data = raw.get("entries")
305 entries = len(entries_data) if isinstance(entries_data, dict) else 0
306 result.append(IndexInfoEntry(
307 name=name,
308 status="present",
309 updated_at=updated_at,
310 entries=entries,
311 ))
312 except Exception:
313 result.append(IndexInfoEntry(
314 name=name, status="corrupt", updated_at=None, entries=0,
315 ))
316 else:
317 result.append(IndexInfoEntry(
318 name=name, status="absent", updated_at=None, entries=0,
319 ))
320 return result
321
322 def purge_index(root: pathlib.Path, name: str) -> bool:
323 """Delete the on-disk file for the named index.
324
325 Returns ``True`` if the file existed and was deleted, ``False`` if it was
326 already absent. Raises ``ValueError`` for unknown index names.
327 """
328 if name not in KNOWN_INDEX_NAMES:
329 raise ValueError(f"Unknown index name '{name}'. Valid names: {', '.join(KNOWN_INDEX_NAMES)}")
330 path = _index_path(root, name)
331 try:
332 path.unlink()
333 logger.debug("🗑️ Purged index: %s", name)
334 return True
335 except FileNotFoundError:
336 return False
File History 1 commit
sha256:057be43e401106b2b4d877e8028ae67d16a966562d7429707b71a5eaacd5027b docs(mwp-master): tick all ACs green; mark MWP-7 complete w… Sonnet 4.6 20 days ago