"""muse code blame -- symbol-level attribution. ``git blame`` attributes every *line* to a commit -- a 300-line class gives you 300 attribution entries. ``muse code blame`` attributes the *symbol* as a semantic unit: one answer per function, class, or method, regardless of how many lines it occupies. Rename tracking --------------- ``muse code blame`` follows renames automatically, in both directions: * Blaming the **current** name (post-rename) walks backward through the rename event, then continues tracking the symbol's earlier history under its old name. * Blaming the **original** name (pre-rename) finds both the creation event and the rename event, then follows the symbol forward under its new name. Early-exit optimisation ----------------------- The scan stops as soon as a ``"created"`` event is found for the symbol. At that point the full lineage is known; continuing would yield no new events. For long-lived symbols this can reduce scan work dramatically. Security model -------------- - The ``ADDRESS`` argument is validated to reject null bytes and ANSI/control characters before any processing occurs. - All user-controlled values (address, commit references, author strings, commit messages, change details) are sanitized via ``sanitize_display()`` before appearing in human-readable output. - The ``from_ref`` argument is sanitized in error messages. - JSON output carries raw stored values (no terminal sanitization applied) — agents receive the exact data as committed. - All error messages go to **stderr**; **stdout** carries only data. Agent UX -------- Pass ``--json`` for a stable machine-readable object. All fields are always present. Filter by ``--kind`` and/or ``--author`` to narrow output further. Usage:: muse code blame "src/billing.py::compute_invoice_total" muse code blame "api/server.go::Server.HandleRequest" muse code blame "src/models.py::User.save" --all muse code blame "src/billing.py::run" --from feat/my-branch muse code blame "src/billing.py::run" --json muse code blame "src/billing.py::run" --kind created --kind renamed muse code blame "src/billing.py::run" --author alice JSON schema (``--json``):: { "address": "src/billing.py::compute_invoice_total", "start_ref": "main", "total_commits_scanned": 42, "truncated": false, "events": [ { "event": "created", "commit_id": "<64-char hex>", "author": "alice", "message": "initial implementation", "committed_at": "2026-01-15T12:00:00+00:00", "address": "src/billing.py::compute_invoice_total", "detail": "created", "new_address": null } ] } Exit codes ---------- - 0 — success - 1 — invalid arguments (bad address, bad --kind, bad ref) - 2 — not inside a Muse repository - 4 — commit reference not found """ from __future__ import annotations import argparse import json import logging import sys from dataclasses import dataclass from typing import TYPE_CHECKING, Literal, TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import CommitRecord, read_current_branch, resolve_commit_ref from muse.domain import DomainOp from muse.plugins.code._query import walk_commits_bfs from muse.core.validation import clamp_int, sanitize_display, sanitize_provenance if TYPE_CHECKING: import pathlib logger = logging.getLogger(__name__) _DEFAULT_MAX = 500 SymbolEventKind = Literal[ "created", "modified", "renamed", "moved", "deleted", "signature" ] _ALL_KINDS: frozenset[str] = frozenset( ("created", "modified", "renamed", "moved", "deleted", "signature") ) # --------------------------------------------------------------------------- # JSON TypedDicts — stable machine-readable output schemas # --------------------------------------------------------------------------- class _BlameEventJson(TypedDict): """One event in the blame history of a symbol.""" event: SymbolEventKind commit_id: str author: str message: str committed_at: str address: str detail: str new_address: str | None class _BlameResultJson(TypedDict): """Top-level JSON output of ``muse code blame --json``.""" address: str start_ref: str total_commits_scanned: int truncated: bool events: list[_BlameEventJson] # --------------------------------------------------------------------------- # Internal dataclass # --------------------------------------------------------------------------- @dataclass class _BlameEvent: """One attributed change event for a symbol across the commit history.""" kind: SymbolEventKind commit: CommitRecord address: str detail: str new_address: str | None = None def to_dict(self) -> _BlameEventJson: """Serialise to the stable ``_BlameEventJson`` schema.""" return _BlameEventJson( event=self.kind, commit_id=self.commit.commit_id, author=self.commit.author, message=self.commit.message, committed_at=self.commit.committed_at.isoformat(), address=self.address, detail=self.detail, new_address=self.new_address, ) # --------------------------------------------------------------------------- # Core event-extraction logic # --------------------------------------------------------------------------- def _flat_ops(ops: list[DomainOp]) -> list[DomainOp]: """Flatten patch ops so that every leaf ``DomainOp`` is at the top level.""" result: list[DomainOp] = [] for op in ops: if op["op"] == "patch": result.extend(op["child_ops"]) else: result.append(op) return result def _events_in_commit( commit: CommitRecord, address: str, file_prefix: str, bare_name: str, ) -> tuple[list[_BlameEvent], str]: """Scan *commit* for events touching *address*. Returns ``(events, next_address)`` where ``next_address`` is the symbol name to search for in older commits (changes only when a rename is detected while walking newest-first). Args: commit: The commit record to scan. address: The full symbol address currently being tracked (e.g. ``"src/billing.py::compute_invoice_total"``). file_prefix: The file part of *address* (``"src/billing.py"``). Pre-split by the caller to avoid redundant splits. bare_name: The symbol name part of *address* (``"compute_invoice_total"``). Pre-split by the caller. Rename semantics (walking newest-first) ---------------------------------------- Renames are stored as ``{op: "replace", address: OLD_NAME, new_summary: "renamed to NEW_NAME"}``. * **Direct match** (``op.address == address``): we found an event for the symbol we are currently tracking. If it is a rename, older commits had the symbol under ``address`` (the old name) — ``next_address`` is unchanged. * **Reverse rename** (``op.address == some_old_name``, ``new_summary == "renamed to "``): the symbol we are tracking was previously named ``some_old_name``. Switch ``next_address`` to ``some_old_name`` so we pick up its earlier history. """ events: list[_BlameEvent] = [] next_address = address if commit.structured_delta is None: return events, next_address for op in _flat_ops(commit.structured_delta["ops"]): op_address: str = op["address"] if op_address == address: # ── Direct match ───────────────────────────────────────────────── if op["op"] == "insert": events.append(_BlameEvent( "created", commit, address, op.get("content_summary", "created"), )) elif op["op"] == "delete": detail: str = op.get("content_summary", "deleted") kind: SymbolEventKind = "moved" if "moved to" in detail else "deleted" events.append(_BlameEvent(kind, commit, address, detail)) elif op["op"] == "replace": ns: str = op.get("new_summary", "") if ns.startswith("renamed to "): new_name = ns.removeprefix("renamed to ").strip() new_addr = f"{file_prefix}::{new_name}" events.append(_BlameEvent( "renamed", commit, address, f"renamed to {new_name}", new_addr, )) # Walking backward: older commits had the symbol as # *address* (the old name). Do NOT update next_address. elif ns.startswith("moved to "): events.append(_BlameEvent("moved", commit, address, ns)) elif "signature" in ns: events.append(_BlameEvent( "signature", commit, address, ns or "signature changed", )) else: events.append(_BlameEvent( "modified", commit, address, ns or "modified", )) elif op["op"] == "replace": # ── Reverse rename detection ────────────────────────────────────── # Was some other symbol (op_address) renamed TO the symbol we are # currently tracking (address)? e.g.: # op = {address: "billing.py::compute_total", # new_summary: "renamed to compute_invoice_total"} # and address = "billing.py::compute_invoice_total" ns_other: str = op.get("new_summary", "") if ns_other.startswith("renamed to ") and "::" in op_address: renamed_to = ns_other.removeprefix("renamed to ").strip() op_file = op_address.rsplit("::", 1)[0] if op_file == file_prefix and renamed_to == bare_name: # Found: op_address (old name) was renamed to address. old_name = op_address.rsplit("::", 1)[-1] events.append(_BlameEvent( "renamed", commit, op_address, f"renamed from {old_name} to {bare_name}", address, )) # Walking backward: switch to the old name. next_address = op_address return events, next_address # --------------------------------------------------------------------------- # Argument parser registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``blame`` subcommand. All error messages are routed to stderr. ``--json`` emits a stable ``_BlameResultJson`` object to stdout. """ parser = subparsers.add_parser( "blame", help="Show which commit last touched a specific symbol.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "address", metavar="ADDRESS", help='Symbol address, e.g. "src/billing.py::compute_invoice_total".', ) parser.add_argument( "--from", default=None, metavar="REF", dest="from_ref", help="Start walking from this commit / branch (default: HEAD).", ) parser.add_argument( "--all", "-a", action="store_true", dest="show_all", help="Show the full change history, not just the three most recent events.", ) parser.add_argument( "--max", default=_DEFAULT_MAX, type=int, metavar="N", dest="max_commits", help=f"Maximum commits to scan (default: {_DEFAULT_MAX}).", ) parser.add_argument( "--kind", action="append", dest="kinds", default=None, metavar="KIND", help=( "Filter output to events of this kind. Accepted values: " "created, modified, renamed, moved, deleted, signature. " "Repeat to include multiple kinds." ), ) parser.add_argument( "--author", default=None, metavar="PATTERN", dest="author_filter", help=( "Filter output to events from commits whose author contains " "PATTERN (case-insensitive substring match)." ), ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit attribution as structured JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Show which commit last touched a specific symbol. ``muse code blame`` attributes the symbol as a semantic unit — one answer per function, class, or method, regardless of line count. Renames are tracked automatically. Blaming a symbol that was renamed from an earlier name will follow the rename backward and continue tracking the symbol's history under the old name. Uses a BFS walk that follows both parents of merge commits, so events on merged feature branches are never missed. Scanning stops as soon as a ``"created"`` event is found — at that point the full lineage is established and further commits cannot add new events. Patterns are searched against the full commit DAG up to ``--max`` commits. """ address: str = args.address from_ref: str | None = args.from_ref show_all: bool = args.show_all max_commits: int = clamp_int(args.max_commits, 1, 100_000, "max_commits") as_json: bool = args.as_json kinds_raw: list[str] | None = args.kinds author_filter: str | None = args.author_filter # ── Validation ──────────────────────────────────────────────────────────── # Reject control characters and null bytes in the address before any use. sanitised_addr = sanitize_provenance(address) if sanitised_addr != address or "\x00" in address: print( "❌ ADDRESS contains control characters or null bytes.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if "::" not in address: print( f"❌ Invalid address {address!r} — expected 'file.py::SymbolName'.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Validate --kind values. kind_filter: frozenset[str] | None = None if kinds_raw: invalid = [k for k in kinds_raw if k not in _ALL_KINDS] if invalid: print( f"❌ Unknown --kind value(s): {', '.join(invalid)}. " f"Accepted: {', '.join(sorted(_ALL_KINDS))}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) kind_filter = frozenset(kinds_raw) # ── Repo / commit resolution ────────────────────────────────────────────── root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) start_commit = resolve_commit_ref(root, repo_id, branch, from_ref) if start_commit is None: ref_display = sanitize_display(from_ref or "HEAD") print(f"❌ Commit {ref_display!r} not found.", file=sys.stderr) raise SystemExit(ExitCode.NOT_FOUND) # ── BFS walk ────────────────────────────────────────────────────────────── commits, truncated = walk_commits_bfs(root, start_commit.commit_id, max_commits) # ── Event collection (rename-aware, early-exit on "created") ───────────── current_address = address all_events: list[_BlameEvent] = [] for commit in commits: # Pre-split current_address once per iteration (address may change on # rename detection — splitting here avoids double-split inside the # inner function). if "::" not in current_address: break cur_file, cur_bare = current_address.rsplit("::", 1) evs, current_address = _events_in_commit( commit, current_address, cur_file, cur_bare ) all_events.extend(evs) # Early exit: once the symbol's creation is found, full lineage is # established — no older commit can contribute new events. if any(ev.kind == "created" for ev in evs): break # ── Apply filters ───────────────────────────────────────────────────────── filtered: list[_BlameEvent] = all_events if kind_filter is not None: filtered = [ev for ev in filtered if ev.kind in kind_filter] if author_filter is not None: needle = author_filter.lower() filtered = [ ev for ev in filtered if needle in (ev.commit.author or "").lower() ] # ── Output ──────────────────────────────────────────────────────────────── start_ref = sanitize_display(from_ref or branch) if as_json: result = _BlameResultJson( address=address, start_ref=from_ref or branch, total_commits_scanned=len(commits), truncated=truncated, events=[e.to_dict() for e in reversed(filtered)], ) print(json.dumps(result, indent=2)) return print(f"\n{sanitize_display(address)}") print("─" * 62) if truncated: print( f" ⚠️ History may be incomplete — scanned {len(commits)} commits " f"(--max {max_commits}).", ) if not filtered: if all_events: print(" (no events match the active filters)") else: print( " (no events found — symbol may not exist or have no recorded history)" ) return events_to_show = filtered if show_all else filtered[:3] _LABELS = ("last touched:", "previous: ", "before that: ") for idx, ev in enumerate(events_to_show): if idx < len(_LABELS): label = _LABELS[idx] else: label = f"#{idx + 1}:".ljust(13) date_str = ev.commit.committed_at.strftime("%Y-%m-%d") short_id = sanitize_display(ev.commit.commit_id[:8]) print(f"{label} {short_id} {date_str}") print(f" author: {sanitize_display(ev.commit.author or 'unknown')}") print(f' message: "{sanitize_display(ev.commit.message)}"') print(f" change: {sanitize_display(ev.detail)}") if ev.new_address: print(f" → tracking continues as {sanitize_display(ev.new_address)}") print("") if not show_all and len(filtered) > 3: remaining = len(filtered) - 3 print( f" … {remaining} older event(s) — pass --all to see the full history." ) _ = start_ref # used for JSON; keep linter happy