"""muse code query — symbol graph predicate query (v2). SQL for your codebase. A full predicate DSL over the typed, content-addressed symbol graph — with OR, NOT, grouping, and an expanded field set. v2 grammar:: expr = or_expr or_expr = and_expr ( OR and_expr )* and_expr = not_expr ( [AND] not_expr )* # implicit AND not_expr = NOT primary | primary primary = "(" expr ")" | atom atom = KEY OP VALUE Supported operators:: = exact match ~= contains (case-insensitive) ^= starts with (case-insensitive) $= ends with (case-insensitive) != not equal Supported keys:: kind function | class | method | variable | import | … language Python | Go | Rust | TypeScript | … name bare symbol name qualified_name dotted name (User.save) file file path hash content_id prefix (exact-body match) body_hash body_hash prefix signature_id signature_id prefix lineno_gt symbol starts after line N lineno_lt symbol starts before line N size_gt symbol body exceeds N lines (end_lineno − lineno > N) size_lt symbol body shorter than N lines Usage:: muse code query "kind=function" "language=Python" "name~=validate" muse code query "(kind=function OR kind=method) name^=_" muse code query "NOT kind=import" "file~=billing" muse code query "hash=a3f2c9" muse code query "kind=function" "name$=_test" --commit HEAD~10 muse code query "kind=function" "name~=validate" --all-commits muse code query "kind=function" "size_gt=50" --sort size # biggest fns muse code query "kind=function" "name~=compute" --count # just the count muse code query "kind=function" --unique-bodies # find clones muse code query "kind=function" --all-commits --since 2026-01-01 # added this year """ from __future__ import annotations import argparse import collections.abc import datetime import json import logging import pathlib import sys from muse._version import __version__ from muse.core.errors import ExitCode from muse.core.repo import parse_date_arg, read_repo_id, require_repo from muse.core.store import ( CommitRecord, get_all_commits, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.plugins.code._predicate import Predicate, PredicateError, parse_query from muse.plugins.code._query import language_of, symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolRecord from muse.core.validation import clamp_int, sanitize_display type _QueryResult = dict[str, str | int | bool] type _StrMap = dict[str, str] type _IconMap = dict[str, str] logger = logging.getLogger(__name__) _KIND_ICON: _IconMap = { "function": "fn", "async_function": "fn~", "class": "class", "method": "method", "async_method": "method~", "variable": "var", "import": "import", } _VALID_SORT_FIELDS = frozenset({"file", "name", "kind", "lineno", "size"}) class _HistoricalMatch: """A symbol match found in a historical commit (--all-commits mode).""" def __init__( self, address: str, rec: SymbolRecord, commit: CommitRecord, first_seen: bool, ) -> None: self.address = address self.rec = rec self.commit = commit self.first_seen = first_seen def to_dict(self) -> _QueryResult: return { "address": self.address, "kind": self.rec["kind"], "name": self.rec["name"], "content_id": self.rec["content_id"], "first_seen": self.first_seen, "commit_id": self.commit.commit_id, "commit_message": self.commit.message, "committed_at": self.commit.committed_at.isoformat(), "branch": self.commit.branch, } def _query_all_commits( root: pathlib.Path, filters: list[Predicate], max_commits: int, since: datetime.date | None, until: datetime.date | None, ) -> tuple[list[_HistoricalMatch], bool]: """Walk every commit oldest-first, apply predicates against each snapshot. Shares one ``SymbolCache`` instance across all snapshot loads so the cache is read from disk exactly once and written back at most once — instead of once per snapshot. On a warm cache this reduces wall time from O(n×200ms) to O(1×load + n×dict_lookup). Deduplicates on ``snapshot_id`` — commits sharing a snapshot (e.g. merge commits with no file changes) are processed exactly once. Returns: ``(matches, truncated)`` — ``truncated`` is True when the walk was capped at ``max_commits``. """ all_commits = get_all_commits(root) if not all_commits: return [], False sorted_commits = sorted(all_commits, key=lambda c: c.committed_at) # Apply date filters early to avoid unnecessary snapshot loading. if since is not None: sorted_commits = [ c for c in sorted_commits if c.committed_at.date() >= since ] if until is not None: sorted_commits = [ c for c in sorted_commits if c.committed_at.date() <= until ] truncated = len(sorted_commits) > max_commits sorted_commits = sorted_commits[:max_commits] results: list[_HistoricalMatch] = [] first_seen_map: _StrMap = {} seen_snapshots: set[str] = set() # Load the symbol cache once; share it across all snapshot iterations. shared_cache: SymbolCache = load_symbol_cache(root) try: for commit in sorted_commits: # Skip commits whose snapshot was already processed. if commit.snapshot_id in seen_snapshots: continue seen_snapshots.add(commit.snapshot_id) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} if not manifest: continue symbol_map = symbols_for_snapshot(root, manifest, cache=shared_cache) for file_path, tree in sorted(symbol_map.items()): for addr, rec in sorted(tree.items(), key=lambda kv: kv[1]["lineno"]): if not all(f(file_path, rec) for f in filters): continue cid = rec["content_id"] is_first = cid not in first_seen_map if is_first: first_seen_map[cid] = commit.commit_id results.append(_HistoricalMatch(addr, rec, commit, is_first)) finally: # Persist any newly parsed entries even if we exit early. shared_cache.save() return results, truncated _SortTuple = tuple[str, str, SymbolRecord] def _sort_key(sort_by: str) -> collections.abc.Callable[[_SortTuple], tuple[str | int, ...]]: """Return a sort key function for a list of ``(file_path, addr, rec)`` tuples.""" if sort_by == "name": return lambda t: (t[2]["name"].lower(), t[0], t[2]["lineno"]) if sort_by == "kind": return lambda t: (t[2]["kind"], t[0], t[2]["lineno"]) if sort_by == "lineno": return lambda t: (t[2]["lineno"], t[0]) if sort_by == "size": # Negate size so largest comes first. return lambda t: (-(t[2]["end_lineno"] - t[2]["lineno"]), t[0]) # Default: file then lineno. return lambda t: (t[0], t[2]["lineno"]) def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the query subcommand.""" parser = subparsers.add_parser( "query", help="Query the symbol graph with a predicate DSL.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "predicates", nargs="+", metavar="PREDICATE", help='One or more predicates, e.g. "kind=function" "name~=validate".', ) parser.add_argument( "--commit", "-c", dest="ref", default=None, metavar="REF", help="Query a historical snapshot instead of HEAD.", ) parser.add_argument( "--all-commits", action="store_true", help=( "Search across ALL commits (every branch). Enables temporal" " hash= queries: find when a function body first appeared." " Mutually exclusive with --commit." ), ) parser.add_argument( "--since", metavar="YYYY-MM-DD", default=None, help=( "With --all-commits: only consider commits on or after this date." ), ) parser.add_argument( "--until", metavar="YYYY-MM-DD", default=None, help=( "With --all-commits: only consider commits on or before this date." ), ) parser.add_argument( "--max-commits", type=int, default=10_000, metavar="N", help=( "With --all-commits: cap the number of commits walked" " (default: 10000)." ), ) parser.add_argument( "--limit", type=int, default=0, metavar="N", help="Cap the number of results returned (0 = unlimited).", ) parser.add_argument( "--sort", default="file", metavar="FIELD", choices=sorted(_VALID_SORT_FIELDS), help=( f"Sort results by field: {', '.join(sorted(_VALID_SORT_FIELDS))}" " (default: file)." ), ) parser.add_argument( "--count", action="store_true", help="Print only the count of matching symbols — no symbol list.", ) parser.add_argument( "--unique-bodies", action="store_true", help=( "Deduplicate by content_id — show only unique implementations." " Turns muse query into a clone detector." ), ) parser.add_argument( "--hashes", dest="show_hashes", action="store_true", help="Include content hashes in output.", ) parser.add_argument( "--json", dest="as_json", action="store_true", help="Emit results as JSON.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Query the symbol graph with a predicate DSL. ``muse query`` is SQL for your codebase. Every predicate is evaluated against the typed, content-addressed symbol graph — not raw text. New in v2.1: ``size_gt=N`` / ``size_lt=N`` — filter by symbol body line count. ``--count`` — emit only the result count. ``--limit N`` — cap results (like SQL LIMIT). ``--sort FIELD`` — sort by file, name, kind, lineno, or size. ``--unique-bodies`` — deduplicate by content_id (clone detector mode). ``--since / --until YYYY-MM-DD`` — temporal range for --all-commits. """ predicates: list[str] = args.predicates ref: str | None = args.ref all_commits: bool = args.all_commits show_hashes: bool = args.show_hashes as_json: bool = args.as_json count_only: bool = args.count limit: int = clamp_int(args.limit, 0, 10000, 'limit') sort_by: str = args.sort unique_bodies: bool = args.unique_bodies max_commits: int = clamp_int(args.max_commits, 1, 100000, 'max_commits') root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) if not predicates: print("❌ At least one predicate is required.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if all_commits and ref is not None: print( "❌ --all-commits and --commit are mutually exclusive.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if limit < 0: print("❌ --limit must be >= 0.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if max_commits < 1: print("❌ --max-commits must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Parse --since / --until date filters. since_date: datetime.date | None = ( parse_date_arg(args.since, "--since").date() if args.since else None ) until_date: datetime.date | None = ( parse_date_arg(args.until, "--until").date() if args.until else None ) if (since_date or until_date) and not all_commits: print( "❌ --since / --until require --all-commits.", file=sys.stderr ) raise SystemExit(ExitCode.USER_ERROR) # Parse predicates via the v2 grammar. try: combined_predicate: Predicate = parse_query(predicates) except PredicateError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) filters: list[Predicate] = [combined_predicate] # ── --all-commits mode ──────────────────────────────────────────────────── if all_commits: historical, truncated = _query_all_commits( root, filters, max_commits, since_date, until_date ) if as_json: print( json.dumps( { "schema_version": __version__, "mode": "all-commits", "truncated": truncated, "results": [h.to_dict() for h in historical], }, indent=2, ) ) return if not historical: pred_display = " AND ".join(predicates) print( f" (no symbols matching: {pred_display}" f" [searched all commits])" ) return # Deduplicate for display: show unique addresses with first-seen commit. seen_addrs: set[str] = set() unique: list[_HistoricalMatch] = [] for h in historical: if h.first_seen and h.address not in seen_addrs: seen_addrs.add(h.address) unique.append(h) if limit > 0: unique = unique[:limit] if count_only: print(len(unique)) return pred_display = " AND ".join(predicates) trunc_note = " ⚠️ truncated" if truncated else "" print( f"\n{len(unique)} unique symbol(s) matching" f" [{pred_display}] across all commits{trunc_note}\n" ) for h in unique: date_str = h.commit.committed_at.strftime("%Y-%m-%d") short_id = h.commit.commit_id[:8] icon = _KIND_ICON.get(h.rec["kind"], h.rec["kind"]) hash_part = f" {h.rec['content_id'][:8]}.." if show_hashes else "" branch_label = ( f" [{h.commit.branch}]" if h.commit.branch else "" ) print( f" {h.address:<60} {icon:<8}" f" first seen {short_id} {date_str}" f"{branch_label}{hash_part}" ) return # ── Single-snapshot mode (default) ──────────────────────────────────────── commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr ) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} symbol_map = symbols_for_snapshot(root, manifest) # Collect matches. matches: list[tuple[str, str, SymbolRecord]] = [] for file_path, tree in symbol_map.items(): for addr, rec in tree.items(): if all(f(file_path, rec) for f in filters): matches.append((file_path, addr, rec)) # Sort. matches.sort(key=_sort_key(sort_by)) # Unique-bodies: deduplicate by content_id. if unique_bodies: seen_cids: set[str] = set() deduped: list[tuple[str, str, SymbolRecord]] = [] for fp, addr, rec in matches: cid = rec["content_id"] if cid not in seen_cids: seen_cids.add(cid) deduped.append((fp, addr, rec)) matches = deduped # Apply limit. limited = limit > 0 and len(matches) > limit if limited: matches = matches[:limit] # Count-only output. if count_only: print(len(matches)) return # JSON output. if as_json: out: list[dict[str, str | int]] = [] for fp, addr, rec in matches: out.append( { "address": addr, "kind": rec["kind"], "name": rec["name"], "qualified_name": rec["qualified_name"], "file": fp, "lineno": rec["lineno"], "end_lineno": rec["end_lineno"], "size": rec["end_lineno"] - rec["lineno"], "language": language_of(fp), "content_id": rec["content_id"], "body_hash": rec["body_hash"], "signature_id": rec["signature_id"], } ) print( json.dumps( { "schema_version": __version__, "commit": commit.commit_id[:8], "sort": sort_by, "unique_bodies": unique_bodies, "truncated": limited, "results": out, }, indent=2, ) ) return # Human-readable output. if not matches: pred_str = " AND ".join(predicates) print(f" (no symbols matching: {pred_str})") return files_seen: set[str] = set() for fp, addr, rec in matches: files_seen.add(fp) icon = _KIND_ICON.get(rec["kind"], rec["kind"]) line = rec["lineno"] size = rec["end_lineno"] - rec["lineno"] hash_part = f" {rec['content_id'][:8]}.." if show_hashes else "" size_part = f" {size:>3}L" if sort_by == "size" else "" print(f" {sanitize_display(addr):<60} {icon:<10} line {line:>4}{size_part}{hash_part}") pred_display = " AND ".join(predicates) trunc_note = f" (limited to {limit})" if limited else "" print( f"\n{len(matches)} match(es) across {len(files_seen)} file(s)" f" [{pred_display}]{trunc_note}" )