"""muse code grep -- semantic symbol search across the symbol graph. Unlike ``git grep`` which searches raw text lines, ``muse code grep`` searches the *typed symbol graph* -- only returning actual symbol declarations with their kind, file, line number, and stable content hash. No false positives from comments, string literals, or call sites. Every result is a real symbol that exists in the repository. By default, PATTERN is matched case-insensitively against the bare symbol name. When PATTERN contains a ``.`` or ``::`` it is also matched against the fully-qualified name, so ``Invoice.validate`` finds only that specific method rather than every symbol named ``validate``. Usage:: muse code grep "validate" # symbols whose name contains "validate" muse code grep "Invoice.validate" # exact qualified-name match muse code grep "^handle" --regex # names matching regex "^handle" muse code grep "Invoice" --kind class # only class symbols muse code grep "compute" --language go # only Go symbols (case-insensitive) muse code grep "total" --file billing # scope to one file (fast) muse code grep "total" --commit HEAD~5 # search a historical snapshot muse code grep "validate" --count # just the total count muse code grep "validate" --json # machine-readable output for agents Output:: muse/billing.py::validate_amount fn line 8 muse/auth.py::validate_token fn line 14 muse/auth.py::Validator class line 22 muse/auth.py::Validator.validate method line 28 4 match(es) across 2 file(s) Security note: patterns are capped at 512 characters to prevent ReDoS. Invalid regex syntax is caught and reported as exit 1 rather than crashing. """ from __future__ import annotations import argparse import json import logging import pathlib import re import sys from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref from muse.plugins.code._query import language_of, symbols_for_snapshot from muse.plugins.code._query import _SUFFIX_LANG # for case-insensitive lang normalisation from muse.plugins.code.ast_parser import SymbolRecord from muse.core.validation import sanitize_display from muse.core.store import Manifest type _IconMap = dict[str, str] logger = logging.getLogger(__name__) # Guard against ReDoS: reject patterns longer than this before compiling. _MAX_PATTERN_LEN: int = 512 _KIND_ICON: _IconMap = { "function": "fn", "async_function": "fn~", "class": "class", "method": "method", "async_method": "method~", "variable": "var", "import": "import", } # Canonical map for case-insensitive --language matching. _LANG_CANONICAL: _IconMap = {lang.lower(): lang for lang in set(_SUFFIX_LANG.values())} def _normalise_language(lang: str) -> str: return _LANG_CANONICAL.get(lang.strip().lower(), lang.strip()) # --------------------------------------------------------------------------- # Repository helpers # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # File-filter helpers (same as symbols.py) # --------------------------------------------------------------------------- def _file_matches(file_path: str, file_filter: str) -> bool: """True if *file_path* equals or ends with ``/``.""" if file_path == file_filter: return True normalized = file_filter.replace("\\", "/") return file_path.endswith("/" + normalized) def _resolve_file_filter( file_filter: str, manifest: Manifest, ) -> str | None: """Resolve a partial path suffix to the exact manifest key. Exits non-zero on ambiguity; returns ``None`` when there is no match (caller handles the empty result). """ matching = [p for p in sorted(manifest) if _file_matches(p, file_filter)] if len(matching) == 1: return matching[0] if len(matching) > 1: print( f"❌ '{file_filter}' is ambiguous — matches {len(matching)} files. " "Use a more specific path:", file=sys.stderr, ) for m in matching[:10]: print(f" {m}", file=sys.stderr) if len(matching) > 10: print(f" … and {len(matching) - 10} more", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return None # no match — caller handles empty result # --------------------------------------------------------------------------- # Argument parser registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the grep subcommand.""" parser = subparsers.add_parser( "grep", help="Search the symbol graph by name — not file text.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "pattern", metavar="PATTERN", help="Name pattern to search for.", ) parser.add_argument( "--regex", "-e", action="store_true", dest="use_regex", help="Treat PATTERN as a regular expression (default: substring match).", ) parser.add_argument( "--kind", "-k", default=None, metavar="KIND", dest="kind_filter", help="Restrict to symbols of this kind (function, class, method, …).", ) parser.add_argument( "--language", "-l", default=None, metavar="LANG", dest="language_filter", help="Restrict to symbols from files of this language (case-insensitive).", ) parser.add_argument( "--file", "-f", default=None, metavar="PATH", dest="file_filter", help=( "Scope to a single file. Accepts an exact path or a unique suffix " "(e.g. 'billing.py' matches 'src/billing.py'). Up to 24x faster." ), ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help="Search a historical commit instead of the working tree.", ) parser.add_argument( "--hashes", action="store_true", dest="show_hashes", help="Include content hashes in output.", ) output_group = parser.add_mutually_exclusive_group() output_group.add_argument( "--count", action="store_true", dest="count_only", help="Print only the total match count.", ) output_group.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as structured JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Search the symbol graph by name — not file text. ``muse code grep`` searches the typed, content-addressed symbol graph. Every result is a real symbol declaration — no false positives from comments, string literals, or call sites. The ``--regex`` flag enables full Python regex syntax. Without it, PATTERN is matched as a case-insensitive substring. When PATTERN contains a ``.`` or ``::`` it is also matched against the fully-qualified symbol name, so ``Invoice.validate`` returns exactly that method rather than every symbol named ``validate``. Use ``--file`` to scope the search to one file — this is up to 24x faster than a full-codebase scan and sufficient for most targeted lookups. When ``--commit`` is omitted the working tree is searched, reflecting edits not yet committed. Patterns are capped at 512 characters to guard against ReDoS. """ pattern: str = args.pattern use_regex: bool = args.use_regex kind_filter: str | None = args.kind_filter language_filter: str | None = args.language_filter file_filter: str | None = args.file_filter ref: str | None = args.ref show_hashes: bool = args.show_hashes count_only: bool = args.count_only as_json: bool = args.as_json # ── Input validation ────────────────────────────────────────────────────── if len(pattern) > _MAX_PATTERN_LEN: print( f"❌ Pattern too long ({len(pattern)} chars) — maximum is {_MAX_PATTERN_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if language_filter is not None: language_filter = _normalise_language(language_filter) # When pattern contains a separator, also search qualified names. search_qualified = "." in pattern or "::" in pattern try: regex = ( re.compile(pattern, re.IGNORECASE) if use_regex else re.compile(re.escape(pattern), re.IGNORECASE) ) except re.error as exc: print(f"❌ Invalid regex pattern: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Repo / commit resolution ────────────────────────────────────────────── root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) 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 {} # ── File-filter resolution ──────────────────────────────────────────────── resolved_file_filter = file_filter if file_filter is not None: found = _resolve_file_filter(file_filter, manifest) if found is not None: resolved_file_filter = found # None → no match; pass original so symbols_for_snapshot returns {} # ── Working-tree vs object-store mode ──────────────────────────────────── working_tree = ref is None workdir = root if working_tree else None source_ref = "working-tree" if working_tree else commit.commit_id[:8] # ── Symbol extraction ───────────────────────────────────────────────────── symbol_map = symbols_for_snapshot( root, manifest, kind_filter=kind_filter, file_filter=resolved_file_filter, language_filter=language_filter, workdir=workdir, ) # ── Pattern matching ────────────────────────────────────────────────────── matches: list[tuple[str, str, SymbolRecord]] = [] for file_path, tree in sorted(symbol_map.items()): for addr, rec in sorted(tree.items(), key=lambda kv: kv[1]["lineno"]): name_hit = regex.search(rec["name"]) qual_hit = search_qualified and regex.search(rec["qualified_name"]) if name_hit or qual_hit: matches.append((file_path, addr, rec)) # ── Output ──────────────────────────────────────────────────────────────── if count_only: print(f"{len(matches)} match(es)") return if as_json: results: list[dict[str, str | int | bool]] = [] for _fp, addr, rec in matches: results.append({ "address": addr, "kind": rec["kind"], "name": rec["name"], "qualified_name": rec["qualified_name"], "file": addr.split("::")[0], "lineno": rec["lineno"], "language": language_of(addr.split("::")[0]), "content_id": rec["content_id"], }) print(json.dumps({ "source_ref": source_ref, "working_tree": working_tree, "pattern": pattern, "total_matches": len(matches), "results": results, }, indent=2)) return if not matches: print(f" (no symbols matching '{sanitize_display(pattern)}')") return files_seen: set[str] = set() for file_path, addr, rec in matches: files_seen.add(file_path) icon = _KIND_ICON.get(rec["kind"], rec["kind"]) line = rec["lineno"] hash_part = f" {rec['content_id'][:8]}.." if show_hashes else "" print(f" {sanitize_display(addr):<60} {icon:<10} line {line:>4}{hash_part}") print(f"\n{len(matches)} match(es) across {len(files_seen)} file(s)")