"""``muse code code-query`` — predicate query over code commit history. Search the commit graph for code changes matching a structured predicate:: muse code code-query "symbol == 'my_function' and change == 'added'" muse code code-query "language == 'Python' and author == 'agent-x'" muse code code-query "agent_id == 'claude' and sem_ver_bump == 'major'" muse code code-query "file == 'src/core.py'" muse code code-query "change == 'removed' and kind == 'class'" muse code code-query "model_id contains 'claude'" muse code code-query "symbol endswith _handler" muse code code-query "author == 'gabriel'" --since 2026-01-01 muse code code-query "sem_ver_bump == 'major'" --count Fields ------ ``symbol`` Qualified symbol name (e.g. ``"MyClass.method"``). ``file`` Workspace-relative file path. ``language`` Language name (``"Python"``, ``"TypeScript"``…). ``kind`` Symbol kind (``"function"``, ``"class"``, ``"method"``…). ``change`` ``"added"``, ``"removed"``, or ``"modified"``. ``author`` Commit author string. ``agent_id`` Agent identity from commit provenance. ``model_id`` Model ID from commit provenance. ``toolchain_id`` Toolchain string from commit provenance. ``sem_ver_bump`` ``"none"``, ``"patch"``, ``"minor"``, or ``"major"``. ``branch`` Branch name. Operators: ``==``, ``!=``, ``contains``, ``startswith``, ``endswith`` Usage:: muse code code-query QUERY muse code code-query QUERY --branch dev --max 100 muse code code-query QUERY --since 2026-01-01 --until 2026-06-30 muse code code-query QUERY --limit 10 muse code code-query QUERY --count muse code code-query QUERY --json """ from __future__ import annotations import argparse import datetime import json import logging import pathlib import sys from muse.core.query_engine import format_matches, walk_history from muse.core.repo import parse_date_arg, require_repo from muse.core.store import read_current_branch from muse.plugins.code._code_query import build_evaluator from muse.core.validation import clamp_int logger = logging.getLogger(__name__) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the code-query subcommand.""" parser = subparsers.add_parser( "code-query", help="Query the code commit history using a structured predicate.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "query", help="Query expression (see muse code code-query --help).", ) parser.add_argument( "--branch", default=None, help="Branch to search (default: HEAD branch).", ) parser.add_argument( "--max", type=int, default=200, dest="max_commits", help="Maximum commits to inspect (walk depth). Default: 200.", ) parser.add_argument( "--limit", type=int, default=None, dest="limit", help="Maximum matches to display. Does not affect walk depth (see --max).", ) parser.add_argument( "--since", default=None, metavar="DATE", help="Only include commits on or after DATE (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS).", ) parser.add_argument( "--until", default=None, metavar="DATE", help="Only include commits on or before DATE (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS).", ) parser.add_argument( "--count", action="store_true", help="Print only the total match count, not the matches themselves.", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit JSON array of matches.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Query the code commit history using a structured predicate. Walks up to *max* commits from HEAD on the specified branch and returns all commits (and symbol-level changes) matching the predicate. Examples:: muse code code-query "symbol == 'parse_query' and change == 'added'" muse code code-query "agent_id contains 'claude' and sem_ver_bump == 'major'" muse code code-query "file == 'muse/core/store.py'" muse code code-query "author == 'gabriel'" --since 2026-01-01 muse code code-query "sem_ver_bump == 'major'" --count """ query: str = args.query branch: str | None = args.branch max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') limit: int | None = args.limit as_json: bool = args.as_json count_only: bool = args.count since: datetime.datetime | None = ( parse_date_arg(args.since, "--since") if args.since else None ) until: datetime.datetime | None = ( parse_date_arg(args.until, "--until") if args.until else None ) root: pathlib.Path = require_repo() try: evaluator = build_evaluator(query) except ValueError as exc: print(f"❌ Query parse error: {exc}", file=sys.stderr) raise SystemExit(1) from exc resolved_branch = branch or read_current_branch(root) # The code evaluator reads commit.structured_delta only — it never touches # the snapshot manifest. Skipping manifest I/O cuts one file-read per commit. matches = walk_history( root, resolved_branch, evaluator, max_commits=max_commits, load_manifest=False, since=since, until=until, ) if count_only: print(len(matches)) return if as_json: print(json.dumps({"total": len(matches), "results": matches})) return display_limit = limit if limit is not None else 50 print(format_matches(matches, max_results=display_limit))