"""muse code index — manage and rebuild the optional local index layer. Indexes live under ``.muse/indices/`` and are fully derived from the commit history. They are optional — all commands work without them, but indexes dramatically accelerate repeated queries on large repositories. Available indexes ----------------- ``symbol_history`` Maps every symbol address to its full event timeline across all commits. Reduces ``muse code symbol-log``, ``muse code lineage``, and ``muse code query-history`` from O(commits × files) to O(1) lookups. ``hash_occurrence`` Maps every ``body_hash`` to the list of addresses that share it. Reduces ``muse code clones`` and ``muse code find-symbol hash=`` to O(1). Sub-commands ------------ ``muse code index status`` Show the status, entry count, and last-updated time of each index. ``muse code index rebuild [--index NAME] [--dry-run]`` Rebuild one or all indexes by walking the entire commit history. Safe to run multiple times. Pass ``--dry-run`` to see what would be built without writing anything. ``muse code index purge [--index NAME]`` Delete one or all local index files. The next rebuild recreates them. Usage:: muse code index status muse code index status --json muse code index rebuild muse code index rebuild --json muse code index rebuild --index symbol_history muse code index rebuild --index hash_occurrence muse code index rebuild --dry-run muse code index purge muse code index purge --index symbol_history JSON output — ``muse code index status --json``:: [ {"name": "symbol_history", "status": "present", "entries": 1024, "updated_at": "2026-03-21T12:00:00+00:00"}, {"name": "hash_occurrence", "status": "absent", "entries": 0, "updated_at": null} ] JSON output — ``muse code index rebuild --json``:: {"schema_version": "0.1.5", "rebuilt": ["symbol_history", "hash_occurrence"], "symbol_history_addresses": 512, "symbol_history_events": 2048, "hash_occurrence_clusters": 31, "hash_occurrence_addresses": 87} """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import TypedDict from muse import __version__ from muse.core.errors import ExitCode from muse.core.indices import ( KNOWN_INDEX_NAMES, HashOccurrenceIndex, IndexInfoEntry, SymbolHistoryEntry, SymbolHistoryIndex, index_info, purge_index, save_hash_occurrence, save_symbol_history, ) from muse.core.object_store import read_object from muse.core.repo import require_repo from muse.core.store import get_all_commits, get_commit_snapshot_manifest, read_current_branch from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.plugins.code._query import is_semantic from muse.plugins.code.ast_parser import parse_symbols from muse.core.validation import sanitize_display type _BlobCache = dict[str, bytes] type _ManifestCache = dict[str, dict[str, str]] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # TypedDict for rebuild JSON output schema # --------------------------------------------------------------------------- class _RebuildResult(TypedDict, total=False): schema_version: str dry_run: bool rebuilt: list[str] symbol_history_addresses: int symbol_history_events: int hash_occurrence_clusters: int hash_occurrence_addresses: int # Knowtation domain extension (Phase 3.4) knowtation_link_graph_notes: int knowtation_link_graph_edges_resolved: int knowtation_link_graph_edges_broken: int knowtation_link_graph_entities: int # --------------------------------------------------------------------------- # Index build logic # --------------------------------------------------------------------------- def _build_symbol_history( root: pathlib.Path, symbol_cache: SymbolCache | None = None, ) -> SymbolHistoryIndex: """Walk all commits oldest-first and build the symbol history index. Performance notes ----------------- * A ``manifest_cache`` (keyed by commit_id) ensures that each snapshot manifest is fetched at most once per rebuild, regardless of how many symbol ops share the same commit. * A ``blob_cache`` (keyed by obj_id) ensures that each source blob is ``read_object``'d at most once per rebuild. * ``symbol_cache`` is the **persistent** cross-run parse cache (:class:`~muse.core.symbol_cache.SymbolCache`). Because obj_id is the SHA-256 of file bytes (content-addressed), a file that did not change between two commits always produces a cache hit — its AST is never re-parsed. On a warm cache a 300-commit rebuild drops from minutes to seconds. Together these three layers reduce I/O from O(commits × ops × files) on the first run to O(1) per unique (obj_id, address) pair on subsequent runs. """ all_commits = sorted( get_all_commits(root), key=lambda c: c.committed_at, ) index: SymbolHistoryIndex = {} # manifest_cache: commit_id → {file_path: obj_id} manifest_cache: _ManifestCache = {} # blob_cache: obj_id → raw bytes (within this run; SymbolCache handles cross-run) blob_cache: _BlobCache = {} for commit in all_commits: if commit.structured_delta is None: continue committed_at = commit.committed_at.isoformat() ops = commit.structured_delta.get("ops", []) # Fetch the manifest once per commit, not once per child op. if commit.commit_id not in manifest_cache: raw_manifest = get_commit_snapshot_manifest(root, commit.commit_id) if raw_manifest is None: logger.debug( "Missing snapshot manifest for commit %s — skipping", commit.commit_id[:8], ) continue manifest_cache[commit.commit_id] = raw_manifest manifest = manifest_cache[commit.commit_id] for op in ops: if op["op"] != "patch": continue for child in op.get("child_ops", []): addr = child["address"] if "::" not in addr: continue file_path = addr.split("::")[0] if not is_semantic(file_path): continue child_op = child["op"] if child_op not in ("insert", "delete", "replace"): continue # Extract hash fields from the snapshot blob. # Resolution order: # 1. symbol_cache (persistent msgpack cache, keyed by obj_id) # 2. blob_cache (in-run bytes cache, avoids duplicate reads) # 3. read_object + parse_symbols (cache miss — first encounter) obj_id = manifest.get(file_path) body_hash = "" signature_id = "" content_id = "" if obj_id: # Try the persistent SymbolCache first. tree = symbol_cache.get(obj_id) if symbol_cache else None if tree is None: # Fetch bytes (in-run blob_cache avoids duplicate reads). if obj_id not in blob_cache: raw = read_object(root, obj_id) if raw is not None: blob_cache[obj_id] = raw blob = blob_cache.get(obj_id) if blob is not None: tree = parse_symbols(blob, file_path) if symbol_cache is not None: symbol_cache.put(obj_id, tree) if tree is not None: rec = tree.get(addr) if rec: body_hash = rec["body_hash"] signature_id = rec["signature_id"] content_id = rec["content_id"] if not content_id: # Fall back to the content_id stored in the delta itself. if child_op == "insert": content_id = str(child.get("content_id") or "") elif child_op == "delete": content_id = str(child.get("content_id") or "") elif child_op == "replace": content_id = str(child.get("new_content_id") or "") entry = SymbolHistoryEntry( commit_id=commit.commit_id, committed_at=committed_at, op=child_op, content_id=content_id, body_hash=body_hash, signature_id=signature_id, ) index.setdefault(addr, []).append(entry) return index def _build_hash_occurrence(root: pathlib.Path) -> HashOccurrenceIndex: """Walk the HEAD snapshot and build the hash occurrence index.""" try: muse_dir = root / ".muse" branch = read_current_branch(root) head_commit_id = (muse_dir / "refs" / "heads" / branch).read_text().strip() except OSError as exc: logger.debug("Could not determine HEAD commit for hash_occurrence build: %s", exc) return {} raw_manifest = get_commit_snapshot_manifest(root, head_commit_id) if raw_manifest is None: logger.debug( "Missing snapshot manifest for HEAD %s — hash_occurrence will be empty", head_commit_id[:8], ) return {} index: HashOccurrenceIndex = {} for file_path, obj_id in sorted(raw_manifest.items()): if not is_semantic(file_path): continue raw = read_object(root, obj_id) if raw is None: continue tree = parse_symbols(raw, file_path) for addr, rec in tree.items(): if rec["kind"] == "import": continue bh = rec["body_hash"] index.setdefault(bh, []).append(addr) # Remove trivial (size-1) entries — they are not clones. return {h: addrs for h, addrs in index.items() if len(addrs) > 1} # --------------------------------------------------------------------------- # Sub-commands # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the index subcommand.""" parser = subparsers.add_parser( "index", help="Manage the optional local index layer.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") subs.required = True # --- purge --- purge_p = subs.add_parser( "purge", help="Delete one or all local index files.", description=( "Delete index files under .muse/indices/. The canonical commit\n" "history and object store are never touched — indexes are fully\n" "rebuildable at any time with 'muse code index rebuild'.\n" "Absent indexes are silently skipped (exit 0).\n\n" "Agent quickstart\n" "----------------\n" " muse code index purge --json\n" " muse code index purge -j\n" " muse code index purge -j | jq .purged\n" " muse code index purge --index symbol_history -j\n\n" "JSON output schema\n" "------------------\n" ' {"schema_version": "",\n' ' "purged": ["symbol_history", ...],\n' ' "skipped": ["hash_occurrence", ...]}\n\n' "Exit codes\n" "----------\n" " 0 — operation completed (skipped indexes do not cause failure)\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) purge_p.add_argument( "--index", "-i", dest="index_name", default=None, metavar="NAME", choices=list(KNOWN_INDEX_NAMES), help="Purge a specific index. Default: purge all.", ) purge_p.add_argument("--json", "-j", dest="as_json", action="store_true", help="Emit purge summary as JSON.") purge_p.set_defaults(func=run_purge) # --- rebuild --- rebuild_p = subs.add_parser( "rebuild", help="Rebuild local indexes from the full commit history.", description=( "Rebuild symbol_history and/or hash_occurrence under .muse/indices/.\n" "Safe to run any number of times — atomic writes prevent corruption.\n" "A warm SymbolCache means unchanged files are never re-parsed.\n\n" "Use --dry-run to compute statistics without writing to disk.\n\n" "Agent quickstart\n" "----------------\n" " muse code index rebuild --json\n" " muse code index rebuild -j\n" " muse code index rebuild -j | jq .rebuilt\n" " muse code index rebuild --index symbol_history -j\n" " muse code index rebuild --dry-run -j\n\n" "JSON output schema\n" "------------------\n" ' {"schema_version": "", "dry_run": ,\n' ' "rebuilt": ["symbol_history", "hash_occurrence"],\n' ' "symbol_history_addresses": , "symbol_history_events": ,\n' ' "hash_occurrence_clusters": , "hash_occurrence_addresses": }\n\n' " Note: *_addresses / *_events / *_clusters keys are absent when the\n" " corresponding index was not rebuilt.\n\n" "Exit codes\n" "----------\n" " 0 — rebuild (or dry run) completed successfully\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) rebuild_p.add_argument( "--index", "-i", dest="index_name", default=None, metavar="NAME", choices=list(KNOWN_INDEX_NAMES), help="Rebuild a specific index. Default: rebuild all.", ) rebuild_p.add_argument("--dry-run", action="store_true", help="Compute what would be built without writing anything.") rebuild_p.add_argument("--verbose", "-v", action="store_true", help="Show progress.") rebuild_p.add_argument("--json", "-j", dest="as_json", action="store_true", help="Emit rebuild summary as JSON.") rebuild_p.set_defaults(func=run_rebuild) # --- status --- status_p = subs.add_parser( "status", help="Show the status and entry count of each local index.", description=( "Report the on-disk state of every index under .muse/indices/.\n" "Each index is either present (valid), absent (not yet built),\n" "or corrupt (exists but failed to parse).\n\n" "Agent quickstart\n" "----------------\n" " muse code index status --json\n" " muse code index status -j\n" " muse code index status -j | jq '.[].status'\n" " muse code index status -j | jq '.[] | select(.status != \"present\")'\n\n" "JSON output schema\n" "------------------\n" " [{\"name\": \"symbol_history\", \"status\": \"present|absent|corrupt\",\n" ' "entries": , "updated_at": "|null"}, ...]\n\n' "Exit codes\n" "----------\n" " 0 — status emitted (absent/corrupt indexes do NOT cause non-zero exit)\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) status_p.add_argument("--json", "-j", dest="as_json", action="store_true", help="Emit index status as JSON.") status_p.set_defaults(func=run_status) def run_status(args: argparse.Namespace) -> None: """Show the status and entry count of each local Muse index. Reports the on-disk state of every index under ``.muse/indices/``. Each index is reported as ``present`` (readable and valid), ``absent`` (file not found — not yet built), or ``corrupt`` (file exists but failed to parse). Indexes are derived entirely from commit history and can be rebuilt at any time without modifying the canonical store. Security: all text-mode output passes through ``sanitize_display()`` so index names cannot inject ANSI or control characters into the terminal. Index names in JSON output come exclusively from ``KNOWN_INDEX_NAMES`` (a static compile-time tuple), not from user input. The ``updated_at`` and ``entries`` values are read from trusted local msgpack files in ``.muse/indices/``; they are not derived from user-supplied data. JSON output fields (``--json`` / ``-j``) ----------------------------------------- Returns a JSON array, one object per known index: ``name`` Registry key for the index (``"symbol_history"`` or ``"hash_occurrence"``). ``status`` ``"present"`` — built and readable. ``"absent"`` — not yet built; run ``muse code index rebuild``. ``"corrupt"`` — file exists but failed to parse; run rebuild. ``entries`` Number of top-level entries in the index (``0`` when absent or corrupt). ``updated_at`` ISO-8601 timestamp of the last rebuild, or ``null``. Exit codes ---------- 0 — status emitted (even if indexes are absent or corrupt) 2 — not inside a Muse repository """ as_json: bool = args.as_json root = require_repo() muse_dir = root / ".muse" infos: list[IndexInfoEntry] = index_info(root) if as_json: out = [ { "name": info["name"], "status": info["status"], "entries": info["entries"], "updated_at": info["updated_at"], } for info in infos ] print(json.dumps(out)) return print("\nLocal index status:") print("─" * 50) for info in infos: status = info["status"] name = info["name"] updated = (info["updated_at"] or "")[:19] entries = info["entries"] if status == "present": print(f" ✅ {sanitize_display(name):<20} {entries:>8} entries (updated {updated})") elif status == "absent": print(f" ⬜ {sanitize_display(name):<20} (not built — run: muse code index rebuild)") else: print(f" ❌ {sanitize_display(name):<20} corrupt — run: muse code index rebuild") print() def run_rebuild(args: argparse.Namespace) -> None: """Rebuild local indexes from the full commit history. Walks all commits oldest-first to build ``symbol_history`` and/or ``hash_occurrence`` under ``.muse/indices/``. Safe to run any number of times — existing index files are atomically overwritten. A shared ``SymbolCache`` (keyed by content-addressed object ID) means AST parses are never repeated for unchanged files. On a warm cache a 300-commit rebuild drops from minutes to seconds. With ``--dry-run`` the full build is performed in memory and statistics are reported, but nothing is written to disk and the symbol cache is not saved. Security: the ``--index`` argument is validated by argparse against ``KNOWN_INDEX_NAMES`` before ``run_rebuild`` is called — no caller-supplied name reaches the file system unvalidated. Writes use atomic msgpack saves (``save_symbol_history`` / ``save_hash_occurrence``) so a crash cannot corrupt existing index data. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``schema_version`` Muse version string at build time. ``dry_run`` ``true`` when ``--dry-run`` was passed; no files were written. ``rebuilt`` List of index names that were (or would be) rebuilt. ``symbol_history_addresses`` Number of distinct symbol addresses in the new index. Present only when ``symbol_history`` was rebuilt. ``symbol_history_events`` Total number of insert/delete/replace events across all addresses. Present only when ``symbol_history`` was rebuilt. ``hash_occurrence_clusters`` Number of hash clusters with more than one address (clone groups). Present only when ``hash_occurrence`` was rebuilt. ``hash_occurrence_addresses`` Total address count across all clone clusters. Present only when ``hash_occurrence`` was rebuilt. Exit codes ---------- 0 — rebuild completed (or dry run computed) successfully 2 — not inside a Muse repository """ index_name: str | None = args.index_name dry_run: bool = args.dry_run verbose: bool = args.verbose as_json: bool = args.as_json root = require_repo() muse_dir = root / ".muse" build_all = index_name is None built: list[str] = [] result: _RebuildResult = { "schema_version": __version__, "dry_run": dry_run, } # Load the persistent SymbolCache once — it is shared across both index # builds and saved at the end. This gives cross-run caching of AST parses # so that unchanged files are never re-parsed. sym_cache = load_symbol_cache(root) if build_all or index_name == "symbol_history": if verbose and not as_json: print("Building symbol_history index…") idx = _build_symbol_history(root, symbol_cache=sym_cache) if not dry_run: save_symbol_history(root, idx) n_events = sum(len(evts) for evts in idx.values()) result["symbol_history_addresses"] = len(idx) result["symbol_history_events"] = n_events if not as_json: tag = " (dry run)" if dry_run else "" print(f" ✅ symbol_history — {len(idx)} addresses, {n_events} events{tag}") built.append("symbol_history") if build_all or index_name == "hash_occurrence": if verbose and not as_json: print("Building hash_occurrence index…") idx2 = _build_hash_occurrence(root) if not dry_run: save_hash_occurrence(root, idx2) n_clones = sum(len(addrs) for addrs in idx2.values()) result["hash_occurrence_clusters"] = len(idx2) result["hash_occurrence_addresses"] = n_clones if not as_json: tag = " (dry run)" if dry_run else "" print(f" ✅ hash_occurrence — {len(idx2)} clone clusters, {n_clones} addresses{tag}") built.append("hash_occurrence") # ── Knowtation-domain extension (Phase 3.4) ────────────────────────────── # When the active domain is "knowtation", also rebuild the in-memory link # graph (wikilinks + markdown refs + entity refs). This currently runs # on-the-fly per CLI call (build_link_index < 10s on 5k notes), so this # rebuild simply verifies it can be constructed cleanly and reports stats. try: from muse.plugins.registry import read_domain # noqa: PLC0415 active_domain = read_domain(root) except Exception: # noqa: BLE001 — fall through to Code-domain rebuild active_domain = "code" if active_domain == "knowtation": try: from muse.plugins.knowtation.link_index import build_link_index # noqa: PLC0415 if verbose and not as_json: print("Building knowtation_link_graph index (in-memory verification)…") link_idx = build_link_index(root) n_resolved = sum( 1 for links in link_idx.forward.values() for link in links if link.target is not None ) n_broken = len(link_idx.broken_links) n_entities = len(link_idx.entity_index) result["knowtation_link_graph_notes"] = link_idx.notes_indexed result["knowtation_link_graph_edges_resolved"] = n_resolved result["knowtation_link_graph_edges_broken"] = n_broken result["knowtation_link_graph_entities"] = n_entities built.append("knowtation_link_graph") if not as_json: tag = " (dry run)" if dry_run else "" print( f" ✅ knowtation_link_graph — {link_idx.notes_indexed} notes, " f"{n_resolved} resolved edges, {n_broken} broken, " f"{n_entities} entities{tag}" ) except Exception as exc: # noqa: BLE001 logger.debug("knowtation index rebuild failed: %s", exc) if not as_json: print( f" ⚠️ knowtation_link_graph — rebuild failed: {exc}", file=sys.stderr, ) result["rebuilt"] = built # Persist any newly cached parse results so the next rebuild is faster. if not dry_run: sym_cache.save() if as_json: print(json.dumps(result)) return action = "Computed" if dry_run else "Rebuilt" print(f"\n{action} {len(built)} index(es)" + (" — no files written (dry run)" if dry_run else " under .muse/indices/")) if not dry_run: print("Run 'muse code index status' to verify.") def run_purge(args: argparse.Namespace) -> None: """Delete one or all local Muse index files under ``.muse/indices/``. Indexes are derived data — the canonical commit history and object store are never modified. Any purged index can be fully recreated at any time with ``muse code index rebuild``. Indexes that are not present are silently skipped and reported under ``skipped`` rather than causing an error. Security: the ``--index`` argument is validated by argparse against ``KNOWN_INDEX_NAMES`` before ``run_purge`` is called, and ``purge_index`` performs a second validation before any ``unlink`` — no caller-supplied path fragment can reach the file system. Only files under ``.muse/indices/`` are ever deleted; the rest of the object store is untouched. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``schema_version`` Muse version string. ``purged`` List of index names whose files were found and deleted. ``skipped`` List of index names that were not present (nothing to delete). Exit codes ---------- 0 — operation completed (skipped indexes do NOT cause non-zero exit) 2 — not inside a Muse repository """ index_name: str | None = args.index_name as_json: bool = args.as_json root = require_repo() muse_dir = root / ".muse" names = list(KNOWN_INDEX_NAMES) if index_name is None else [index_name] purged: list[str] = [] skipped: list[str] = [] for name in names: if purge_index(root, name): purged.append(name) else: skipped.append(name) if as_json: print(json.dumps({ "schema_version": __version__, "purged": purged, "skipped": skipped, })) return for name in purged: print(f" 🗑️ {name} — deleted") for name in skipped: print(f" ⬜ {name} — not present, nothing to delete") if purged: print(f"\nPurged {len(purged)} index(es). Run 'muse code index rebuild' to recreate.")