"""Domain-agnostic commit-history query engine for Muse. Any domain can walk the commit graph, evaluate a predicate per commit, and collect structured matches — without reimplementing the graph-traversal loop. Three-tier walker contract -------------------------- Muse has three distinct commit-graph walkers, each serving a different tier: 1. **``walk_history``** (this module) — *porcelain tier*. Use for any query that produces :class:`QueryMatch` results consumed by users or agents. Supports ``follow_merges=True`` (BFS over the full DAG), ``since``/``until`` time filters, ``load_manifest=False`` optimisation, and the standard evaluator protocol. This is the right choice for all ``muse code`` porcelain commands. 2. **``store.walk_commits_between``** — *plumbing tier, range-bounded*. Use when you need a raw ``list[CommitRecord]`` for a specific linear range (``from_commit_id`` exclusive → ``to_commit_id`` inclusive, first-parent only). Used by ``find_symbol``, ``lineage``, ``query_history``, and ``status`` where the full match protocol is not needed. 3. **``_query.walk_commits_bfs``** — *low-level DAG primitive*. Available as a low-level escape hatch during the migration period. Prefer ``walk_history(follow_merges=True)`` for new code. Architecture ------------ :: muse/core/query_engine.py ← this file: generic history walker muse/plugins/midi/_midi_query.py ← MIDI predicate evaluator muse/plugins/code/_code_query.py ← code predicate evaluator muse/cli/commands/midi_query.py ← CLI for MIDI query muse/cli/commands/code_query.py ← CLI for code query Usage pattern:: from muse.core.query_engine import walk_history, QueryMatch def my_evaluator( commit: CommitRecord, manifest: Manifest, repo_root: pathlib.Path, ) -> list[QueryMatch]: matches = [] if "interesting-file.py" in manifest: matches.append(QueryMatch( commit_id=commit.commit_id, author=commit.author, committed_at=commit.committed_at.isoformat(), branch=commit.branch, detail="found interesting-file.py", extra={}, )) return matches results = walk_history(repo_root, branch="main", evaluator=my_evaluator) # DAG walk with follow_merges: results = walk_history( repo_root, branch="main", evaluator=my_evaluator, follow_merges=True ) Public API ---------- - :class:`QueryMatch` — one result row from the evaluator. - :class:`CommitEvaluator` — type alias for the evaluator callable. - :func:`walk_history` — traverse commits and collect matches. """ from __future__ import annotations import datetime import logging import pathlib from collections import deque from collections.abc import Callable from typing import TypedDict from muse.core._types import Manifest from muse.core.store import ( CommitRecord, get_commit_snapshot_manifest, get_head_commit_id, read_commit, ) logger = logging.getLogger(__name__) _DEFAULT_MAX_COMMITS = 500 # --------------------------------------------------------------------------- # Result type # --------------------------------------------------------------------------- class QueryMatch(TypedDict, total=False): """One match returned by a predicate evaluator. Required fields: ``commit_id`` The commit that produced this match. ``author`` Commit author string. ``committed_at`` ISO-8601 timestamp string. ``branch`` Branch name. ``detail`` Short human-readable description of what matched. Optional: ``extra`` Domain-specific data (e.g. ``{"symbol": "my_fn"}``). ``agent_id`` Agent identity from commit provenance (if present). ``model_id`` Model ID from commit provenance (if present). """ commit_id: str author: str committed_at: str branch: str detail: str extra: Manifest agent_id: str model_id: str # --------------------------------------------------------------------------- # Evaluator type alias # --------------------------------------------------------------------------- #: Signature every domain evaluator must satisfy. #: Returns a (possibly empty) list of :class:`QueryMatch` for the commit. CommitEvaluator = Callable[ [CommitRecord, dict[str, str], pathlib.Path], list[QueryMatch], ] # --------------------------------------------------------------------------- # Core history walker # --------------------------------------------------------------------------- def walk_history( repo_root: pathlib.Path, branch: str, evaluator: CommitEvaluator, *, max_commits: int = _DEFAULT_MAX_COMMITS, head_commit_id: str | None = None, load_manifest: bool = True, since: datetime.datetime | None = None, until: datetime.datetime | None = None, follow_merges: bool = False, ) -> list[QueryMatch]: """Walk the commit graph from HEAD and collect matches from *evaluator*. For each commit the evaluator receives the :class:`~muse.core.store.CommitRecord`, the raw file manifest (path → SHA-256 hash), and the repository root. It returns a list of :class:`QueryMatch` dicts (empty list if the commit has no matches). When *follow_merges* is ``False`` (default), only the main parent chain (``parent_commit_id``) is followed — sufficient for single-branch queries and avoids loading the full DAG. When *follow_merges* is ``True``, BFS is used so that merge commits' second parents (``parent2_commit_id``) are also visited. This is necessary for commands like ``hotspots`` and ``blame`` where missing feature-branch commits would give incorrect results. Args: repo_root: Repository root containing ``.muse/``. branch: Branch to start from (used to resolve HEAD when *head_commit_id* is ``None``). evaluator: Domain-specific callable — see :data:`CommitEvaluator`. max_commits: Maximum commits to inspect. Default 500. head_commit_id: Override the starting commit. ``None`` → resolve HEAD via the store (same logic as all other commands). load_manifest: When ``False``, passes an empty manifest ``{}`` to the evaluator and skips the snapshot-manifest I/O entirely. Safe for evaluators that only inspect commit-level fields (``author``, ``agent_id``, ``sem_ver_bump``, etc.). Provides a significant speed-up on large repos. since: Skip commits older than this timestamp (inclusive). until: Skip commits newer than this timestamp (inclusive). follow_merges: When ``True``, follow ``parent2_commit_id`` at merge commits (BFS). When ``False`` (default), follow only the linear ``parent_commit_id`` chain. Returns: All :class:`QueryMatch` records collected, ordered by walk order (newest-first for the main parent chain). """ if head_commit_id is None: resolved = get_head_commit_id(repo_root, branch) if not resolved: logger.warning("Branch '%s' has no commits.", branch) return [] head_commit_id = resolved results: list[QueryMatch] = [] if follow_merges: results = _walk_history_bfs( repo_root, head_commit_id, evaluator, max_commits=max_commits, load_manifest=load_manifest, since=since, until=until, ) else: results = _walk_history_linear( repo_root, head_commit_id, evaluator, max_commits=max_commits, load_manifest=load_manifest, since=since, until=until, ) return results def _make_tz_aware(dt: datetime.datetime) -> datetime.datetime: """Return *dt* with UTC timezone attached if it is naive.""" return dt if dt.tzinfo else dt.replace(tzinfo=datetime.timezone.utc) def _passes_time_filter( commit: CommitRecord, since: datetime.datetime | None, until: datetime.datetime | None, ) -> bool: """Return True if *commit* falls within the [since, until] window.""" ts = _make_tz_aware(commit.committed_at) if since is not None and ts < _make_tz_aware(since): return False if until is not None and ts > _make_tz_aware(until): return False return True def _evaluate_commit( repo_root: pathlib.Path, commit: CommitRecord, evaluator: CommitEvaluator, load_manifest: bool, ) -> list[QueryMatch]: """Load manifest if needed and run *evaluator* on *commit*.""" if load_manifest: manifest_rec = get_commit_snapshot_manifest(repo_root, commit.commit_id) manifest: Manifest = dict(manifest_rec) if manifest_rec else {} else: manifest = {} try: return evaluator(commit, manifest, repo_root) except Exception: logger.exception("Evaluator error on commit %s", commit.commit_id) return [] def _walk_history_linear( repo_root: pathlib.Path, start_commit_id: str, evaluator: CommitEvaluator, *, max_commits: int, load_manifest: bool, since: datetime.datetime | None, until: datetime.datetime | None, ) -> list[QueryMatch]: """Linear first-parent walk — internal implementation.""" results: list[QueryMatch] = [] current_id: str | None = start_commit_id seen = 0 while current_id and seen < max_commits: commit = read_commit(repo_root, current_id) if commit is None: break seen += 1 if _passes_time_filter(commit, since, until): results.extend(_evaluate_commit(repo_root, commit, evaluator, load_manifest)) current_id = commit.parent_commit_id return results def _walk_history_bfs( repo_root: pathlib.Path, start_commit_id: str, evaluator: CommitEvaluator, *, max_commits: int, load_manifest: bool, since: datetime.datetime | None, until: datetime.datetime | None, ) -> list[QueryMatch]: """BFS DAG walk following both parents — internal implementation.""" results: list[QueryMatch] = [] queue: deque[str] = deque([start_commit_id]) seen_ids: set[str] = set() visited = 0 while queue and visited < max_commits: current_id = queue.popleft() if current_id in seen_ids: continue seen_ids.add(current_id) commit = read_commit(repo_root, current_id) if commit is None: continue visited += 1 if _passes_time_filter(commit, since, until): results.extend(_evaluate_commit(repo_root, commit, evaluator, load_manifest)) if commit.parent_commit_id: queue.append(commit.parent_commit_id) if commit.parent2_commit_id: queue.append(commit.parent2_commit_id) return results def format_matches(matches: list[QueryMatch], *, max_results: int = 50) -> str: """Format a list of matches as a human-readable table. Args: matches: The results from :func:`walk_history`. max_results: Maximum rows to show before the "… N more" truncation line. Returns: Multi-line string ready for printing to stdout. """ if not matches: return "No matches found." lines: list[str] = [f"Found {len(matches)} match(es):\n"] for m in matches[:max_results]: cid = m.get("commit_id", "?")[:8] author = m.get("author", "unknown") ts = m.get("committed_at", "")[:10] detail = m.get("detail", "") agent = m.get("agent_id", "") agent_str = f" [{agent}]" if agent else "" lines.append(f" {cid} {ts} {author}{agent_str} — {detail}") if len(matches) > max_results: lines.append(f"\n … {len(matches) - max_results} more (use --limit to raise the cap)") return "\n".join(lines)