"""``muse content-grep`` — full-text search across tracked files. Searches the content of every tracked file for a pattern. By default the search runs against the committed HEAD snapshot (reading from the object store). Pass ``--working-tree`` to search the actual files on disk, including uncommitted edits — essential for agents verifying their own changes before committing. ``--working-tree`` and ``--ref`` are mutually exclusive. Binary files and non-UTF-8 files are silently skipped. Regex safety: patterns are compiled with a 500-character length limit to prevent catastrophic backtracking (ReDoS). Performance (snapshot mode): object reads run in parallel using a bounded ``ThreadPoolExecutor`` (``min(8, cpu_count())`` workers). File filtering: ``--include`` and ``--exclude`` accept ``fnmatch``-style glob patterns applied to relative file paths. Usage:: muse content-grep "Cm7" # literal substring (HEAD) muse content-grep "TODO" --working-tree # search working tree (disk) muse content-grep "tempo:\\s+\\d+" # regex muse content-grep "TODO" --ignore-case # case-insensitive muse content-grep "chorus" --files-only # only file paths muse content-grep "bass" --ref feat/audio # search a branch tip muse content-grep "note" --include "*.txt" # only .txt files muse content-grep "debug" --exclude "*.min.js" muse content-grep "TODO" --max-matches 20 # cap results muse content-grep "verse" --context 2 # 2 lines of context muse content-grep "chord" --json # machine-readable JSON output schema (``--json``):: { "source": "commit" | "working-tree", "commit_id": "<64-char hex>" | null, "snapshot_id": "<64-char hex>" | null, "pattern": "", "total_files_matched": , "total_matches": , "results": [ { "path": "", "object_id": "<64-char hex>" | null, "match_count": , "matches": [ { "line_number": , "text": "", "context_before": ["", ...], "context_after": ["", ...] }, ... ] }, ... ] } Exit codes:: 0 — pattern found in at least one file 1 — no matches (or no commits) 3 — I/O error """ from __future__ import annotations import argparse import concurrent.futures import fnmatch import json import logging import os import pathlib import re import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.object_store import read_object from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_head_commit_id, read_commit, read_current_branch, read_snapshot, resolve_commit_ref, ) from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) _BINARY_CHUNK = 8192 _MAX_PATTERN_LEN = 500 # reject patterns that could cause catastrophic backtracking _DEFAULT_MAX_WORKERS = min(8, (os.cpu_count() or 1)) # Directories to skip when walking the working tree. _SKIP_DIRS: frozenset[str] = frozenset({ ".muse", ".git", "__pycache__", ".mypy_cache", ".pytest_cache", ".tox", "node_modules", ".venv", "venv", ".env", }) # --------------------------------------------------------------------------- # TypedDicts for structured output # --------------------------------------------------------------------------- class GrepMatch(TypedDict): """A single matching line within a file, with optional surrounding context.""" line_number: int text: str context_before: list[str] context_after: list[str] class GrepFileResult(TypedDict): """All matches within a single file.""" path: str object_id: str | None # None when source is working-tree match_count: int matches: list[GrepMatch] class _ContentGrepJson(TypedDict): """Top-level JSON envelope for ``muse content-grep --json``.""" source: str # "commit" | "working-tree" commit_id: str | None snapshot_id: str | None pattern: str total_files_matched: int total_matches: int results: list[GrepFileResult] # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _is_binary(data: bytes) -> bool: """Return ``True`` if *data* (the first chunk) contains null bytes.""" return b"\x00" in data def _path_matches_globs(rel_path: str, include: str | None, exclude: str | None) -> bool: """Return ``True`` if *rel_path* passes the include/exclude glob filters. ``--include`` and ``--exclude`` use ``fnmatch`` on the basename **and** on the full relative path so that patterns like ``*.py`` and ``src/*.py`` both work intuitively. """ basename = pathlib.PurePosixPath(rel_path).name if include is not None: if not (fnmatch.fnmatch(basename, include) or fnmatch.fnmatch(rel_path, include)): return False if exclude is not None: if fnmatch.fnmatch(basename, exclude) or fnmatch.fnmatch(rel_path, exclude): return False return True def _search_lines( raw: bytes, pattern: re.Pattern[str], files_only: bool, count_only: bool, context_lines: int, ) -> tuple[int, list[GrepMatch]]: """Search *raw* bytes for *pattern*; return ``(match_count, matches)``. Binary content and non-UTF-8 content return ``(0, [])``. """ probe = raw[:_BINARY_CHUNK] if _is_binary(probe): return 0, [] text = raw.decode("utf-8", errors="replace") all_lines = text.splitlines() matches: list[GrepMatch] = [] total = 0 for lineno, line in enumerate(all_lines, start=1): if pattern.search(line): total += 1 if not files_only and not count_only: before: list[str] = [] after: list[str] = [] if context_lines > 0: idx = lineno - 1 # 0-based index before = [ l.rstrip("\r") for l in all_lines[max(0, idx - context_lines) : idx] ] after = [ l.rstrip("\r") for l in all_lines[idx + 1 : idx + 1 + context_lines] ] matches.append( GrepMatch( line_number=lineno, text=line.rstrip("\r"), context_before=before, context_after=after, ) ) return total, matches def _search_object( root_path: pathlib.Path, object_id: str, pattern: re.Pattern[str], files_only: bool, count_only: bool, context_lines: int, ) -> tuple[int, list[GrepMatch]]: """Search a committed object for *pattern*; return ``(match_count, matches)``.""" try: raw = read_object(root_path, object_id) except OSError as exc: logger.warning("⚠️ grep: could not read object %s: %s", object_id[:12], exc) return 0, [] if raw is None: return 0, [] return _search_lines(raw, pattern, files_only, count_only, context_lines) def _search_disk_file( abs_path: pathlib.Path, pattern: re.Pattern[str], files_only: bool, count_only: bool, context_lines: int, ) -> tuple[int, list[GrepMatch]]: """Search a file on disk for *pattern*; return ``(match_count, matches)``.""" try: raw = abs_path.read_bytes() except OSError as exc: logger.warning("⚠️ grep: could not read %s: %s", abs_path, exc) return 0, [] return _search_lines(raw, pattern, files_only, count_only, context_lines) def _walk_working_tree( root: pathlib.Path, include_glob: str | None, exclude_glob: str | None, ) -> list[tuple[str, pathlib.Path]]: """Walk *root* recursively, skipping VCS/cache dirs, return ``(rel_path, abs_path)`` pairs.""" results: list[tuple[str, pathlib.Path]] = [] for dirpath, dirnames, filenames in os.walk(root): # Prune directories in-place so os.walk skips them entirely. dirnames[:] = sorted( d for d in dirnames if d not in _SKIP_DIRS and not d.startswith(".") and not os.path.isdir(os.path.join(dirpath, d, ".muse")) ) for filename in sorted(filenames): abs_path = pathlib.Path(dirpath) / filename try: rel_path = abs_path.relative_to(root).as_posix() except ValueError: continue if _path_matches_globs(rel_path, include_glob, exclude_glob): results.append((rel_path, abs_path)) return results # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the content-grep subcommand.""" parser = subparsers.add_parser( "content-grep", help="Search tracked file content for a pattern.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "pattern", help="Regular expression pattern to search for.", ) parser.add_argument( "--working-tree", "-w", action="store_true", dest="working_tree", help=( "Search files on disk (working tree) instead of the committed HEAD snapshot. " "Finds matches in uncommitted edits. Mutually exclusive with --ref." ), ) parser.add_argument( "--ref", default=None, help="Branch, tag, or commit SHA to search (default: HEAD). Mutually exclusive with --working-tree.", ) parser.add_argument( "--ignore-case", "-i", action="store_true", dest="ignore_case", help="Case-insensitive matching.", ) parser.add_argument( "--files-only", "-l", action="store_true", dest="files_only", help="Print only file paths with matches.", ) parser.add_argument( "--count", "-c", action="store_true", dest="count_mode", help="Print only match counts per file.", ) parser.add_argument( "--include", default=None, dest="include_glob", help="Only search files whose path matches this fnmatch glob (e.g. '*.py').", ) parser.add_argument( "--exclude", default=None, dest="exclude_glob", help="Skip files whose path matches this fnmatch glob (e.g. '*.min.js').", ) parser.add_argument( "--max-matches", "-m", type=int, default=None, dest="max_matches", help="Stop after this many total matches across all files.", ) parser.add_argument( "--context", "-C", type=int, default=0, dest="context_lines", help="Number of surrounding lines to include with each match (like grep -C).", ) parser.add_argument( "--json", action="store_true", dest="json_out", help="Emit machine-readable JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Subcommand handler # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Search tracked file content for a pattern. Pattern validation (length + ``re.compile``) is performed *before* any I/O so that invalid patterns are rejected cheaply. When ``--working-tree`` is set, files are read directly from disk so that uncommitted edits are visible. This is the correct mode for agents verifying changes before committing. Object reads (snapshot mode) are parallelised with a ``ThreadPoolExecutor`` for I/O efficiency. Binary files and non-UTF-8 files are silently skipped. Exit code 0 means at least one match was found; exit code 1 means no matches. Examples:: muse content-grep "chorus" muse content-grep "TODO" --working-tree muse content-grep "TODO|FIXME" --files-only muse content-grep "tempo" --ignore-case --json muse content-grep "chord" --ref feat/harmony muse content-grep "note" --include "*.txt" --max-matches 10 muse content-grep "verse" --context 2 """ pattern: str = args.pattern working_tree: bool = args.working_tree ref: str | None = args.ref ignore_case: bool = args.ignore_case files_only: bool = args.files_only count_mode: bool = args.count_mode include_glob: str | None = args.include_glob exclude_glob: str | None = args.exclude_glob max_matches: int | None = args.max_matches context_lines: int = max(0, args.context_lines) json_out: bool = args.json_out if working_tree and ref is not None: print("❌ --working-tree and --ref are mutually exclusive.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Validate pattern BEFORE any I/O — cheap rejection of bad inputs. if len(pattern) > _MAX_PATTERN_LEN: print( f"❌ Pattern too long ({len(pattern)} chars, max {_MAX_PATTERN_LEN}). " "Use a shorter pattern or re.escape() for literal matches.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Normalize BRE-style \| to ERE-style | so agents with grep muscle memory # get the expected behaviour without needing to know the distinction. pattern = pattern.replace(r"\|", "|") flags = re.IGNORECASE if ignore_case else 0 try: compiled: re.Pattern[str] = re.compile(pattern, flags) except re.error as exc: print(f"❌ Invalid regex: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from exc root = require_repo() # ── Working-tree mode ──────────────────────────────────────────────────── if working_tree: disk_files = _walk_working_tree(root, include_glob, exclude_glob) file_results: list[GrepFileResult] = [] total_matches = 0 for rel_path, abs_path in disk_files: if max_matches is not None and total_matches >= max_matches: break match_count, matches = _search_disk_file( abs_path, compiled, files_only, count_mode, context_lines ) if match_count > 0: if max_matches is not None: remaining = max_matches - total_matches if match_count > remaining: matches = matches[:remaining] match_count = len(matches) file_results.append( GrepFileResult( path=rel_path, object_id=None, match_count=match_count, matches=matches, ) ) total_matches += match_count _emit( file_results=file_results, total_matches=total_matches, source="working-tree", commit_id=None, snapshot_id=None, pattern=pattern, json_out=json_out, files_only=files_only, count_mode=count_mode, context_lines=context_lines, ) if not file_results: raise SystemExit(ExitCode.USER_ERROR) # exit 1 = no matches return # ── Snapshot mode (default) ────────────────────────────────────────────── repo_id = read_repo_id(root) branch = read_current_branch(root) if ref is None: commit_id = get_head_commit_id(root, branch) if commit_id is None: print("❌ No commits on current branch.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) else: commit_rec = resolve_commit_ref(root, repo_id, branch, ref) if commit_rec is None: print(f"❌ Ref '{sanitize_display(ref)}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) commit_id = commit_rec.commit_id commit = read_commit(root, commit_id) if commit is None: print(f"❌ Commit {commit_id[:12]} not found.", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) snap = read_snapshot(root, commit.snapshot_id) if snap is None: print(f"❌ Snapshot {commit.snapshot_id[:12]} not found.", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) filtered: list[tuple[str, str]] = [ (rel_path, object_id) for rel_path, object_id in sorted(snap.manifest.items()) if _path_matches_globs(rel_path, include_glob, exclude_glob) ] def _search(item: tuple[str, str]) -> tuple[str, str, int, list[GrepMatch]]: rel_path, object_id = item cnt, ms = _search_object(root, object_id, compiled, files_only, count_mode, context_lines) return rel_path, object_id, cnt, ms snap_results: list[GrepFileResult] = [] snap_total = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=_DEFAULT_MAX_WORKERS) as pool: for rel_path, object_id, match_count, matches in pool.map(_search, filtered): if match_count > 0: if max_matches is not None: remaining = max_matches - snap_total if remaining <= 0: break if match_count > remaining: matches = matches[:remaining] match_count = len(matches) snap_results.append( GrepFileResult( path=rel_path, object_id=object_id, match_count=match_count, matches=matches, ) ) snap_total += match_count _emit( file_results=snap_results, total_matches=snap_total, source="commit", commit_id=commit_id, snapshot_id=commit.snapshot_id, pattern=pattern, json_out=json_out, files_only=files_only, count_mode=count_mode, context_lines=context_lines, ) if not snap_results: raise SystemExit(ExitCode.USER_ERROR) # exit 1 = no matches # --------------------------------------------------------------------------- # Output helper (shared between modes) # --------------------------------------------------------------------------- def _emit( *, file_results: list[GrepFileResult], total_matches: int, source: str, commit_id: str | None, snapshot_id: str | None, pattern: str, json_out: bool, files_only: bool, count_mode: bool, context_lines: int, ) -> None: """Render search results to stdout in text or JSON format.""" if json_out: payload: _ContentGrepJson = { "source": source, "commit_id": commit_id, "snapshot_id": snapshot_id, "pattern": pattern, "total_files_matched": len(file_results), "total_matches": total_matches, "results": file_results, } print(json.dumps(payload, indent=2)) else: for fr in file_results: safe_path = sanitize_display(fr["path"]) if files_only: print(safe_path) elif count_mode: print(f"{safe_path}:{fr['match_count']}") else: for m in fr["matches"]: if context_lines > 0: for ctx in m["context_before"]: print(f"{safe_path}:{m['line_number']}-{sanitize_display(ctx)}") print(f"{safe_path}:{m['line_number']}:{sanitize_display(m['text'])}") for ctx in m["context_after"]: print(f"{safe_path}:{m['line_number']}+{sanitize_display(ctx)}") else: print(f"{safe_path}:{m['line_number']}:{sanitize_display(m['text'])}")