"""muse code entangle — hidden symbol-pair entanglement detector. ``muse code coupling`` works at the file level. ``muse code entangle`` goes one level deeper and works at the **symbol level**: it finds *pairs of symbols* that consistently change in the same commit but have **no structural import or call-graph link** between them. These are the hidden "keep-in-sync" relationships that nobody documented — the ones that cause mysterious bugs when one side is updated and the other is forgotten. Why this is impossible in Git ------------------------------ Git sees ``billing.py`` and ``serializers.py`` changed in the same commit. It has no way to say *which functions* changed, so it cannot report symbol-pair co-change. Muse stores a ``structured_delta`` with every commit. Each delta records every symbol operation — insert, delete, replace — with its precise address. ``entangle`` mines this history to surface the statistical signal that file diffing cannot see. Exclusions ---------- - Import pseudo-symbols (``::import::``) are excluded. - Commits touching more than ``_MAX_SYMBOLS_PER_COMMIT`` distinct symbols are skipped — they are mass-refactors that produce O(N²) noise. - The same-file constraint can be relaxed with ``--include-same-file``. - Results where either symbol is in a test file are surfaced with a ``[test]`` badge. Structural-link check --------------------- Two symbols are considered **structurally linked** if the file that contains *symbol A* has an import record pointing to the module/file of *symbol B*, or vice versa. This check is intentionally conservative — it only looks at top-level import symbols in the current snapshot. A pair that is flagged as "entangled" despite an obvious import may be using an indirect import chain; that is still interesting to know. Usage:: muse code entangle muse code entangle --top 20 muse code entangle --min-rate 0.5 # only pairs that co-changed > 50 % of the time muse code entangle --symbol billing.py::Invoice.compute_total muse code entangle --include-same-file # include pairs in the same file muse code entangle --since HEAD~30 muse code entangle --json Output:: Symbol entanglement — HEAD (47 commits · 3 entangled pairs) # SYMBOL A ↔ SYMBOL B RATE CO-CHANGES 1 billing.py::Invoice.compute_total ↔ serializers.py::to_json 91% 10 / 11 2 auth.py::verify_identity ↔ middleware.py::check_token 80% 8 / 10 3 models.py::User.save ↔ events.py::emit_user_event 75% 6 / 8 [cross-file, no import link] ⚠️ These pairs change together but have no structural import link. Consider adding documentation, a shared interface, or an explicit test. JSON output (--json):: { "ref": "HEAD", "commits_analysed": 47, "truncated": false, "filters": { "min_rate": 0.5, "min_co_changes": 2, "symbol": null, "include_same_file": false }, "pairs": [ { "symbol_a": "billing.py::Invoice.compute_total", "symbol_b": "serializers.py::to_json", "file_a": "billing.py", "file_b": "serializers.py", "same_file": false, "structurally_linked": false, "co_changes": 10, "commits_both_active": 11, "co_change_rate": 0.91, "a_in_test": false, "b_in_test": false } ] } """ from __future__ import annotations import argparse import ast import json import logging import pathlib import re import sys 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 ( JsonValue, Manifest, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.symbol_cache import load_symbol_cache from muse.domain import DomainOp type _ImportMap = dict[str, set[str]] type _CounterMap = dict[str, int] type _JsonFilters = dict[str, JsonValue] 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__) # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_TOP = 20 _DEFAULT_MIN_CO_CHANGES = 2 _DEFAULT_MIN_RATE = 0.0 _DEFAULT_MAX_COMMITS = 10_000 # Skip commits with too many distinct symbols — they're mass-refactors. _MAX_SYMBOLS_PER_COMMIT = 200 # Test-file heuristics. _TEST_RE: tuple[re.Pattern[str], ...] = ( re.compile(r"(^|/)test_"), re.compile(r"_test\.py$"), re.compile(r"(^|/)tests/"), re.compile(r"(^|/)spec/"), re.compile(r"(^|/)conftest\.py$"), ) def _is_test_file(path: str) -> bool: return any(p.search(path) for p in _TEST_RE) # ── TypedDict ────────────────────────────────────────────────────────────────── class _EntangledPair(TypedDict): symbol_a: str symbol_b: str file_a: str file_b: str same_file: bool structurally_linked: bool co_changes: int commits_both_active: int co_change_rate: float a_in_test: bool b_in_test: bool # ── Helpers ──────────────────────────────────────────────────────────────────── def _module_name_from_path(file_path: str) -> str: """Convert a file path to a rough Python module name. ``muse/core/store.py`` → ``muse.core.store`` ``billing.py`` → ``billing`` """ stem = file_path.removesuffix(".py").removesuffix(".pyi") return stem.replace("/", ".").replace("\\", ".") def _build_import_map( root: pathlib.Path, manifest: Manifest, ) -> _ImportMap: """Return ``{file_path: {imported_module_name, ...}}`` from the snapshot. We extract import pseudo-symbols from the symbol cache / AST parser. These records have ``kind == "import"`` and ``qualified_name`` set to the imported module name (e.g. ``muse.core.store`` or just ``store``). """ cache = load_symbol_cache(root) all_trees = symbols_for_snapshot(root, manifest, cache=cache) import_map: _ImportMap = {} for file_path, tree in all_trees.items(): mods: set[str] = set() for rec in tree.values(): if rec.get("kind") == "import": qn: str = rec.get("qualified_name") or rec.get("name") or "" if qn: mods.add(qn) # Also record the top-level package for coarse matching. mods.add(qn.split(".")[0]) import_map[file_path] = mods return import_map def _are_structurally_linked( file_a: str, file_b: str, import_map: _ImportMap, ) -> bool: """Return True if either file imports the other (even partially). We check whether any module name in the import set of file_a resembles the module path of file_b, or vice versa. We use a suffix check rather than exact match to handle aliased or re-exported imports. """ mod_a = _module_name_from_path(file_a) mod_b = _module_name_from_path(file_b) imports_of_a = import_map.get(file_a, set()) imports_of_b = import_map.get(file_b, set()) # file_a imports file_b? for imp in imports_of_a: if mod_b.endswith(imp) or imp.endswith(mod_b) or imp == mod_b.split(".")[-1]: return True # file_b imports file_a? for imp in imports_of_b: if mod_a.endswith(imp) or imp.endswith(mod_a) or imp == mod_a.split(".")[-1]: return True return False # ── Core algorithm ───────────────────────────────────────────────────────────── def _collect_symbol_pairs( root: pathlib.Path, head_commit_id: str, stop_at: str | None, max_commits: int, include_same_file: bool, ) -> tuple[ dict[tuple[str, str], int], # co-change counts dict[str, int], # how many commits each symbol was active in int, # commits_analysed bool, # truncated ]: """Single BFS pass: count symbol-pair co-changes. Returns: co_changes — ``(addr_a, addr_b) → commit count`` (a < b lexicographically) active — ``addr → commit count where the symbol changed`` commits_analysed, truncated """ commits, truncated = walk_commits_bfs( root, head_commit_id, max_commits, stop_at_commit_id=stop_at ) co_changes: dict[tuple[str, str], int] = {} active: _CounterMap = {} for commit in commits: if commit.structured_delta is None: continue ops: list[DomainOp] = commit.structured_delta["ops"] # Collect distinct symbol addresses changed in this commit, # excluding import pseudo-symbols. changed: list[str] = [] for op in flat_symbol_ops(ops): addr: str = op["address"] if "::import::" not in addr: changed.append(addr) # Skip mass-refactor commits. if len(changed) > _MAX_SYMBOLS_PER_COMMIT: continue # Deduplicate within this commit (a symbol may appear >1 ops). unique = list(dict.fromkeys(changed)) # Track per-symbol activity count. for addr in unique: active[addr] = active.get(addr, 0) + 1 # Count co-changing pairs. for i in range(len(unique)): for j in range(i + 1, len(unique)): a, b = unique[i], unique[j] # Canonical ordering so we always use the same dict key. pair: tuple[str, str] = (a, b) if a <= b else (b, a) fa = a.split("::")[0] fb = b.split("::")[0] if not include_same_file and fa == fb: continue co_changes[pair] = co_changes.get(pair, 0) + 1 return co_changes, active, len(commits), truncated # ── Scoring + filtering ──────────────────────────────────────────────────────── def _build_pairs( co_changes: dict[tuple[str, str], int], active: _CounterMap, import_map: _ImportMap, min_co_changes: int, min_rate: float, symbol_filter: str | None, top: int, include_same_file: bool, ) -> list[_EntangledPair]: """Filter, score, and sort the co-change pairs.""" pairs: list[_EntangledPair] = [] for (a, b), count in co_changes.items(): if count < min_co_changes: continue fa = a.split("::")[0] fb = b.split("::")[0] # --symbol filter: only show pairs involving the requested symbol. if symbol_filter is not None and symbol_filter not in (a, b): continue # How many commits were both symbols active? # Use the lower of the two active counts as the "opportunity window". active_a = active.get(a, count) active_b = active.get(b, count) both_active = min(active_a, active_b) rate = count / both_active if both_active > 0 else 0.0 if rate < min_rate: continue linked = _are_structurally_linked(fa, fb, import_map) # We only surface *unlinked* pairs — that's the entanglement signal. # Unless the user provides --symbol to inspect a specific pair. if linked and symbol_filter is None: continue pairs.append(_EntangledPair( symbol_a=a, symbol_b=b, file_a=fa, file_b=fb, same_file=(fa == fb), structurally_linked=linked, co_changes=count, commits_both_active=both_active, co_change_rate=round(rate, 4), a_in_test=_is_test_file(fa), b_in_test=_is_test_file(fb), )) # Sort: highest co_change_rate first, then co_changes count, then address. pairs.sort(key=lambda p: (-p["co_change_rate"], -p["co_changes"], p["symbol_a"])) return pairs[:top] # ── Formatters ───────────────────────────────────────────────────────────────── def _print_table( pairs: list[_EntangledPair], ref: str, commits_analysed: int, truncated: bool, since: str | None, ) -> None: scope = f"{since}..{ref}" if since else ref trunc = " ⚠️ truncated" if truncated else "" print( f"\nSymbol entanglement — {scope}" f" ({commits_analysed} commits · {len(pairs)} entangled pair(s){trunc})" ) print("") if not pairs: print(" (no entangled symbol pairs found)") print( "\n All co-changing symbol pairs have a structural import link " "— no hidden entanglement detected." ) return max_a = max(len(p["symbol_a"]) for p in pairs) max_b = max(len(p["symbol_b"]) for p in pairs) width = len(str(len(pairs))) hdr = ( f" {'#':>{width}} " f"{'SYMBOL A':<{max_a}} ↔ {'SYMBOL B':<{max_b}} " f"{'RATE':>5} {'CO-CHANGES':>10}" ) print(hdr) print(" " + "─" * (len(hdr) - 2)) for i, p in enumerate(pairs, 1): rate_pct = f"{round(p['co_change_rate'] * 100)} %" co_str = f"{p['co_changes']} / {p['commits_both_active']}" badges: list[str] = [] if p["a_in_test"] or p["b_in_test"]: badges.append("[test]") if p["same_file"]: badges.append("[same-file]") badge_str = " " + " ".join(badges) if badges else "" print( f" {i:>{width}} " f"{sanitize_display(p['symbol_a']):<{max_a}} ↔ {sanitize_display(p['symbol_b']):<{max_b}} " f"{rate_pct:>5} {co_str:>10}{badge_str}" ) print( "\n⚠️ These symbol pairs change together but share no import link." "\n Consider adding docs, a shared interface, or an explicit integration test." ) # ── CLI ──────────────────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the entangle subcommand.""" parser = subparsers.add_parser( "entangle", help=( "Find symbol pairs that always change together but have no " "structural import link — hidden coupling." ), 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( "--min-co-changes", type=int, default=_DEFAULT_MIN_CO_CHANGES, metavar="N", dest="min_co_changes", help=( f"Minimum number of commits where both symbols co-changed " f"(default: {_DEFAULT_MIN_CO_CHANGES})." ), ) parser.add_argument( "--min-rate", type=float, default=_DEFAULT_MIN_RATE, metavar="RATE", dest="min_rate", help=( "Minimum co-change rate 0.0–1.0: fraction of commits where " "both symbols were active that they co-changed " f"(default: {_DEFAULT_MIN_RATE}). " "E.g. --min-rate 0.5 = show only pairs that co-changed >50 %% of the time." ), ) parser.add_argument( "--symbol", "-s", default=None, metavar="ADDRESS", dest="symbol_filter", help=( "Focus on a single symbol: show all pairs it forms with other " "symbols, including structurally-linked ones. " "Address format: file.py::SymbolName" ), ) parser.add_argument( "--since", 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( "--include-same-file", action="store_true", dest="include_same_file", help=( "Include symbol pairs within the same file. By default, " "same-file pairs are excluded since same-file co-change is expected." ), ) 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 symbol pairs that always change together but share no import link. Mines the commit history for symbol-level co-change patterns. Pairs that co-change frequently but have no structural import link are "entangled" — a hidden dependency that only shows up when one side is changed and the other isn't. The co-change *rate* is defined as:: co_changes / min(commits_where_A_changed, commits_where_B_changed) A rate of 1.0 means the two symbols changed together every single time either of them changed. """ top: int = clamp_int(args.top, 1, 10_000, 'top') min_co_changes: int = clamp_int(args.min_co_changes, 1, 100000, 'min_co_changes') min_rate: float = args.min_rate symbol_filter: str | None = args.symbol_filter since: str | None = args.since max_commits: int = clamp_int(args.max_commits, 1, 100_000, 'max_commits') include_same_file: bool = args.include_same_file as_json: bool = args.as_json # ── Validation ──────────────────────────────────────────────────────────── if top < 1: print("❌ --top must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if min_co_changes < 1: print("❌ --min-co-changes must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not (0.0 <= min_rate <= 1.0): print("❌ --min-rate must be between 0.0 and 1.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) if symbol_filter is not None and "::" not in symbol_filter: print( "❌ --symbol must be a qualified address (file.py::SymbolName).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if symbol_filter is not None and len(symbol_filter) > 500: print("❌ --symbol address is too long (max 500 chars).", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── 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_entangle_for_vault run_entangle_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 # ── Phase 1: build import map from HEAD snapshot ────────────────────────── manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} import_map = _build_import_map(root, manifest) # ── Phase 2: mine co-changes in one BFS pass ────────────────────────────── co_changes, active, commits_analysed, truncated = _collect_symbol_pairs( root, head_commit_id=head.commit_id, stop_at=stop_at, max_commits=max_commits, include_same_file=include_same_file, ) # ── Phase 3: filter + rank ──────────────────────────────────────────────── pairs = _build_pairs( co_changes=co_changes, active=active, import_map=import_map, min_co_changes=min_co_changes, min_rate=min_rate, symbol_filter=symbol_filter, top=top, include_same_file=include_same_file, ) # ── Output ──────────────────────────────────────────────────────────────── ref = branch filters: _JsonFilters = { "min_rate": min_rate, "min_co_changes": min_co_changes, "symbol": symbol_filter, "since": since, "include_same_file": include_same_file, "top": top, "max_commits": max_commits, } if as_json: print(json.dumps({ "ref": ref, "commits_analysed": commits_analysed, "truncated": truncated, "filters": filters, "pairs": [dict(p) for p in pairs], }, indent=2)) return _print_table(pairs, ref, commits_analysed, truncated, since)