"""muse code clones — find duplicate and near-duplicate symbols. Detects two tiers of code duplication from committed snapshot data: **Exact clones** Symbols with the same ``body_hash`` at different addresses. The body is character-for-character identical (after normalisation) even if the name or surrounding context differs. These are true copy-paste duplicates. **Near-clones** Symbols with the same ``signature_id`` but different ``body_hash``. Same function signature, different implementation — strong candidates for consolidation behind a shared abstraction. Git has no concept of these. Git stores file diffs; Muse stores symbol identity hashes. Clone detection is a single pass over the snapshot index. Usage:: muse code clones muse code clones --tier exact muse code clones --tier near muse code clones --kind function muse code clones --language Python muse code clones --file muse/core/ muse code clones --exclude-same-file muse code clones --commit HEAD~10 muse code clones --min-cluster 3 muse code clones --json Output:: Clone analysis — commit a1b2c3d4 Exact clones (2 clusters): body_hash a1b2c3d4: src/billing.py::compute_hash function src/utils.py::compute_hash function src/legacy.py::_hash function Near-clones — same signature (3 clusters): signature_id e5f6a7b8: src/billing.py::validate function src/auth.py::validate function Files with most clone members: src/billing.py 4 clone symbols Flags: ``--tier {exact|near|both}`` Which tier to report (default: both). ``--kind KIND`` Restrict to symbols of this kind (function, class, method, …). ``--language LANG`` Restrict to files of this language. ``--file PATH`` Restrict to symbols whose file path starts with PATH. ``--exclude-same-file`` Skip clusters where every member lives in the same file. ``--min-cluster N`` Only show clusters with at least N members (default: 2). ``--commit, -c REF`` Analyse a historical snapshot instead of HEAD. ``--json`` Emit results as JSON. """ from __future__ import annotations import argparse import json import logging import pathlib from typing import Literal, TypedDict from muse._version import __version__ from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core._types import Manifest from muse.core.store import get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref from muse.core.symbol_cache import SymbolCache, load_symbol_cache from muse.plugins.code._query import language_of, symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolRecord from muse.core.validation import clamp_int, sanitize_display type _SymIndex = dict[str, list[tuple[str, SymbolRecord]]] type _FileCountMap = dict[str, int] logger = logging.getLogger(__name__) CloneTier = Literal["exact", "near", "both"] # --------------------------------------------------------------------------- # Typed output shapes # --------------------------------------------------------------------------- class _MemberDict(TypedDict): address: str kind: str language: str body_hash: str signature_id: str content_id: str class _ClusterDict(TypedDict): tier: str hash: str count: int members: list[_MemberDict] class _FileHotspot(TypedDict): file: str clone_symbols: int # --------------------------------------------------------------------------- # Core data model # --------------------------------------------------------------------------- class _CloneCluster: """A group of symbols that are duplicates of each other.""" def __init__( self, tier: CloneTier, hash_value: str, members: list[tuple[str, SymbolRecord]], ) -> None: self.tier = tier self.hash_value = hash_value self.members = members # (address, record) def to_dict(self) -> _ClusterDict: return { "tier": self.tier, "hash": self.hash_value[:8], "count": len(self.members), # int — not str "members": [ _MemberDict( address=addr, kind=rec["kind"], language=language_of(addr.split("::")[0]), body_hash=rec["body_hash"][:8], signature_id=rec["signature_id"][:8], content_id=rec["content_id"][:8], ) for addr, rec in self.members ], } # --------------------------------------------------------------------------- # Detection logic # --------------------------------------------------------------------------- def find_clones( root: pathlib.Path, manifest: Manifest, tier: CloneTier, kind_filter: str | None, min_cluster: int, *, language_filter: str | None = None, file_filter: str | None = None, exclude_same_file: bool = False, cache: SymbolCache | None = None, ) -> list[_CloneCluster]: """Build clone clusters from *manifest*. Args: root: Repository root (object store location). manifest: Snapshot manifest: file path → SHA-256 object ID. tier: Which clone tier(s) to detect: "exact", "near", "both". kind_filter: If set, only analyse symbols of this kind. min_cluster: Minimum cluster size to report (default 2). language_filter: If set, only analyse files of this language. file_filter: If set, only analyse symbols whose file path starts with this prefix (e.g. "muse/core/"). exclude_same_file: If True, skip clusters where all members are in the same file (eliminates test-helper noise). cache: Optional shared ``SymbolCache``. Pass one to avoid re-parsing blobs when the caller has a warm cache. """ sym_map = symbols_for_snapshot( root, manifest, kind_filter=kind_filter, language_filter=language_filter, cache=cache, ) # Flatten to list of (address, record), applying file_filter. all_syms: list[tuple[str, SymbolRecord]] = [ (addr, rec) for fp, tree in sorted(sym_map.items()) if file_filter is None or fp.startswith(file_filter) for addr, rec in sorted(tree.items()) if rec["kind"] != "import" ] clusters: list[_CloneCluster] = [] if tier in ("exact", "both"): body_index: _SymIndex = {} for addr, rec in all_syms: body_index.setdefault(rec["body_hash"], []).append((addr, rec)) for body_hash, members in sorted(body_index.items()): if len(members) < min_cluster: continue if exclude_same_file and _all_same_file(members): continue clusters.append(_CloneCluster("exact", body_hash, members)) if tier in ("near", "both"): sig_index: _SymIndex = {} for addr, rec in all_syms: sig_index.setdefault(rec["signature_id"], []).append((addr, rec)) for sig_id, members in sorted(sig_index.items()): # Near-clone: same signature, at least two DIFFERENT body hashes. unique_bodies = {r["body_hash"] for _, r in members} if len(members) < min_cluster or len(unique_bodies) <= 1: continue if exclude_same_file and _all_same_file(members): continue clusters.append(_CloneCluster("near", sig_id, members)) # Sort: largest clusters first, then by tier (exact before near), then hash. clusters.sort(key=lambda c: (-len(c.members), c.tier, c.hash_value)) return clusters def _all_same_file(members: list[tuple[str, SymbolRecord]]) -> bool: """Return True when every member lives in the same source file.""" files = {addr.split("::")[0] for addr, _ in members} return len(files) == 1 def _file_hotspots( clusters: list[_CloneCluster], top: int = 10, ) -> list[_FileHotspot]: """Rank files by the number of clone-member symbols they contain.""" file_counts: _FileCountMap = {} for cluster in clusters: for addr, _ in cluster.members: fp = addr.split("::")[0] file_counts[fp] = file_counts.get(fp, 0) + 1 ranked = sorted(file_counts.items(), key=lambda t: t[1], reverse=True)[:top] return [_FileHotspot(file=fp, clone_symbols=cnt) for fp, cnt in ranked] # --------------------------------------------------------------------------- # CLI registration and entry point # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the clones subcommand.""" parser = subparsers.add_parser( "clones", help="Find exact and near-duplicate symbols in the committed snapshot.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--tier", "-t", default="both", choices=["exact", "near", "both"], help="Tier to report: exact, near, or both (default: both).", ) 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 files of this language.", ) parser.add_argument( "--file", "-f", default=None, metavar="PATH", dest="file_filter", help="Restrict to symbols whose file path starts with PATH.", ) parser.add_argument( "--exclude-same-file", action="store_true", dest="exclude_same_file", help="Skip clusters where all members are in the same file.", ) parser.add_argument( "--min-cluster", "-m", type=int, default=2, metavar="N", dest="min_cluster", help="Only show clusters with at least N members (default: 2).", ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help="Analyse this commit instead of HEAD.", ) 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 exact and near-duplicate symbols in the committed snapshot. Exact clones share the same ``body_hash`` (identical implementation). Near-clones share the same ``signature_id`` but differ in body — same contract, different implementation. Both are candidates for consolidation behind shared abstractions. Uses content-addressed hashes from the snapshot index — no AST recomputation at query time. A shared SymbolCache ensures each blob is parsed at most once across all analysis passes. """ tier: CloneTier = args.tier kind_filter: str | None = args.kind_filter language_filter: str | None = args.language_filter file_filter: str | None = args.file_filter exclude_same_file: bool = args.exclude_same_file min_cluster: int = clamp_int(args.min_cluster, 1, 10000, 'min_cluster') ref: str | None = args.ref as_json: bool = args.as_json if min_cluster < 2: logger.error("--min-cluster must be at least 2, got %d", min_cluster) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() from muse.plugins.registry import read_domain try: if read_domain(root) == "knowtation": from muse.plugins.knowtation.cli_hooks import run_clones_for_vault run_clones_for_vault(root, args) return except Exception: # noqa: BLE001 pass try: repo_id = read_repo_id(root) except (FileNotFoundError, json.JSONDecodeError, KeyError) as exc: logger.error("Cannot read repo identity: %s", exc) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc branch = read_current_branch(root) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: logger.error("Commit %r not found.", ref or "HEAD") raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} # Shared cache — each blob parsed at most once. cache = load_symbol_cache(root) cluster_list = find_clones( root, manifest, tier, kind_filter, min_cluster, language_filter=language_filter, file_filter=file_filter, exclude_same_file=exclude_same_file, cache=cache, ) exact_clusters = [c for c in cluster_list if c.tier == "exact"] near_clusters = [c for c in cluster_list if c.tier == "near"] total_symbols = sum(len(c.members) for c in cluster_list) hotspots = _file_hotspots(cluster_list) if as_json: print(json.dumps( { "schema_version": __version__, "commit": commit.commit_id[:8], "branch": branch, "tier": tier, "min_cluster": min_cluster, "kind_filter": kind_filter, "language_filter": language_filter, "file_filter": file_filter, "exclude_same_file": exclude_same_file, "exact_clone_clusters": len(exact_clusters), "near_clone_clusters": len(near_clusters), "total_symbols_involved": total_symbols, "file_hotspots": hotspots, "clusters": [c.to_dict() for c in cluster_list], }, indent=2, )) return print(f"\nClone analysis — commit {commit.commit_id[:8]}") if kind_filter: print(f" (kind: {kind_filter})") if language_filter: print(f" (language: {language_filter})") if file_filter: print(f" (file prefix: {file_filter})") if exclude_same_file: print(" (same-file clusters excluded)") print("─" * 62) if not cluster_list: print("\n ✅ No clones detected.") return if exact_clusters and tier in ("exact", "both"): print(f"\nExact clones ({len(exact_clusters)} cluster(s)):") for cl in exact_clusters: print(f" body_hash {cl.hash_value[:8]}:") for addr, rec in cl.members: print(f" {sanitize_display(addr)} {rec['kind']}") if near_clusters and tier in ("near", "both"): print(f"\nNear-clones — same signature ({len(near_clusters)} cluster(s)):") for cl in near_clusters: print(f" signature_id {cl.hash_value[:8]}:") for addr, rec in cl.members: print(f" {sanitize_display(addr)} {rec['kind']} (body {rec['body_hash'][:8]})") print(f"\n {len(cluster_list)} clone cluster(s), {total_symbols} total symbol(s) involved") if hotspots: print(f"\nFiles with most clone members:") for h in hotspots[:5]: print(f" {sanitize_display(h['file'])} {h['clone_symbols']} clone symbol(s)")