blast_risk.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """muse code blast-risk — composite pre-release risk leaderboard. |
| 2 | |
| 3 | Synthesises four dimensions that Muse tracks independently into a single |
| 4 | ranked risk score for every symbol in the snapshot: |
| 5 | |
| 6 | Impact (40 %) — how many *other* symbols directly reference this one |
| 7 | via import, bare name, or attribute access. A bug in a |
| 8 | highly-referenced symbol touches many callsites. |
| 9 | |
| 10 | Churn (30 %) — how many commits in the analysed range modified this |
| 11 | symbol. Frequently-changing code has higher defect |
| 12 | probability (same signal as ``muse code hotspots``). |
| 13 | |
| 14 | Test gap (20 %) — fraction of callers that live in *production* files. |
| 15 | Symbols whose callers are untested carry hidden risk. |
| 16 | |
| 17 | Coupling (10 %) — how many other files co-change with this symbol's file. |
| 18 | High coupling amplifies the blast radius. |
| 19 | |
| 20 | Each dimension is normalised 0–100 across the full symbol set, then |
| 21 | combined with the weights above into a final risk score 0–100. |
| 22 | |
| 23 | No other tool can produce this score because no other tool simultaneously |
| 24 | knows the call graph, the churn history, which callers are tests, and the |
| 25 | file co-change graph. |
| 26 | |
| 27 | Usage:: |
| 28 | |
| 29 | muse code blast-risk |
| 30 | muse code blast-risk --top 20 |
| 31 | muse code blast-risk --since a3f2c9 |
| 32 | muse code blast-risk --kind function |
| 33 | muse code blast-risk --file billing.py |
| 34 | muse code blast-risk --explain billing.py::Invoice.compute_total |
| 35 | muse code blast-risk --min-risk 50 |
| 36 | muse code blast-risk --json |
| 37 | |
| 38 | Output:: |
| 39 | |
| 40 | Symbol blast-risk — HEAD (304 commits · 378 symbols) |
| 41 | Scoring: impact×40% + churn×30% + test-gap×20% + coupling×10% |
| 42 | |
| 43 | # ADDRESS RISK IMPACT CHURN TEST-GAP COUPLING |
| 44 | 1 muse/core/store.py::resolve_commit_ref 94 high 12 87 % high |
| 45 | 2 muse/core/errors.py::ExitCode 87 high 8 72 % med |
| 46 | 3 billing.py::Invoice.compute_total 81 med 15 100 % low |
| 47 | |
| 48 | ⚠️ High-risk symbols are prime targets for test coverage and refactoring. |
| 49 | |
| 50 | JSON output (--json):: |
| 51 | |
| 52 | { |
| 53 | "ref": "HEAD", |
| 54 | "commits_analysed": 304, |
| 55 | "truncated": false, |
| 56 | "filters": { "kind": null, "file": null, "min_risk": 0, "since": null }, |
| 57 | "weights": { "impact": 0.40, "churn": 0.30, "test_gap": 0.20, "coupling": 0.10 }, |
| 58 | "symbols": [ |
| 59 | { |
| 60 | "address": "muse/core/store.py::resolve_commit_ref", |
| 61 | "kind": "function", |
| 62 | "file": "muse/core/store.py", |
| 63 | "risk": 94, |
| 64 | "impact_raw": 47, |
| 65 | "churn_raw": 12, |
| 66 | "test_gap_raw": 0.87, |
| 67 | "coupling_raw": 9, |
| 68 | "impact_score": 100, |
| 69 | "churn_score": 80, |
| 70 | "test_gap_score": 87, |
| 71 | "coupling_score": 75 |
| 72 | } |
| 73 | ] |
| 74 | } |
| 75 | |
| 76 | --explain output:: |
| 77 | |
| 78 | Blast-risk breakdown — muse/core/store.py::resolve_commit_ref |
| 79 | |
| 80 | Risk score: 94 / 100 |
| 81 | Kind: function |
| 82 | |
| 83 | Impact (40 %) score: 100 / 100 raw: 47 direct callers |
| 84 | Churn (30 %) score: 80 / 100 raw: 12 changes in 304 commits |
| 85 | Test gap (20 %) score: 87 / 100 raw: 41 / 47 callers are production (not test) |
| 86 | Coupling (10 %) score: 75 / 100 raw: 9 files co-change with muse/core/store.py |
| 87 | |
| 88 | Top callers (production): |
| 89 | muse/cli/commands/compare.py::run |
| 90 | muse/cli/commands/blame.py::run |
| 91 | muse/plugins/code/_query.py::walk_commits_bfs |
| 92 | … 38 more |
| 93 | """ |
| 94 | |
| 95 | import ast |
| 96 | import argparse |
| 97 | import json |
| 98 | import logging |
| 99 | import pathlib |
| 100 | import re |
| 101 | import sys |
| 102 | from collections.abc import Callable |
| 103 | from typing import TypedDict |
| 104 | |
| 105 | from muse.core.envelope import EnvelopeJson, make_envelope |
| 106 | from muse.core.errors import ExitCode |
| 107 | from muse.core.repo import require_repo |
| 108 | from muse.core.types import Manifest |
| 109 | from muse.core.refs import read_current_branch |
| 110 | from muse.core.commits import resolve_commit_ref |
| 111 | from muse.core.snapshots import get_commit_snapshot_manifest |
| 112 | from muse.core.symbol_cache import load_symbol_cache |
| 113 | from muse.core.timing import start_timer |
| 114 | from muse.domain import DomainOp |
| 115 | from muse.plugins.code._query import ( |
| 116 | file_pairs, |
| 117 | flat_symbol_ops, |
| 118 | language_of, |
| 119 | symbols_for_snapshot, |
| 120 | touched_files, |
| 121 | walk_commits_bfs, |
| 122 | ) |
| 123 | from muse.core.validation import MAX_AST_BYTES, clamp_int, sanitize_display |
| 124 | |
| 125 | logger = logging.getLogger(__name__) |
| 126 | |
| 127 | type CallerMap = dict[str, list[str]] # name → list of caller file_paths |
| 128 | type ChurnMap = dict[str, int] # file_path or address → churn count |
| 129 | type CouplingMap = dict[str, int] # file_path → co-change partner count |
| 130 | type SymbolRecordMap = dict[str, tuple[str, str]] # address → (name, kind) |
| 131 | type CallerListMap = dict[str, list[str]] # address → list of caller files |
| 132 | type TestGapMap = dict[str, float] # address → test gap fraction |
| 133 | |
| 134 | # ── Constants ────────────────────────────────────────────────────────────────── |
| 135 | |
| 136 | _DEFAULT_TOP = 20 |
| 137 | _DEFAULT_MAX_COMMITS = 10_000 |
| 138 | _MAX_FILES_PER_COMMIT = 50 # skip mass-edit commits (noisy for coupling) |
| 139 | |
| 140 | _W_IMPACT: float = 0.40 |
| 141 | _W_CHURN: float = 0.30 |
| 142 | _W_TEST_GAP: float = 0.20 |
| 143 | _W_COUPLING: float = 0.10 |
| 144 | |
| 145 | # Test-file heuristics — callers in these files are considered "tested". |
| 146 | _TEST_PATTERNS: tuple[re.Pattern[str], ...] = ( |
| 147 | re.compile(r"(^|/)test_"), |
| 148 | re.compile(r"_test\.py$"), |
| 149 | re.compile(r"(^|/)tests/"), |
| 150 | re.compile(r"(^|/)spec/"), |
| 151 | re.compile(r"(^|/)conftest\.py$"), |
| 152 | ) |
| 153 | |
| 154 | # ── TypedDicts ───────────────────────────────────────────────────────────────── |
| 155 | |
| 156 | class _SymbolRisk(TypedDict): |
| 157 | """Fully-scored risk record for one symbol.""" |
| 158 | |
| 159 | address: str |
| 160 | kind: str |
| 161 | file: str |
| 162 | risk: int # final composite score 0-100 |
| 163 | impact_raw: int # direct caller count |
| 164 | churn_raw: int # commit-change count |
| 165 | test_gap_raw: float # fraction of callers that are production (0-1) |
| 166 | coupling_raw: int # co-change partner count for the file |
| 167 | impact_score: int # normalised 0-100 |
| 168 | churn_score: int # normalised 0-100 |
| 169 | test_gap_score: int # = round(test_gap_raw * 100) |
| 170 | coupling_score: int # normalised 0-100 |
| 171 | # Only present in --explain output. |
| 172 | production_callers: list[str] |
| 173 | test_callers: list[str] |
| 174 | |
| 175 | class _SymbolRiskJson(TypedDict): |
| 176 | """JSON-serialisable risk record — ``production_callers``/``test_callers`` excluded.""" |
| 177 | |
| 178 | address: str |
| 179 | kind: str |
| 180 | file: str |
| 181 | risk: int |
| 182 | risk_label: str |
| 183 | impact_raw: int |
| 184 | churn_raw: int |
| 185 | test_gap_raw: float |
| 186 | coupling_raw: int |
| 187 | impact_score: int |
| 188 | churn_score: int |
| 189 | test_gap_score: int |
| 190 | coupling_score: int |
| 191 | |
| 192 | class _BlastRiskFilters(TypedDict): |
| 193 | kind: str | None |
| 194 | file: str | None |
| 195 | min_risk: int |
| 196 | since: str | None |
| 197 | top: int |
| 198 | max_commits: int |
| 199 | |
| 200 | class _BlastRiskWeights(TypedDict): |
| 201 | impact: float |
| 202 | churn: float |
| 203 | test_gap: float |
| 204 | coupling: float |
| 205 | |
| 206 | class _BlastRiskOutput(EnvelopeJson): |
| 207 | """JSON payload for ``muse code blast-risk`` table output.""" |
| 208 | |
| 209 | ref: str |
| 210 | commits_analysed: int |
| 211 | truncated: bool |
| 212 | filters: _BlastRiskFilters |
| 213 | weights: _BlastRiskWeights |
| 214 | symbols: list[_SymbolRiskJson] |
| 215 | |
| 216 | class _ExplainJson(EnvelopeJson): |
| 217 | """JSON payload for ``muse code blast-risk --explain`` output.""" |
| 218 | |
| 219 | address: str |
| 220 | kind: str |
| 221 | file: str |
| 222 | risk: int |
| 223 | impact_raw: int |
| 224 | churn_raw: int |
| 225 | test_gap_raw: float |
| 226 | coupling_raw: int |
| 227 | impact_score: int |
| 228 | churn_score: int |
| 229 | test_gap_score: int |
| 230 | coupling_score: int |
| 231 | production_callers: list[str] |
| 232 | test_callers: list[str] |
| 233 | |
| 234 | # ── Helpers ──────────────────────────────────────────────────────────────────── |
| 235 | |
| 236 | def _is_test_file(path: str) -> bool: |
| 237 | return any(p.search(path) for p in _TEST_PATTERNS) |
| 238 | |
| 239 | def _normalise_score(raw: int | float, maximum: int | float) -> int: |
| 240 | if maximum <= 0: |
| 241 | return 0 |
| 242 | return round(min(100, (raw / maximum) * 100)) |
| 243 | |
| 244 | # ── Phase 1: caller map ──────────────────────────────────────────────────────── |
| 245 | |
| 246 | def _build_caller_map( |
| 247 | root: pathlib.Path, |
| 248 | manifest: Manifest, |
| 249 | known_names: frozenset[str], |
| 250 | ) -> CallerMap: |
| 251 | """Return ``name → [caller_file_path, ...]`` by scanning all Python files. |
| 252 | |
| 253 | Uses AST-level ``ast.Name`` and ``ast.Attribute`` nodes so only |
| 254 | identifier references are counted — string literals and comments are |
| 255 | never included. |
| 256 | |
| 257 | Only names that appear in *known_names* are tracked to avoid building an |
| 258 | arbitrarily large map. |
| 259 | """ |
| 260 | caller_map: CallerMap = {} |
| 261 | for file_path in sorted(manifest): |
| 262 | if language_of(file_path) != "Python": |
| 263 | continue |
| 264 | disk_path = root / file_path |
| 265 | if not disk_path.is_file(): |
| 266 | continue |
| 267 | try: |
| 268 | source = disk_path.read_bytes() |
| 269 | if len(source) > MAX_AST_BYTES: |
| 270 | return {} |
| 271 | tree = ast.parse(source, filename=file_path) |
| 272 | except (OSError, SyntaxError): |
| 273 | continue |
| 274 | for node in ast.walk(tree): |
| 275 | name: str | None = None |
| 276 | if isinstance(node, ast.Name) and node.id in known_names: |
| 277 | name = node.id |
| 278 | elif isinstance(node, ast.Attribute) and node.attr in known_names: |
| 279 | name = node.attr |
| 280 | if name is not None: |
| 281 | caller_map.setdefault(name, []).append(file_path) |
| 282 | return caller_map |
| 283 | |
| 284 | # ── Phase 2: churn + coupling in one BFS pass ───────────────────────────────── |
| 285 | |
| 286 | def _collect_churn_and_coupling( |
| 287 | root: pathlib.Path, |
| 288 | head_commit_id: str, |
| 289 | stop_at: str | None, |
| 290 | max_commits: int, |
| 291 | ) -> tuple[ChurnMap, CouplingMap, int, bool]: |
| 292 | """Single BFS pass collecting churn and file-coupling simultaneously. |
| 293 | |
| 294 | Returns: |
| 295 | churn — ``address → commit-change count`` |
| 296 | file_coupling — ``file_path → number of distinct co-change partners`` |
| 297 | commits_analysed — number of commits walked |
| 298 | truncated — whether the BFS hit max_commits |
| 299 | """ |
| 300 | commits, truncated = walk_commits_bfs( |
| 301 | root, head_commit_id, max_commits, stop_at_commit_id=stop_at |
| 302 | ) |
| 303 | |
| 304 | churn: ChurnMap = {} |
| 305 | pair_counts: dict[tuple[str, str], int] = {} |
| 306 | |
| 307 | for commit in commits: |
| 308 | if commit.structured_delta is None: |
| 309 | continue |
| 310 | ops: list[DomainOp] = commit.structured_delta["ops"] |
| 311 | |
| 312 | # ── churn ────────────────────────────────────────────────────────── |
| 313 | for op in flat_symbol_ops(ops): |
| 314 | addr: str = op["address"] |
| 315 | if "::import::" not in addr: # exclude import pseudo-symbols |
| 316 | churn[addr] = churn.get(addr, 0) + 1 |
| 317 | |
| 318 | # ── coupling ─────────────────────────────────────────────────────── |
| 319 | files = touched_files(ops) |
| 320 | if 2 <= len(files) <= _MAX_FILES_PER_COMMIT: |
| 321 | for fa, fb in file_pairs(files): |
| 322 | pair_counts[(fa, fb)] = pair_counts.get((fa, fb), 0) + 1 |
| 323 | |
| 324 | # Collapse pair counts to per-file partner counts. |
| 325 | file_coupling: CouplingMap = {} |
| 326 | for (fa, fb), cnt in pair_counts.items(): |
| 327 | if cnt >= 1: |
| 328 | file_coupling[fa] = file_coupling.get(fa, 0) + 1 |
| 329 | file_coupling[fb] = file_coupling.get(fb, 0) + 1 |
| 330 | |
| 331 | return churn, file_coupling, len(commits), truncated |
| 332 | |
| 333 | # ── Phase 3: score + rank ────────────────────────────────────────────────────── |
| 334 | |
| 335 | def _score_symbols( |
| 336 | symbol_records: SymbolRecordMap, |
| 337 | caller_map: CallerMap, |
| 338 | churn: ChurnMap, |
| 339 | file_coupling: CouplingMap, |
| 340 | kind_filter: str | None, |
| 341 | file_filter: str | None, |
| 342 | min_risk: int, |
| 343 | top: int, |
| 344 | include_callers: bool, |
| 345 | ) -> list[_SymbolRisk]: |
| 346 | """Compute normalised scores and return the ranked list.""" |
| 347 | |
| 348 | # ── Gather raw values ────────────────────────────────────────────────── |
| 349 | raw_impacts: ChurnMap = {} |
| 350 | raw_test_gaps: TestGapMap = {} |
| 351 | prod_callers_map: CallerListMap = {} |
| 352 | test_callers_map: CallerListMap = {} |
| 353 | |
| 354 | for address, (name, kind) in symbol_records.items(): |
| 355 | file_path = address.split("::")[0] |
| 356 | callers = caller_map.get(name, []) |
| 357 | |
| 358 | prod = [c for c in callers if not _is_test_file(c) and c != file_path] |
| 359 | tests = [c for c in callers if _is_test_file(c)] |
| 360 | total = len(prod) + len(tests) |
| 361 | |
| 362 | raw_impacts[address] = total |
| 363 | raw_test_gaps[address] = len(prod) / total if total > 0 else 0.0 |
| 364 | prod_callers_map[address] = sorted(set(prod)) |
| 365 | test_callers_map[address] = sorted(set(tests)) |
| 366 | |
| 367 | # ── Normalisation maximums ───────────────────────────────────────────── |
| 368 | max_impact = max(raw_impacts.values(), default=1) or 1 |
| 369 | max_churn = max(churn.values(), default=1) or 1 |
| 370 | max_coupling = max(file_coupling.values(), default=1) or 1 |
| 371 | |
| 372 | # ── Score every symbol ───────────────────────────────────────────────── |
| 373 | scored: list[_SymbolRisk] = [] |
| 374 | for address, (name, kind) in symbol_records.items(): |
| 375 | file_path = address.split("::")[0] |
| 376 | |
| 377 | if kind_filter and kind.lower() != kind_filter: |
| 378 | continue |
| 379 | if file_filter and not ( |
| 380 | file_path == file_filter or file_path.endswith(f"/{file_filter}") |
| 381 | ): |
| 382 | continue |
| 383 | |
| 384 | impact_raw = raw_impacts[address] |
| 385 | churn_raw = churn.get(address, 0) |
| 386 | test_gap_raw = raw_test_gaps[address] |
| 387 | coupling_raw = file_coupling.get(file_path, 0) |
| 388 | |
| 389 | impact_s = _normalise_score(impact_raw, max_impact) |
| 390 | churn_s = _normalise_score(churn_raw, max_churn) |
| 391 | test_gap_s = round(test_gap_raw * 100) |
| 392 | coupling_s = _normalise_score(coupling_raw, max_coupling) |
| 393 | |
| 394 | risk = round( |
| 395 | _W_IMPACT * impact_s |
| 396 | + _W_CHURN * churn_s |
| 397 | + _W_TEST_GAP * test_gap_s |
| 398 | + _W_COUPLING * coupling_s |
| 399 | ) |
| 400 | |
| 401 | if risk < min_risk: |
| 402 | continue |
| 403 | |
| 404 | scored.append(_SymbolRisk( |
| 405 | address=address, |
| 406 | kind=kind, |
| 407 | file=file_path, |
| 408 | risk=risk, |
| 409 | impact_raw=impact_raw, |
| 410 | churn_raw=churn_raw, |
| 411 | test_gap_raw=test_gap_raw, |
| 412 | coupling_raw=coupling_raw, |
| 413 | impact_score=impact_s, |
| 414 | churn_score=churn_s, |
| 415 | test_gap_score=test_gap_s, |
| 416 | coupling_score=coupling_s, |
| 417 | production_callers=prod_callers_map[address] if include_callers else [], |
| 418 | test_callers=test_callers_map[address] if include_callers else [], |
| 419 | )) |
| 420 | |
| 421 | scored.sort(key=lambda r: (-r["risk"], r["address"])) |
| 422 | return scored[:top] |
| 423 | |
| 424 | # ── Formatters ───────────────────────────────────────────────────────────────── |
| 425 | |
| 426 | def _level(score: int) -> str: |
| 427 | if score >= 70: |
| 428 | return "high" |
| 429 | if score >= 35: |
| 430 | return " med" |
| 431 | return " low" |
| 432 | |
| 433 | def _print_table( |
| 434 | records: list[_SymbolRisk], |
| 435 | ref: str, |
| 436 | commits_analysed: int, |
| 437 | truncated: bool, |
| 438 | since: str | None, |
| 439 | ) -> None: |
| 440 | scope = f"{since}..{ref}" if since else ref |
| 441 | trunc_note = " ⚠️ truncated" if truncated else "" |
| 442 | print( |
| 443 | f"\nSymbol blast-risk — {scope}" |
| 444 | f" ({commits_analysed} commits{trunc_note})" |
| 445 | ) |
| 446 | print( |
| 447 | f"Scoring: impact×{int(_W_IMPACT*100)}%" |
| 448 | f" + churn×{int(_W_CHURN*100)}%" |
| 449 | f" + test-gap×{int(_W_TEST_GAP*100)}%" |
| 450 | f" + coupling×{int(_W_COUPLING*100)}%\n" |
| 451 | ) |
| 452 | |
| 453 | if not records: |
| 454 | print(" (no symbols meet the filter criteria)") |
| 455 | return |
| 456 | |
| 457 | max_addr = max(len(r["address"]) for r in records) |
| 458 | hdr = ( |
| 459 | f" {'#':>3} {'ADDRESS':<{max_addr}} {'RISK':>4} " |
| 460 | f"{'IMPACT':>6} {'CHURN':>5} {'TEST-GAP':>8} {'COUPLING':>8}" |
| 461 | ) |
| 462 | print(hdr) |
| 463 | print(f" {'─' * (len(hdr) - 2)}") |
| 464 | |
| 465 | for i, rec in enumerate(records, 1): |
| 466 | tg = f"{rec['test_gap_score']:>3} %" |
| 467 | print( |
| 468 | f" {i:>3} {rec['address']:<{max_addr}} {rec['risk']:>4} " |
| 469 | f"{_level(rec['impact_score']):>6} {rec['churn_raw']:>5} " |
| 470 | f"{tg:>8} {_level(rec['coupling_score']):>8}" |
| 471 | ) |
| 472 | |
| 473 | print( |
| 474 | "\n⚠️ High-risk symbols are prime targets for test coverage " |
| 475 | "and refactoring." |
| 476 | ) |
| 477 | |
| 478 | def _print_explain(rec: _SymbolRisk, commits_analysed: int) -> None: |
| 479 | addr = rec["address"] |
| 480 | print(f"\nBlast-risk breakdown — {sanitize_display(addr)}\n") |
| 481 | print(f" Risk score: {rec['risk']} / 100") |
| 482 | print(f" Kind: {rec['kind']}") |
| 483 | print() |
| 484 | |
| 485 | rows = [ |
| 486 | ("Impact", _W_IMPACT, rec["impact_score"], f"{rec['impact_raw']} direct caller(s)"), |
| 487 | ("Churn", _W_CHURN, rec["churn_score"], f"{rec['churn_raw']} change(s) in {commits_analysed} commits"), |
| 488 | ("Test gap", _W_TEST_GAP, rec["test_gap_score"], f"{rec['test_gap_score']}% of callers are production (no test)"), |
| 489 | ("Coupling", _W_COUPLING, rec["coupling_score"], f"{rec['coupling_raw']} file(s) co-change with {rec['file']}"), |
| 490 | ] |
| 491 | for label, weight, score, detail in rows: |
| 492 | pct = f"{int(weight*100)}%" |
| 493 | print(f" {label:<10} ({pct:>4}) score: {score:>3} / 100 {detail}") |
| 494 | |
| 495 | prod = rec["production_callers"] |
| 496 | if prod: |
| 497 | print(f"\n Production callers ({len(prod)}):") |
| 498 | for c in prod[:10]: |
| 499 | print(f" {c}") |
| 500 | if len(prod) > 10: |
| 501 | print(f" … {len(prod) - 10} more") |
| 502 | |
| 503 | tests = rec["test_callers"] |
| 504 | if tests: |
| 505 | print(f"\n Test callers ({len(tests)}):") |
| 506 | for c in tests[:5]: |
| 507 | print(f" {c}") |
| 508 | if len(tests) > 5: |
| 509 | print(f" … {len(tests) - 5} more") |
| 510 | |
| 511 | # ── CLI ──────────────────────────────────────────────────────────────────────── |
| 512 | |
| 513 | def register( |
| 514 | subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", |
| 515 | ) -> None: |
| 516 | """Register the blast-risk subcommand.""" |
| 517 | parser = subparsers.add_parser( |
| 518 | "blast-risk", |
| 519 | help=( |
| 520 | "Composite pre-release risk leaderboard: impact × churn × " |
| 521 | "test-gap × coupling." |
| 522 | ), |
| 523 | description=__doc__, |
| 524 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 525 | ) |
| 526 | parser.add_argument( |
| 527 | "--top", "-t", |
| 528 | type=int, default=_DEFAULT_TOP, metavar="N", |
| 529 | help=f"Show the top N riskiest symbols (default: {_DEFAULT_TOP}).", |
| 530 | ) |
| 531 | parser.add_argument( |
| 532 | "--since", "-s", |
| 533 | default=None, metavar="REF", dest="since", |
| 534 | help=( |
| 535 | "Limit churn and coupling analysis to commits reachable from HEAD " |
| 536 | "but not from REF. Useful for release-scoped risk: " |
| 537 | "--since v1.0" |
| 538 | ), |
| 539 | ) |
| 540 | parser.add_argument( |
| 541 | "--max-commits", |
| 542 | type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", |
| 543 | help=f"Maximum commits to scan for churn/coupling (default: {_DEFAULT_MAX_COMMITS}).", |
| 544 | ) |
| 545 | parser.add_argument( |
| 546 | "--kind", "-k", |
| 547 | default=None, metavar="KIND", dest="kind_filter", |
| 548 | help="Restrict to symbols of this kind (function, class, method, …).", |
| 549 | ) |
| 550 | parser.add_argument( |
| 551 | "--file", |
| 552 | default=None, metavar="FILE", dest="file_filter", |
| 553 | help="Restrict to symbols in this file (accepts a path suffix).", |
| 554 | ) |
| 555 | parser.add_argument( |
| 556 | "--min-risk", |
| 557 | type=int, default=0, metavar="N", dest="min_risk", |
| 558 | help="Only show symbols with a composite risk score >= N (default: 0).", |
| 559 | ) |
| 560 | parser.add_argument( |
| 561 | "--explain", |
| 562 | default=None, metavar="ADDRESS", dest="explain", |
| 563 | help=( |
| 564 | "Print a detailed per-dimension breakdown for a single symbol " |
| 565 | "instead of the leaderboard table." |
| 566 | ), |
| 567 | ) |
| 568 | parser.add_argument( |
| 569 | "--json", "-j", |
| 570 | action="store_true", dest="json_out", |
| 571 | help="Emit results as JSON (agent-friendly; -j is a shorthand alias).", |
| 572 | ) |
| 573 | parser.set_defaults(func=run) |
| 574 | |
| 575 | def run(args: argparse.Namespace) -> None: |
| 576 | """Rank every symbol by composite blast-risk score. |
| 577 | |
| 578 | Combines four dimensions: impact (callers), churn (commit frequency), test |
| 579 | gap (untested callers), and coupling (co-changing files). Each dimension is |
| 580 | normalised 0–100 then weighted into a single ``risk`` score. |
| 581 | |
| 582 | Agent quickstart |
| 583 | ---------------- |
| 584 | :: |
| 585 | |
| 586 | muse code blast-risk --json |
| 587 | muse code blast-risk --top 20 --min-risk 50 --json |
| 588 | muse code blast-risk --explain src/billing.py::compute_total --json |
| 589 | |
| 590 | JSON fields |
| 591 | ----------- |
| 592 | ref Branch or ref analysed. |
| 593 | commits_analysed Number of commits scanned for churn. |
| 594 | truncated ``true`` if ``--max-commits`` was reached. |
| 595 | filters Echo of CLI filter arguments. |
| 596 | weights Weight constants for each dimension. |
| 597 | symbols Ranked list of symbol risk records. |
| 598 | |
| 599 | Each symbol record: |
| 600 | |
| 601 | address Fully qualified symbol address. |
| 602 | kind Symbol kind: function, class, method, … |
| 603 | file Source file path. |
| 604 | risk Composite risk score 0–100. |
| 605 | impact_raw Raw caller count. |
| 606 | churn_raw Raw commit count touching this symbol. |
| 607 | test_gap_raw Number of untested callers. |
| 608 | coupling_raw Co-change file count. |
| 609 | impact_score Normalised impact 0–100. |
| 610 | churn_score Normalised churn 0–100. |
| 611 | test_gap_score Normalised test gap 0–100. |
| 612 | coupling_score Normalised coupling 0–100. |
| 613 | |
| 614 | Exit codes |
| 615 | ---------- |
| 616 | 0 Analysis complete. |
| 617 | 1 Invalid arguments or symbol not found (``--explain``). |
| 618 | 2 Not inside a Muse repository. |
| 619 | """ |
| 620 | elapsed = start_timer() |
| 621 | top: int = clamp_int(args.top, 1, 10_000, 'top') |
| 622 | since: str | None = args.since |
| 623 | max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') |
| 624 | kind_filter: str | None = args.kind_filter |
| 625 | file_filter: str | None = args.file_filter |
| 626 | min_risk: int = clamp_int(args.min_risk, 0, 100, 'min_risk') |
| 627 | explain_addr: str | None = args.explain |
| 628 | json_out: bool = args.json_out |
| 629 | |
| 630 | # ── Validation ──────────────────────────────────────────────────────────── |
| 631 | |
| 632 | if top < 1: |
| 633 | print("❌ --top must be >= 1.", file=sys.stderr) |
| 634 | raise SystemExit(ExitCode.USER_ERROR) |
| 635 | if max_commits < 1: |
| 636 | print("❌ --max-commits must be >= 1.", file=sys.stderr) |
| 637 | raise SystemExit(ExitCode.USER_ERROR) |
| 638 | if not (0 <= min_risk <= 100): |
| 639 | print("❌ --min-risk must be between 0 and 100.", file=sys.stderr) |
| 640 | raise SystemExit(ExitCode.USER_ERROR) |
| 641 | if kind_filter is not None: |
| 642 | kind_filter = kind_filter.strip().lower() |
| 643 | |
| 644 | # ── Repo setup ──────────────────────────────────────────────────────────── |
| 645 | |
| 646 | root = require_repo() |
| 647 | branch = read_current_branch(root) |
| 648 | |
| 649 | head = resolve_commit_ref(root, branch, None) |
| 650 | if head is None: |
| 651 | print("❌ HEAD commit not found.", file=sys.stderr) |
| 652 | raise SystemExit(ExitCode.USER_ERROR) |
| 653 | |
| 654 | stop_at: str | None = None |
| 655 | if since is not None: |
| 656 | since_commit = resolve_commit_ref(root, branch, since) |
| 657 | if since_commit is None: |
| 658 | print(f"❌ Commit '{since}' not found.", file=sys.stderr) |
| 659 | raise SystemExit(ExitCode.USER_ERROR) |
| 660 | stop_at = since_commit.commit_id |
| 661 | |
| 662 | manifest: Manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} |
| 663 | |
| 664 | # ── Phase 1: load symbol trees (cached) ────────────────────────────────── |
| 665 | |
| 666 | shared_cache = load_symbol_cache(root) |
| 667 | all_trees = symbols_for_snapshot(root, manifest, cache=shared_cache) |
| 668 | |
| 669 | # Build: address → (name, kind) and name → [addresses] |
| 670 | symbol_records: SymbolRecordMap = {} |
| 671 | known_names: set[str] = set() |
| 672 | for file_path, tree in all_trees.items(): |
| 673 | for addr, sym in tree.items(): |
| 674 | sym_name: str = sym["name"] |
| 675 | sym_kind: str = sym["kind"] |
| 676 | if "::import::" in addr: |
| 677 | continue |
| 678 | symbol_records[addr] = (sym_name, sym_kind) |
| 679 | known_names.add(sym_name) |
| 680 | |
| 681 | if not symbol_records: |
| 682 | _emit_empty(json_out, branch, 0, False, since, kind_filter, file_filter, min_risk, top, max_commits, elapsed) |
| 683 | return |
| 684 | |
| 685 | # ── Phase 2: caller map (AST scan of all Python files) ─────────────────── |
| 686 | |
| 687 | caller_map = _build_caller_map(root, manifest, frozenset(known_names)) |
| 688 | |
| 689 | # ── Phase 3: churn + coupling in one BFS pass ───────────────────────────── |
| 690 | |
| 691 | churn, file_coupling, commits_analysed, truncated = _collect_churn_and_coupling( |
| 692 | root, head.commit_id, stop_at, max_commits |
| 693 | ) |
| 694 | |
| 695 | # ── Phase 4: --explain mode ─────────────────────────────────────────────── |
| 696 | |
| 697 | if explain_addr is not None: |
| 698 | if explain_addr not in symbol_records: |
| 699 | print( |
| 700 | f"❌ Symbol '{explain_addr}' not found in HEAD snapshot.", |
| 701 | file=sys.stderr, |
| 702 | ) |
| 703 | print(" Hint: muse code symbols --json | grep address", file=sys.stderr) |
| 704 | raise SystemExit(ExitCode.USER_ERROR) |
| 705 | |
| 706 | # Score just this one symbol with full caller lists. |
| 707 | ranked = _score_symbols( |
| 708 | {explain_addr: symbol_records[explain_addr]}, |
| 709 | caller_map, churn, file_coupling, |
| 710 | kind_filter=None, file_filter=None, |
| 711 | min_risk=0, top=1, |
| 712 | include_callers=True, |
| 713 | ) |
| 714 | if not ranked: |
| 715 | print(f" (no risk data available for '{explain_addr}')") |
| 716 | return |
| 717 | rec = ranked[0] |
| 718 | if json_out: |
| 719 | print(json.dumps(_ExplainJson( |
| 720 | **make_envelope(elapsed), |
| 721 | address=rec["address"], |
| 722 | kind=rec["kind"], |
| 723 | file=rec["file"], |
| 724 | risk=rec["risk"], |
| 725 | impact_raw=rec["impact_raw"], |
| 726 | churn_raw=rec["churn_raw"], |
| 727 | test_gap_raw=rec["test_gap_raw"], |
| 728 | coupling_raw=rec["coupling_raw"], |
| 729 | impact_score=rec["impact_score"], |
| 730 | churn_score=rec["churn_score"], |
| 731 | test_gap_score=rec["test_gap_score"], |
| 732 | coupling_score=rec["coupling_score"], |
| 733 | production_callers=rec.get("production_callers", []), |
| 734 | test_callers=rec.get("test_callers", []), |
| 735 | ))) |
| 736 | else: |
| 737 | _print_explain(rec, commits_analysed) |
| 738 | return |
| 739 | |
| 740 | # ── Phase 5: score + rank all symbols ───────────────────────────────────── |
| 741 | |
| 742 | ranked = _score_symbols( |
| 743 | symbol_records, caller_map, churn, file_coupling, |
| 744 | kind_filter=kind_filter, file_filter=file_filter, |
| 745 | min_risk=min_risk, top=top, |
| 746 | include_callers=False, |
| 747 | ) |
| 748 | |
| 749 | # ── Output ──────────────────────────────────────────────────────────────── |
| 750 | |
| 751 | ref = branch |
| 752 | filters = _BlastRiskFilters( |
| 753 | kind=kind_filter, |
| 754 | file=file_filter, |
| 755 | min_risk=min_risk, |
| 756 | since=since, |
| 757 | top=top, |
| 758 | max_commits=max_commits, |
| 759 | ) |
| 760 | |
| 761 | if json_out: |
| 762 | def _risk_label(score: int) -> str: |
| 763 | if score >= 71: |
| 764 | return "high" |
| 765 | if score >= 31: |
| 766 | return "medium" |
| 767 | return "low" |
| 768 | |
| 769 | symbols: list[_SymbolRiskJson] = [ |
| 770 | _SymbolRiskJson( |
| 771 | address=r["address"], kind=r["kind"], file=r["file"], |
| 772 | risk=r["risk"], risk_label=_risk_label(r["risk"]), |
| 773 | impact_raw=r["impact_raw"], churn_raw=r["churn_raw"], |
| 774 | test_gap_raw=r["test_gap_raw"], coupling_raw=r["coupling_raw"], |
| 775 | impact_score=r["impact_score"], churn_score=r["churn_score"], |
| 776 | test_gap_score=r["test_gap_score"], coupling_score=r["coupling_score"], |
| 777 | ) |
| 778 | for r in ranked |
| 779 | ] |
| 780 | print(json.dumps(_BlastRiskOutput( |
| 781 | **make_envelope(elapsed), |
| 782 | ref=ref, |
| 783 | commits_analysed=commits_analysed, |
| 784 | truncated=truncated, |
| 785 | filters=filters, |
| 786 | weights=_BlastRiskWeights( |
| 787 | impact=_W_IMPACT, |
| 788 | churn=_W_CHURN, |
| 789 | test_gap=_W_TEST_GAP, |
| 790 | coupling=_W_COUPLING, |
| 791 | ), |
| 792 | symbols=symbols, |
| 793 | ))) |
| 794 | return |
| 795 | |
| 796 | _print_table(ranked, ref, commits_analysed, truncated, since) |
| 797 | |
| 798 | def _emit_empty( |
| 799 | json_out: bool, |
| 800 | ref: str, |
| 801 | commits_analysed: int, |
| 802 | truncated: bool, |
| 803 | since: str | None, |
| 804 | kind_filter: str | None, |
| 805 | file_filter: str | None, |
| 806 | min_risk: int, |
| 807 | top: int, |
| 808 | max_commits: int, |
| 809 | elapsed: Callable[[], float], |
| 810 | ) -> None: |
| 811 | if json_out: |
| 812 | print(json.dumps(_BlastRiskOutput( |
| 813 | **make_envelope(elapsed), |
| 814 | ref=ref, |
| 815 | commits_analysed=commits_analysed, |
| 816 | truncated=truncated, |
| 817 | filters=_BlastRiskFilters( |
| 818 | kind=kind_filter, file=file_filter, |
| 819 | min_risk=min_risk, since=since, |
| 820 | top=top, max_commits=max_commits, |
| 821 | ), |
| 822 | weights=_BlastRiskWeights( |
| 823 | impact=_W_IMPACT, churn=_W_CHURN, |
| 824 | test_gap=_W_TEST_GAP, coupling=_W_COUPLING, |
| 825 | ), |
| 826 | symbols=[], |
| 827 | ))) |
| 828 | else: |
| 829 | print("\n(no symbols found in the current snapshot)") |