"""muse code predict — predict which symbols will change next. Every commit leaves a fingerprint. Symbols that churn repeatedly, that always change together, whose signatures keep shifting, whose modules are accelerating — all of these are signals that a change is coming. ``muse code predict`` synthesises five independent signals mined from the full commit DAG and surfaces a ranked leaderboard of symbols statistically most likely to change in the next sprint, with named confidence bands and per-symbol reasoning so you know *why* each prediction was made. The five signals ---------------- 1. **Recency** — a symbol touched 2 commits ago has more editing momentum than one touched 200 commits ago. Decays linearly over the ``--horizon`` window. 2. **Frequency** — symbols changed many times within the analysis window are in active development and will continue to evolve. 3. **Co-change entanglement** — pairs of symbols that co-change in commits at a high Jaccard rate are implicitly coupled. If your partner just changed, you are likely next. This is the hidden-dependency signal only Muse can compute at symbol granularity. 4. **Signature instability** — a symbol whose signature keeps shifting is still being negotiated. High ratio of signature ops to total ops → likely to change again. 5. **Module velocity** — symbols living in an accelerating module are more likely to change because the module is still being developed. Output:: Predicted changes (horizon: 50 commits · 315 analysed) HIGH CONFIDENCE (score >= 0.70) ───────────────────────────────────────────────────────────────────── 0.91 muse/core/store.py::resolve_commit_ref ↳ changed 3 commits ago — editing momentum ↳ entangled with resolve_commit_sha (88% co-change) ↳ muse/core/ velocity: +12 symbols/window 0.84 muse/cli/commands/dead.py::run ↳ changed 12× in last 50 commits ↳ signature changed 3× — API still evolving MEDIUM CONFIDENCE (score 0.45 – 0.70) ───────────────────────────────────────────────────────────────────── 0.62 muse/plugins/code/_query.py::flat_symbol_ops ↳ entangled with walk_commits_bfs (75% co-change) Explain mode (``--explain ADDRESS``) shows the full signal breakdown for a specific symbol:: billing.py::Invoice.compute_total — signal breakdown Score: 0.91 HIGH ───────────────────────────────────────────────────── recency: 0.95 [████████████████████] changed 3 commits ago frequency: 0.88 [██████████████████ ] 9× in last 50 commits co_change: 0.92 [████████████████████] entangled with process_order sig_instability: 0.50 [██████████ ] 2/4 changes were sig ops module_velocity: 0.75 [███████████████ ] billing/ accelerating Top co-change partners billing.py::process_order 100% (9 commits) services.py::create_invoice 78% (7 commits) JSON output (``--json``) provides full provenance for every prediction:: { "generated_at": "2026-03-24T18:00:00", "horizon_commits": 50, "max_commits": 2000, "commits_analysed": 315, "truncated": false, "predictions": [ { "address": "muse/core/store.py::resolve_commit_ref", "name": "resolve_commit_ref", "kind": "function", "file": "muse/core/store.py", "score": 0.91, "confidence": "high", "reasons": [ "changed 3 commits ago — editing momentum", ... ], "signals": { "recency": 0.95, "frequency": 0.88, "co_change": 0.92, "sig_instability": 0.5, "module_velocity": 0.75 }, "last_changed_commit": "f66b7b1d...", "last_changed_date": "2026-03-24", "top_partners": [ { "address": "...", "co_change_rate": 0.88, "co_change_commits": 9 } ] } ] } Security note ------------- Symbol addresses in the output are derived from the content-addressed object store, not from user input. Commit messages used in reason text are sanitised (control characters stripped). ``--explain`` validates the supplied address format before any store access. All file I/O goes through the object store; no user-supplied paths are opened directly. """ from __future__ import annotations import argparse import datetime import json import logging import pathlib import re import sys from collections import Counter from typing import Iterator, TypedDict from muse.core._types import Metadata from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.domain import DomainOp from muse.core.store import CommitRecord from muse.domain import StructuredDelta from muse.plugins.code._query import flat_symbol_ops, symbols_for_snapshot, walk_commits_bfs from muse.core.validation import clamp_int, sanitize_display logger = logging.getLogger(__name__) type _CounterMap = dict[str, int] type _DateMap = dict[str, datetime.datetime] type _CoCountMap = dict[str, Counter[str]] # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_HORIZON = 50 # recent commits used as prediction window _DEFAULT_MAX_COMMITS = 2_000 # total commits to walk for history _DEFAULT_TOP = 15 _MAX_REASONS = 4 _MAX_PARTNERS = 5 _MAX_CO_MATCH = 20 # top co-change partners to consider per symbol # Co-change Jaccard cap: only consider recent_set × recent_set pairs to # keep the matrix bounded at O(|recent_set|²). _MAX_RECENT_SET = 500 # Signal weights — must sum to 1.0. _W_RECENCY = 0.30 _W_FREQUENCY = 0.25 _W_CO_CHANGE = 0.25 _W_SIG_INSTAB = 0.10 _W_MOD_VELOCITY = 0.10 _CONFIDENCE_HIGH = 0.70 _CONFIDENCE_MED = 0.45 _CTRL_RE = re.compile(r"[\x00-\x09\x0b-\x1f\x7f-\x9f\x1b]") _SIG_KW: frozenset[str] = frozenset({"signature"}) # ── TypedDicts ───────────────────────────────────────────────────────────────── class _SignalSet(TypedDict): recency: float frequency: float co_change: float sig_instability: float module_velocity: float class _PartnerRecord(TypedDict): address: str co_change_rate: float co_change_commits: int class _PredictionRecord(TypedDict): address: str name: str kind: str file: str score: float confidence: str reasons: list[str] signals: _SignalSet last_changed_commit: str last_changed_date: str top_partners: list[_PartnerRecord] class _PredictJson(TypedDict): generated_at: str horizon_commits: int max_commits: int commits_analysed: int truncated: bool predictions: list[_PredictionRecord] # ── Helpers ──────────────────────────────────────────────────────────────────── def _sanitise(s: str, max_len: int = 80) -> str: clean = _CTRL_RE.sub("", s).strip() return clean[: max_len - 1] + "…" if len(clean) > max_len else clean def _module_key(address: str, depth: int) -> str: """Return the module prefix for *address* at directory depth *depth*.""" fp = address.split("::")[0] parts = pathlib.PurePosixPath(fp).parts if len(parts) <= 1: return fp return str(pathlib.PurePosixPath(*parts[: min(depth, len(parts) - 1)])) + "/" def _pct_bar(value: float, width: int = 20) -> str: filled = round(max(0.0, min(1.0, value)) * width) return "█" * filled + "░" * (width - filled) def _confidence_label(score: float) -> str: if score >= _CONFIDENCE_HIGH: return "high" if score >= _CONFIDENCE_MED: return "medium" return "low" def _iter_symbol_ops( structured_delta: StructuredDelta | None, ) -> Iterator[DomainOp]: """Safely iterate symbol-level ops from a structured delta, handling None.""" if structured_delta is None: return yield from flat_symbol_ops(structured_delta["ops"]) # ── Core prediction engine ───────────────────────────────────────────────────── def _build_predictions( commits: list[CommitRecord], horizon: int, module_depth: int, ) -> list[_PredictionRecord]: """Compute prediction records from a commit walk. Single-pass design: collects all signals in one iteration over the commits list, then assembles scores and reasons without a second traversal. Args: commits: All commits, newest-first (from ``walk_commits_bfs``). horizon: Number of recent commits used as the prediction window. module_depth: Directory depth for module grouping. Returns: Unsorted list of ``_PredictionRecord``; caller sorts by score. """ # ── Phase 1 — single pass over all commits ──────────────────────────────── # Per-symbol accumulators. sym_first_pos: _CounterMap = {} # lowest commit index (=most recent) sym_freq: Counter[str] = Counter() # times touched in [0, horizon) sym_sig_changes: Counter[str] = Counter() # sig ops in full history sym_total_changes: Counter[str] = Counter() # all ops in full history sym_last_commit: Metadata = {} sym_last_date: _DateMap = {} # Per-commit symbol sets (needed for co-change computation). commit_sym_sets: list[set[str]] = [] # Module velocity: net symbol changes in current and prior window. window = max(horizon // 2, 1) mod_net_current: Counter[str] = Counter() mod_net_prior: Counter[str] = Counter() for i, commit in enumerate(commits): sym_set: set[str] = set() for op in _iter_symbol_ops(commit.structured_delta): addr: str = op["address"] sym_set.add(addr) sym_total_changes[addr] += 1 # Track first (most recent) occurrence. if addr not in sym_first_pos: sym_first_pos[addr] = i sym_last_commit[addr] = commit.commit_id sym_last_date[addr] = commit.committed_at # Frequency inside the horizon window. if i < horizon: sym_freq[addr] += 1 # Signature instability. summary = str(op.get("new_summary") or "").lower() if any(kw in summary for kw in _SIG_KW): sym_sig_changes[addr] += 1 # Module velocity. op_kind = op.get("op", "") delta = ( 1 if op_kind == "insert" else (-1 if op_kind == "delete" else 0) ) if delta != 0: mod = _module_key(addr, module_depth) if i < window: mod_net_current[mod] += delta elif i < 2 * window: mod_net_prior[mod] += delta commit_sym_sets.append(sym_set) # ── Phase 2 — co-change matrix (recent_set × recent_set) ───────────────── # Build the candidate set: symbols touched in the horizon window. # Cap at _MAX_RECENT_SET to bound matrix size. recent_candidates: list[str] = [ addr for addr, freq in sym_freq.most_common(_MAX_RECENT_SET) ] recent_set: frozenset[str] = frozenset(recent_candidates) # co_count[a][b] = commits where both a and b (in recent_set) appeared. co_count: _CoCountMap = {sym: Counter() for sym in recent_set} touch_count: Counter[str] = Counter() for sym_set in commit_sym_sets: recent_in_commit = sym_set & recent_set for sym in recent_in_commit: touch_count[sym] += 1 for partner in recent_in_commit: if partner != sym: co_count[sym][partner] += 1 # ── Phase 3 — score each candidate ─────────────────────────────────────── max_freq = max(sym_freq.values()) if sym_freq else 1 max_mod_net = max( (abs(v) for v in mod_net_current.values()), default=1 ) max_mod_net = max(max_mod_net, 1) predictions: list[_PredictionRecord] = [] for addr in recent_set: if "::" not in addr: continue file_path = addr.split("::")[0] sym_name = addr.split("::", 1)[1] # Signal 1 — recency. pos = sym_first_pos.get(addr, horizon) recency = max(0.0, 1.0 - pos / max(horizon, 1)) # Signal 2 — frequency. frequency = sym_freq[addr] / max_freq # Signal 3 — co-change (best Jaccard with any recent partner). best_rate = 0.0 best_partner: str | None = None best_partner_co = 0 for partner, co in co_count[addr].most_common(_MAX_CO_MATCH): t_a = touch_count[addr] t_b = touch_count.get(partner, 0) denom = t_a + t_b - co if denom > 0: rate = co / denom if rate > best_rate: best_rate = rate best_partner = partner best_partner_co = co co_change_score = best_rate # Signal 4 — signature instability. total = sym_total_changes[addr] sig_instab = sym_sig_changes[addr] / total if total > 0 else 0.0 # Signal 5 — module velocity. mod = _module_key(addr, module_depth) mod_vel_raw = mod_net_current.get(mod, 0) mod_vel_score = min(1.0, abs(mod_vel_raw) / max_mod_net) # Composite score. score = ( _W_RECENCY * recency + _W_FREQUENCY * frequency + _W_CO_CHANGE * co_change_score + _W_SIG_INSTAB * sig_instab + _W_MOD_VELOCITY * mod_vel_score ) confidence = _confidence_label(score) # Reasons — pick the strongest signals. reasons: list[str] = [] if recency >= 0.8: n_ago = pos + 1 reasons.append( f"changed {n_ago} commit{'s' if n_ago != 1 else ''} ago" " — editing momentum" ) if frequency >= 0.60: reasons.append( f"changed {sym_freq[addr]}× in last {horizon} commits" ) if co_change_score >= 0.50 and best_partner is not None: partner_bare = best_partner.split("::")[-1] reasons.append( f"entangled with {partner_bare}" f" ({round(best_rate * 100)}% co-change," f" {best_partner_co} commits)" ) if sig_instab >= 0.50: reasons.append( f"signature changed {sym_sig_changes[addr]}×" " — API still evolving" ) if mod_vel_raw != 0: sign = "+" if mod_vel_raw > 0 else "" reasons.append( f"{mod} velocity: {sign}{mod_vel_raw} symbols/window" ) if not reasons: reasons.append(f"touched within last {horizon} commits") reasons = reasons[:_MAX_REASONS] # Top co-change partners (for --explain and JSON). top_partners: list[_PartnerRecord] = [] for partner, co in co_count[addr].most_common(_MAX_PARTNERS): t_a = touch_count[addr] t_b = touch_count.get(partner, 0) denom = t_a + t_b - co rate = co / denom if denom > 0 else 0.0 if rate > 0: top_partners.append( _PartnerRecord( address=partner, co_change_rate=round(rate, 3), co_change_commits=co, ) ) predictions.append( _PredictionRecord( address=addr, name=sym_name.split(".")[-1], kind="symbol", # enriched later from HEAD snapshot file=file_path, score=round(score, 4), confidence=confidence, reasons=reasons, signals=_SignalSet( recency=round(recency, 3), frequency=round(frequency, 3), co_change=round(co_change_score, 3), sig_instability=round(sig_instab, 3), module_velocity=round(mod_vel_score, 3), ), last_changed_commit=sym_last_commit.get(addr, ""), last_changed_date=( sym_last_date[addr].strftime("%Y-%m-%d") if addr in sym_last_date else "" ), top_partners=top_partners, ) ) predictions.sort(key=lambda r: r["score"], reverse=True) return predictions # ── Formatters ───────────────────────────────────────────────────────────────── def _print_leaderboard( predictions: list[_PredictionRecord], top: int, min_confidence: float, horizon: int, commits_analysed: int, file_filter: str | None, kind_filter: str | None, ) -> None: """Print the ranked prediction leaderboard to stdout.""" filtered = [ r for r in predictions if r["score"] >= min_confidence and (file_filter is None or file_filter in r["file"]) and (kind_filter is None or r["kind"] == kind_filter) ] if top > 0: filtered = filtered[:top] if not filtered: print(" No predictions meet the current filters.") return print( f"\n Predicted changes" f" (horizon: {horizon} commits · {commits_analysed} analysed)\n" ) bands: list[tuple[str, float, float]] = [ ("HIGH CONFIDENCE", _CONFIDENCE_HIGH, 1.01), ("MEDIUM CONFIDENCE", _CONFIDENCE_MED, _CONFIDENCE_HIGH), ("LOW CONFIDENCE", 0.0, _CONFIDENCE_MED), ] for band_label, band_lo, band_hi in bands: band_rows = [r for r in filtered if band_lo <= r["score"] < band_hi] if not band_rows: continue header_suffix = ( "score >= 0.70" if band_lo >= _CONFIDENCE_HIGH else f"score {band_lo:.2f} – {band_hi:.2f}" if band_hi < 1.0 else f"score < {_CONFIDENCE_MED:.2f}" ) print(f" {band_label} ({header_suffix})") print(" " + "─" * 69) for row in band_rows: print(f" {row['score']:.2f} {sanitize_display(row['address'])}") for reason in row["reasons"]: print(f" ↳ {reason}") print() def _print_explain(record: _PredictionRecord, horizon: int) -> None: """Print a detailed signal breakdown for a single symbol.""" confidence_label = record["confidence"].upper() print(f"\n {sanitize_display(record['address'])} — signal breakdown\n") print(f" Score: {record['score']:.2f} {confidence_label}") print(" " + "─" * 53) sig_labels: list[tuple[str, float, str]] = [ ("recency", record["signals"]["recency"], f"changed recently (horizon={horizon})"), ("frequency", record["signals"]["frequency"], "change frequency in window"), ("co_change", record["signals"]["co_change"], "co-change entanglement"), ("sig_instability", record["signals"]["sig_instability"], "signature churn ratio"), ("module_velocity", record["signals"]["module_velocity"], "module growth acceleration"), ] for label, value, desc in sig_labels: bar = _pct_bar(value, width=20) print(f" {label:<20} {value:.2f} [{bar}] {desc}") print() if record["reasons"]: print(" Reasons") for r in record["reasons"]: print(f" ↳ {r}") print() if record["top_partners"]: print(f" Top co-change partners") for p in record["top_partners"]: bare = p["address"].split("::")[-1] pct = round(p["co_change_rate"] * 100) print( f" {bare:<40} {pct:>3}% ({p['co_change_commits']} commits)" ) print() # ── Entry point ──────────────────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Entry point for ``muse code predict``.""" root = require_repo() # ── Validate arguments ──────────────────────────────────────────────────── horizon: int = clamp_int(args.horizon, 1, 1000, 'horizon') max_commits: int = clamp_int(args.max_commits, 1, 100000, 'max_commits') top: int = clamp_int(args.top, 0, 10000, 'top') min_confidence: float = args.min_confidence module_depth: int = clamp_int(args.module_depth, 1, 50, 'module_depth') if horizon < 1: print("❌ --horizon 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.0 <= min_confidence <= 1.0): print("❌ --min-confidence must be between 0.0 and 1.0.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if module_depth < 1: print("❌ --module-depth must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) explain_addr: str | None = getattr(args, "explain", None) if explain_addr is not None and "::" not in explain_addr: print( "❌ --explain ADDRESS must be in ``file::Symbol`` format.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Resolve HEAD ────────────────────────────────────────────────────────── 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 — is this an empty repository?", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Walk commits ────────────────────────────────────────────────────────── commits, truncated = walk_commits_bfs(root, head.commit_id, max_commits) if not commits: print(" No commits found — nothing to predict.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Build predictions ───────────────────────────────────────────────────── predictions = _build_predictions(commits, horizon, module_depth) # ── Enrich kind from HEAD snapshot ─────────────────────────────────────── manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} all_trees = symbols_for_snapshot(root, manifest) addr_to_kind: Metadata = {} for sym_tree in all_trees.values(): for addr, rec in sym_tree.items(): addr_to_kind[addr] = rec.get("kind", "symbol") for pred in predictions: pred["kind"] = addr_to_kind.get(pred["address"], "symbol") # ── Apply kind/file filters ─────────────────────────────────────────────── file_filter: str | None = getattr(args, "file", None) kind_filter: str | None = getattr(args, "kind", None) # ── Output ──────────────────────────────────────────────────────────────── if args.json: filtered = [ r for r in predictions if r["score"] >= min_confidence and (file_filter is None or file_filter in r["file"]) and (kind_filter is None or r["kind"] == kind_filter) ] if top > 0: filtered = filtered[:top] out: _PredictJson = _PredictJson( generated_at=datetime.datetime.now(datetime.timezone.utc) .strftime("%Y-%m-%dT%H:%M:%S"), horizon_commits=horizon, max_commits=max_commits, commits_analysed=len(commits), truncated=truncated, predictions=filtered, ) print(json.dumps(out, indent=2)) return if explain_addr is not None: # Find the prediction record for this address. match = next( (r for r in predictions if r["address"] == explain_addr), None ) if match is None: print( f"❌ {explain_addr!r} not found in prediction set." f" It may not have been touched in the last" f" {horizon} commits.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) _print_explain(match, horizon) return _print_leaderboard( predictions, top, min_confidence, horizon, len(commits), file_filter, kind_filter, ) # ── CLI registration ─────────────────────────────────────────────────────────── def register( sub: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Register ``predict`` under the ``code`` subcommand group.""" p = sub.add_parser( "predict", help=( "Predict which symbols will change next based on recency," " frequency, co-change entanglement, signature instability," " and module velocity." ), description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument( "--horizon", type=int, default=_DEFAULT_HORIZON, metavar="N", help=( f"Number of recent commits used as the prediction window" f" (default: {_DEFAULT_HORIZON})." ), ) p.add_argument( "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", help=( f"Maximum commits to walk for full history signals" f" (default: {_DEFAULT_MAX_COMMITS})." ), ) p.add_argument( "--top", type=int, default=_DEFAULT_TOP, metavar="N", help=( f"Show top N predictions (default: {_DEFAULT_TOP};" " 0 = all)." ), ) p.add_argument( "--min-confidence", type=float, default=0.0, metavar="F", help=( "Minimum score threshold in [0.0, 1.0] (default: 0.0 = show all)." ), ) p.add_argument( "--explain", metavar="ADDRESS", help=( "Show the full signal breakdown for a specific symbol" " (e.g. billing.py::Invoice.compute_total)." ), ) p.add_argument( "--kind", metavar="KIND", help=( "Filter predictions to a specific symbol kind" " (function, method, class, …)." ), ) p.add_argument( "--file", metavar="PATTERN", help="Filter predictions to symbols whose file path contains PATTERN.", ) p.add_argument( "--module-depth", type=int, default=2, metavar="D", help=( "Directory depth used to group symbols into modules" " for the velocity signal (default: 2)." ), ) p.add_argument( "--json", action="store_true", help="Emit JSON instead of human-readable text.", ) p.set_defaults(func=run)