"""muse code impact — transitive blast-radius analysis. Answers the question every engineer asks before touching a function: *"If I change this, what else could break?"* ``muse code impact`` builds the reverse call graph for the committed snapshot, then performs a BFS from the target symbol's bare name through every caller, then every caller's callers, until the full transitive closure is reached. The result is a depth-ordered blast-radius map: depth 1 = direct callers, depth 2 = callers of callers, and so on. This tells you exactly how far a change propagates through the codebase. With ``--compare REF``, the command diffs the blast radius between two commits, showing exactly which callers were *added* (new risk) and which were *removed* (coupling reduced) — the ideal pre-merge signal for any proposal that touches a widely-called symbol. With ``--forward``, the direction reverses: instead of "who calls this?", it answers "what does this call?" — the full transitive dependency fan-out of the target symbol. This is structurally impossible in Git. Git stores files as blobs — it has no concept of call relationships between functions. You would need an external static-analysis tool and a separate dependency graph. In Muse, the symbol graph is a first-class citizen of every committed snapshot. Usage:: muse code impact "src/billing.py::compute_invoice_total" muse code impact "src/billing.py::compute_invoice_total" --depth 2 muse code impact "src/auth.py::validate_token" --commit HEAD~5 muse code impact "src/core.py::content_hash" --json muse code impact "src/billing.py::process_order" --compare HEAD~10 muse code impact "src/billing.py::process_order" --forward --depth 3 muse code impact "src/api.py::handle_request" --file src/billing.py muse code impact "src/core.py::content_hash" --count Output:: Impact analysis: src/billing.py::compute_invoice_total ────────────────────────────────────────────────────────────── Depth 1 — direct callers (2): src/api.py::create_invoice src/billing.py::process_order Depth 2 — callers of callers (1): src/api.py::handle_request ────────────────────────────────────────────────────────────── Total blast radius: 3 symbols across 2 files 🔴 High impact — add tests before changing this symbol. Flags: ``--depth, -d N`` Stop BFS after N levels (default: 0 = unlimited). ``--commit, -c REF`` Analyse a historical snapshot instead of HEAD. ``--compare REF`` Diff blast radius between HEAD (or ``--commit``) and REF. Shows added callers (new risk) and removed callers (reduced coupling). ``--forward`` Show callees (dependencies) instead of callers. Answers: "what does this symbol depend on?" rather than "what depends on this symbol?" ``--file PATH`` Restrict the blast-radius output to callers from this file only. ``--count`` Print only the total count of callers (scriptable). ``--json`` Emit the full blast-radius map as JSON. """ import argparse import json import logging import pathlib import sys from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.timing import start_timer from typing import TypedDict from muse.core.refs import read_current_branch from muse.core.commits import ( CommitRecord, resolve_commit_ref, ) from muse.core.snapshots import get_commit_snapshot_manifest from muse.core.symbol_cache import load_symbol_cache from muse.core.callgraph_cache import load_callgraph_cache from muse.core.implicit_edge_cache import load_implicit_edge_cache from muse.plugins.code._callgraph import ( ForwardGraph, ReverseGraph, build_forward_graph, build_reverse_graph, transitive_callees, transitive_callers, ) from muse.plugins.code._framework import ( ImplicitEntryEdge, ImplicitEdgeGraph, build_implicit_edge_graph, ) from muse.plugins.code._query import dir_of, language_of from muse.core.validation import clamp_int, sanitize_display type _BlastRadius = dict[str, list[str]] type _StrMeta = dict[str, str] logger = logging.getLogger(__name__) class _EntryPointJson(TypedDict): """One framework entry-point edge in JSON output.""" framework_id: str kind: str metadata: _StrMeta class _ImpactJsonBase(EnvelopeJson): """Required fields for ``muse code impact`` JSON output.""" mode: str address: str target_name: str commit_id: str depth_limit: int total: int class _ImpactJson(_ImpactJsonBase, total=False): """Full JSON payload for ``muse code impact`` output. Inherits required envelope + domain fields from :class:`_ImpactJsonBase`. Fields (all optional) --------------------- file_filter File scope applied, or ``None``. blast_radius Depth-keyed map of caller addresses (reverse mode). callees Depth-keyed map of callee names (forward mode). entry_points Framework entry-point edges, if any. compare_commit_id Commit ID of the ``--compare`` snapshot. added_callers Callers present in HEAD but not in compare. removed_callers Callers present in compare but not in HEAD. net_change ``len(added_callers) - len(removed_callers)``. """ file_filter: str | None blast_radius: _BlastRadius callees: dict[str, list[str]] entry_points: list[_EntryPointJson] compare_commit_id: str added_callers: list[str] removed_callers: list[str] net_change: int def _flat_addrs(blast: dict[int, list[str]]) -> set[str]: """Return the flat set of all addresses across all depths.""" return {addr for addrs in blast.values() for addr in addrs} def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the impact subcommand.""" parser = subparsers.add_parser( "impact", help="Show the transitive blast-radius of changing a 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( "--depth", "-d", type=int, default=None, metavar="N", help="Maximum BFS depth (default: 5, max: 50). Use --transitive for full traversal.", ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help="Analyse a historical snapshot instead of HEAD.", ) parser.add_argument( "--compare", default=None, metavar="REF", dest="compare_ref", help="Diff blast radius against this commit reference.", ) parser.add_argument( "--forward", action="store_true", dest="forward", help="Show callees (dependencies) instead of callers.", ) parser.add_argument( "--file", "-f", default=None, metavar="PATH", dest="file_filter", help="Restrict blast-radius output to callers from this file.", ) parser.add_argument( "--count", action="store_true", dest="count_only", help="Print only the total count of matching symbols.", ) parser.add_argument( "--roll-up-to", default=None, choices=("directory",), metavar="LEVEL", dest="roll_up_to", help="Roll up blast-radius addresses to the given granularity (e.g. 'directory').", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit results as JSON.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Show the transitive blast-radius of changing a symbol. Builds the reverse call graph for the committed snapshot, then BFS-walks it outward from the target symbol. Depth 1 = direct callers; depth 2 = callers of callers; and so on. Use ``--compare`` to diff the blast radius against another commit; ``--forward`` to walk callees instead of callers. Agent quickstart ---------------- :: muse code impact "src/billing.py::compute_total" --json muse code impact "src/billing.py::compute_total" --depth 3 --json muse code impact "src/billing.py::compute_total" --forward --json muse code impact "src/billing.py::compute_total" --compare HEAD~10 --json JSON fields (reverse / blast-radius) ------------------------------------- mode ``"reverse"``. address Symbol address analysed. commit_id Commit snapshot used. blast_radius Depth-keyed map of caller addresses (strings ``"1"``, ``"2"``…). total Total caller count across all depths. JSON fields (``--forward`` / callees) --------------------------------------- mode ``"forward"``. callees Depth-keyed map of callee names. total Total callee count. Exit codes ---------- 0 Analysis complete. 1 Symbol not found or invalid arguments. 2 Not inside a Muse repository. """ elapsed = start_timer() address: str = args.address _depth_raw: int | None = args.depth depth: int = clamp_int(_depth_raw, 1, 50, "depth") if _depth_raw is not None else 5 ref: str | None = args.ref compare_ref: str | None = args.compare_ref forward: bool = args.forward file_filter: str | None = args.file_filter count_only: bool = args.count_only json_out: bool = args.json_out roll_up_to: str | None = getattr(args, "roll_up_to", None) if forward and compare_ref: print("❌ --forward and --compare are mutually exclusive.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() branch = read_current_branch(root) lang = language_of(address.split("::")[0]) if "::" in address else "" if lang and lang != "Python": print( f"⚠️ Impact analysis is currently Python-only. '{address}' is {lang}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) commit = resolve_commit_ref(root, branch, ref) if commit is None: print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} cache = load_symbol_cache(root) cg_cache = load_callgraph_cache(root) implicit_cache = load_implicit_edge_cache(root) target_name = address.split("::")[-1].split(".")[-1] if "::" in address else address if forward: fwd: ForwardGraph = build_forward_graph(root, manifest, cache=cache, callgraph_cache=cg_cache) callee_map = transitive_callees(address, fwd, max_depth=depth) cache.save() cg_cache.save() _render_forward( address=address, target_name=target_name, commit=commit, callee_map=callee_map, depth_limit=depth, count_only=count_only, json_out=json_out, duration_ms=elapsed(), ) return rev: ReverseGraph = build_reverse_graph(root, manifest, cache=cache, callgraph_cache=cg_cache) implicit: ImplicitEdgeGraph = build_implicit_edge_graph(root, manifest, cache=cache, implicit_cache=implicit_cache) blast = transitive_callers(target_name, rev, max_depth=depth) if file_filter: blast = { d: [a for a in addrs if "::" in a and a.split("::")[0] == file_filter] for d, addrs in blast.items() } blast = {d: addrs for d, addrs in blast.items() if addrs} compare_commit: CommitRecord | None = None added: set[str] = set() removed: set[str] = set() if compare_ref: compare_commit = resolve_commit_ref(root, branch, compare_ref) if compare_commit is None: print(f"❌ --compare commit '{compare_ref}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) compare_manifest = get_commit_snapshot_manifest(root, compare_commit.commit_id) or {} compare_rev: ReverseGraph = build_reverse_graph(root, compare_manifest, cache=cache, callgraph_cache=cg_cache) compare_blast = transitive_callers(target_name, compare_rev, max_depth=depth) if file_filter: compare_blast = { d: [a for a in addrs if "::" in a and a.split("::")[0] == file_filter] for d, addrs in compare_blast.items() } compare_blast = {d: addrs for d, addrs in compare_blast.items() if addrs} current_flat = _flat_addrs(blast) compare_flat = _flat_addrs(compare_blast) added = current_flat - compare_flat removed = compare_flat - current_flat cache.save() cg_cache.save() implicit_cache.save() entry_points = implicit.get(address, []) _render_reverse( address=address, target_name=target_name, commit=commit, blast=blast, entry_points=entry_points, depth_limit=depth, file_filter=file_filter, compare_commit=compare_commit, added=added, removed=removed, count_only=count_only, json_out=json_out, roll_up_to=roll_up_to, duration_ms=elapsed(), ) # --------------------------------------------------------------------------- # Renderers # --------------------------------------------------------------------------- def _render_forward( *, address: str, target_name: str, commit: CommitRecord, callee_map: dict[int, list[str]], depth_limit: int, count_only: bool, json_out: bool, duration_ms: float = 0.0, ) -> None: """Render forward (callees) analysis.""" total = sum(len(v) for v in callee_map.values()) if count_only and not json_out: print(total) return if json_out: out = _ImpactJson( **make_envelope(lambda: duration_ms), mode="forward", address=address, target_name=target_name, commit_id=commit.commit_id, depth_limit=depth_limit, total=total, ) out["callees"] = {str(d): names for d, names in sorted(callee_map.items())} print(json.dumps(out)) return print(f"\nDependency fan-out: {sanitize_display(address)}") print("─" * 62) if not callee_map: print(f"\n ('{target_name}' calls nothing detectable — leaf function or dynamic dispatch)") return for d in sorted(callee_map.keys()): label = "direct callees" if d == 1 else f"depth-{d} callees" print(f"\nDepth {d} — {label} ({len(callee_map[d])}):") for name in sorted(callee_map[d]): print(f" {sanitize_display(name)}") print(f"\n{'─' * 62}") print(f"Total dependency fan-out: {total} callee(s)") print("\nNote: analysis covers Python call-sites only.") def _render_reverse( *, address: str, target_name: str, commit: CommitRecord, blast: dict[int, list[str]], entry_points: list[ImplicitEntryEdge], depth_limit: int, file_filter: str | None, compare_commit: CommitRecord | None, added: set[str], removed: set[str], count_only: bool, json_out: bool, roll_up_to: str | None = None, duration_ms: float = 0.0, ) -> None: """Render reverse (callers / blast-radius) analysis.""" total = sum(len(v) for v in blast.values()) if count_only and not json_out: print(total) return if json_out: payload = _ImpactJson( **make_envelope(lambda: duration_ms), mode="reverse", address=address, target_name=target_name, commit_id=commit.commit_id, depth_limit=depth_limit, total=total, ) payload["file_filter"] = file_filter payload["blast_radius"] = {str(d): list(addrs) for d, addrs in sorted(blast.items())} if roll_up_to == "directory": dir_counts: dict[str, int] = {} for addrs in blast.values(): for addr in addrs: d = dir_of(addr.split("::")[0]) if "::" in addr else dir_of(addr) dir_counts[d] = dir_counts.get(d, 0) + 1 payload["directory_blast_radius"] = dir_counts if entry_points: payload["entry_points"] = [ _EntryPointJson( framework_id=ep.framework_id, kind=ep.kind, metadata=ep.metadata, ) for ep in entry_points ] if compare_commit is not None: payload["compare_commit_id"] = compare_commit.commit_id payload["added_callers"] = sorted(added) payload["removed_callers"] = sorted(removed) payload["net_change"] = len(added) - len(removed) print(json.dumps(payload)) return print(f"\nImpact analysis: {sanitize_display(address)}") if file_filter: print(f"(filtered to callers in: {file_filter})") print("─" * 62) if entry_points: print("\n Framework entry point — externally reachable via:") for ep in entry_points: meta_str = " ".join(f"{k}={v}" for k, v in ep.metadata.items() if v) print(f" [{ep.framework_id}] kind={ep.kind} {meta_str}") if not blast: if entry_points: print("\n (no explicit callers in user code — wired by the framework above)") else: print( f"\n (no callers detected — '{target_name}' may be an entry point or dead code)" ) print("\n Note: analysis covers Python only; external callers are not detected.") else: all_files: set[str] = set() for d in sorted(blast.keys()): callers = blast[d] if d == 1: label = "direct callers" elif d == 2: label = "callers of callers" else: label = f"depth-{d} callers" print(f"\nDepth {d} — {label} ({len(callers)}):") for addr in sorted(callers): marker = " ← NEW" if addr in added else "" print(f" {sanitize_display(addr)}{marker}") if "::" in addr: all_files.add(addr.split("::")[0]) print(f"\n{'─' * 62}") file_label = "file" if len(all_files) == 1 else "files" print(f"Total blast radius: {total} symbol(s) across {len(all_files)} {file_label}") if total >= 10: print("🔴 High impact — add tests before changing this symbol.") elif total >= 3: print("🟡 Medium impact — review callers before changing this symbol.") else: print("🟢 Low impact — change is well-contained.") print( "\nNote: analysis covers Python call-sites only." " Dynamic dispatch (getattr, decorators) is not detected." ) if compare_commit is not None: print(f"\nBlast-radius diff vs {compare_commit.commit_id}:") print("─" * 62) if not added and not removed: print(" No change in blast radius.") else: net = len(added) - len(removed) sign = "+" if net >= 0 else "" print(f" Net change: {sign}{net} caller(s)") if added: print(f"\n Added ({len(added)}):") for a in sorted(added): print(f" + {a}") if removed: print(f"\n Removed ({len(removed)}):") for r in sorted(removed): print(f" - {r}")