impact.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """muse code impact — transitive blast-radius analysis. |
| 2 | |
| 3 | Answers the question every engineer asks before touching a function: |
| 4 | *"If I change this, what else could break?"* |
| 5 | |
| 6 | ``muse code impact`` builds the reverse call graph for the committed snapshot, |
| 7 | then performs a BFS from the target symbol's bare name through every caller, |
| 8 | then every caller's callers, until the full transitive closure is reached. |
| 9 | |
| 10 | The result is a depth-ordered blast-radius map: depth 1 = direct callers, |
| 11 | depth 2 = callers of callers, and so on. This tells you exactly how far a |
| 12 | change propagates through the codebase. |
| 13 | |
| 14 | With ``--compare REF``, the command diffs the blast radius between two commits, |
| 15 | showing exactly which callers were *added* (new risk) and which were *removed* |
| 16 | (coupling reduced) — the ideal pre-merge signal for any proposal that touches a |
| 17 | widely-called symbol. |
| 18 | |
| 19 | With ``--forward``, the direction reverses: instead of "who calls this?", it |
| 20 | answers "what does this call?" — the full transitive dependency fan-out of the |
| 21 | target symbol. |
| 22 | |
| 23 | This is structurally impossible in Git. Git stores files as blobs — it has |
| 24 | no concept of call relationships between functions. You would need an |
| 25 | external static-analysis tool and a separate dependency graph. In Muse, |
| 26 | the symbol graph is a first-class citizen of every committed snapshot. |
| 27 | |
| 28 | Usage:: |
| 29 | |
| 30 | muse code impact "src/billing.py::compute_invoice_total" |
| 31 | muse code impact "src/billing.py::compute_invoice_total" --depth 2 |
| 32 | muse code impact "src/auth.py::validate_token" --commit HEAD~5 |
| 33 | muse code impact "src/core.py::content_hash" --json |
| 34 | muse code impact "src/billing.py::process_order" --compare HEAD~10 |
| 35 | muse code impact "src/billing.py::process_order" --forward --depth 3 |
| 36 | muse code impact "src/api.py::handle_request" --file src/billing.py |
| 37 | muse code impact "src/core.py::content_hash" --count |
| 38 | |
| 39 | Output:: |
| 40 | |
| 41 | Impact analysis: src/billing.py::compute_invoice_total |
| 42 | ────────────────────────────────────────────────────────────── |
| 43 | |
| 44 | Depth 1 — direct callers (2): |
| 45 | src/api.py::create_invoice |
| 46 | src/billing.py::process_order |
| 47 | |
| 48 | Depth 2 — callers of callers (1): |
| 49 | src/api.py::handle_request |
| 50 | |
| 51 | ────────────────────────────────────────────────────────────── |
| 52 | Total blast radius: 3 symbols across 2 files |
| 53 | 🔴 High impact — add tests before changing this symbol. |
| 54 | |
| 55 | Flags: |
| 56 | |
| 57 | ``--depth, -d N`` |
| 58 | Stop BFS after N levels (default: 0 = unlimited). |
| 59 | |
| 60 | ``--commit, -c REF`` |
| 61 | Analyse a historical snapshot instead of HEAD. |
| 62 | |
| 63 | ``--compare REF`` |
| 64 | Diff blast radius between HEAD (or ``--commit``) and REF. |
| 65 | Shows added callers (new risk) and removed callers (reduced coupling). |
| 66 | |
| 67 | ``--forward`` |
| 68 | Show callees (dependencies) instead of callers. Answers: "what does |
| 69 | this symbol depend on?" rather than "what depends on this symbol?" |
| 70 | |
| 71 | ``--file PATH`` |
| 72 | Restrict the blast-radius output to callers from this file only. |
| 73 | |
| 74 | ``--count`` |
| 75 | Print only the total count of callers (scriptable). |
| 76 | |
| 77 | ``--json`` |
| 78 | Emit the full blast-radius map as JSON. |
| 79 | """ |
| 80 | |
| 81 | from __future__ import annotations |
| 82 | |
| 83 | import argparse |
| 84 | import json |
| 85 | import logging |
| 86 | import pathlib |
| 87 | import sys |
| 88 | |
| 89 | from muse.core.errors import ExitCode |
| 90 | from muse.core.repo import read_repo_id, require_repo |
| 91 | from typing import TypedDict |
| 92 | |
| 93 | from muse.core.store import ( |
| 94 | CommitRecord, |
| 95 | get_commit_snapshot_manifest, |
| 96 | read_current_branch, |
| 97 | resolve_commit_ref, |
| 98 | ) |
| 99 | from muse.core.symbol_cache import load_symbol_cache |
| 100 | from muse.plugins.registry import read_domain |
| 101 | from muse.plugins.code._callgraph import ( |
| 102 | ForwardGraph, |
| 103 | ReverseGraph, |
| 104 | build_forward_graph, |
| 105 | build_reverse_graph, |
| 106 | transitive_callees, |
| 107 | transitive_callers, |
| 108 | ) |
| 109 | from muse.plugins.code._framework import ( |
| 110 | ImplicitEntryEdge, |
| 111 | ImplicitEdgeGraph, |
| 112 | build_implicit_edge_graph, |
| 113 | ) |
| 114 | from muse.plugins.code._query import language_of |
| 115 | from muse.core.validation import clamp_int, sanitize_display |
| 116 | |
| 117 | |
| 118 | type _BlastRadius = dict[str, list[str]] |
| 119 | type _StrMeta = dict[str, str] |
| 120 | logger = logging.getLogger(__name__) |
| 121 | |
| 122 | |
| 123 | class _EntryPointJson(TypedDict): |
| 124 | """One framework entry-point edge in JSON output.""" |
| 125 | |
| 126 | framework_id: str |
| 127 | kind: str |
| 128 | metadata: _StrMeta |
| 129 | |
| 130 | |
| 131 | class _ImpactJson(TypedDict, total=False): |
| 132 | """JSON payload for ``muse code impact`` output.""" |
| 133 | |
| 134 | mode: str |
| 135 | address: str |
| 136 | target_name: str |
| 137 | commit_id: str |
| 138 | depth_limit: int |
| 139 | file_filter: str | None |
| 140 | blast_radius: _BlastRadius |
| 141 | total: int |
| 142 | entry_points: list[_EntryPointJson] |
| 143 | compare_commit_id: str |
| 144 | added_callers: list[str] |
| 145 | removed_callers: list[str] |
| 146 | net_change: int |
| 147 | |
| 148 | |
| 149 | def _flat_addrs(blast: dict[int, list[str]]) -> set[str]: |
| 150 | """Return the flat set of all addresses across all depths.""" |
| 151 | return {addr for addrs in blast.values() for addr in addrs} |
| 152 | |
| 153 | |
| 154 | # --------------------------------------------------------------------------- |
| 155 | # Knowtation domain path |
| 156 | # --------------------------------------------------------------------------- |
| 157 | |
| 158 | |
| 159 | def _run_knowtation_impact( |
| 160 | *, |
| 161 | root: pathlib.Path, |
| 162 | address: str, |
| 163 | forward: bool, |
| 164 | depth: int, |
| 165 | count_only: bool, |
| 166 | as_json: bool, |
| 167 | include_attachments: bool = True, |
| 168 | ) -> None: |
| 169 | """Knowtation-domain path for ``muse code impact``. |
| 170 | |
| 171 | Treats *address* as a vault-relative note path and walks the link graph |
| 172 | transitively. Default direction is reverse BFS (notes that depend on the |
| 173 | target); pass ``forward=True`` for forward BFS (notes the target links |
| 174 | to). |
| 175 | |
| 176 | Args: |
| 177 | root: Repository root. |
| 178 | address: Vault-relative note path. |
| 179 | forward: Forward BFS instead of reverse. |
| 180 | depth: BFS depth cap (``<= 0`` interpreted as unlimited). |
| 181 | count_only: Emit only the total count of notes. |
| 182 | as_json: Emit JSON instead of text. |
| 183 | include_attachments: Honour the include_attachments flag. |
| 184 | """ |
| 185 | from muse.plugins.knowtation.code_intel import note_impact |
| 186 | from muse.plugins.knowtation.link_index import build_link_index |
| 187 | from muse.plugins.knowtation._query import is_note |
| 188 | |
| 189 | if not is_note(address): |
| 190 | print( |
| 191 | f"❌ '{address}' is not a Markdown note. " |
| 192 | "Knowtation impact requires a .md / .markdown / .mdx path.", |
| 193 | file=sys.stderr, |
| 194 | ) |
| 195 | raise SystemExit(ExitCode.USER_ERROR) |
| 196 | |
| 197 | candidate = root / address |
| 198 | if not candidate.is_file(): |
| 199 | print(f"❌ Note '{address}' not found in vault.", file=sys.stderr) |
| 200 | raise SystemExit(ExitCode.USER_ERROR) |
| 201 | |
| 202 | try: |
| 203 | index = build_link_index(root) |
| 204 | except ValueError as exc: |
| 205 | print(f"❌ Could not index vault: {exc}", file=sys.stderr) |
| 206 | raise SystemExit(ExitCode.USER_ERROR) |
| 207 | |
| 208 | bfs_depth = depth if depth > 0 else 0 |
| 209 | result = note_impact( |
| 210 | index, |
| 211 | address, |
| 212 | forward=forward, |
| 213 | depth=bfs_depth, |
| 214 | include_attachments=include_attachments, |
| 215 | ) |
| 216 | |
| 217 | if count_only: |
| 218 | print(result["total"]) |
| 219 | return |
| 220 | |
| 221 | if as_json: |
| 222 | result["domain"] = "knowtation" |
| 223 | print(json.dumps(result, indent=2)) |
| 224 | return |
| 225 | |
| 226 | direction = "Forward dependencies of" if forward else "Blast radius (callers) of" |
| 227 | print(f"\n{direction} {sanitize_display(address)}") |
| 228 | print("─" * 62) |
| 229 | if not result["by_depth"]: |
| 230 | msg = "no outgoing links" if forward else "no incoming links" |
| 231 | print(f" ({msg})") |
| 232 | else: |
| 233 | for d_str in sorted(result["by_depth"], key=int): |
| 234 | notes = result["by_depth"][d_str] |
| 235 | print(f"\n depth {d_str}:") |
| 236 | for note in notes: |
| 237 | print(f" {sanitize_display(note)}") |
| 238 | print(f"\n{result['total']} note(s) across {len(result['by_depth'])} depth(s)") |
| 239 | |
| 240 | |
| 241 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 242 | """Register the impact subcommand.""" |
| 243 | parser = subparsers.add_parser( |
| 244 | "impact", |
| 245 | help="Show the transitive blast-radius of changing a symbol.", |
| 246 | description=__doc__, |
| 247 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 248 | ) |
| 249 | parser.add_argument( |
| 250 | "address", metavar="ADDRESS", |
| 251 | help='Symbol address, e.g. "src/billing.py::compute_invoice_total".', |
| 252 | ) |
| 253 | parser.add_argument( |
| 254 | "--depth", "-d", type=int, default=None, metavar="N", |
| 255 | help="Maximum BFS depth (default: 5, max: 50). Use --transitive for full traversal.", |
| 256 | ) |
| 257 | parser.add_argument( |
| 258 | "--commit", "-c", default=None, metavar="REF", dest="ref", |
| 259 | help="Analyse a historical snapshot instead of HEAD.", |
| 260 | ) |
| 261 | parser.add_argument( |
| 262 | "--compare", default=None, metavar="REF", dest="compare_ref", |
| 263 | help="Diff blast radius against this commit reference.", |
| 264 | ) |
| 265 | parser.add_argument( |
| 266 | "--forward", action="store_true", dest="forward", |
| 267 | help="Show callees (dependencies) instead of callers.", |
| 268 | ) |
| 269 | parser.add_argument( |
| 270 | "--file", "-f", default=None, metavar="PATH", dest="file_filter", |
| 271 | help="Restrict blast-radius output to callers from this file.", |
| 272 | ) |
| 273 | parser.add_argument( |
| 274 | "--count", action="store_true", dest="count_only", |
| 275 | help="Print only the total count of matching symbols.", |
| 276 | ) |
| 277 | parser.add_argument( |
| 278 | "--json", action="store_true", dest="as_json", |
| 279 | help="Emit results as JSON.", |
| 280 | ) |
| 281 | parser.add_argument( |
| 282 | "--include-attachments", |
| 283 | action=argparse.BooleanOptionalAction, |
| 284 | default=True, |
| 285 | dest="include_attachments", |
| 286 | help=( |
| 287 | "Knowtation only: follow mist-attachment co-references during impact " |
| 288 | "(default on). Use --no-include-attachments to disable." |
| 289 | ), |
| 290 | ) |
| 291 | parser.set_defaults(func=run) |
| 292 | |
| 293 | |
| 294 | def run(args: argparse.Namespace) -> None: |
| 295 | """Show the transitive blast-radius of changing a symbol. |
| 296 | |
| 297 | Builds the reverse call graph for the committed snapshot, then BFS-walks |
| 298 | it from the target symbol outwards. Depth 1 = direct callers; depth 2 = |
| 299 | callers of callers; and so on until no new callers are found. |
| 300 | |
| 301 | With ``--compare`` the blast radius is diffed against another commit, |
| 302 | revealing exactly which callers were added or removed. |
| 303 | |
| 304 | With ``--forward`` the direction inverts: depth 1 = direct callees. |
| 305 | |
| 306 | Python only (call-graph analysis uses stdlib ``ast``). |
| 307 | """ |
| 308 | address: str = args.address |
| 309 | _depth_raw: int | None = args.depth |
| 310 | depth: int = clamp_int(_depth_raw, 1, 50, "depth") if _depth_raw is not None else 5 |
| 311 | ref: str | None = args.ref |
| 312 | compare_ref: str | None = args.compare_ref |
| 313 | forward: bool = args.forward |
| 314 | file_filter: str | None = args.file_filter |
| 315 | count_only: bool = args.count_only |
| 316 | as_json: bool = args.as_json |
| 317 | |
| 318 | if forward and compare_ref: |
| 319 | print("❌ --forward and --compare are mutually exclusive.", file=sys.stderr) |
| 320 | raise SystemExit(ExitCode.USER_ERROR) |
| 321 | |
| 322 | root = require_repo() |
| 323 | repo_id = read_repo_id(root) |
| 324 | branch = read_current_branch(root) |
| 325 | |
| 326 | try: |
| 327 | active_domain = read_domain(root) |
| 328 | except Exception: |
| 329 | active_domain = "code" |
| 330 | |
| 331 | if active_domain == "knowtation": |
| 332 | _run_knowtation_impact( |
| 333 | root=root, |
| 334 | address=address, |
| 335 | forward=forward, |
| 336 | depth=depth, |
| 337 | count_only=count_only, |
| 338 | as_json=as_json, |
| 339 | include_attachments=getattr(args, "include_attachments", True), |
| 340 | ) |
| 341 | return |
| 342 | |
| 343 | lang = language_of(address.split("::")[0]) if "::" in address else "" |
| 344 | if lang and lang != "Python": |
| 345 | print( |
| 346 | f"⚠️ Impact analysis is currently Python-only. '{address}' is {lang}.", |
| 347 | file=sys.stderr, |
| 348 | ) |
| 349 | raise SystemExit(ExitCode.USER_ERROR) |
| 350 | |
| 351 | commit = resolve_commit_ref(root, repo_id, branch, ref) |
| 352 | if commit is None: |
| 353 | print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr) |
| 354 | raise SystemExit(ExitCode.USER_ERROR) |
| 355 | |
| 356 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} |
| 357 | |
| 358 | cache = load_symbol_cache(root) |
| 359 | |
| 360 | target_name = address.split("::")[-1].split(".")[-1] if "::" in address else address |
| 361 | |
| 362 | if forward: |
| 363 | fwd: ForwardGraph = build_forward_graph(root, manifest, cache=cache) |
| 364 | callee_map = transitive_callees(address, fwd, max_depth=depth) |
| 365 | cache.save() |
| 366 | _render_forward( |
| 367 | address=address, |
| 368 | target_name=target_name, |
| 369 | commit=commit, |
| 370 | callee_map=callee_map, |
| 371 | depth_limit=depth, |
| 372 | count_only=count_only, |
| 373 | as_json=as_json, |
| 374 | ) |
| 375 | return |
| 376 | |
| 377 | rev: ReverseGraph = build_reverse_graph(root, manifest, cache=cache) |
| 378 | implicit: ImplicitEdgeGraph = build_implicit_edge_graph(root, manifest, cache=cache) |
| 379 | blast = transitive_callers(target_name, rev, max_depth=depth) |
| 380 | |
| 381 | if file_filter: |
| 382 | blast = { |
| 383 | d: [a for a in addrs if "::" in a and a.split("::")[0] == file_filter] |
| 384 | for d, addrs in blast.items() |
| 385 | } |
| 386 | blast = {d: addrs for d, addrs in blast.items() if addrs} |
| 387 | |
| 388 | compare_commit: CommitRecord | None = None |
| 389 | added: set[str] = set() |
| 390 | removed: set[str] = set() |
| 391 | |
| 392 | if compare_ref: |
| 393 | compare_commit = resolve_commit_ref(root, repo_id, branch, compare_ref) |
| 394 | if compare_commit is None: |
| 395 | print(f"❌ --compare commit '{compare_ref}' not found.", file=sys.stderr) |
| 396 | raise SystemExit(ExitCode.USER_ERROR) |
| 397 | compare_manifest = get_commit_snapshot_manifest(root, compare_commit.commit_id) or {} |
| 398 | compare_rev: ReverseGraph = build_reverse_graph(root, compare_manifest, cache=cache) |
| 399 | compare_blast = transitive_callers(target_name, compare_rev, max_depth=depth) |
| 400 | |
| 401 | if file_filter: |
| 402 | compare_blast = { |
| 403 | d: [a for a in addrs if "::" in a and a.split("::")[0] == file_filter] |
| 404 | for d, addrs in compare_blast.items() |
| 405 | } |
| 406 | compare_blast = {d: addrs for d, addrs in compare_blast.items() if addrs} |
| 407 | |
| 408 | current_flat = _flat_addrs(blast) |
| 409 | compare_flat = _flat_addrs(compare_blast) |
| 410 | added = current_flat - compare_flat |
| 411 | removed = compare_flat - current_flat |
| 412 | |
| 413 | cache.save() |
| 414 | |
| 415 | entry_points = implicit.get(address, []) |
| 416 | |
| 417 | _render_reverse( |
| 418 | address=address, |
| 419 | target_name=target_name, |
| 420 | commit=commit, |
| 421 | blast=blast, |
| 422 | entry_points=entry_points, |
| 423 | depth_limit=depth, |
| 424 | file_filter=file_filter, |
| 425 | compare_commit=compare_commit, |
| 426 | added=added, |
| 427 | removed=removed, |
| 428 | count_only=count_only, |
| 429 | as_json=as_json, |
| 430 | ) |
| 431 | |
| 432 | |
| 433 | # --------------------------------------------------------------------------- |
| 434 | # Renderers |
| 435 | # --------------------------------------------------------------------------- |
| 436 | |
| 437 | |
| 438 | def _render_forward( |
| 439 | *, |
| 440 | address: str, |
| 441 | target_name: str, |
| 442 | commit: CommitRecord, |
| 443 | callee_map: dict[int, list[str]], |
| 444 | depth_limit: int, |
| 445 | count_only: bool, |
| 446 | as_json: bool, |
| 447 | ) -> None: |
| 448 | """Render forward (callees) analysis.""" |
| 449 | total = sum(len(v) for v in callee_map.values()) |
| 450 | |
| 451 | if count_only and not as_json: |
| 452 | print(total) |
| 453 | return |
| 454 | |
| 455 | if as_json: |
| 456 | print(json.dumps( |
| 457 | { |
| 458 | "mode": "forward", |
| 459 | "address": address, |
| 460 | "target_name": target_name, |
| 461 | "commit_id": commit.commit_id, |
| 462 | "depth_limit": depth_limit, |
| 463 | "callees": {str(d): names for d, names in sorted(callee_map.items())}, |
| 464 | "total": total, |
| 465 | }, |
| 466 | indent=2, |
| 467 | )) |
| 468 | return |
| 469 | |
| 470 | print(f"\nDependency fan-out: {sanitize_display(address)}") |
| 471 | print("─" * 62) |
| 472 | |
| 473 | if not callee_map: |
| 474 | print(f"\n ('{target_name}' calls nothing detectable — leaf function or dynamic dispatch)") |
| 475 | return |
| 476 | |
| 477 | for d in sorted(callee_map.keys()): |
| 478 | label = "direct callees" if d == 1 else f"depth-{d} callees" |
| 479 | print(f"\nDepth {d} — {label} ({len(callee_map[d])}):") |
| 480 | for name in sorted(callee_map[d]): |
| 481 | print(f" {sanitize_display(name)}") |
| 482 | |
| 483 | print("\n" + "─" * 62) |
| 484 | print(f"Total dependency fan-out: {total} callee(s)") |
| 485 | print("\nNote: analysis covers Python call-sites only.") |
| 486 | |
| 487 | |
| 488 | def _render_reverse( |
| 489 | *, |
| 490 | address: str, |
| 491 | target_name: str, |
| 492 | commit: CommitRecord, |
| 493 | blast: dict[int, list[str]], |
| 494 | entry_points: list[ImplicitEntryEdge], |
| 495 | depth_limit: int, |
| 496 | file_filter: str | None, |
| 497 | compare_commit: CommitRecord | None, |
| 498 | added: set[str], |
| 499 | removed: set[str], |
| 500 | count_only: bool, |
| 501 | as_json: bool, |
| 502 | ) -> None: |
| 503 | """Render reverse (callers / blast-radius) analysis.""" |
| 504 | total = sum(len(v) for v in blast.values()) |
| 505 | |
| 506 | if count_only and not as_json: |
| 507 | print(total) |
| 508 | return |
| 509 | |
| 510 | if as_json: |
| 511 | payload = _ImpactJson( |
| 512 | mode="reverse", |
| 513 | address=address, |
| 514 | target_name=target_name, |
| 515 | commit_id=commit.commit_id, |
| 516 | depth_limit=depth_limit, |
| 517 | file_filter=file_filter, |
| 518 | blast_radius={str(d): list(addrs) for d, addrs in sorted(blast.items())}, |
| 519 | total=total, |
| 520 | ) |
| 521 | if entry_points: |
| 522 | payload["entry_points"] = [ |
| 523 | _EntryPointJson( |
| 524 | framework_id=ep.framework_id, |
| 525 | kind=ep.kind, |
| 526 | metadata=ep.metadata, |
| 527 | ) |
| 528 | for ep in entry_points |
| 529 | ] |
| 530 | if compare_commit is not None: |
| 531 | payload["compare_commit_id"] = compare_commit.commit_id |
| 532 | payload["added_callers"] = sorted(added) |
| 533 | payload["removed_callers"] = sorted(removed) |
| 534 | payload["net_change"] = len(added) - len(removed) |
| 535 | print(json.dumps(payload, indent=2)) |
| 536 | return |
| 537 | |
| 538 | print(f"\nImpact analysis: {sanitize_display(address)}") |
| 539 | if file_filter: |
| 540 | print(f"(filtered to callers in: {file_filter})") |
| 541 | print("─" * 62) |
| 542 | |
| 543 | if entry_points: |
| 544 | print("\n Framework entry point — externally reachable via:") |
| 545 | for ep in entry_points: |
| 546 | meta_str = " ".join(f"{k}={v}" for k, v in ep.metadata.items() if v) |
| 547 | print(f" [{ep.framework_id}] kind={ep.kind} {meta_str}") |
| 548 | |
| 549 | if not blast: |
| 550 | if entry_points: |
| 551 | print("\n (no explicit callers in user code — wired by the framework above)") |
| 552 | else: |
| 553 | print( |
| 554 | f"\n (no callers detected — '{target_name}' may be an entry point or dead code)" |
| 555 | ) |
| 556 | print("\n Note: analysis covers Python only; external callers are not detected.") |
| 557 | else: |
| 558 | all_files: set[str] = set() |
| 559 | for d in sorted(blast.keys()): |
| 560 | callers = blast[d] |
| 561 | if d == 1: |
| 562 | label = "direct callers" |
| 563 | elif d == 2: |
| 564 | label = "callers of callers" |
| 565 | else: |
| 566 | label = f"depth-{d} callers" |
| 567 | print(f"\nDepth {d} — {label} ({len(callers)}):") |
| 568 | for addr in sorted(callers): |
| 569 | marker = " ← NEW" if addr in added else "" |
| 570 | print(f" {sanitize_display(addr)}{marker}") |
| 571 | if "::" in addr: |
| 572 | all_files.add(addr.split("::")[0]) |
| 573 | |
| 574 | print("\n" + "─" * 62) |
| 575 | file_label = "file" if len(all_files) == 1 else "files" |
| 576 | print(f"Total blast radius: {total} symbol(s) across {len(all_files)} {file_label}") |
| 577 | if total >= 10: |
| 578 | print("🔴 High impact — add tests before changing this symbol.") |
| 579 | elif total >= 3: |
| 580 | print("🟡 Medium impact — review callers before changing this symbol.") |
| 581 | else: |
| 582 | print("🟢 Low impact — change is well-contained.") |
| 583 | print( |
| 584 | "\nNote: analysis covers Python call-sites only." |
| 585 | " Dynamic dispatch (getattr, decorators) is not detected." |
| 586 | ) |
| 587 | |
| 588 | if compare_commit is not None: |
| 589 | print(f"\nBlast-radius diff vs {compare_commit.commit_id[:8]}:") |
| 590 | print("─" * 62) |
| 591 | if not added and not removed: |
| 592 | print(" No change in blast radius.") |
| 593 | else: |
| 594 | net = len(added) - len(removed) |
| 595 | sign = "+" if net >= 0 else "" |
| 596 | print(f" Net change: {sign}{net} caller(s)") |
| 597 | if added: |
| 598 | print(f"\n Added ({len(added)}):") |
| 599 | for a in sorted(added): |
| 600 | print(f" + {a}") |
| 601 | if removed: |
| 602 | print(f"\n Removed ({len(removed)}):") |
| 603 | for r in sorted(removed): |
| 604 | print(f" - {r}") |