"""muse code coupling — file co-change analysis. Identifies files that change together most often. High co-change frequency between two files signals a hidden dependency — they are logically coupled even if there is no explicit import between them. This is structurally impossible in Git at the semantic level: Git could count raw file modifications, but ``muse code coupling`` counts only *semantic* co-changes — commits where both files had AST-level symbol modifications, not formatting-only edits (which Muse already separates from real changes). Commits that touch more than 50 files semantically are skipped — they are almost always mass-renames or initial imports whose coupling signal is noise, not signal. Usage:: muse code coupling muse code coupling --top 20 muse code coupling --from HEAD~30 muse code coupling --file muse/cli/commands/stable.py # focus on one file muse code coupling --json # machine-readable Output:: File co-change analysis — top 10 most coupled pairs Commits analysed: 302 1 muse/cli/commands/symbol_log.py ↔ tests/test_code_commands.py co-changed in 3 commits 2 muse/plugins/code/_query.py ↔ tests/test_code_commands.py co-changed in 2 commits High coupling = hidden dependency. Consider extracting a shared interface. """ 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 JsonValue, read_current_branch, resolve_commit_ref from muse.plugins.code._query import file_pairs, touched_files, walk_commits_bfs from muse.core.validation import clamp_int, sanitize_display logger = logging.getLogger(__name__) _DEFAULT_TOP = 20 _DEFAULT_MIN = 2 _DEFAULT_MAX_COMMITS = 10_000 # Commits touching more than this many files semantically are skipped — # they are mass-renames or bulk imports with no meaningful coupling signal, # and they would generate O(N²) pair combinations. _MAX_FILES_PER_COMMIT = 50 def _resolve_file_suffix( file_filter: str, all_files: set[str], ) -> str | None: """Return the unique file from *all_files* whose path ends with *file_filter*. Returns ``None`` if no file matches. Prints a diagnostic and exits if more than one file matches (ambiguous). """ matches = [f for f in all_files if f == file_filter or f.endswith("/" + file_filter)] if len(matches) == 1: return matches[0] if len(matches) > 1: print(f"❌ --file {file_filter!r} is ambiguous — multiple matches:", file=sys.stderr) for m in sorted(matches): print(f" {m}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return None def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the coupling subcommand.""" parser = subparsers.add_parser( "coupling", help="Find files that change together most often — hidden dependencies.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--top", "-n", type=int, default=_DEFAULT_TOP, metavar="N", help=f"Number of pairs to show (default: {_DEFAULT_TOP}).", ) 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( "--min", type=int, default=_DEFAULT_MIN, metavar="N", dest="min_count", help=f"Minimum co-change count to include (default: {_DEFAULT_MIN}).", ) parser.add_argument( "--file", "-f", default=None, metavar="FILE", dest="file_filter", help=( "Focus on a single file: show only its coupling partners " "and rank them by co-change count. " "Accepts a suffix (e.g. 'billing.py') or a full path." ), ) parser.add_argument( "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", 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: """Find files that change together most often — hidden dependencies. ``muse code coupling`` identifies semantic co-change: file pairs that had AST-level symbol modifications in the same commit. This is stricter than raw file co-change — formatting-only edits and non-code files are excluded. Use ``--file`` to focus on a single file and see all its coupling partners ranked. Use ``--from`` / ``--to`` to scope the analysis to a sprint or release. Use ``--min`` to raise the minimum co-change threshold. """ top: int = clamp_int(args.top, 1, 10_000, 'top') from_ref: str | None = args.from_ref to_ref: str | None = args.to_ref min_count: int = clamp_int(args.min_count, 1, 100_000, 'min_count') file_filter: str | None = args.file_filter max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') as_json: bool = args.as_json if top < 1: print("❌ --top must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if min_count < 1: print("❌ --min 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) root = require_repo() 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) stop_at: 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) stop_at = from_commit.commit_id commits, truncated = walk_commits_bfs( root, to_commit.commit_id, max_commits=max_commits, stop_at_commit_id=stop_at, ) # Collect all file paths seen across any commit (for --file suffix matching). all_seen_files: set[str] = set() for commit in commits: if commit.structured_delta is None: continue all_seen_files.update(touched_files(commit.structured_delta["ops"])) # Resolve --file suffix to a canonical path, if requested. resolved_file: str | None = None if file_filter is not None: resolved_file = _resolve_file_suffix(file_filter, all_seen_files) if resolved_file is None: # No commits touched this file — either the path is wrong or it has # never had a semantic change. if as_json: print(json.dumps({ "from_ref": from_ref, "to_ref": to_ref or branch, "commits_analysed": len(commits), "truncated": truncated, "filters": { "top": top, "min_count": min_count, "file": file_filter, "max_commits": max_commits, }, "file": file_filter, "pairs": [], }, indent=2)) else: print(f"\nNo semantic co-changes found for {file_filter!r}.") print("The file may not exist or may never have had symbol-level changes.") return # Count co-changing pairs. pair_counts: dict[tuple[str, str], int] = {} for commit in commits: if commit.structured_delta is None: continue files = touched_files(commit.structured_delta["ops"]) if len(files) < 2: continue # Skip commits that touched too many files — they add noise, not signal. if len(files) > _MAX_FILES_PER_COMMIT: continue for a, b in file_pairs(files): # When --file is set, only count pairs involving that file. if resolved_file is not None and resolved_file not in (a, b): continue pair_counts[(a, b)] = pair_counts.get((a, b), 0) + 1 filtered = {pair: cnt for pair, cnt in pair_counts.items() if cnt >= min_count} ranked = sorted(filtered.items(), key=lambda kv: kv[1], reverse=True)[:top] if as_json: pairs_out: list[dict[str, JsonValue]] if resolved_file is not None: # When --file is set, emit partner + count rather than a/b pair. pairs_out = [ { "file": resolved_file, "partner": b if a == resolved_file else a, "co_changes": c, } for (a, b), c in ranked ] else: pairs_out = [ {"file_a": a, "file_b": b, "co_changes": c} for (a, b), c in ranked ] print(json.dumps( { "from_ref": from_ref, "to_ref": to_ref or branch, "commits_analysed": len(commits), "truncated": truncated, "filters": { "top": top, "min_count": min_count, "file": file_filter, "max_commits": max_commits, }, "pairs": pairs_out, }, indent=2, )) return # Human-readable output. if resolved_file is not None: print(f"\nCoupling partners of {resolved_file}") else: print(f"\nFile co-change analysis — top {len(ranked)} most coupled pairs") print(f"Commits analysed: {len(commits)}") if truncated: print(f"⚠️ Scan capped at {max_commits} commits — pass --max-commits to extend.") print("") if not ranked: threshold_msg = f"{min_count}+" if min_count > 1 else "2+" if resolved_file: print(f" (no files co-changed with {sanitize_display(str(resolved_file))!r} {threshold_msg} times)") else: print(f" (no file pairs co-changed {threshold_msg} times)") return width = len(str(len(ranked))) if resolved_file is not None: # Partner-focused display. max_partner = max(len(b if a == resolved_file else a) for (a, b), _ in ranked) for rank, ((a, b), count) in enumerate(ranked, 1): partner = b if a == resolved_file else a label = "commit" if count == 1 else "commits" print(f" {rank:>{width}} {sanitize_display(partner):<{max_partner}} co-changed in {count:>3} {label}") else: max_a = max(len(a) for (a, _), _ in ranked) for rank, ((a, b), count) in enumerate(ranked, 1): label = "commit" if count == 1 else "commits" print( f" {rank:>{width}} {sanitize_display(a):<{max_a}} ↔ {sanitize_display(b):<50} " f"co-changed in {count:>3} {label}" ) print("") if resolved_file: print(f"These files always change with {sanitize_display(str(resolved_file))}. Hidden coupling.") else: print("High coupling = hidden dependency. Consider extracting a shared interface.")