"""muse code narrative — plain-English story of a symbol's life. Every other analysis command answers a *dimensional* question: ``age`` → how old? ``hotspots`` → how churned? ``blast-risk`` → how risky? ``semantic-test-coverage`` → how tested? ``narrative`` answers a *biographical* question: **"What happened to this symbol, and why?"** It reads every structured delta that ever touched the symbol, walks back through the commit graph, and composes a timeline or flowing prose story of the symbol's complete life — from the commit that created it to the last commit that changed it. This is only possible because Muse records *why* each change happened at the semantic level (body rewrite, signature change, rename) in addition to *what* changed. A line-diff VCS can tell you bytes changed; Muse tells you the story. Output modes ------------ ``--format timeline`` (default) Chronological event list with dates, commit fragments, and SemVer bumps. ``--format prose`` Flowing paragraph — suitable for proposal summaries, code reviews, or documentation. ``--show-source`` Append the symbol's source at birth (first version) and at HEAD (current version), if the source can be read from the object store. Usage:: muse code narrative "billing.py::Invoice.compute_total" muse code narrative "billing.py::Invoice.compute_total" --format prose muse code narrative "billing.py::Invoice.compute_total" --show-source muse code narrative "billing.py::Invoice.compute_total" --since v1.0 muse code narrative "billing.py::Invoice.compute_total" --max-commits 500 muse code narrative "billing.py::Invoice.compute_total" --json Timeline output:: Invoice.compute_total (method) · billing.py 🌱 Born Jan 12, 2026 · commit a3f2c9e1 · MINOR "feat: initial invoice model" Created as a method on Invoice, taking 2 parameters. ✏️ Signature Feb 3, 2026 · commit b7e3d012 · MINOR "feat: add optional currency parameter" 🔧 Body Feb 14, 2026 · commit c1a9f055 · PATCH "perf: vectorise invoice computation" 🏷️ Renamed Mar 1, 2026 · commit d4b2e733 · PATCH "refactor: clean up public API names" compute_invoice_total → compute_total 🔧 Body Mar 8, 2026 · commit e8c6a191 · MINOR "feat: add currency conversion logic" Largest change (+47 lines) ✏️ Signature Mar 20, 2026 · commit f2d9b447 · MAJOR ⚠️ breaking "breaking: currency parameter now required" ───────────────────────────────────────────────────────────────── Life summary Born: Jan 12, 2026 (97 days ago) Last change: Mar 20, 2026 (commit f2d9b447) Events: 3 body rewrites · 2 signature changes · 1 rename Survival: ~25% of original implementation remains Status: Alive Prose output (``--format prose``):: Invoice.compute_total was born on January 12th, 2026 as a method on the Invoice class (commit a3f2c9e1). Three weeks later its signature was extended with an optional currency parameter — a MINOR bump. On February 14th the body was rewritten for performance, replacing a manual loop with a sum() comprehension. On March 1st the function was renamed from compute_invoice_total to compute_total during an API cleanup. The most significant event came on March 8th: a full body rewrite to add currency conversion logic, the largest change in the symbol's history. Finally, on March 20th the currency parameter was made required — a MAJOR breaking change. At 97 days old, the method has survived 3 body rewrites and retains an estimated 25% of its original implementation. JSON output (``--json``):: { "address": "billing.py::Invoice.compute_total", "name": "compute_total", "kind": "method", "status": "alive", "born_date": "2026-01-12", "born_commit": "a3f2c9e1", "last_change_date": "2026-03-20", "last_change_commit": "f2d9b447", "calendar_age_days": 97, "genetic_age_days": 12, "impl_changes": 3, "sig_changes": 2, "renames": 1, "est_survival_pct": 25, "commits_analysed": 304, "truncated": false, "events": [ { "date": "2026-01-12", "commit_id": "a3f2c9e1", "commit_msg": "feat: initial invoice model", "event_type": "create", "sem_ver_bump": "minor", "detail": "Created as a method taking 2 parameters." } ] } Security note ------------- Symbol addresses are validated to contain ``::`` and both path components are stripped of leading/trailing whitespace before any file-system access. Commit messages are sanitised (control characters removed) before display so a malicious commit message cannot inject terminal escape sequences. """ import argparse import ast import datetime import json import logging import pathlib import re import sys from dataclasses import dataclass, field from typing import Literal, TypedDict from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.refs import read_current_branch from muse.core.commits import resolve_commit_ref from muse.core.snapshots import get_commit_snapshot_manifest from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.timing import start_timer from muse.domain import DomainOp from muse.plugins.code._callgraph import find_func_node from muse.plugins.code._query import flat_symbol_ops, symbols_for_snapshot, walk_commits_bfs from muse.core.validation import clamp_int, MAX_AST_BYTES, sanitize_display type _StrMap = dict[str, str] logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_MAX_COMMITS = 10_000 # Op summary keywords — same as used by ``age`` for consistency. _IMPL_KEYWORDS: frozenset[str] = frozenset( {"implementation", "modified", "body", "reformatted"} ) _SIG_KEYWORDS: frozenset[str] = frozenset({"signature"}) _RENAME_KEYWORDS: frozenset[str] = frozenset({"renamed", "moved"}) # Control-character sanitisation pattern — strip anything that could be a # terminal escape sequence (ESC and all C0/C1 control chars except newline). _CTRL_RE = re.compile(r"[\x00-\x09\x0b-\x1f\x7f-\x9f\x1b]") EventType = Literal["create", "impl", "sig", "rename", "delete", "other"] OutputFormat = Literal["timeline", "prose"] _EVENT_ICON: _StrMap = { "create": "🌱", "impl": "🔧", "sig": "✏️ ", "rename": "🏷️ ", "delete": "💀", "other": " ", } _EVENT_LABEL: _StrMap = { "create": "Born ", "impl": "Body ", "sig": "Signature", "rename": "Renamed ", "delete": "Deleted ", "other": "Changed ", } # ── TypedDicts ───────────────────────────────────────────────────────────────── class _EventRecord(TypedDict): """A single event in a symbol's life.""" date: str commit_id: str commit_msg: str event_type: str sem_ver_bump: str detail: str class _NarrativeJson(EnvelopeJson): """Top-level JSON output structure. Machine-readable envelope emitted by ``muse code narrative --json``. Fields ------ address: Full ``file::Symbol`` address narrated. name: Bare symbol name (last ``::`` component). kind: Symbol kind as reported by the code plugin (e.g. ``"function"``). file: File portion of *address*. status: ``"alive"`` or ``"deleted"``. born_date: ISO-8601 date of first appearance (``"YYYY-MM-DD"``). born_commit: Full commit ID of the birth commit. last_change_date: ISO-8601 date of the most recent change. last_change_commit: Full commit ID of the most recent change. last_impl_date: ISO-8601 date of the most recent impl/sig change (empty if none). last_impl_commit: Full commit ID of the most recent impl/sig change (empty if none). calendar_age_days: Days since birth (wall-clock). genetic_age_days: Days since last impl/sig change (proxy for "genetic age"). impl_changes: Count of body-rewrite events. sig_changes: Count of signature-change events. renames: Count of rename events. est_survival_pct: Estimated percentage of original implementation remaining. commits_analysed: Number of commits walked during the BFS. truncated: ``true`` if the walk was capped by ``--max-commits``. events: Chronological list of ``_EventRecord`` dicts. """ address: str name: str kind: str file: str status: str born_date: str born_commit: str last_change_date: str last_change_commit: str last_impl_date: str last_impl_commit: str calendar_age_days: int genetic_age_days: int impl_changes: int sig_changes: int renames: int est_survival_pct: int commits_analysed: int truncated: bool events: list[_EventRecord] # ── Internal accumulators ────────────────────────────────────────────────────── @dataclass class _RawEvent: """An event captured during the BFS walk, before prose generation.""" ts: datetime.datetime commit_id: str commit_msg: str sem_ver_bump: str event_type: EventType # Extra context extracted from the op summaries. op_old_summary: str = "" op_new_summary: str = "" @dataclass class _SymbolHistory: """Accumulated history for one symbol.""" born_ts: datetime.datetime | None = None born_commit: str = "" last_change_ts: datetime.datetime | None = None last_change_commit: str = "" last_impl_ts: datetime.datetime | None = None last_impl_commit: str = "" impl_changes: int = 0 sig_changes: int = 0 renames: int = 0 status: str = "alive" # "alive" | "deleted" events: list[_RawEvent] = field(default_factory=list) # ── Helpers ──────────────────────────────────────────────────────────────────── def _sanitise_msg(msg: str, max_len: int = 72) -> str: """Strip control characters and truncate commit message for display.""" clean = _CTRL_RE.sub("", msg).strip() if len(clean) > max_len: clean = f"{clean[:max_len - 1]}…" return clean def _classify_op( op: DomainOp, ) -> EventType: """Classify a symbol op into a change category. Returns one of: ``create``, ``impl``, ``sig``, ``rename``, ``delete``, ``other``. """ kind = op.get("op", "") if kind == "insert": return "create" if kind == "delete": return "delete" if kind != "replace": return "other" new_sum = str(op.get("new_summary") or op.get("content_summary") or "").lower() old_sum = str(op.get("old_summary") or "").lower() if any(kw in new_sum for kw in _RENAME_KEYWORDS): return "rename" if any(kw in new_sum for kw in _SIG_KEYWORDS): return "sig" if any(kw in new_sum for kw in _IMPL_KEYWORDS) or any( kw in old_sum for kw in _IMPL_KEYWORDS ): return "impl" # Unknown replace → conservative impl change. return "impl" def _format_date(dt: datetime.datetime) -> str: """Format a datetime as ``Mon DD, YYYY`` (e.g. ``Jan 12, 2026``).""" return dt.strftime("%b %d, %Y").replace(" ", " ") def _format_date_long(dt: datetime.datetime) -> str: """Format as ``January 12th, 2026``.""" day = dt.day suffix = ( "th" if 11 <= day <= 13 else {1: "st", 2: "nd", 3: "rd"}.get(day % 10, "th") ) return dt.strftime(f"%B {day}{suffix}, %Y") def _days_ago(dt: datetime.datetime | None) -> str: """Return a human-readable relative string for *dt*.""" if dt is None: return "unknown" now = datetime.datetime.now(tz=datetime.timezone.utc) aware = dt if dt.tzinfo else dt.replace(tzinfo=datetime.timezone.utc) delta = now - aware days = max(0, delta.days) if days == 0: return "today" if days == 1: return "1 day ago" if days < 7: return f"{days} days ago" if days < 30: return f"{days // 7} wk ago" if days < 365: return f"{days // 30} mo ago" yr = days // 365 mo = (days % 365) // 30 return f"{yr} yr {mo} mo ago" if mo else f"{yr} yr ago" def _relative_to(earlier: datetime.datetime, later: datetime.datetime) -> str: """Return human-readable delta between two datetimes (e.g. '3 weeks later').""" earlier_n = earlier.replace(tzinfo=None) if earlier.tzinfo else earlier later_n = later.replace(tzinfo=None) if later.tzinfo else later days = max(0, (later_n - earlier_n).days) if days == 0: return "the same day" if days == 1: return "1 day later" if days < 7: return f"{days} days later" if days < 30: weeks = days // 7 return f"{weeks} week{'s' if weeks > 1 else ''} later" if days < 365: months = days // 30 return f"{months} month{'s' if months > 1 else ''} later" years = days // 365 return f"{years} year{'s' if years > 1 else ''} later" def _extract_rename(new_summary: str, old_summary: str) -> tuple[str, str]: """Try to extract old_name → new_name from rename op summaries. Returns (old_name, new_name) or ("", "") if not extractable. """ # Try pattern: "renamed X to Y" or "moved X to Y" m = re.search(r"(?:renamed?|moved?)\s+(\S+)\s+to\s+(\S+)", new_summary, re.IGNORECASE) if m: return m.group(1), m.group(2) # Try pattern from old_summary "X" → new_summary "Y" (simple name extraction) old_name = old_summary.split("::")[1] if "::" in old_summary else "" new_name = new_summary.split("::")[1] if "::" in new_summary else "" return old_name, new_name def _extract_source( root: pathlib.Path, snapshot_commit_id: str, address: str, max_lines: int = 10, ) -> str | None: """Extract the source of a symbol at a given commit. Reads the file blob from the object store for *snapshot_commit_id*, parses the AST, and extracts the function/class body lines. Args: root: Repository root. snapshot_commit_id: Commit ID to read the snapshot from. address: Full symbol address (``file::QualifiedName``). max_lines: Maximum source lines to return. Returns: Indented source string, or ``None`` if not extractable. """ from muse.core.object_store import read_object from muse.core.snapshots import get_commit_snapshot_manifest if "::" not in address: return None file_path, sym_name = address.split("::", 1) manifest = get_commit_snapshot_manifest(root, snapshot_commit_id) if not manifest: return None obj_id = manifest.get(file_path) if obj_id is None: return None raw = read_object(root, obj_id) if raw is None: return None try: if len(raw) > MAX_AST_BYTES: return None tree = ast.parse(raw) except SyntaxError: return None func_node = find_func_node(tree.body, sym_name.split(".")) if func_node is None: # Try class-level return None # Extract source lines try: source_lines = raw.decode("utf-8", errors="replace").splitlines() except Exception: return None start = func_node.lineno - 1 end = func_node.end_lineno or (start + max_lines) snippet = source_lines[start : min(end, start + max_lines)] if end - start > max_lines: snippet.append(" …") return "\n".join(f" {line}" for line in snippet) # ── Core BFS walk ────────────────────────────────────────────────────────────── def _collect_history( root: pathlib.Path, head_commit_id: str, address: str, max_commits: int, stop_at: str | None, ) -> tuple[_SymbolHistory, int, bool]: """Walk the commit DAG and build a ``_SymbolHistory`` for *address*. Args: root: Repository root. head_commit_id: SHA-256 of HEAD commit. address: Symbol address to trace. max_commits: BFS depth cap. stop_at: Optional exclusive lower-bound commit ID. Returns: ``(history, commits_analysed, truncated)`` """ commits, truncated = walk_commits_bfs( root, head_commit_id, max_commits, stop_at_commit_id=stop_at ) history = _SymbolHistory() for commit in commits: if commit.structured_delta is None: continue ops: list[DomainOp] = commit.structured_delta["ops"] ts = commit.committed_at cid = commit.commit_id msg = _sanitise_msg(commit.message) bump = (commit.sem_ver_bump if commit.sem_ver_bump else "none").lower() for op in flat_symbol_ops(ops): if op["address"] != address: continue if "::import::" in op["address"]: continue event_type = _classify_op(op) old_sum = str(op.get("old_summary") or "").strip() new_sum = str(op.get("new_summary") or op.get("content_summary") or "").strip() raw_event = _RawEvent( ts=ts, commit_id=cid, commit_msg=msg, sem_ver_bump=bump, event_type=event_type, op_old_summary=old_sum, op_new_summary=new_sum, ) history.events.append(raw_event) # Update lifecycle metadata. ts_naive = ts.replace(tzinfo=None) if ts.tzinfo else ts if history.born_ts is None or ts_naive < ( history.born_ts.replace(tzinfo=None) if history.born_ts.tzinfo else history.born_ts ): history.born_ts = ts history.born_commit = cid if history.last_change_ts is None or ts_naive > ( history.last_change_ts.replace(tzinfo=None) if history.last_change_ts.tzinfo else history.last_change_ts ): history.last_change_ts = ts history.last_change_commit = cid if event_type in ("impl", "sig"): if history.last_impl_ts is None or ts_naive > ( history.last_impl_ts.replace(tzinfo=None) if history.last_impl_ts.tzinfo else history.last_impl_ts ): history.last_impl_ts = ts history.last_impl_commit = cid if event_type == "impl": history.impl_changes += 1 elif event_type == "sig": history.sig_changes += 1 elif event_type == "rename": history.renames += 1 elif event_type == "delete": history.status = "deleted" # Sort events oldest-first for display. history.events.sort(key=lambda e: e.ts) return history, len(commits), truncated # ── Prose generation ─────────────────────────────────────────────────────────── _BIRTH_OPENERS: tuple[str, ...] = ( "was born", "was introduced", "first appeared", "was created", ) _IMPL_VERBS: tuple[str, ...] = ( "its body was rewritten", "the implementation was overhauled", "it underwent a body rewrite", "the logic was reworked", ) _SIG_VERBS: tuple[str, ...] = ( "its signature changed", "the interface was updated", "the signature was modified", ) def _prose_sentence( event: _RawEvent, index: int, born_ts: datetime.datetime | None, prev_ts: datetime.datetime | None, name: str, kind: str, ) -> str: """Generate one prose sentence for an event.""" date_str = _format_date_long(event.ts) bump_str = ( f" — a {event.sem_ver_bump.upper()} bump" if event.sem_ver_bump not in ("none", "") else "" ) if event.event_type == "create": opener = _BIRTH_OPENERS[index % len(_BIRTH_OPENERS)] rel = "" return ( f"**{name}** {opener} on {date_str} as a {kind}" f" (commit {event.commit_id}){bump_str}." ) # Relative time since birth or previous event. relative: str if prev_ts is not None: relative = _relative_to(prev_ts, event.ts) elif born_ts is not None: relative = _relative_to(born_ts, event.ts) else: relative = f"On {date_str}," if event.event_type == "impl": verb = _IMPL_VERBS[index % len(_IMPL_VERBS)] return f"{relative.capitalize()}, {verb}{bump_str}." if event.event_type == "sig": verb = _SIG_VERBS[index % len(_SIG_VERBS)] return f"{relative.capitalize()}, {verb}{bump_str}." if event.event_type == "rename": old, new = _extract_rename(event.op_new_summary, event.op_old_summary) rename_detail = f" from {old} to {new}" if old and new else "" return f"{relative.capitalize()}, it was renamed{rename_detail}{bump_str}." if event.event_type == "delete": return f"{relative.capitalize()}, the symbol was deleted{bump_str}." return f"{relative.capitalize()}, it changed{bump_str}." def _generate_prose( history: _SymbolHistory, address: str, name: str, kind: str, ) -> str: """Generate a flowing paragraph narrative for *history*.""" if not history.events: return f"No history found for {address}." sentences: list[str] = [] prev_ts: datetime.datetime | None = None for i, event in enumerate(history.events): sentence = _prose_sentence(event, i, history.born_ts, prev_ts, name, kind) sentences.append(sentence) prev_ts = event.ts # Summary sentence. parts: list[str] = [] if history.born_ts: now = datetime.datetime.now(tz=datetime.timezone.utc) aware = ( history.born_ts if history.born_ts.tzinfo else history.born_ts.replace(tzinfo=datetime.timezone.utc) ) age_days = (now - aware).days parts.append(f"At {age_days} days old") if history.impl_changes > 0: rw = history.impl_changes parts_body = f"survived {rw} body rewrite{'s' if rw != 1 else ''}" survival = round(100 / (history.impl_changes + 1)) parts_body += f" and retains an estimated {survival}% of its original implementation" if parts: sentences.append(f" {parts[0]}, the {kind} has {parts_body}.") else: sentences.append(f" The {kind} has {parts_body}.") elif parts: if history.status == "deleted": sentences.append(f" {parts[0]}, the {kind} is now deleted.") else: sentences.append(f" {parts[0]}, the {kind} remains unchanged.") return f" {'\n '.join(sentences)}" # ── Event detail builder ─────────────────────────────────────────────────────── def _event_detail(event: _RawEvent) -> str: """Generate a short detail line for a timeline event.""" if event.event_type == "rename": old, new = _extract_rename(event.op_new_summary, event.op_old_summary) if old and new: return f"{old} → {new}" if event.event_type == "create": summary = event.op_new_summary or event.op_old_summary if summary: return summary[:80] return "" # ── Formatters ───────────────────────────────────────────────────────────────── def _print_timeline( history: _SymbolHistory, address: str, name: str, kind: str, commits_analysed: int, truncated: bool, show_source: bool, source_birth: str | None, source_head: str | None, ) -> None: """Print the human-readable timeline view.""" file_part = address.split("::")[0] if "::" in address else address print(f"\n {sanitize_display(name)} ({sanitize_display(kind)}) · {sanitize_display(file_part)}\n") if not history.events: print(" No history found in the analysed commit range.") return # Detect largest single change (impl events by line count proxy — use index). impl_events = [e for e in history.events if e.event_type == "impl"] largest_idx: int | None = None if len(impl_events) >= 2: # We don't have line counts, so mark the last body rewrite as notable # when there are multiple — common pattern. largest_idx = history.events.index(impl_events[-1]) for i, event in enumerate(history.events): icon = _EVENT_ICON.get(event.event_type, " ") label = _EVENT_LABEL.get(event.event_type, "Changed ") date_str = _format_date(event.ts) bump = event.sem_ver_bump bump_str = f" · {bump.upper()}" if bump and bump != "none" else "" breaking = " ⚠️ breaking" if bump == "major" else "" print(f" {icon} {label} {date_str} · commit {event.commit_id}{bump_str}{breaking}") print(f" \"{event.commit_msg}\"") detail = _event_detail(event) if detail: print(f" {detail}") if largest_idx is not None and i == largest_idx: print(" Largest change in symbol's history") print() # Summary bar. sep = "─" * 65 print(f" {sep}") print(" Life summary") if history.born_ts: print( f" Born: {_format_date(history.born_ts)}" f" ({_days_ago(history.born_ts)}) · commit {history.born_commit}" ) if history.last_change_ts: print( f" Last change: {_format_date(history.last_change_ts)}" f" · commit {history.last_change_commit}" ) counts: list[str] = [] if history.impl_changes: counts.append(f"{history.impl_changes} body rewrite{'s' if history.impl_changes != 1 else ''}") if history.sig_changes: counts.append(f"{history.sig_changes} signature change{'s' if history.sig_changes != 1 else ''}") if history.renames: counts.append(f"{history.renames} rename{'s' if history.renames != 1 else ''}") if counts: print(f" Events: {' · '.join(counts)}") survival = round(100 / (history.impl_changes + 1)) print(f" Survival: ~{survival}% of original implementation remains") print(f" Status: {history.status.capitalize()}") if truncated: print(f"\n ⚠️ History truncated at {commits_analysed} commits.") print() # Optional source code snippets. if show_source: if source_birth: print(f" First version (commit {history.born_commit})") print(source_birth) print() if source_head: print(" Current version (HEAD)") print(source_head) print() def _print_prose( history: _SymbolHistory, address: str, name: str, kind: str, truncated: bool, ) -> None: """Print the prose narrative view.""" print(f"\n {sanitize_display(name)} ({sanitize_display(kind)})\n") prose = _generate_prose(history, address, name, kind) # Wrap at ~80 chars for readability. print(prose) if truncated: print(f"\n ⚠️ History was truncated; narrative may be incomplete.") print() def _build_json_events(history: _SymbolHistory) -> list[_EventRecord]: """Convert internal events to JSON-ready records.""" records: list[_EventRecord] = [] for event in history.events: detail = _event_detail(event) if not detail and event.op_new_summary: detail = event.op_new_summary[:80] records.append( _EventRecord( date=event.ts.strftime("%Y-%m-%d"), commit_id=event.commit_id, commit_msg=event.commit_msg, event_type=event.event_type, sem_ver_bump=event.sem_ver_bump, detail=detail, ) ) return records # ── Entry point ──────────────────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Render the life story of a symbol across commit history. Walks commit history for a ``file::Symbol`` address, builds a chronological event log of implementation changes, signature changes, and renames, and computes evolutionary metrics (calendar age, genetic age, survival estimate). Agent quickstart ---------------- :: muse code narrative "billing.py::Invoice" --json muse code narrative "billing.py::compute_total" -n 50 --json muse code narrative "billing.py::Invoice" --format prose --json JSON fields ----------- address Symbol address (``file::Symbol``). name Symbol short name. kind Symbol kind (``function``, ``class``, …). file File part of the address. status ``"active"`` or ``"deleted"``. born_date ISO date the symbol was first introduced. born_commit Commit ID where the symbol was born. last_change_date ISO date of the last implementation change. calendar_age_days Days since the symbol was born. genetic_age_days Days since the last implementation change. impl_changes Number of implementation-level changes. sig_changes Number of signature changes. renames Number of renames. est_survival_pct Estimated survival probability (0–100). commits_analysed Number of commits walked. truncated ``true`` if the walk hit the ``-n`` cap. events Chronological event list. Exit codes ---------- 0 Success. 1 Invalid address format or no history found. 2 Not inside a Muse repository. """ elapsed = start_timer() root = require_repo() # ── Argument validation ──────────────────────────────────────────────────── address: str = args.address.strip() if "::" not in address: print( "❌ ADDRESS must be in ``file::Symbol`` format" " (e.g. billing.py::Invoice.compute_total).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) max_commits: int = clamp_int(args.max_commits, 1, 100000, 'max_commits') if max_commits < 1: print("❌ --max-commits must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) output_format: OutputFormat = args.format show_source: bool = args.show_source # ── Resolve HEAD ─────────────────────────────────────────────────────────── branch = read_current_branch(root) head = resolve_commit_ref(root, branch, None) if head is None: print("❌ HEAD commit not found — is this an empty repository?", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Resolve --since ──────────────────────────────────────────────────────── stop_at: str | None = None if args.since: since_commit = resolve_commit_ref(root, branch, args.since) if since_commit is None: print(f"❌ Could not resolve --since ref: {args.since!r}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) stop_at = since_commit.commit_id # ── Verify symbol exists in HEAD snapshot ───────────────────────────────── manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} all_trees = symbols_for_snapshot(root, manifest) # Build flat address → kind map. addr_to_kind: _StrMap = {} for sym_tree in all_trees.values(): for addr, rec in sym_tree.items(): addr_to_kind[addr] = rec["kind"] file_part = address.split("::")[0] sym_part = address.split("::", 1)[1] name = sym_part.split(".")[-1] # bare symbol name (last component) # Be tolerant: the symbol may have been deleted (status=deleted) — still # show its history. But try to get the kind from HEAD first. kind = addr_to_kind.get(address, "symbol") # ── Collect commit history ───────────────────────────────────────────────── history, commits_analysed, truncated = _collect_history( root, head.commit_id, address, max_commits, stop_at ) if not history.events: print( f"❌ No history found for {address!r} in the last" f" {commits_analysed} commit{'s' if commits_analysed != 1 else ''}.", file=sys.stderr, ) print( " Check that the address is correct (file::ClassName.method_name)", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Optional source extraction ──────────────────────────────────────────── source_birth: str | None = None source_head: str | None = None if show_source and history.born_commit: from muse.core.commits import resolve_commit_ref as _rcr birth_commit = _rcr(root, branch, history.born_commit) if birth_commit: source_birth = _extract_source(root, birth_commit.commit_id, address) source_head = _extract_source(root, head.commit_id, address) # ── Output ──────────────────────────────────────────────────────────────── if args.json_out: now = datetime.datetime.now(tz=datetime.timezone.utc) def _days(dt: datetime.datetime | None) -> int: if dt is None: return 0 aware = dt if dt.tzinfo else dt.replace(tzinfo=datetime.timezone.utc) return max(0, (now - aware).days) def _date_str(dt: datetime.datetime | None) -> str: return dt.strftime("%Y-%m-%d") if dt else "" calendar_age = _days(history.born_ts) genetic_age = _days(history.last_impl_ts) survival = round(100 / (history.impl_changes + 1)) print(json.dumps(_NarrativeJson( **make_envelope(elapsed), address=address, name=name, kind=kind, file=file_part, status=history.status, born_date=_date_str(history.born_ts), born_commit=history.born_commit, last_change_date=_date_str(history.last_change_ts), last_change_commit=history.last_change_commit, last_impl_date=_date_str(history.last_impl_ts), last_impl_commit=history.last_impl_commit, calendar_age_days=calendar_age, genetic_age_days=genetic_age, impl_changes=history.impl_changes, sig_changes=history.sig_changes, renames=history.renames, est_survival_pct=survival, commits_analysed=commits_analysed, truncated=truncated, events=_build_json_events(history), ))) return if output_format == "prose": _print_prose(history, address, name, kind, truncated) else: _print_timeline( history, address, name, kind, commits_analysed, truncated, show_source, source_birth, source_head, ) # ── CLI registration ─────────────────────────────────────────────────────────── def register( sub: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Register ``narrative`` under the ``code`` subcommand group. Args: sub: The subparser action from the ``code`` command group. """ p = sub.add_parser( "narrative", help=( "Plain-English story of a symbol's life, built from structured" " commit deltas." ), description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument( "address", metavar="ADDRESS", help=( "Full symbol address to narrate" " (e.g. billing.py::Invoice.compute_total)." ), ) p.add_argument( "--format", metavar="FMT", choices=["timeline", "prose"], default="timeline", help=( "Output format: ``timeline`` (default) shows a chronological" " event list; ``prose`` generates a flowing paragraph story." ), ) p.add_argument( "--show-source", action="store_true", help=( "Append the symbol's source code at birth and at HEAD" " (Python only; requires the blob to be in the object store)." ), ) p.add_argument( "--since", metavar="REF", help=( "Limit history to commits after REF" " (branch name, commit SHA, or tag)." ), ) p.add_argument( "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", help=f"Maximum commits to walk (default: {_DEFAULT_MAX_COMMITS}).", ) p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit JSON instead of human-readable text.", ) p.set_defaults(func=run)