"""muse code blast-risk — composite pre-release risk leaderboard. Synthesises four dimensions that Muse tracks independently into a single ranked risk score for every symbol in the snapshot: Impact (40 %) — how many *other* symbols directly reference this one via import, bare name, or attribute access. A bug in a highly-referenced symbol touches many callsites. Churn (30 %) — how many commits in the analysed range modified this symbol. Frequently-changing code has higher defect probability (same signal as ``muse code hotspots``). Test gap (20 %) — fraction of callers that live in *production* files. Symbols whose callers are untested carry hidden risk. Coupling (10 %) — how many other files co-change with this symbol's file. High coupling amplifies the blast radius. Each dimension is normalised 0–100 across the full symbol set, then combined with the weights above into a final risk score 0–100. No other tool can produce this score because no other tool simultaneously knows the call graph, the churn history, which callers are tests, and the file co-change graph. Usage:: muse code blast-risk muse code blast-risk --top 20 muse code blast-risk --since a3f2c9 muse code blast-risk --kind function muse code blast-risk --file billing.py muse code blast-risk --explain billing.py::Invoice.compute_total muse code blast-risk --min-risk 50 muse code blast-risk --json Output:: Symbol blast-risk — HEAD (304 commits · 378 symbols) Scoring: impact×40% + churn×30% + test-gap×20% + coupling×10% # ADDRESS RISK IMPACT CHURN TEST-GAP COUPLING 1 muse/core/store.py::resolve_commit_ref 94 high 12 87 % high 2 muse/core/errors.py::ExitCode 87 high 8 72 % med 3 billing.py::Invoice.compute_total 81 med 15 100 % low ⚠️ High-risk symbols are prime targets for test coverage and refactoring. JSON output (--json):: { "ref": "HEAD", "commits_analysed": 304, "truncated": false, "filters": { "kind": null, "file": null, "min_risk": 0, "since": null }, "weights": { "impact": 0.40, "churn": 0.30, "test_gap": 0.20, "coupling": 0.10 }, "symbols": [ { "address": "muse/core/store.py::resolve_commit_ref", "kind": "function", "file": "muse/core/store.py", "risk": 94, "impact_raw": 47, "churn_raw": 12, "test_gap_raw": 0.87, "coupling_raw": 9, "impact_score": 100, "churn_score": 80, "test_gap_score": 87, "coupling_score": 75 } ] } --explain output:: Blast-risk breakdown — muse/core/store.py::resolve_commit_ref Risk score: 94 / 100 Kind: function Impact (40 %) score: 100 / 100 raw: 47 direct callers Churn (30 %) score: 80 / 100 raw: 12 changes in 304 commits Test gap (20 %) score: 87 / 100 raw: 41 / 47 callers are production (not test) Coupling (10 %) score: 75 / 100 raw: 9 files co-change with muse/core/store.py Top callers (production): muse/cli/commands/compare.py::run muse/cli/commands/blame.py::run muse/plugins/code/_query.py::walk_commits_bfs … 38 more """ from __future__ import annotations import ast import argparse import json import logging import pathlib import re import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( Manifest, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.symbol_cache import load_symbol_cache from muse.domain import DomainOp from muse.plugins.code._query import ( file_pairs, flat_symbol_ops, language_of, symbols_for_snapshot, touched_files, walk_commits_bfs, ) from muse.core.validation import MAX_AST_BYTES, clamp_int, sanitize_display logger = logging.getLogger(__name__) type CallerMap = dict[str, list[str]] # name → list of caller file_paths type ChurnMap = dict[str, int] # file_path or address → churn count type CouplingMap = dict[str, int] # file_path → co-change partner count type SymbolRecordMap = dict[str, tuple[str, str]] # address → (name, kind) type CallerListMap = dict[str, list[str]] # address → list of caller files type TestGapMap = dict[str, float] # address → test gap fraction # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_TOP = 20 _DEFAULT_MAX_COMMITS = 10_000 _MAX_FILES_PER_COMMIT = 50 # skip mass-edit commits (noisy for coupling) _W_IMPACT: float = 0.40 _W_CHURN: float = 0.30 _W_TEST_GAP: float = 0.20 _W_COUPLING: float = 0.10 # Test-file heuristics — callers in these files are considered "tested". _TEST_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"(^|/)test_"), re.compile(r"_test\.py$"), re.compile(r"(^|/)tests/"), re.compile(r"(^|/)spec/"), re.compile(r"(^|/)conftest\.py$"), ) # ── TypedDicts ───────────────────────────────────────────────────────────────── class _SymbolRisk(TypedDict): """Fully-scored risk record for one symbol.""" address: str kind: str file: str risk: int # final composite score 0-100 impact_raw: int # direct caller count churn_raw: int # commit-change count test_gap_raw: float # fraction of callers that are production (0-1) coupling_raw: int # co-change partner count for the file impact_score: int # normalised 0-100 churn_score: int # normalised 0-100 test_gap_score: int # = round(test_gap_raw * 100) coupling_score: int # normalised 0-100 # Only present in --explain output. production_callers: list[str] test_callers: list[str] class _SymbolRiskJson(TypedDict): """JSON-serialisable risk record — ``production_callers``/``test_callers`` excluded.""" address: str kind: str file: str risk: int impact_raw: int churn_raw: int test_gap_raw: float coupling_raw: int impact_score: int churn_score: int test_gap_score: int coupling_score: int class _BlastRiskFilters(TypedDict): kind: str | None file: str | None min_risk: int since: str | None top: int max_commits: int class _BlastRiskWeights(TypedDict): impact: float churn: float test_gap: float coupling: float class _BlastRiskOutput(TypedDict): """JSON payload for ``muse code blast-risk`` output.""" ref: str commits_analysed: int truncated: bool filters: _BlastRiskFilters weights: _BlastRiskWeights symbols: list[_SymbolRiskJson] # ── Helpers ──────────────────────────────────────────────────────────────────── def _is_test_file(path: str) -> bool: return any(p.search(path) for p in _TEST_PATTERNS) def _normalise_score(raw: int | float, maximum: int | float) -> int: if maximum <= 0: return 0 return round(min(100, (raw / maximum) * 100)) # ── Phase 1: caller map ──────────────────────────────────────────────────────── def _build_caller_map( root: pathlib.Path, manifest: Manifest, known_names: frozenset[str], ) -> CallerMap: """Return ``name → [caller_file_path, ...]`` by scanning all Python files. Uses AST-level ``ast.Name`` and ``ast.Attribute`` nodes so only identifier references are counted — string literals and comments are never included. Only names that appear in *known_names* are tracked to avoid building an arbitrarily large map. """ caller_map: CallerMap = {} for file_path in sorted(manifest): if language_of(file_path) != "Python": continue disk_path = root / file_path if not disk_path.is_file(): continue try: source = disk_path.read_bytes() if len(source) > MAX_AST_BYTES: return {} tree = ast.parse(source, filename=file_path) except (OSError, SyntaxError): continue for node in ast.walk(tree): name: str | None = None if isinstance(node, ast.Name) and node.id in known_names: name = node.id elif isinstance(node, ast.Attribute) and node.attr in known_names: name = node.attr if name is not None: caller_map.setdefault(name, []).append(file_path) return caller_map # ── Phase 2: churn + coupling in one BFS pass ───────────────────────────────── def _collect_churn_and_coupling( root: pathlib.Path, head_commit_id: str, stop_at: str | None, max_commits: int, ) -> tuple[dict[str, int], dict[str, int], int, bool]: """Single BFS pass collecting churn and file-coupling simultaneously. Returns: churn — ``address → commit-change count`` file_coupling — ``file_path → number of distinct co-change partners`` commits_analysed — number of commits walked truncated — whether the BFS hit max_commits """ commits, truncated = walk_commits_bfs( root, head_commit_id, max_commits, stop_at_commit_id=stop_at ) churn: ChurnMap = {} pair_counts: dict[tuple[str, str], int] = {} for commit in commits: if commit.structured_delta is None: continue ops: list[DomainOp] = commit.structured_delta["ops"] # ── churn ────────────────────────────────────────────────────────── for op in flat_symbol_ops(ops): addr: str = op["address"] if "::import::" not in addr: # exclude import pseudo-symbols churn[addr] = churn.get(addr, 0) + 1 # ── coupling ─────────────────────────────────────────────────────── files = touched_files(ops) if 2 <= len(files) <= _MAX_FILES_PER_COMMIT: for fa, fb in file_pairs(files): pair_counts[(fa, fb)] = pair_counts.get((fa, fb), 0) + 1 # Collapse pair counts to per-file partner counts. file_coupling: CouplingMap = {} for (fa, fb), cnt in pair_counts.items(): if cnt >= 1: file_coupling[fa] = file_coupling.get(fa, 0) + 1 file_coupling[fb] = file_coupling.get(fb, 0) + 1 return churn, file_coupling, len(commits), truncated # ── Phase 3: score + rank ────────────────────────────────────────────────────── def _score_symbols( symbol_records: SymbolRecordMap, caller_map: CallerMap, churn: ChurnMap, file_coupling: CouplingMap, kind_filter: str | None, file_filter: str | None, min_risk: int, top: int, include_callers: bool, ) -> list[_SymbolRisk]: """Compute normalised scores and return the ranked list.""" # ── Gather raw values ────────────────────────────────────────────────── raw_impacts: ChurnMap = {} raw_test_gaps: TestGapMap = {} prod_callers_map: CallerListMap = {} test_callers_map: CallerListMap = {} for address, (name, kind) in symbol_records.items(): file_path = address.split("::")[0] callers = caller_map.get(name, []) prod = [c for c in callers if not _is_test_file(c) and c != file_path] tests = [c for c in callers if _is_test_file(c)] total = len(prod) + len(tests) raw_impacts[address] = total raw_test_gaps[address] = len(prod) / total if total > 0 else 0.0 prod_callers_map[address] = sorted(set(prod)) test_callers_map[address] = sorted(set(tests)) # ── Normalisation maximums ───────────────────────────────────────────── max_impact = max(raw_impacts.values(), default=1) or 1 max_churn = max(churn.values(), default=1) or 1 max_coupling = max(file_coupling.values(), default=1) or 1 # ── Score every symbol ───────────────────────────────────────────────── scored: list[_SymbolRisk] = [] for address, (name, kind) in symbol_records.items(): file_path = address.split("::")[0] if kind_filter and kind.lower() != kind_filter: continue if file_filter and not ( file_path == file_filter or file_path.endswith("/" + file_filter) ): continue impact_raw = raw_impacts[address] churn_raw = churn.get(address, 0) test_gap_raw = raw_test_gaps[address] coupling_raw = file_coupling.get(file_path, 0) impact_s = _normalise_score(impact_raw, max_impact) churn_s = _normalise_score(churn_raw, max_churn) test_gap_s = round(test_gap_raw * 100) coupling_s = _normalise_score(coupling_raw, max_coupling) risk = round( _W_IMPACT * impact_s + _W_CHURN * churn_s + _W_TEST_GAP * test_gap_s + _W_COUPLING * coupling_s ) if risk < min_risk: continue scored.append(_SymbolRisk( address=address, kind=kind, file=file_path, risk=risk, impact_raw=impact_raw, churn_raw=churn_raw, test_gap_raw=test_gap_raw, coupling_raw=coupling_raw, impact_score=impact_s, churn_score=churn_s, test_gap_score=test_gap_s, coupling_score=coupling_s, production_callers=prod_callers_map[address] if include_callers else [], test_callers=test_callers_map[address] if include_callers else [], )) scored.sort(key=lambda r: (-r["risk"], r["address"])) return scored[:top] # ── Formatters ───────────────────────────────────────────────────────────────── def _level(score: int) -> str: if score >= 70: return "high" if score >= 35: return " med" return " low" def _print_table( records: list[_SymbolRisk], ref: str, commits_analysed: int, truncated: bool, since: str | None, ) -> None: scope = f"{since}..{ref}" if since else ref trunc_note = " ⚠️ truncated" if truncated else "" print( f"\nSymbol blast-risk — {scope}" f" ({commits_analysed} commits{trunc_note})" ) print( f"Scoring: impact×{int(_W_IMPACT*100)}%" f" + churn×{int(_W_CHURN*100)}%" f" + test-gap×{int(_W_TEST_GAP*100)}%" f" + coupling×{int(_W_COUPLING*100)}%\n" ) if not records: print(" (no symbols meet the filter criteria)") return max_addr = max(len(r["address"]) for r in records) hdr = ( f" {'#':>3} {'ADDRESS':<{max_addr}} {'RISK':>4} " f"{'IMPACT':>6} {'CHURN':>5} {'TEST-GAP':>8} {'COUPLING':>8}" ) print(hdr) print(" " + "─" * (len(hdr) - 2)) for i, rec in enumerate(records, 1): tg = f"{rec['test_gap_score']:>3} %" print( f" {i:>3} {rec['address']:<{max_addr}} {rec['risk']:>4} " f"{_level(rec['impact_score']):>6} {rec['churn_raw']:>5} " f"{tg:>8} {_level(rec['coupling_score']):>8}" ) print( "\n⚠️ High-risk symbols are prime targets for test coverage " "and refactoring." ) def _print_explain(rec: _SymbolRisk, commits_analysed: int) -> None: addr = rec["address"] print(f"\nBlast-risk breakdown — {sanitize_display(addr)}\n") print(f" Risk score: {rec['risk']} / 100") print(f" Kind: {rec['kind']}") print() rows = [ ("Impact", _W_IMPACT, rec["impact_score"], f"{rec['impact_raw']} direct caller(s)"), ("Churn", _W_CHURN, rec["churn_score"], f"{rec['churn_raw']} change(s) in {commits_analysed} commits"), ("Test gap", _W_TEST_GAP, rec["test_gap_score"], f"{rec['test_gap_score']}% of callers are production (no test)"), ("Coupling", _W_COUPLING, rec["coupling_score"], f"{rec['coupling_raw']} file(s) co-change with {rec['file']}"), ] for label, weight, score, detail in rows: pct = f"{int(weight*100)}%" print(f" {label:<10} ({pct:>4}) score: {score:>3} / 100 {detail}") prod = rec["production_callers"] if prod: print(f"\n Production callers ({len(prod)}):") for c in prod[:10]: print(f" {c}") if len(prod) > 10: print(f" … {len(prod) - 10} more") tests = rec["test_callers"] if tests: print(f"\n Test callers ({len(tests)}):") for c in tests[:5]: print(f" {c}") if len(tests) > 5: print(f" … {len(tests) - 5} more") # ── CLI ──────────────────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the blast-risk subcommand.""" parser = subparsers.add_parser( "blast-risk", help=( "Composite pre-release risk leaderboard: impact × churn × " "test-gap × coupling." ), description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--top", "-t", type=int, default=_DEFAULT_TOP, metavar="N", help=f"Show the top N riskiest symbols (default: {_DEFAULT_TOP}).", ) parser.add_argument( "--since", "-s", default=None, metavar="REF", dest="since", help=( "Limit churn and coupling analysis to commits reachable from HEAD " "but not from REF. Useful for release-scoped risk: " "--since v1.0" ), ) parser.add_argument( "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", help=f"Maximum commits to scan for churn/coupling (default: {_DEFAULT_MAX_COMMITS}).", ) parser.add_argument( "--kind", "-k", default=None, metavar="KIND", dest="kind_filter", help="Restrict to symbols of this kind (function, class, method, …).", ) parser.add_argument( "--file", "-f", default=None, metavar="FILE", dest="file_filter", help="Restrict to symbols in this file (accepts a path suffix).", ) parser.add_argument( "--min-risk", type=int, default=0, metavar="N", dest="min_risk", help="Only show symbols with a composite risk score >= N (default: 0).", ) parser.add_argument( "--explain", default=None, metavar="ADDRESS", dest="explain", help=( "Print a detailed per-dimension breakdown for a single symbol " "instead of the leaderboard table." ), ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as JSON.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Rank every symbol by composite blast-risk. The score combines four dimensions that Muse tracks natively: how many symbols reference this one (impact), how often it changes (churn), how many of its callers are untested (test gap), and how many files co-change with its file (coupling). Each dimension is normalised 0–100 then weighted. """ top: int = clamp_int(args.top, 1, 10_000, 'top') since: str | None = args.since max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') kind_filter: str | None = args.kind_filter file_filter: str | None = args.file_filter min_risk: int = clamp_int(args.min_risk, 0, 100, 'min_risk') explain_addr: str | None = args.explain as_json: bool = args.as_json # ── Validation ──────────────────────────────────────────────────────────── if top < 1: print("❌ --top must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if max_commits < 1: print("❌ --max-commits must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not (0 <= min_risk <= 100): print("❌ --min-risk must be between 0 and 100.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if kind_filter is not None: kind_filter = kind_filter.strip().lower() # ── Repo setup ──────────────────────────────────────────────────────────── root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) head = resolve_commit_ref(root, repo_id, branch, None) if head is None: print("❌ HEAD commit not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) stop_at: str | None = None if since is not None: since_commit = resolve_commit_ref(root, repo_id, branch, since) if since_commit is None: print(f"❌ Commit '{since}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) stop_at = since_commit.commit_id manifest: Manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} # ── Phase 1: load symbol trees (cached) ────────────────────────────────── shared_cache = load_symbol_cache(root) all_trees = symbols_for_snapshot(root, manifest, cache=shared_cache) # Build: address → (name, kind) and name → [addresses] symbol_records: SymbolRecordMap = {} known_names: set[str] = set() for file_path, tree in all_trees.items(): for addr, sym in tree.items(): sym_name: str = sym["name"] sym_kind: str = sym["kind"] if "::import::" in addr: continue symbol_records[addr] = (sym_name, sym_kind) known_names.add(sym_name) if not symbol_records: _emit_empty(as_json, branch, 0, False, since, kind_filter, file_filter, min_risk) return # ── Phase 2: caller map (AST scan of all Python files) ─────────────────── caller_map = _build_caller_map(root, manifest, frozenset(known_names)) # ── Phase 3: churn + coupling in one BFS pass ───────────────────────────── churn, file_coupling, commits_analysed, truncated = _collect_churn_and_coupling( root, head.commit_id, stop_at, max_commits ) # ── Phase 4: --explain mode ─────────────────────────────────────────────── if explain_addr is not None: if explain_addr not in symbol_records: print( f"❌ Symbol '{explain_addr}' not found in HEAD snapshot.", file=sys.stderr, ) print(" Hint: muse code symbols --json | grep address", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Score just this one symbol with full caller lists. ranked = _score_symbols( {explain_addr: symbol_records[explain_addr]}, caller_map, churn, file_coupling, kind_filter=None, file_filter=None, min_risk=0, top=1, include_callers=True, ) if not ranked: print(f" (no risk data available for '{explain_addr}')") return rec = ranked[0] if as_json: print(json.dumps(dict(rec), indent=2)) else: _print_explain(rec, commits_analysed) return # ── Phase 5: score + rank all symbols ───────────────────────────────────── ranked = _score_symbols( symbol_records, caller_map, churn, file_coupling, kind_filter=kind_filter, file_filter=file_filter, min_risk=min_risk, top=top, include_callers=False, ) # ── Output ──────────────────────────────────────────────────────────────── ref = branch filters = _BlastRiskFilters( kind=kind_filter, file=file_filter, min_risk=min_risk, since=since, top=top, max_commits=max_commits, ) if as_json: symbols: list[_SymbolRiskJson] = [ _SymbolRiskJson( address=r["address"], kind=r["kind"], file=r["file"], risk=r["risk"], impact_raw=r["impact_raw"], churn_raw=r["churn_raw"], test_gap_raw=r["test_gap_raw"], coupling_raw=r["coupling_raw"], impact_score=r["impact_score"], churn_score=r["churn_score"], test_gap_score=r["test_gap_score"], coupling_score=r["coupling_score"], ) for r in ranked ] print(json.dumps(_BlastRiskOutput( ref=ref, commits_analysed=commits_analysed, truncated=truncated, filters=filters, weights=_BlastRiskWeights( impact=_W_IMPACT, churn=_W_CHURN, test_gap=_W_TEST_GAP, coupling=_W_COUPLING, ), symbols=symbols, ), indent=2)) return _print_table(ranked, ref, commits_analysed, truncated, since) def _emit_empty( as_json: bool, ref: str, commits_analysed: int, truncated: bool, since: str | None, kind_filter: str | None, file_filter: str | None, min_risk: int, ) -> None: if as_json: print(json.dumps({ "ref": ref, "commits_analysed": commits_analysed, "truncated": truncated, "filters": { "kind": kind_filter, "file": file_filter, "min_risk": min_risk, "since": since, }, "weights": { "impact": _W_IMPACT, "churn": _W_CHURN, "test_gap": _W_TEST_GAP, "coupling": _W_COUPLING, }, "symbols": [], }, indent=2)) else: print("\n(no symbols found in the current snapshot)")