"""Shared query helpers for the code-domain CLI commands. This module provides the low-level primitives that multiple code-domain commands need — symbol extraction from snapshots, commit-graph walking, and language classification — so each command can stay thin. None of these functions are part of the public ``CodePlugin`` API. They are internal helpers for the CLI layer and must not be imported by any core module. """ import itertools import logging import pathlib import re from collections.abc import Iterator from muse.core.types import Manifest, Metadata, blob_id, short_id from muse.core.object_store import read_object from muse.core.stat_cache import StatCache from muse.core.commits import ( CommitRecord, read_commit, ) from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.domain import DomainOp from muse.plugins.code.ast_parser import ( SymbolTree, parse_symbols, ) type _SymbolTreeMap = dict[str, SymbolTree] type _CommitIndex = dict[str, CommitRecord] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Language classification # --------------------------------------------------------------------------- _SUFFIX_LANG: Metadata = { ".py": "Python", ".pyi": "Python", ".ts": "TypeScript", ".tsx": "TypeScript", ".js": "JavaScript", ".jsx": "JavaScript", ".mjs": "JavaScript", ".cjs": "JavaScript", ".go": "Go", ".rs": "Rust", ".java": "Java", ".cs": "C#", ".c": "C", ".h": "C", ".cpp": "C++", ".cc": "C++", ".cxx": "C++", ".hpp": "C++", ".hxx": "C++", ".rb": "Ruby", ".kt": "Kotlin", ".kts": "Kotlin", ".swift": "Swift", ".md": "Markdown", ".rst": "reStructuredText", ".txt": "Text", ".toml": "TOML", ".yaml": "YAML", ".yml": "YAML", ".json": "JSON", ".jsonc": "JSON", ".css": "CSS", ".scss": "SCSS", ".html": "HTML", ".htm": "HTML", ".sql": "SQL", ".sh": "Shell", ".bash": "Shell", ".zsh": "Shell", ".proto": "Protobuf", ".tf": "Terraform", } def language_of(file_path: str) -> str: """Return a display language name for *file_path* based on its suffix.""" suffix = pathlib.PurePosixPath(file_path).suffix.lower() return _SUFFIX_LANG.get(suffix, suffix or "(no ext)") def is_semantic(file_path: str) -> bool: """Return ``True`` if *file_path* has a suffix with AST-level support.""" from muse.plugins.code.ast_parser import SEMANTIC_EXTENSIONS suffix = pathlib.PurePosixPath(file_path).suffix.lower() return suffix in SEMANTIC_EXTENSIONS # --------------------------------------------------------------------------- # Language normalisation # --------------------------------------------------------------------------- _LANG_CANONICAL: dict[str, str] = {lang.lower(): lang for lang in set(_SUFFIX_LANG.values())} def normalise_language(lang: str) -> str: """Normalise a user-supplied language name to its canonical display form. Case-insensitive. Returns the canonical capitalisation (e.g. ``"python"`` → ``"Python"``) or the stripped input unchanged if unrecognised. """ return _LANG_CANONICAL.get(lang.strip().lower(), lang.strip()) # --------------------------------------------------------------------------- # Test-file detection # --------------------------------------------------------------------------- _TEST_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"(^|/)test_"), re.compile(r"_test\.py$"), re.compile(r"(^|/)tests/"), re.compile(r"(^|/)spec/"), re.compile(r"(^|/)conftest\.py$"), ) def is_test_file(file_path: str) -> bool: """Return ``True`` if *file_path* looks like a test or conftest file.""" return any(p.search(file_path) for p in _TEST_PATTERNS) # --------------------------------------------------------------------------- # Symbol extraction from a snapshot manifest # --------------------------------------------------------------------------- def _rekey_tree(tree: SymbolTree, file_path: str) -> SymbolTree: """Re-key *tree* so every address starts with *file_path*. The ``SymbolCache`` is keyed by content SHA-256. When two files have identical bytes they share one cache entry whose addresses carry the path of whichever file was parsed first. This function patches those addresses to the actual *file_path* being processed — an O(n) pass that is only triggered on a cache hit where the prefix differs. """ first = next(iter(tree), None) if first is None: return tree cached_prefix = first.split("::")[0] if "::" in first else None if cached_prefix == file_path: return tree # Addresses already correct — common case. return { f"{file_path}::{addr.split('::', 1)[1]}" if "::" in addr else addr: rec for addr, rec in tree.items() } def symbols_for_snapshot( root: pathlib.Path, manifest: Manifest, *, kind_filter: str | None = None, file_filter: str | None = None, language_filter: str | None = None, workdir: pathlib.Path | None = None, cache: SymbolCache | None = None, stat_cache: StatCache | None = None, stats: dict[str, int] | None = None, ) -> _SymbolTreeMap: """Extract symbol trees for all semantic files in *manifest*. Results are served from the persistent symbol cache when available, cutting a full-snapshot scan from ~1,300 ms to ~22 ms on a warm cache. The cache is keyed by the SHA-256 of the file bytes (``object_id`` for committed files; freshly computed SHA-256 for working-tree reads), so every hit is guaranteed correct. Args: root: Repository root (used to locate the object store and load/save the symbol cache). manifest: Snapshot manifest mapping file path → SHA-256. kind_filter: If set, only include symbols with this ``kind``. file_filter: If set, only include symbols from this exact file path. language_filter: If set, only include symbols from files of this language. workdir: When set, file content is read from *workdir* first (working-tree mode), falling back to the object store if the file does not exist on disk. Pass ``root`` to get live, uncommitted changes reflected immediately. cache: Optional pre-loaded ``SymbolCache`` instance. When ``None`` the cache is loaded from disk automatically and saved at the end of the call. Pass an explicit instance when batching multiple calls to share one cache load/save cycle. stat_cache: Optional ``StatCache`` instance. When supplied and ``workdir`` is set, the SHA-256 cache key for each disk file is derived from ``(ino, mtime, size)`` without reading the file bytes. On a warm ``SymbolCache`` hit this means zero file reads for unchanged files — the dominant cost of the working-tree diff path. Has no effect when ``workdir`` is ``None``. stats: Optional mutable dict the caller can pass to observe per-file cache hit/miss counts (muse#49 Phase 2) — increments ``stats["symbol_hits"]`` and ``stats["symbol_misses"]`` in place. ``None`` (the default) skips this bookkeeping, so existing callers see zero behavior change. Returns: Dict mapping ``file_path → SymbolTree``; empty trees are omitted. """ if cache is None: active_cache: SymbolCache = load_symbol_cache(root) own_cache = True else: active_cache = cache own_cache = False result: _SymbolTreeMap = {} for file_path, object_id in sorted(manifest.items()): if not is_semantic(file_path): continue if file_filter and file_path != file_filter: continue if language_filter and language_of(file_path) != language_filter: continue # --- Determine cache key and read bytes if needed ---------------- # For working-tree files the object_id is the committed version; # the disk content may differ. We compute the SHA-256 of whatever # bytes we actually parse so the cache key is always content-addressed. cache_key = object_id raw: bytes | None = None # When stat_cache is active we defer reading disk bytes until we know # whether the SymbolCache will miss. This variable holds the path to # read from if a sym-cache miss forces us to load and parse the file. _disk_for_miss: pathlib.Path | None = None if workdir is not None: disk_path = workdir / file_path if disk_path.is_file(): try: if stat_cache is not None: st = disk_path.stat() # get_cached returns the SHA-256 (sha256:… prefix). # On a stat-cache hit it does so without reading bytes. # On a stat-cache miss it reads and hashes internally. cache_key = stat_cache.get_cached( file_path, str(disk_path), st.st_mtime, st.st_size, st.st_ino, ) _disk_for_miss = disk_path # bytes deferred else: raw = disk_path.read_bytes() # Recompute the key for the actual disk content. cache_key = blob_id(raw) except OSError as exc: logger.debug("Could not read %s from workdir: %s", file_path, exc) # --- Cache lookup ------------------------------------------------ tree = active_cache.get(cache_key) if tree is None: # Cache miss — fetch bytes and parse. if stats is not None: stats["symbol_misses"] = stats.get("symbol_misses", 0) + 1 if raw is None: if _disk_for_miss is not None: # stat_cache path: bytes were not pre-read; load them now. try: raw = _disk_for_miss.read_bytes() except OSError as exc: logger.debug("Could not read %s from disk: %s", file_path, exc) if raw is None: raw = read_object(root, object_id) if raw is None: logger.debug("Object %s missing for %s — skipping", short_id(object_id), file_path) continue tree = parse_symbols(raw, file_path) active_cache.put(cache_key, tree) else: # Cache hit — re-key addresses when two files share the same # SHA-256 (identical content). The cache stores the tree built # for whichever file was parsed first; its addresses carry that # file's path prefix. Patch them to the current file_path so # cross-file clone detection and all downstream consumers get # correct symbol addresses. if stats is not None: stats["symbol_hits"] = stats.get("symbol_hits", 0) + 1 tree = _rekey_tree(tree, file_path) # --- Apply filters on top of the (potentially cached) full tree -- if kind_filter: tree = {addr: rec for addr, rec in tree.items() if rec["kind"] == kind_filter} if tree: result[file_path] = tree if own_cache: active_cache.save() return result # --------------------------------------------------------------------------- # Commit-graph walking # --------------------------------------------------------------------------- def walk_commits_bfs( root: pathlib.Path, start_commit_id: str, max_commits: int = 500, stop_at_commit_id: str | None = None, ) -> tuple[list[CommitRecord], bool]: """BFS walk of the commit DAG from *start_commit_id*, newest-first. Unlike the linear :func:`walk_commits`, this follows ``parent2_commit_id`` at merge commits so events on merged feature branches are never missed. Args: root: Repository root. start_commit_id: SHA-256 of the commit to start from. max_commits: Safety cap — returns ``truncated=True`` if reached. stop_at_commit_id: When set, stop BFS *before* entering this commit (exclusive lower bound, useful for range queries). Returns: ``(commits, truncated)`` — list sorted newest-first by ``committed_at``; ``truncated=True`` when ``max_commits`` was hit before exhausting the DAG. """ from muse.core.graph import iter_ancestors prune = (lambda cid: cid == stop_at_commit_id) if stop_at_commit_id else None truncated = False commits_by_id: _CommitIndex = {} for commit in iter_ancestors(root, start_commit_id, prune=prune): if len(commits_by_id) >= max_commits: truncated = True break commits_by_id[commit.commit_id] = commit return ( sorted(commits_by_id.values(), key=lambda c: c.committed_at, reverse=True), truncated, ) # --------------------------------------------------------------------------- # Op traversal helpers # --------------------------------------------------------------------------- def dir_of(path: str) -> str: """Return the immediate parent directory of *path* as a POSIX string. - ``"src/billing.py"`` → ``"src"`` - ``"muse/cli/commands/cat.py"`` → ``"muse/cli/commands"`` - ``"main.py"`` → ``"."`` - ``"src/"`` → ``"src"`` (trailing slash stripped first) """ if path.endswith("/"): return path.rstrip("/") or "." parent = str(pathlib.PurePosixPath(path).parent) return parent def flat_directory_ops(ops: list[DomainOp]) -> Iterator[DomainOp]: """Yield directory-level ops from *ops*. Yields: - ``RenameOp`` (op == "rename") - ``InsertOp`` / ``DeleteOp`` whose address ends with "/" or whose content_summary starts with "directory:" Skips file-level and symbol-level ops. """ for op in ops: op_type = op["op"] if op_type == "rename": yield op elif op_type in ("insert", "delete"): addr: str = op["address"] summary: str = op.get("content_summary", "") # type: ignore[arg-type] if addr.endswith("/") or summary.startswith("directory:"): yield op def touched_directories(ops: list[DomainOp]) -> frozenset[str]: """Return directories affected by the ops in *ops*. Counts: - The parent dir of every file with semantic child ops in a PatchOp. - Both ``address`` and ``from_address`` of RenameOps. Ignores PatchOps with empty child_ops (no semantic change). """ dirs: set[str] = set() for op in ops: op_type = op["op"] if op_type == "patch": if op["child_ops"]: # type: ignore[index] dirs.add(dir_of(op["address"])) # type: ignore[arg-type] elif op_type == "rename": dirs.add(op["address"]) # type: ignore[arg-type] dirs.add(op["from_address"]) # type: ignore[union-attr] return frozenset(dirs) def flat_symbol_ops(ops: list[DomainOp]) -> Iterator[DomainOp]: """Yield all leaf ops, recursing into PatchOp.child_ops. Only yields ops that have a symbol-level address (i.e. contain ``::``). """ for op in ops: if op["op"] == "patch": for child in op["child_ops"]: if "::" in child["address"]: yield child elif "::" in op["address"]: yield op def touched_files(ops: list[DomainOp]) -> frozenset[str]: """Return the set of file paths that appear as PatchOp addresses in *ops*. Only counts files that had symbol-level child ops (semantic changes), not coarse file-level replace/insert/delete ops. """ files: set[str] = set() for op in ops: if op["op"] == "patch" and op["child_ops"]: files.add(op["address"]) return frozenset(files) def file_pairs(files: frozenset[str]) -> Iterator[tuple[str, str]]: """Yield all ordered pairs ``(a, b)`` with ``a < b`` from *files*.""" yield from itertools.combinations(sorted(files), 2)