"""muse code query-history — temporal symbol search across commit history. Searches the commit history for symbols matching a predicate expression, bounded by a commit range. Unlike ``muse code query --all-commits``, this command is focused on *change events* — it shows when each symbol first appeared, when it was last seen, how many commits it survived, and how many distinct implementations it had. It answers questions that are impossible in Git: * "Find all public Python functions introduced after tag v1.0" * "Show me every class whose signature changed in the last 50 commits" * "Which functions were present in tag v1.0 but are gone in tag v2.0?" * "Find the most volatile symbols in the last 100 commits" * "What was deleted between the last two releases?" Usage:: muse code query-history "kind=function" "language=Python" muse code query-history "name~=validate" --from v1.0 --to HEAD muse code query-history "kind=class" --from abc12345 muse code query-history "file~=billing" "kind=function" --json muse code query-history "kind=function" --changed-only --sort changes muse code query-history "kind=function" --removed-only --from v1.0 --to v2.0 muse code query-history "kind=function" --introduced-only --from v1.0 Output:: Symbol history — kind=function language=Python (42 commits) ────────────────────────────────────────────────────────────── src/billing.py::compute_total function [12 commits] 3 versions 2026-01-01..2026-03-10 src/billing.py::compute_tax function [ 8 commits] stable 2026-01-15..2026-03-10 └─ introduced: a1b2c3d4 2026-01-15 └─ last seen: f7a8b9c0 2026-03-10 Flags: ``--from REF`` Start of the commit range (exclusive; default: initial commit). ``--to REF`` End of the commit range (inclusive; default: HEAD). ``--changed-only`` Show only symbols that changed at least once (change_count > 1). ``--introduced-only`` Show only symbols that exist in ``--to`` but not in ``--from`` (net-new). ``--removed-only`` Show only symbols that exist in ``--from`` but not in ``--to`` (deleted). ``--sort FIELD`` Sort results by: address (default), commits, changes, first, last. ``--min-changes N`` Only show symbols with at least N distinct versions. ``--count`` Emit only the count of matching symbols. ``--limit N`` Cap the number of results returned. ``--max-commits N`` Cap the number of commits walked (default: 10000). ``--json`` Emit results as JSON. """ from __future__ import annotations import argparse import collections.abc import json import logging import pathlib import sys from muse._version import __version__ from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( CommitRecord, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, walk_commits_between, ) 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 _HistoryDict = dict[str, str | int | None] type _SymbolDict = dict[str, str | int] type _AddrRecordMap = dict[str, tuple[SymbolRecord, str]] type _HistoryIndex = dict[str, "_SymbolHistory"] logger = logging.getLogger(__name__) _VALID_SORT_FIELDS = frozenset({"address", "commits", "changes", "first", "last"}) class _SymbolHistory: """Accumulated history of one symbol across a commit range.""" def __init__(self, address: str, kind: str, language: str) -> None: self.address = address self.kind = kind self.language = language self.first_commit_id: str = "" self.first_committed_at: str = "" self.last_commit_id: str = "" self.last_committed_at: str = "" self.commit_count: int = 0 self.content_ids: set[str] = set() @property def change_count(self) -> int: """Number of distinct content_ids seen — 1 means body never changed.""" return len(self.content_ids) def record(self, commit_id: str, committed_at: str, content_id: str) -> None: """Record the symbol's presence in one commit.""" if not self.first_commit_id: self.first_commit_id = commit_id self.first_committed_at = committed_at self.last_commit_id = commit_id self.last_committed_at = committed_at self.commit_count += 1 self.content_ids.add(content_id) def to_dict(self) -> _HistoryDict: return { "address": self.address, "kind": self.kind, "language": self.language, "commit_count": self.commit_count, "change_count": self.change_count, "first_commit_id": self.first_commit_id, "first_commit_id_short": self.first_commit_id[:8], "first_committed_at": self.first_committed_at[:10], "last_commit_id": self.last_commit_id, "last_commit_id_short": self.last_commit_id[:8], "last_committed_at": self.last_committed_at[:10], "stable": self.change_count == 1, } class _RemovedSymbol: """A symbol present in the --from snapshot but absent in --to.""" def __init__(self, address: str, rec: SymbolRecord, language: str) -> None: self.address = address self.rec = rec self.language = language def to_dict(self) -> _SymbolDict: return { "address": self.address, "kind": self.rec["kind"], "language": self.language, "status": "removed", } class _IntroducedSymbol: """A symbol present in the --to snapshot but absent in --from.""" def __init__(self, address: str, rec: SymbolRecord, language: str) -> None: self.address = address self.rec = rec self.language = language def to_dict(self) -> _SymbolDict: return { "address": self.address, "kind": self.rec["kind"], "language": self.language, "status": "introduced", } def _collect_addresses( root: pathlib.Path, commit_id: str, predicate: Predicate, cache: SymbolCache | None = None, ) -> _AddrRecordMap: """Return {address: (rec, language)} for all symbols in *commit_id* matching *predicate*.""" manifest = get_commit_snapshot_manifest(root, commit_id) or {} sym_map = symbols_for_snapshot(root, manifest, cache=cache) result: _AddrRecordMap = {} for file_path, tree in sym_map.items(): lang = language_of(file_path) for addr, rec in tree.items(): if predicate(file_path, rec): result[addr] = (rec, lang) return result def _sort_key_fn( sort_by: str, ) -> collections.abc.Callable[[_SymbolHistory], tuple[str | int, ...]]: """Return a sort key for ``_SymbolHistory`` objects.""" if sort_by == "commits": return lambda h: (-h.commit_count, h.address) if sort_by == "changes": return lambda h: (-h.change_count, h.address) if sort_by == "first": return lambda h: (h.first_committed_at, h.address) if sort_by == "last": return lambda h: (h.last_committed_at, h.address) return lambda h: (h.address,) def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the query-history subcommand.""" parser = subparsers.add_parser( "query-history", help="Search commit history for symbols matching a predicate expression.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "predicates", nargs="+", metavar="PREDICATE", help='One or more predicates, e.g. "kind=function" "language=Python".', ) parser.add_argument( "--from", dest="from_ref", default=None, metavar="REF", help="Start of range (exclusive; default: initial commit).", ) parser.add_argument( "--to", dest="to_ref", default=None, metavar="REF", help="End of range (inclusive; default: HEAD).", ) parser.add_argument( "--changed-only", action="store_true", help="Show only symbols with more than one distinct implementation (change_count > 1).", ) parser.add_argument( "--introduced-only", action="store_true", help=( "Show only symbols present in --to but absent in --from" " (net-new symbols)." ), ) parser.add_argument( "--removed-only", action="store_true", help=( "Show only symbols present in --from but absent in --to" " (deleted symbols)." ), ) parser.add_argument( "--sort", default="address", metavar="FIELD", choices=sorted(_VALID_SORT_FIELDS), help=( f"Sort results by field:" f" {', '.join(sorted(_VALID_SORT_FIELDS))} (default: address)." ), ) parser.add_argument( "--min-changes", type=int, default=1, metavar="N", help="Only show symbols with at least N distinct versions (default: 1).", ) parser.add_argument( "--count", action="store_true", help="Emit only the count of matching symbols.", ) parser.add_argument( "--limit", type=int, default=0, metavar="N", help="Cap the number of results returned (0 = unlimited).", ) parser.add_argument( "--max-commits", type=int, default=10_000, metavar="N", help="Cap the number of commits walked (default: 10000).", ) 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: """Search commit history for symbols matching a predicate expression. Walks the commit range ``(--from, --to]`` oldest-first, accumulating presence data for every matching symbol. The predicate grammar is identical to ``muse code query`` — supports OR, NOT, parentheses, and all field keys including ``size_gt`` / ``size_lt``. Modes:: (default) All symbols seen anywhere in the range. --changed-only Symbols whose body changed at least once. --introduced-only Symbols present in --to but not in --from (born). --removed-only Symbols present in --from but not in --to (deleted). Examples:: muse code query-history "kind=function" "language=Python" muse code query-history "name~=validate" --from v1.0 --to HEAD muse code query-history "kind=class" --json muse code query-history "kind=function" --changed-only --sort changes muse code query-history "kind=function" --removed-only --from v1.0 --to v2.0 """ predicates: list[str] = args.predicates from_ref: str | None = args.from_ref to_ref: str | None = args.to_ref as_json: bool = args.as_json changed_only: bool = args.changed_only introduced_only: bool = args.introduced_only removed_only: bool = args.removed_only sort_by: str = args.sort min_changes: int = clamp_int(args.min_changes, 0, 100000, 'min_changes') count_only: bool = args.count limit: int = clamp_int(args.limit, 0, 10000, 'limit') max_commits: int = clamp_int(args.max_commits, 1, 100000, 'max_commits') # Validate mutually exclusive mode flags. mode_flags = sum([changed_only, introduced_only, removed_only]) if mode_flags > 1: print( "❌ --changed-only, --introduced-only, and --removed-only" " 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 min_changes < 1: print("❌ --min-changes must be >= 1.", 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) 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) try: predicate = parse_query(predicates) except PredicateError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Resolve range endpoints. to_commit = resolve_commit_ref(root, repo_id, branch, to_ref) if to_commit is None: print(f"❌ --to ref '{to_ref or 'HEAD'}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from_commit_id: str | None = None if from_ref is not None: from_c = resolve_commit_ref(root, repo_id, branch, from_ref) if from_c is None: print(f"❌ --from ref '{from_ref}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from_commit_id = from_c.commit_id # Load the symbol cache once — shared across all snapshot operations in this # run. On a warm cache this eliminates O(n × 170ms) disk I/O in favour of # a single load + single save, regardless of how many snapshots are walked. shared_cache: SymbolCache = load_symbol_cache(root) # ── --introduced-only / --removed-only: snapshot diff mode ─────────────── if introduced_only or removed_only: to_addrs = _collect_addresses( root, to_commit.commit_id, predicate, cache=shared_cache ) # For --from: use from_commit_id if provided, else fall back to the # oldest commit reachable from to_commit. if from_commit_id is not None: from_addrs = _collect_addresses( root, from_commit_id, predicate, cache=shared_cache ) else: # No --from: treat the very first commit as the baseline. all_in_range: list[CommitRecord] = sorted( walk_commits_between(root, to_commit.commit_id, None), key=lambda c: c.committed_at, ) if all_in_range: from_addrs = _collect_addresses( root, all_in_range[0].commit_id, predicate, cache=shared_cache ) else: from_addrs = {} # All snapshot loading for diff mode is complete — persist any new entries. shared_cache.save() if introduced_only: introduced_addrs = set(to_addrs) - set(from_addrs) intro_symbols: list[_IntroducedSymbol] = sorted( [ _IntroducedSymbol(addr, rec, lang) for addr, (rec, lang) in to_addrs.items() if addr in introduced_addrs ], key=lambda s: s.address, ) if limit > 0: intro_symbols = intro_symbols[:limit] if count_only: print(len(intro_symbols)) return if as_json: print( json.dumps( { "schema_version": __version__, "mode": "introduced-only", "to_commit": to_commit.commit_id[:8], "from_commit": from_commit_id[:8] if from_commit_id else None, "symbols_found": len(intro_symbols), "results": [s.to_dict() for s in intro_symbols], }, indent=2, ) ) return pred_display = " AND ".join(predicates) from_label = from_commit_id[:8] if from_commit_id else "initial" print( f"\n Introduced symbols — {pred_display}" f" ({from_label}..{to_commit.commit_id[:8]})\n" ) if not intro_symbols: print(" (no new symbols found)") return for sym in intro_symbols: print(f" + {sanitize_display(sym.address)} [{sym.rec['kind']}]") print(f"\n {len(intro_symbols)} symbol(s) introduced") return # removed_only removed_addrs = set(from_addrs) - set(to_addrs) _removed_list: list[_RemovedSymbol] = [ _RemovedSymbol(addr, rec, lang) for addr, (rec, lang) in from_addrs.items() if addr in removed_addrs ] _removed_list.sort(key=lambda s: s.address) removed_symbols: list[_RemovedSymbol] = ( _removed_list[:limit] if limit > 0 else _removed_list ) if count_only: print(len(removed_symbols)) return if as_json: print( json.dumps( { "schema_version": __version__, "mode": "removed-only", "to_commit": to_commit.commit_id[:8], "from_commit": from_commit_id[:8] if from_commit_id else None, "symbols_found": len(removed_symbols), "results": [s.to_dict() for s in removed_symbols], }, indent=2, ) ) return pred_display = " AND ".join(predicates) from_label = from_commit_id[:8] if from_commit_id else "initial" print( f"\n Removed symbols — {pred_display}" f" ({from_label}..{to_commit.commit_id[:8]})\n" ) if not removed_symbols: print(" (no symbols removed in this range)") return for rem in removed_symbols: print(f" - {sanitize_display(rem.address)} [{rem.rec['kind']}]") print(f"\n {len(removed_symbols)} symbol(s) removed") return # ── Default / --changed-only: walk commit range ─────────────────────────── raw_commits: list[CommitRecord] = sorted( walk_commits_between(root, to_commit.commit_id, from_commit_id), key=lambda c: c.committed_at, ) truncated = len(raw_commits) > max_commits commits = raw_commits[:max_commits] # Accumulate per-symbol history — deduplicate on snapshot_id. history: _HistoryIndex = {} seen_snapshots: set[str] = set() for commit in commits: if commit.snapshot_id in seen_snapshots: continue seen_snapshots.add(commit.snapshot_id) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} sym_map = symbols_for_snapshot(root, manifest, cache=shared_cache) for file_path, tree in sym_map.items(): for addr, rec in tree.items(): if not predicate(file_path, rec): continue if addr not in history: history[addr] = _SymbolHistory( address=addr, kind=rec["kind"], language=language_of(file_path), ) history[addr].record( commit.commit_id, commit.committed_at.isoformat(), rec["content_id"], ) # All snapshot loading for walk mode is complete — persist any new entries. shared_cache.save() results: list[_SymbolHistory] = list(history.values()) # Apply filters. if changed_only: results = [r for r in results if r.change_count > 1] if min_changes > 1: results = [r for r in results if r.change_count >= min_changes] # Sort. results.sort(key=_sort_key_fn(sort_by)) # Apply limit. limited = limit > 0 and len(results) > limit if limited: results = results[:limit] # Count-only output. if count_only: print(len(results)) return # JSON output. if as_json: print( json.dumps( { "schema_version": __version__, "mode": "changed-only" if changed_only else "default", "to_commit": to_commit.commit_id[:8], "from_commit": from_commit_id[:8] if from_commit_id else None, "commits_scanned": len(commits), "truncated": truncated, "symbols_found": len(results), "results": [r.to_dict() for r in results], }, indent=2, ) ) return # Human-readable output. pred_display = " AND ".join(predicates) trunc_note = f" ⚠️ capped at {max_commits}" if truncated else "" print( f"\nSymbol history — {pred_display}" f" ({len(commits)} commit(s) scanned{trunc_note})" ) print("─" * 62) if not results: print(" (no matching symbols found in range)") return max_addr = max(len(r.address) for r in results) for r in results: versions = ( f"{r.change_count} version(s)" if r.change_count > 1 else "stable" ) span = f"{r.first_committed_at[:10]}..{r.last_committed_at[:10]}" print( f" {r.address:<{max_addr}} {r.kind:<14} " f"[{r.commit_count:>3} commit(s)] {versions:<14} {span}" ) if r.first_commit_id: print(f" └─ introduced: {r.first_commit_id[:8]}") if r.first_commit_id != r.last_commit_id: print(f" └─ last seen: {r.last_commit_id[:8]}") lim_note = f" (limited to {limit})" if limited else "" print(f"\n {len(results)} symbol(s) found{lim_note}")