coupling.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """muse code coupling — file co-change analysis. |
| 2 | |
| 3 | Identifies files that change together most often. High co-change frequency |
| 4 | between two files signals a hidden dependency — they are logically coupled |
| 5 | even if there is no explicit import between them. |
| 6 | |
| 7 | This is structurally impossible in Git at the semantic level: Git could |
| 8 | count raw file modifications, but ``muse code coupling`` counts only |
| 9 | *semantic* co-changes — commits where both files had AST-level symbol |
| 10 | modifications, not formatting-only edits (which Muse already separates |
| 11 | from real changes). |
| 12 | |
| 13 | Commits that touch more than 50 files semantically are skipped — they |
| 14 | are almost always mass-renames or initial imports whose coupling signal is |
| 15 | noise, not signal. |
| 16 | |
| 17 | Usage:: |
| 18 | |
| 19 | muse code coupling |
| 20 | muse code coupling --top 20 |
| 21 | muse code coupling --from HEAD~30 |
| 22 | muse code coupling --file muse/cli/commands/stable.py # focus on one file |
| 23 | muse code coupling --json # machine-readable |
| 24 | |
| 25 | Output:: |
| 26 | |
| 27 | File co-change analysis — top 10 most coupled pairs |
| 28 | Commits analysed: 302 |
| 29 | |
| 30 | 1 muse/cli/commands/symbol_log.py ↔ tests/test_code_commands.py co-changed in 3 commits |
| 31 | 2 muse/plugins/code/_query.py ↔ tests/test_code_commands.py co-changed in 2 commits |
| 32 | |
| 33 | High coupling = hidden dependency. Consider extracting a shared interface. |
| 34 | """ |
| 35 | |
| 36 | from __future__ import annotations |
| 37 | |
| 38 | import argparse |
| 39 | import json |
| 40 | import logging |
| 41 | import pathlib |
| 42 | import sys |
| 43 | |
| 44 | from muse.core.errors import ExitCode |
| 45 | from muse.core.repo import read_repo_id, require_repo |
| 46 | from muse.core.store import JsonValue, read_current_branch, resolve_commit_ref |
| 47 | from muse.plugins.code._query import file_pairs, touched_files, walk_commits_bfs |
| 48 | from muse.core.validation import clamp_int, sanitize_display |
| 49 | |
| 50 | logger = logging.getLogger(__name__) |
| 51 | |
| 52 | _DEFAULT_TOP = 20 |
| 53 | _DEFAULT_MIN = 2 |
| 54 | _DEFAULT_MAX_COMMITS = 10_000 |
| 55 | # Commits touching more than this many files semantically are skipped — |
| 56 | # they are mass-renames or bulk imports with no meaningful coupling signal, |
| 57 | # and they would generate O(N²) pair combinations. |
| 58 | _MAX_FILES_PER_COMMIT = 50 |
| 59 | |
| 60 | |
| 61 | |
| 62 | def _resolve_file_suffix( |
| 63 | file_filter: str, |
| 64 | all_files: set[str], |
| 65 | ) -> str | None: |
| 66 | """Return the unique file from *all_files* whose path ends with *file_filter*. |
| 67 | |
| 68 | Returns ``None`` if no file matches. Prints a diagnostic and exits if |
| 69 | more than one file matches (ambiguous). |
| 70 | """ |
| 71 | matches = [f for f in all_files if f == file_filter or f.endswith("/" + file_filter)] |
| 72 | if len(matches) == 1: |
| 73 | return matches[0] |
| 74 | if len(matches) > 1: |
| 75 | print(f"❌ --file {file_filter!r} is ambiguous — multiple matches:", file=sys.stderr) |
| 76 | for m in sorted(matches): |
| 77 | print(f" {m}", file=sys.stderr) |
| 78 | raise SystemExit(ExitCode.USER_ERROR) |
| 79 | return None |
| 80 | |
| 81 | |
| 82 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 83 | """Register the coupling subcommand.""" |
| 84 | parser = subparsers.add_parser( |
| 85 | "coupling", |
| 86 | help="Find files that change together most often — hidden dependencies.", |
| 87 | description=__doc__, |
| 88 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 89 | ) |
| 90 | parser.add_argument( |
| 91 | "--top", "-n", type=int, default=_DEFAULT_TOP, metavar="N", |
| 92 | help=f"Number of pairs to show (default: {_DEFAULT_TOP}).", |
| 93 | ) |
| 94 | parser.add_argument( |
| 95 | "--from", default=None, metavar="REF", dest="from_ref", |
| 96 | help="Exclusive start of the commit range (default: initial commit).", |
| 97 | ) |
| 98 | parser.add_argument( |
| 99 | "--to", default=None, metavar="REF", dest="to_ref", |
| 100 | help="Inclusive end of the commit range (default: HEAD).", |
| 101 | ) |
| 102 | parser.add_argument( |
| 103 | "--min", type=int, default=_DEFAULT_MIN, metavar="N", dest="min_count", |
| 104 | help=f"Minimum co-change count to include (default: {_DEFAULT_MIN}).", |
| 105 | ) |
| 106 | parser.add_argument( |
| 107 | "--file", "-f", default=None, metavar="FILE", dest="file_filter", |
| 108 | help=( |
| 109 | "Focus on a single file: show only its coupling partners " |
| 110 | "and rank them by co-change count. " |
| 111 | "Accepts a suffix (e.g. 'billing.py') or a full path." |
| 112 | ), |
| 113 | ) |
| 114 | parser.add_argument( |
| 115 | "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", |
| 116 | help=f"Maximum commits to scan (default: {_DEFAULT_MAX_COMMITS}).", |
| 117 | ) |
| 118 | parser.add_argument( |
| 119 | "--json", action="store_true", dest="as_json", |
| 120 | help="Emit results as JSON.", |
| 121 | ) |
| 122 | parser.set_defaults(func=run) |
| 123 | |
| 124 | |
| 125 | def run(args: argparse.Namespace) -> None: |
| 126 | """Find files that change together most often — hidden dependencies. |
| 127 | |
| 128 | ``muse code coupling`` identifies semantic co-change: file pairs that had |
| 129 | AST-level symbol modifications in the same commit. This is stricter |
| 130 | than raw file co-change — formatting-only edits and non-code files |
| 131 | are excluded. |
| 132 | |
| 133 | Use ``--file`` to focus on a single file and see all its coupling partners |
| 134 | ranked. Use ``--from`` / ``--to`` to scope the analysis to a sprint or |
| 135 | release. Use ``--min`` to raise the minimum co-change threshold. |
| 136 | """ |
| 137 | top: int = clamp_int(args.top, 1, 10_000, 'top') |
| 138 | from_ref: str | None = args.from_ref |
| 139 | to_ref: str | None = args.to_ref |
| 140 | min_count: int = clamp_int(args.min_count, 1, 100_000, 'min_count') |
| 141 | file_filter: str | None = args.file_filter |
| 142 | max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') |
| 143 | as_json: bool = args.as_json |
| 144 | |
| 145 | if top < 1: |
| 146 | print("❌ --top must be >= 1.", file=sys.stderr) |
| 147 | raise SystemExit(ExitCode.USER_ERROR) |
| 148 | if min_count < 1: |
| 149 | print("❌ --min must be >= 1.", file=sys.stderr) |
| 150 | raise SystemExit(ExitCode.USER_ERROR) |
| 151 | if max_commits < 1: |
| 152 | print("❌ --max-commits must be >= 1.", file=sys.stderr) |
| 153 | raise SystemExit(ExitCode.USER_ERROR) |
| 154 | |
| 155 | root = require_repo() |
| 156 | repo_id = read_repo_id(root) |
| 157 | branch = read_current_branch(root) |
| 158 | |
| 159 | to_commit = resolve_commit_ref(root, repo_id, branch, to_ref) |
| 160 | if to_commit is None: |
| 161 | print(f"❌ Commit '{to_ref or 'HEAD'}' not found.", file=sys.stderr) |
| 162 | raise SystemExit(ExitCode.USER_ERROR) |
| 163 | |
| 164 | stop_at: str | None = None |
| 165 | if from_ref is not None: |
| 166 | from_commit = resolve_commit_ref(root, repo_id, branch, from_ref) |
| 167 | if from_commit is None: |
| 168 | print(f"❌ Commit '{from_ref}' not found.", file=sys.stderr) |
| 169 | raise SystemExit(ExitCode.USER_ERROR) |
| 170 | stop_at = from_commit.commit_id |
| 171 | |
| 172 | commits, truncated = walk_commits_bfs( |
| 173 | root, |
| 174 | to_commit.commit_id, |
| 175 | max_commits=max_commits, |
| 176 | stop_at_commit_id=stop_at, |
| 177 | ) |
| 178 | |
| 179 | # Collect all file paths seen across any commit (for --file suffix matching). |
| 180 | all_seen_files: set[str] = set() |
| 181 | for commit in commits: |
| 182 | if commit.structured_delta is None: |
| 183 | continue |
| 184 | all_seen_files.update(touched_files(commit.structured_delta["ops"])) |
| 185 | |
| 186 | # Resolve --file suffix to a canonical path, if requested. |
| 187 | resolved_file: str | None = None |
| 188 | if file_filter is not None: |
| 189 | resolved_file = _resolve_file_suffix(file_filter, all_seen_files) |
| 190 | if resolved_file is None: |
| 191 | # No commits touched this file — either the path is wrong or it has |
| 192 | # never had a semantic change. |
| 193 | if as_json: |
| 194 | print(json.dumps({ |
| 195 | "from_ref": from_ref, |
| 196 | "to_ref": to_ref or branch, |
| 197 | "commits_analysed": len(commits), |
| 198 | "truncated": truncated, |
| 199 | "filters": { |
| 200 | "top": top, "min_count": min_count, |
| 201 | "file": file_filter, "max_commits": max_commits, |
| 202 | }, |
| 203 | "file": file_filter, |
| 204 | "pairs": [], |
| 205 | }, indent=2)) |
| 206 | else: |
| 207 | print(f"\nNo semantic co-changes found for {file_filter!r}.") |
| 208 | print("The file may not exist or may never have had symbol-level changes.") |
| 209 | return |
| 210 | |
| 211 | # Count co-changing pairs. |
| 212 | pair_counts: dict[tuple[str, str], int] = {} |
| 213 | for commit in commits: |
| 214 | if commit.structured_delta is None: |
| 215 | continue |
| 216 | files = touched_files(commit.structured_delta["ops"]) |
| 217 | if len(files) < 2: |
| 218 | continue |
| 219 | # Skip commits that touched too many files — they add noise, not signal. |
| 220 | if len(files) > _MAX_FILES_PER_COMMIT: |
| 221 | continue |
| 222 | for a, b in file_pairs(files): |
| 223 | # When --file is set, only count pairs involving that file. |
| 224 | if resolved_file is not None and resolved_file not in (a, b): |
| 225 | continue |
| 226 | pair_counts[(a, b)] = pair_counts.get((a, b), 0) + 1 |
| 227 | |
| 228 | filtered = {pair: cnt for pair, cnt in pair_counts.items() if cnt >= min_count} |
| 229 | ranked = sorted(filtered.items(), key=lambda kv: kv[1], reverse=True)[:top] |
| 230 | |
| 231 | if as_json: |
| 232 | pairs_out: list[dict[str, JsonValue]] |
| 233 | if resolved_file is not None: |
| 234 | # When --file is set, emit partner + count rather than a/b pair. |
| 235 | pairs_out = [ |
| 236 | { |
| 237 | "file": resolved_file, |
| 238 | "partner": b if a == resolved_file else a, |
| 239 | "co_changes": c, |
| 240 | } |
| 241 | for (a, b), c in ranked |
| 242 | ] |
| 243 | else: |
| 244 | pairs_out = [ |
| 245 | {"file_a": a, "file_b": b, "co_changes": c} |
| 246 | for (a, b), c in ranked |
| 247 | ] |
| 248 | print(json.dumps( |
| 249 | { |
| 250 | "from_ref": from_ref, |
| 251 | "to_ref": to_ref or branch, |
| 252 | "commits_analysed": len(commits), |
| 253 | "truncated": truncated, |
| 254 | "filters": { |
| 255 | "top": top, |
| 256 | "min_count": min_count, |
| 257 | "file": file_filter, |
| 258 | "max_commits": max_commits, |
| 259 | }, |
| 260 | "pairs": pairs_out, |
| 261 | }, |
| 262 | indent=2, |
| 263 | )) |
| 264 | return |
| 265 | |
| 266 | # Human-readable output. |
| 267 | if resolved_file is not None: |
| 268 | print(f"\nCoupling partners of {resolved_file}") |
| 269 | else: |
| 270 | print(f"\nFile co-change analysis — top {len(ranked)} most coupled pairs") |
| 271 | print(f"Commits analysed: {len(commits)}") |
| 272 | if truncated: |
| 273 | print(f"⚠️ Scan capped at {max_commits} commits — pass --max-commits to extend.") |
| 274 | print("") |
| 275 | |
| 276 | if not ranked: |
| 277 | threshold_msg = f"{min_count}+" if min_count > 1 else "2+" |
| 278 | if resolved_file: |
| 279 | print(f" (no files co-changed with {sanitize_display(str(resolved_file))!r} {threshold_msg} times)") |
| 280 | else: |
| 281 | print(f" (no file pairs co-changed {threshold_msg} times)") |
| 282 | return |
| 283 | |
| 284 | width = len(str(len(ranked))) |
| 285 | |
| 286 | if resolved_file is not None: |
| 287 | # Partner-focused display. |
| 288 | max_partner = max(len(b if a == resolved_file else a) for (a, b), _ in ranked) |
| 289 | for rank, ((a, b), count) in enumerate(ranked, 1): |
| 290 | partner = b if a == resolved_file else a |
| 291 | label = "commit" if count == 1 else "commits" |
| 292 | print(f" {rank:>{width}} {sanitize_display(partner):<{max_partner}} co-changed in {count:>3} {label}") |
| 293 | else: |
| 294 | max_a = max(len(a) for (a, _), _ in ranked) |
| 295 | for rank, ((a, b), count) in enumerate(ranked, 1): |
| 296 | label = "commit" if count == 1 else "commits" |
| 297 | print( |
| 298 | f" {rank:>{width}} {sanitize_display(a):<{max_a}} ↔ {sanitize_display(b):<50} " |
| 299 | f"co-changed in {count:>3} {label}" |
| 300 | ) |
| 301 | |
| 302 | print("") |
| 303 | if resolved_file: |
| 304 | print(f"These files always change with {sanitize_display(str(resolved_file))}. Hidden coupling.") |
| 305 | else: |
| 306 | print("High coupling = hidden dependency. Consider extracting a shared interface.") |