"""muse code hotspots -- symbol churn leaderboard. Walks the commit history and counts how many commits touched each symbol. High churn = instability signal. The functions that change most are the ones that need the most attention -- refactoring targets, test coverage gaps, or domain logic under active evolution. Unlike file-level churn metrics, ``muse code hotspots`` operates at the *symbol* level: a 5,000-line module with one unstable function shows that function at the top, not the whole file. Import pseudo-symbols (``::import::*``) are excluded by default because they almost always reflect dependency management rather than logic churn. Pass ``--include-imports`` to include them. Usage:: muse code hotspots muse code hotspots --top 20 muse code hotspots --kind function --language Python muse code hotspots --from HEAD~30 --to HEAD muse code hotspots --min 3 # only symbols that changed >= 3 times muse code hotspots --json # machine-readable for agents Output:: Symbol churn -- top 10 most-changed symbols Commits analysed: 47 1 src/billing.py::compute_invoice_total 12 changes 2 src/api.py::handle_request 9 changes 3 src/auth.py::validate_token 7 changes 4 src/models.py::User.save 5 changes High churn = instability signal. """ from __future__ import annotations import argparse import json import logging import pathlib import sys 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, language_of, walk_commits_bfs, ) logger = logging.getLogger(__name__) _DEFAULT_TOP = 20 _DEFAULT_MAX_COMMITS = 10_000 # Canonical kind names produced by Muse's AST parser summaries. _KNOWN_KINDS: frozenset[str] = frozenset({ "function", "async_function", "class", "method", "async_method", "variable", "import", "section", "rule", }) # Language normalisation (case-insensitive --language matching). from muse.plugins.code._query import _SUFFIX_LANG from muse.core.validation import clamp_int, sanitize_display type _IntMap = dict[str, int] type _StrMap = dict[str, str] _LANG_CANONICAL: _StrMap = {lang.lower(): lang for lang in set(_SUFFIX_LANG.values())} def _normalise_language(lang: str) -> str: return _LANG_CANONICAL.get(lang.strip().lower(), lang.strip()) def _kind_from_op(op: DomainOp) -> str: """Extract the symbol kind from the op's summary fields. ``replace`` ops carry the kind in ``old_summary`` as the first word:: "function _collect_paths (implementation)" → "function" ``insert`` / ``delete`` ops carry it in ``content_summary`` as the second word (after "added" / "removed"):: "added function test_fn L10–20" → "function" "removed import json L5–5" → "import" """ if op["op"] == "replace": raw = op.get("old_summary") summary: str = raw if isinstance(raw, str) else "" parts = summary.split(None, 1) if parts and parts[0] in _KNOWN_KINDS: return parts[0] else: raw2 = op.get("content_summary") summary2: str = raw2 if isinstance(raw2, str) else "" parts2 = summary2.split() if len(parts2) >= 2 and parts2[1] in _KNOWN_KINDS: return parts2[1] return "" # --------------------------------------------------------------------------- # Repository helpers # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Churn collection # --------------------------------------------------------------------------- def _collect_churn( root: pathlib.Path, to_commit_id: str, from_commit_id: str | None, kind_filter: str | None, language_filter: str | None, include_imports: bool, max_commits: int, ) -> tuple[dict[str, int], int, bool]: """Return ``(churn_counts, commits_analysed, truncated)``. Uses a BFS walk that follows both ``parent_commit_id`` and ``parent2_commit_id``, so events on merged feature branches are included. """ commits, truncated = walk_commits_bfs( root, to_commit_id, max_commits, stop_at_commit_id=from_commit_id ) counts: _IntMap = {} for commit in commits: if commit.structured_delta is None: continue for op in flat_symbol_ops(commit.structured_delta["ops"]): addr: str = op["address"] # Exclude import pseudo-symbols unless requested. if not include_imports and "::import::" in addr: continue file_path = addr.split("::")[0] if language_filter and language_of(file_path) != language_filter: continue if kind_filter: if _kind_from_op(op) != kind_filter: continue counts[addr] = counts.get(addr, 0) + 1 return counts, len(commits), truncated # --------------------------------------------------------------------------- # Argument parser registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the hotspots subcommand.""" parser = subparsers.add_parser( "hotspots", help="Show the symbols that change most often — the churn leaderboard.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--top", "-n", type=int, default=_DEFAULT_TOP, metavar="N", dest="top", help=f"Number of symbols to show (default: {_DEFAULT_TOP}).", ) parser.add_argument( "--min", type=int, default=1, metavar="N", dest="min_changes", help="Only show symbols that changed at least N times (default: 1).", ) 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( "--language", "-l", default=None, metavar="LANG", dest="language_filter", help="Restrict to symbols from files of this language (case-insensitive).", ) parser.add_argument( "--include-imports", action="store_true", dest="include_imports", help="Include import pseudo-symbols (excluded by default).", ) parser.add_argument( "--from", default=None, metavar="REF", dest="from_ref", help="Exclusive start of the commit range (default: initial commit).", ) parser.add_argument( "--to", default=None, metavar="REF", dest="to_ref", help="Inclusive end of the commit range (default: HEAD).", ) 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 structured JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Show the symbols that change most often — the churn leaderboard. Walks the commit history (BFS, follows both parents of merge commits) and counts how many commits touched each symbol. High churn at the function level reveals instability that file-level metrics miss: a single file may be stable while one specific function inside it burns. Import pseudo-symbols (``::import::*``) are excluded by default. Use ``--include-imports`` to include them. Use ``--from`` / ``--to`` to scope the analysis to a sprint or release. Use ``--kind function`` to focus on functions only. Use ``--min 3`` to surface only the truly hot symbols. """ top: int = clamp_int(args.top, 1, 10_000, 'top') min_changes: int = clamp_int(args.min_changes, 0, 100000, 'min_changes') kind_filter: str | None = args.kind_filter language_filter: str | None = args.language_filter include_imports: bool = args.include_imports from_ref: str | None = args.from_ref to_ref: str | None = args.to_ref max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') as_json: bool = args.as_json # ── Validation ──────────────────────────────────────────────────────────── if top < 1: print("❌ --top must be at least 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if min_changes < 1: print("❌ --min must be at least 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if max_commits < 1: print("❌ --max-commits must be at least 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if language_filter is not None: language_filter = _normalise_language(language_filter) # ── Repo / commit resolution ────────────────────────────────────────────── root = require_repo() # Knowtation domain: delegate to the vault-aware note-churn implementation. from muse.plugins.registry import read_domain try: if read_domain(root) == "knowtation": from muse.plugins.knowtation.cli_hooks import run_hotspots_for_vault run_hotspots_for_vault(root, args) return except Exception: # noqa: BLE001 — fall through to Code domain on any failure pass repo_id = read_repo_id(root) branch = read_current_branch(root) to_commit = resolve_commit_ref(root, repo_id, branch, to_ref) if to_commit is None: print(f"❌ Commit '{to_ref or 'HEAD'}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from_commit_id: str | None = None if from_ref is not None: from_commit = resolve_commit_ref(root, repo_id, branch, from_ref) if from_commit is None: print(f"❌ Commit '{from_ref}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from_commit_id = from_commit.commit_id # ── Churn analysis ──────────────────────────────────────────────────────── counts, total_commits, truncated = _collect_churn( root, to_commit.commit_id, from_commit_id, kind_filter, language_filter, include_imports, max_commits, ) # Apply --min filter before ranking. if min_changes > 1: counts = {addr: n for addr, n in counts.items() if n >= min_changes} ranked = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)[:top] # ── Output ──────────────────────────────────────────────────────────────── if as_json: print(json.dumps({ "from_ref": from_ref, "to_ref": to_ref or branch, "commits_analysed": total_commits, "truncated": truncated, "filters": { "kind": kind_filter, "language": language_filter, "include_imports": include_imports, "min_changes": min_changes, }, "hotspots": [{"address": a, "changes": c} for a, c in ranked], }, indent=2)) return if not ranked: print(" (no symbol-level changes found in this range)") return filters_desc = "" if kind_filter: filters_desc += f" kind={kind_filter}" if language_filter: filters_desc += f" language={language_filter}" if min_changes > 1: filters_desc += f" min={min_changes}" print(f"\nSymbol churn — top {len(ranked)} most-changed symbols{filters_desc}") print(f"Commits analysed: {total_commits}", end="") if truncated: print(f" ⚠️ (capped at --max-commits {max_commits})", end="") print("\n") width = len(str(len(ranked))) for rank, (addr, count) in enumerate(ranked, 1): label = "change" if count == 1 else "changes" print(f" {rank:>{width}} {sanitize_display(addr):<60} {count:>4} {label}") print("") print("High churn = instability signal. Consider refactoring or adding tests.")