"""muse code lineage — full symbol provenance chain. Traces the complete life of a symbol through the commit history: created → renamed → moved → copied → modified → deleted, in chronological order. Each transition is classified by comparing hashes across consecutive commits: * **created** — first InsertOp for this address (no prior body_hash match) * **copied_from** — InsertOp whose content_id matches a living symbol at a different address (same body, new address) * **renamed_from** — InsertOp + DeleteOp in same commit with matching content_id (content preserved, address changed within same file) * **moved_from** — InsertOp + DeleteOp in same commit with matching content_id AND different file (cross-file move) * **modified** — ReplaceOp at this address; sub-classified by summary heuristic: ``impl_only``, ``signature_change``, or ``full_rewrite`` * **deleted** — DeleteOp at this address Usage:: muse code lineage "src/billing.py::compute_invoice_total" muse code lineage "src/auth.py::validate_token" --commit HEAD~5 muse code lineage "src/core.py::hash_content" --json muse code lineage "src/billing.py::process_order" --branch feat/payments muse code lineage "src/billing.py::process_order" --since 2025-01-01 muse code lineage "src/billing.py::process_order" --filter modified muse code lineage "src/billing.py::process_order" --stability muse code lineage "src/billing.py::process_order" --count Output:: Lineage: src/billing.py::compute_invoice_total ────────────────────────────────────────────────────────────── 2026-02-01 a1b2c3d4 created "Initial billing module" 2026-02-10 e5f6a7b8 modified (impl_only) "Fix rounding" 2026-02-15 c9d0e1f2 renamed_from "Rename compute_total" └─ src/billing.py::_compute_total 2026-03-10 f7a8b9c0 modified (full_rewrite) "Overhaul billing" 4 events — first seen 2026-02-01 · last seen 2026-03-10 Stability: 50% (2 modification(s) in 4 events) Flags: ``--commit, -c REF`` Walk history starting from this commit (inclusive) instead of HEAD. Only commits reachable from REF are examined. ``--branch BRANCH`` Walk only this branch's linear history instead of all object-store commits. ``--since DATE`` Ignore commits before DATE (YYYY-MM-DD). ``--until DATE`` Ignore commits after DATE (YYYY-MM-DD). ``--filter KIND`` Show only events of this kind: created, modified, deleted, renamed_from, moved_from, copied_from. ``--stability`` Print a stability score: ratio of unmodified commits to total events. ``--count`` Print only the number of events (scriptable). ``--json`` Emit the full provenance chain as JSON. """ from __future__ import annotations import argparse import datetime import json import logging import pathlib import sys from typing import Literal from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( CommitRecord, Metadata, get_all_commits, get_head_commit_id, read_current_branch, resolve_commit_ref, walk_commits_between, ) from muse.plugins.code._query import flat_symbol_ops from muse.core.validation import sanitize_display type _ContentLiveMap = dict[str, set[str]] type _InsertMap = dict[str, "_InsertFields"] type _DeleteMap = dict[str, "_DeleteFields"] type _ReplaceMap = dict[str, "_ReplaceFields"] logger = logging.getLogger(__name__) EventKind = Literal[ "created", "renamed_from", "moved_from", "copied_from", "modified", "deleted", ] _ALL_EVENT_KINDS: frozenset[str] = frozenset({ "created", "renamed_from", "moved_from", "copied_from", "modified", "deleted", }) # --------------------------------------------------------------------------- # Typed op wrappers # --------------------------------------------------------------------------- class _InsertFields: """Extracted fields from an InsertOp.""" __slots__ = ("address", "content_id") def __init__(self, address: str, content_id: str) -> None: self.address = address self.content_id = content_id class _DeleteFields: __slots__ = ("address", "content_id") def __init__(self, address: str, content_id: str) -> None: self.address = address self.content_id = content_id class _ReplaceFields: __slots__ = ("address", "old_content_id", "new_content_id", "old_summary", "new_summary") def __init__( self, address: str, old_content_id: str, new_content_id: str, old_summary: str, new_summary: str, ) -> None: self.address = address self.old_content_id = old_content_id self.new_content_id = new_content_id self.old_summary = old_summary self.new_summary = new_summary # --------------------------------------------------------------------------- # Event type # --------------------------------------------------------------------------- class _LineageEvent: """One classified provenance event for a symbol.""" __slots__ = ( "commit_id", "committed_at", "message", "kind", "detail", "old_content_id", "new_content_id", ) def __init__( self, commit_id: str, committed_at: str, message: str, kind: EventKind, detail: str = "", old_content_id: str = "", new_content_id: str = "", ) -> None: self.commit_id = commit_id self.committed_at = committed_at self.message = message self.kind = kind self.detail = detail self.old_content_id = old_content_id self.new_content_id = new_content_id def to_dict(self) -> Metadata: """Return a JSON-serialisable dict with full (untruncated) IDs.""" d: Metadata = { "commit_id": self.commit_id, "committed_at": self.committed_at, "message": self.message, "event": self.kind, } if self.detail: d["detail"] = self.detail if self.old_content_id: d["old_content_id"] = self.old_content_id if self.new_content_id: d["new_content_id"] = self.new_content_id return d # --------------------------------------------------------------------------- # Classification helpers # --------------------------------------------------------------------------- def _classify_replace(old_summary: str, new_summary: str) -> str: """Classify a ReplaceOp by examining summary strings for change markers.""" if "signature" in old_summary or "signature" in new_summary: return "signature_change" return "full_rewrite" def _stability(events: list[_LineageEvent]) -> tuple[int, int]: """Return ``(modified_count, total_events)``.""" modified = sum(1 for e in events if e.kind == "modified") return modified, len(events) # --------------------------------------------------------------------------- # Core analysis — accepts an explicit commit list (pure, testable) # --------------------------------------------------------------------------- def build_lineage( address: str, commits: list[CommitRecord], ) -> list[_LineageEvent]: """Walk *commits* oldest-first and build the provenance chain for *address*. The caller controls which commits to pass — enabling branch-restricted walks, date-bounded walks, and direct unit testing with synthetic ``CommitRecord`` objects. Copy detection uses an incremental ``content_id → set[address]`` registry updated from ``structured_delta`` ops as commits are processed. This is O(total ops across all commits) — no blob re-parsing, no snapshot scans. Args: address: Full symbol address, e.g. ``"src/billing.py::compute_invoice_total"``. commits: Commits to walk, oldest-first. Returns: List of :class:`_LineageEvent` objects in chronological order. """ events: list[_LineageEvent] = [] address_live = False # content_id → set of currently-live symbol addresses. # Updated incrementally; never cleared — gives O(1) copy detection. live_by_content_id: _ContentLiveMap = {} for commit in commits: if commit.structured_delta is None: continue ops = commit.structured_delta.get("ops", []) committed_at = commit.committed_at.isoformat() message = commit.message inserts: _InsertMap = {} deletes: _DeleteMap = {} replaces: _ReplaceMap = {} for op in flat_symbol_ops(ops): addr = op["address"] if op["op"] == "insert": inserts[addr] = _InsertFields( address=addr, content_id=op["content_id"], ) elif op["op"] == "delete": deletes[addr] = _DeleteFields( address=addr, content_id=op["content_id"], ) elif op["op"] == "replace": replaces[addr] = _ReplaceFields( address=addr, old_content_id=op["old_content_id"], new_content_id=op["new_content_id"], old_summary=op["old_summary"], new_summary=op["new_summary"], ) if address in replaces: rep = replaces[address] detail = _classify_replace(rep.old_summary, rep.new_summary) events.append(_LineageEvent( commit_id=commit.commit_id, committed_at=committed_at, message=message, kind="modified", detail=detail, old_content_id=rep.old_content_id, new_content_id=rep.new_content_id, )) live_by_content_id.get(rep.old_content_id, set()).discard(address) live_by_content_id.setdefault(rep.new_content_id, set()).add(address) if address in inserts: ins = inserts[address] ins_cid = ins.content_id # Rename / move: DeleteOp in same commit with matching content_id. source_addr: str | None = None for del_addr, del_op in deletes.items(): if del_addr != address and del_op.content_id == ins_cid: source_addr = del_addr break if source_addr is not None: del_file = source_addr.split("::")[0] ins_file = address.split("::")[0] ev_kind: EventKind = "moved_from" if del_file != ins_file else "renamed_from" events.append(_LineageEvent( commit_id=commit.commit_id, committed_at=committed_at, message=message, kind=ev_kind, detail=source_addr, new_content_id=ins_cid, )) else: # Copy detection: O(1) lookup in the incremental registry. existing = live_by_content_id.get(ins_cid, set()) - {address} if existing and not address_live: copy_source: str | None = next(iter(sorted(existing))) copy_kind: EventKind = "copied_from" else: copy_source = None copy_kind = "created" events.append(_LineageEvent( commit_id=commit.commit_id, committed_at=committed_at, message=message, kind=copy_kind, detail=copy_source or "", new_content_id=ins_cid, )) live_by_content_id.setdefault(ins_cid, set()).add(address) address_live = True if address in deletes: del_f = deletes[address] events.append(_LineageEvent( commit_id=commit.commit_id, committed_at=committed_at, message=message, kind="deleted", old_content_id=del_f.content_id, )) live_by_content_id.get(del_f.content_id, set()).discard(address) address_live = False # Update registry for all other ops so copy detection stays accurate. for addr, ins in inserts.items(): if addr != address: live_by_content_id.setdefault(ins.content_id, set()).add(addr) for addr, del_op in deletes.items(): if addr != address: live_by_content_id.get(del_op.content_id, set()).discard(addr) return events # --------------------------------------------------------------------------- # Commit collection helpers # --------------------------------------------------------------------------- def _gather_commits( root: pathlib.Path, repo_id: str, branch: str, ref: str | None, branch_filter: str | None, since: datetime.date | None, until: datetime.date | None, ) -> list[CommitRecord] | None: """Resolve the commit list to walk, oldest-first. Returns ``None`` if the requested ref or branch cannot be found. """ commits: list[CommitRecord] if branch_filter is not None: tip = get_head_commit_id(root, branch_filter) if tip is None: return None commits = list(reversed(walk_commits_between(root, tip))) elif ref is not None: commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: return None commits = list(reversed(walk_commits_between(root, commit.commit_id))) else: commits = sorted(get_all_commits(root), key=lambda c: c.committed_at) if since is not None: commits = [c for c in commits if c.committed_at.date() >= since] if until is not None: commits = [c for c in commits if c.committed_at.date() <= until] return commits # --------------------------------------------------------------------------- # CLI registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the lineage subcommand.""" parser = subparsers.add_parser( "lineage", help="Show the full provenance chain of a symbol through commit history.", 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( "--commit", "-c", dest="ref", default=None, metavar="REF", help="Walk only commits reachable from this ref (default: all commits).", ) parser.add_argument( "--branch", "-b", dest="branch_filter", default=None, metavar="BRANCH", help="Walk only this branch's linear history.", ) parser.add_argument( "--since", dest="since", default=None, metavar="DATE", help="Ignore commits before DATE (YYYY-MM-DD).", ) parser.add_argument( "--until", dest="until", default=None, metavar="DATE", help="Ignore commits after DATE (YYYY-MM-DD).", ) parser.add_argument( "--filter", dest="kind_filter", default=None, metavar="KIND", choices=sorted(_ALL_EVENT_KINDS), help="Show only events of this kind (created, modified, deleted, …).", ) parser.add_argument( "--stability", dest="show_stability", action="store_true", help="Print a stability score (ratio of modifications to total events).", ) parser.add_argument( "--count", dest="count_only", action="store_true", help="Print only the total number of events (scriptable).", ) parser.add_argument( "--json", dest="as_json", action="store_true", help="Emit results as JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Show the full provenance chain of a symbol through commit history.""" address: str = args.address ref: str | None = args.ref branch_filter: str | None = args.branch_filter kind_filter: str | None = args.kind_filter show_stability: bool = args.show_stability count_only: bool = args.count_only as_json: bool = args.as_json if "::" not in address: print( "❌ ADDRESS must contain '::' — e.g. 'src/billing.py::compute_invoice_total'.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) since: datetime.date | None = None until: datetime.date | None = None if args.since: try: since = datetime.date.fromisoformat(args.since) except ValueError: print(f"❌ --since: invalid date '{args.since}' (expected YYYY-MM-DD).", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if args.until: try: until = datetime.date.fromisoformat(args.until) except ValueError: print(f"❌ --until: invalid date '{args.until}' (expected YYYY-MM-DD).", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) commits = _gather_commits(root, repo_id, branch, ref, branch_filter, since, until) if commits is None: target = branch_filter or ref or "HEAD" print(f"❌ '{target}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) events = build_lineage(address, commits) if kind_filter: events = [e for e in events if e.kind == kind_filter] modified_count, total_count = _stability(events) if count_only and not as_json: print(len(events)) return if as_json: pct = round((total_count - modified_count) / total_count * 100) if total_count else 100 print(json.dumps( { "address": address, "total": len(events), "events": [e.to_dict() for e in events], "stability_pct": pct, "modified_count": modified_count, }, indent=2, )) return print(f"\nLineage: {sanitize_display(address)}") print("─" * 62) if not events: print( "\n (no events found — address may not exist in this repository's history," "\n or the structured_delta does not carry symbol-level ops)" ) return for ev in events: date = ev.committed_at[:10] short = ev.commit_id[:8] kind_label: str = ev.kind if ev.detail and ev.kind == "modified": kind_label = f"modified ({ev.detail})" msg_str = f' "{sanitize_display(ev.message)}"' if ev.message else "" print(f" {date} {short} {kind_label:<28}{msg_str}") if ev.detail and ev.kind in ("renamed_from", "moved_from", "copied_from"): print(f"{'':34}└─ {sanitize_display(ev.detail)}") print() first = events[0].committed_at[:10] last = events[-1].committed_at[:10] suffix = "" if first == last else f" · last seen {last}" print(f" {len(events)} event(s) — first seen {first}{suffix}") if show_stability and total_count: pct = round((total_count - modified_count) / total_count * 100) print(f" Stability: {pct}% ({modified_count} modification(s) in {total_count} events)")