"""muse code velocity — symbol-growth rate by module, with acceleration and next-change prediction. This is the *energy map* of a codebase. ``muse code hotspots`` tells you which symbols changed most. ``muse code velocity`` tells you where the codebase is growing, where it is shrinking, where it has stalled — and how fast each trend is accelerating. It also answers a forward-looking question: **which symbols are most likely to change in the next few commits?** This is computed statistically from recency, frequency, and module acceleration. Why this matters ---------------- File commit counts or line-change counts obscure the signal. A file with 500 line changes might just be a re-format. A module that gained 12 new symbols across 8 commits is actively expanding its API surface — that is genuine architectural investment. Three windows ------------- Velocity is computed over two consecutive commit windows of equal size (``--window N``, default 20): * **Current window** — the most recent N commits. * **Prior window** — the N commits before that. The difference between the two gives **acceleration**: * Positive acceleration = module growing faster than before. * Negative acceleration = module is decelerating / winding down. * Zero acceleration from zero velocity = stagnation. Prediction ---------- ``--predict K`` outputs the top K symbols most likely to change in the next commit, ranked by a composite score: ``score = frequency × recency_weight × module_velocity_weight`` where ``recency_weight = 1 / (1 + rank)`` (rank 0 = most recent commit). This is a statistical signal, not a guarantee. Usage:: muse code velocity muse code velocity --window 10 --top 10 muse code velocity --predict 5 muse code velocity --since v1.0 muse code velocity --json Output:: Symbol velocity — HEAD (40 commits · window: 20) Sorted by: net growth, current window MODULE ADD DEL NET MOD ACCEL BAR muse/core/ +14 -2 +12 47 ▲ +4 ████████████ muse/cli/commands/ +11 -1 +10 31 ▲ +2 ██████████ tests/ +8 -0 +8 12 ▲ +3 ████████ muse/plugins/ +3 -5 -2 8 ▼ -1 ▏ (net negative) docs/ 0 -0 0 0 ─ (stagnant 15 commits) Acceleration leaders: muse/core/ (+4 net vs prior window) Stagnant modules: docs/ (0 changes in 18 commits) --predict output:: Next-change predictions (top 5 by statistical likelihood): 1 muse/core/store.py::resolve_commit_ref score: 0.91 2 muse/cli/commands/dead.py::run score: 0.84 3 tests/test_code_commands.py::TestHotspots score: 0.71 JSON output (--json):: { "ref": "HEAD", "window_size": 20, "commits_analysed": 40, "truncated": false, "filters": { "top": 20, "since": null }, "modules": [ { "module": "muse/core/", "current": { "added": 14, "removed": 2, "net": 12, "modified": 47, "active_commits": 18 }, "prior": { "added": 10, "removed": 2, "net": 8, "modified": 33, "active_commits": 14 }, "acceleration": 4, "stagnant_commits": 0 } ], "predictions": [ { "address": "muse/core/store.py::resolve_commit_ref", "score": 0.91 } ] } """ from __future__ import annotations import argparse import json import logging import pathlib import sys from dataclasses import dataclass, field 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 read_current_branch, resolve_commit_ref from muse.domain import DomainOp from muse.plugins.code._query import flat_symbol_ops, walk_commits_bfs from muse.core.validation import clamp_int, sanitize_display type _CounterMap = dict[str, int] type _ModAccumMap = dict[str, "_ModuleAccumulator"] type _SymFreqMap = dict[str, "_SymbolFreq"] logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_WINDOW = 20 _DEFAULT_TOP = 20 _DEFAULT_MAX_COMMITS = 10_000 _DEFAULT_PREDICT = 0 # 0 = disabled _BAR_WIDTH = 20 # max bar width in characters # Commits with more than this many symbol ops are mass-refactors — skip for # module velocity (they would unfairly spike a module's counts). _MAX_OPS_PER_COMMIT = 500 # ── Helpers ──────────────────────────────────────────────────────────────────── def _module_of(file_path: str) -> str: """Return the containing directory of a file path. ``muse/core/store.py`` → ``muse/core/`` ``tests/test_billing.py`` → ``tests/`` ``billing.py`` → ``(root)`` """ parts = file_path.replace("\\", "/").rsplit("/", 1) if len(parts) == 1: return "(root)" return parts[0] + "/" def _bar(net: int, max_abs: int) -> str: """Return a unicode block bar proportional to *net* / *max_abs*.""" if max_abs == 0: return "" ratio = abs(net) / max_abs filled = round(ratio * _BAR_WIDTH) bar = "█" * filled if net < 0: bar = bar or "▏" return f"{bar} (net negative)" return bar or "▏" # ── Data types ───────────────────────────────────────────────────────────────── @dataclass class _WindowStats: added: int = 0 removed: int = 0 modified: int = 0 active_commits: int = 0 @property def net(self) -> int: return self.added - self.removed @dataclass class _ModuleAccumulator: current: _WindowStats = field(default_factory=_WindowStats) prior: _WindowStats = field(default_factory=_WindowStats) last_active_rank: int = -1 # commit rank (0 = HEAD) of last activity stagnant_commits: int = 0 # consecutive commits with no activity class _ModuleOut(TypedDict): module: str current: _CounterMap prior: _CounterMap acceleration: int stagnant_commits: int class _PredictionOut(TypedDict): address: str module: str score: float frequency: int last_commit_rank: int # ── Core algorithm ───────────────────────────────────────────────────────────── @dataclass class _SymbolFreq: frequency: int = 0 last_rank: int = 0 # 0 = most recent commit module: str = "" def _walk_and_collect( root: pathlib.Path, head_commit_id: str, stop_at: str | None, window_size: int, max_commits: int, ) -> tuple[ dict[str, _ModuleAccumulator], # per-module stats dict[str, _SymbolFreq], # per-symbol frequency (current window only) int, # total commits analysed bool, # truncated ]: """Single BFS pass building module velocity and symbol frequency data.""" commits, truncated = walk_commits_bfs( root, head_commit_id, max_commits, stop_at_commit_id=stop_at ) modules: _ModAccumMap = {} symbol_freq: _SymFreqMap = {} # Track which modules were active per commit (for stagnation detection). # commits are sorted newest-first after BFS. for rank, commit in enumerate(commits): if commit.structured_delta is None: continue ops: list[DomainOp] = commit.structured_delta["ops"] # Gather all leaf symbol ops for this commit. all_ops = list(flat_symbol_ops(ops)) # Skip mass-refactor commits. if len(all_ops) > _MAX_OPS_PER_COMMIT: continue # Modules that had activity in this commit (for stagnation tracking). active_modules: set[str] = set() for op in all_ops: addr: str = op["address"] if "::import::" in addr: continue file_path = addr.split("::")[0] mod = _module_of(file_path) active_modules.add(mod) acc = modules.setdefault(mod, _ModuleAccumulator()) op_kind = op.get("op", "") # Determine which window this commit belongs to. in_current = rank < window_size in_prior = window_size <= rank < 2 * window_size if in_current: if op_kind == "insert": acc.current.added += 1 elif op_kind == "delete": acc.current.removed += 1 elif op_kind == "replace": acc.current.modified += 1 elif in_prior: if op_kind == "insert": acc.prior.added += 1 elif op_kind == "delete": acc.prior.removed += 1 elif op_kind == "replace": acc.prior.modified += 1 # Symbol frequency (current window only) for prediction. if in_current: sf = symbol_freq.setdefault(addr, _SymbolFreq(module=mod)) sf.frequency += 1 if sf.last_rank == 0 or rank < sf.last_rank: sf.last_rank = rank # Track active_commits per window. for mod in active_modules: acc = modules.setdefault(mod, _ModuleAccumulator()) if rank < window_size: acc.current.active_commits += 1 elif rank < 2 * window_size: acc.prior.active_commits += 1 # Update last active rank. if acc.last_active_rank < 0 or rank < acc.last_active_rank: acc.last_active_rank = rank # Compute stagnant_commits: how many leading commits (from HEAD) had # zero activity for this module. for mod, acc in modules.items(): if acc.last_active_rank < 0: acc.stagnant_commits = len(commits) else: acc.stagnant_commits = acc.last_active_rank return modules, symbol_freq, len(commits), truncated # ── Prediction ───────────────────────────────────────────────────────────────── def _compute_predictions( symbol_freq: _SymFreqMap, modules: _ModAccumMap, window_size: int, top_k: int, ) -> list[_PredictionOut]: """Score each symbol in the current window and return top-K predictions. Score = frequency × recency_weight × module_velocity_weight recency_weight = 1 / (1 + last_commit_rank) (0 = most recent commit → weight 1.0) module_velocity = max(0, net_current) (growth modules get a boost) module_vel_weight = 1.0 + normalised module velocity """ if not symbol_freq or top_k <= 0: return [] # Normalise module velocity (0..1 range). max_net = max( (max(0, modules[sf.module].current.net) for sf in symbol_freq.values() if sf.module in modules), default=0, ) or 1 scored: list[tuple[float, _PredictionOut]] = [] for addr, sf in symbol_freq.items(): if sf.frequency == 0: continue recency_w = 1.0 / (1.0 + sf.last_rank) mod_net = max(0, modules[sf.module].current.net) if sf.module in modules else 0 mod_w = 1.0 + (mod_net / max_net) # 1.0 .. 2.0 score = round(sf.frequency * recency_w * mod_w, 4) scored.append((score, _PredictionOut( address=addr, module=sf.module, score=score, frequency=sf.frequency, last_commit_rank=sf.last_rank, ))) scored.sort(key=lambda t: -t[0]) return [out for _, out in scored[:top_k]] # ── Formatters ───────────────────────────────────────────────────────────────── def _print_table( ranked: list[tuple[str, _ModuleAccumulator]], predictions: list[_PredictionOut], ref: str, commits_analysed: int, window_size: int, truncated: bool, since: str | None, ) -> None: scope = f"{since}..{ref}" if since else ref trunc = " ⚠️ truncated" if truncated else "" print( f"\nSymbol velocity — {scope}" f" ({commits_analysed} commits · window: {window_size}{trunc})" ) print("Sorted by: net growth, current window\n") if not ranked: print(" (no modules with symbol-level changes found)") return max_abs = max(abs(acc.current.net) for _, acc in ranked) or 1 max_mod = max(len(mod) for mod, _ in ranked) hdr = ( f" {'MODULE':<{max_mod}} {'ADD':>5} {'DEL':>5} {'NET':>5} " f"{'MOD':>5} {'ACCEL':>7} BAR" ) print(hdr) print(" " + "─" * (len(hdr) - 2)) accel_leaders: list[str] = [] stagnant: list[tuple[str, int]] = [] for mod, acc in ranked: accel = acc.current.net - acc.prior.net if accel > 0: accel_str = f"▲ +{accel}" elif accel < 0: accel_str = f"▼ {accel}" else: accel_str = "─" bar = _bar(acc.current.net, max_abs) add_str = f"+{acc.current.added}" if acc.current.added else "0" del_str = f"-{acc.current.removed}" if acc.current.removed else "0" net_str = f"+{acc.current.net}" if acc.current.net > 0 else str(acc.current.net) stag = acc.stagnant_commits if stag > 0: stagnant.append((mod, stag)) note = f" (stagnant {stag} commit{'s' if stag != 1 else ''})" print( f" {mod:<{max_mod}} {add_str:>5} {del_str:>5} " f"{net_str:>5} {acc.current.modified:>5} {accel_str:>7} {note}" ) else: print( f" {mod:<{max_mod}} {add_str:>5} {del_str:>5} " f"{net_str:>5} {acc.current.modified:>5} {accel_str:>7} {bar}" ) if accel >= 2: accel_leaders.append(f"{mod} (+{accel} net vs prior window)") print("") if accel_leaders: print(f"Acceleration leaders: {', '.join(accel_leaders[:3])}") stag_str = ", ".join(f"{m} ({n} commits)" for m, n in stagnant[:3]) if stag_str: print(f"Stagnant modules: {stag_str}") if predictions: print(f"\nNext-change predictions (top {len(predictions)}):") for i, pred in enumerate(predictions, 1): print(f" {i:>2} {sanitize_display(pred['address']):<60} score: {pred['score']:.2f}") # ── CLI ──────────────────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the velocity subcommand.""" parser = subparsers.add_parser( "velocity", help=( "Symbol-growth rate by module — where the codebase is growing, " "shrinking, accelerating, or stagnating." ), description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--window", "-w", type=int, default=_DEFAULT_WINDOW, metavar="N", help=( f"Commits per analysis window (default: {_DEFAULT_WINDOW}). " f"Two consecutive windows are compared to compute acceleration." ), ) parser.add_argument( "--top", "-n", type=int, default=_DEFAULT_TOP, metavar="N", help=f"Number of modules to show (default: {_DEFAULT_TOP}).", ) parser.add_argument( "--predict", "-p", type=int, default=_DEFAULT_PREDICT, metavar="K", dest="predict", help=( "Show the top K symbols most likely to change in the next commit " "(default: 0 = disabled). " "Ranked by: recency × frequency × module-velocity." ), ) parser.add_argument( "--since", "-s", default=None, metavar="REF", help="Limit analysis to commits reachable from HEAD but not from REF.", ) parser.add_argument( "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", dest="max_commits", help=f"Maximum commits to scan (default: {_DEFAULT_MAX_COMMITS}).", ) 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: """Compute symbol-growth velocity by module. Mines the commit history for symbol insert/delete/modify operations and aggregates them by module (containing directory). Compares the current window to the prior window to detect acceleration and stagnation. The ``--predict K`` flag uses recency, frequency, and module velocity to rank the top K symbols most likely to change in the next commit. """ window: int = clamp_int(args.window, 1, 1000, 'window') top: int = clamp_int(args.top, 1, 10_000, 'top') predict_k: int = clamp_int(args.predict, 0, 1000, 'predict_k') since: str | None = args.since max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') as_json: bool = args.as_json # ── Validation ──────────────────────────────────────────────────────────── if window < 1: print("❌ --window must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if top < 1: print("❌ --top must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if predict_k < 0: print("❌ --predict must be >= 0.", 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) # Need at least 2 windows to compute acceleration. effective_max = max(max_commits, window * 2) # ── Repo setup ──────────────────────────────────────────────────────────── root = require_repo() from muse.plugins.registry import read_domain try: if read_domain(root) == "knowtation": from muse.plugins.knowtation.cli_hooks import run_velocity_for_vault run_velocity_for_vault(root, args) return except Exception: # noqa: BLE001 pass 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 # ── Main pass ───────────────────────────────────────────────────────────── modules, symbol_freq, commits_analysed, truncated = _walk_and_collect( root, head.commit_id, stop_at, window, effective_max ) # ── Rank modules by current-window net growth ────────────────────────────── ranked = sorted( modules.items(), key=lambda kv: (-kv[1].current.net, -kv[1].current.modified, kv[0]), )[:top] # ── Predictions ─────────────────────────────────────────────────────────── predictions = _compute_predictions(symbol_freq, modules, window, predict_k) # ── Output ──────────────────────────────────────────────────────────────── if as_json: modules_out: list[_ModuleOut] = [ _ModuleOut( module=mod, current={ "added": acc.current.added, "removed": acc.current.removed, "net": acc.current.net, "modified": acc.current.modified, "active_commits": acc.current.active_commits, }, prior={ "added": acc.prior.added, "removed": acc.prior.removed, "net": acc.prior.net, "modified": acc.prior.modified, "active_commits": acc.prior.active_commits, }, acceleration=acc.current.net - acc.prior.net, stagnant_commits=acc.stagnant_commits, ) for mod, acc in ranked ] print(json.dumps({ "ref": branch, "window_size": window, "commits_analysed": commits_analysed, "truncated": truncated, "filters": { "top": top, "since": since, "predict": predict_k, "max_commits": max_commits, }, "modules": [dict(m) for m in modules_out], "predictions": [dict(p) for p in predictions], }, indent=2)) return _print_table(ranked, predictions, branch, commits_analysed, window, truncated, since)