"""muse code dead — dead code detection. Finds symbols that are **never referenced** and whose containing module is **never imported** by anything else in the codebase. A symbol is a dead-code candidate when two independent conditions hold: 1. **No reference**: its bare name does not appear as any ``ast.Name`` id or ``ast.Attribute`` attr anywhere in the codebase. This is broader than call-site detection — it catches attribute accesses, keyword argument values, type annotations, ``isinstance`` checks, and every other form of name usage, not just direct calls. 2. **No import**: its containing file's module name does not appear in any ``import``-kind symbol in any other file. Both conditions must hold simultaneously. A function that is never referenced but lives in a module that *is* imported is still reachable — it may be part of an exported API even if it's not used internally. Performance ----------- All files are processed in a **single parallel pass** and AST-parsed exactly once. Without ``--commit`` the working tree is read from disk, so uncommitted changes are immediately visible. With ``--commit`` the specified historical snapshot is read from the object store. Imports, references, and symbol trees are all extracted in the same pass, then combined. ``--workers`` controls the thread-pool size. Security -------- ``ast.parse`` never executes code. Files exceeding ``--max-file-bytes`` (default 512 KB) are skipped to prevent stalls on generated or minified files. ``--delete`` validates every file path inside the repo root before touching the working tree. Known limitations ----------------- - Symbols whose names are extremely common (e.g. ``run``, ``name``) may appear as false negatives because a matching name exists somewhere else. - Exported APIs: symbols accessed from outside the repo (library code) appear dead because the callers are not in the snapshot. - Entry points: ``main()``, CLI callbacks, and test functions appear dead by design. Use ``--exclude-tests`` to hide test file symbols. - tree-sitter languages: reference extraction is Python-only. Symbols in Go/Rust/TypeScript files are checked for import-graph reachability only. - ``--delete`` is Python-only (requires AST line-range information). Usage:: muse code dead muse code dead --kind function muse code dead --exclude-tests muse code dead --exclude-private muse code dead --high-confidence-only muse code dead --path "musehub/services/*" muse code dead --language Python muse code dead --top 50 muse code dead --group-by-file muse code dead --commit HEAD~10 muse code dead --workers 8 muse code dead --json muse code dead --delete muse code dead --delete --yes muse code dead --allowlist .muse/dead-allowlist.json Confidence levels:: HIGH — not referenced AND module not imported → almost certainly dead MEDIUM — not referenced, but module IS imported → may be exported API surface Flags: ``--kind KIND, -k KIND`` Restrict to symbols of a specific kind (function, class, method, …). ``--exclude-tests`` Exclude symbols in files whose path contains ``test`` or ``spec``. ``--exclude-private`` Exclude symbols whose bare name starts with ``_``. ``--high-confidence-only`` Show only HIGH confidence candidates (module not imported). ``--path GLOB, -p GLOB`` Restrict to files matching this glob pattern (e.g. ``"musehub/services/*"``). ``--language LANG, -l LANG`` Restrict to files of a specific language (e.g. ``Python``, ``TypeScript``). ``--top N`` Show only the top N candidates. ``--group-by-file, -g`` Group output by file instead of a flat sorted list. ``--commit REF, -c REF`` Analyse a historical snapshot instead of HEAD. ``--workers N, -w N`` Number of parallel worker threads for file parsing (default: 8). ``--max-file-bytes N`` Skip files larger than N bytes (default: 524288 = 512 KB). ``--no-color`` Disable ANSI color output. ``--json`` Emit results as JSON. ``--delete`` Interactively delete dead symbols from the working tree (Python only). Prompts for each candidate unless ``--yes`` is also given. ``--yes, -y`` Skip confirmation prompts when used with ``--delete``. ``--allowlist FILE`` JSON file containing a list of symbol addresses to suppress from output. Addresses are matched as exact strings against the ``address`` field. Example file: ``[\"muse/cli/config.py::MuseConfig\"]`` """ from __future__ import annotations import argparse import ast import fnmatch import json import logging import os import pathlib import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import TypedDict from muse.core.errors import ExitCode from muse.core.object_store import read_object from muse.core.repo import read_repo_id, require_repo from muse.core._types import Manifest from muse.core.store import ( CommitRecord, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.plugins.code._framework import ImplicitEdgeGraph, build_implicit_edge_graph from muse.plugins.code._query import language_of from muse.plugins.code.ast_parser import SEMANTIC_EXTENSIONS, SymbolTree, parse_symbols type _BlobMap = dict[str, bytes] type _KindCountMap = dict[str, int] type _DeadByFile = dict[str, list["_DeadCandidate"]] from muse.core.validation import MAX_AST_BYTES, clamp_int, sanitize_display logger = logging.getLogger(__name__) class _DeadCandidateJson(TypedDict): """JSON-serialisable representation of one dead-code candidate.""" address: str file_path: str kind: str referenced: bool module_imported: bool confidence: str reason: str class _DeadPayload(TypedDict, total=False): """JSON payload for ``muse code dead`` output.""" source: str total_files_scanned: int total_symbols_scanned: int elapsed_seconds: float high_confidence_count: int medium_confidence_count: int results: list[_DeadCandidateJson] compare_commit_id: str new_dead: list[_DeadCandidateJson] recovered: list[_DeadCandidateJson] net_change: int class _ScanKwargs(TypedDict): """Keyword arguments forwarded to every :func:`_scan_file_bytes` call. Collected into a TypedDict so the ``**scan_kwargs`` spread is type-safe without a ``# type: ignore`` and the common args are defined once. """ kind_filter: str | None max_file_bytes: int workers: int language_filter: str | None path_filter: str | None exclude_tests: bool exclude_private: bool high_confidence_only: bool allowlist: frozenset[str] _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"}) _MAX_WORKERS: int = 64 _MIN_FILE_BYTES: int = 4_096 # Maximum file size we'll parse (512 KB). Prevents stalling on generated files. _DEFAULT_MAX_FILE_BYTES: int = 524_288 # ── ANSI colours ────────────────────────────────────────────────────────────── _RESET = "\033[0m" _BOLD = "\033[1m" _DIM = "\033[2m" _RED = "\033[31m" _YELLOW = "\033[33m" _CYAN = "\033[36m" _GREEN = "\033[32m" _BLUE = "\033[34m" _MAGENTA = "\033[35m" _WHITE = "\033[37m" _GRAY = "\033[90m" def _c(text: str, *codes: str, use_color: bool = True) -> str: """Wrap *text* with ANSI escape codes if *use_color* is True.""" if not use_color: return text return "".join(codes) + text + _RESET # ── Data structures ─────────────────────────────────────────────────────────── @dataclass class _FileAnalysis: """Everything extracted from a single file in one pass.""" file_path: str lang: str symbol_tree: SymbolTree = field(default_factory=dict) # Every name referenced anywhere in the file (ast.Name ids + ast.Attribute attrs). # This is broader than call-sites: catches attribute access, keyword args, # type annotations, isinstance checks, decorator names, etc. ref_names: set[str] = field(default_factory=set) # Imported module/name strings (from import-kind symbols) imported_names: set[str] = field(default_factory=set) skipped: bool = False error: str | None = None @dataclass class _DeadCandidate: address: str file_path: str kind: str referenced: bool module_imported: bool @property def confidence(self) -> str: return "high" if not self.module_imported else "medium" @property def reason(self) -> str: if not self.referenced and not self.module_imported: return "not referenced, module not imported" return "not referenced (module imported — may be exported API)" def to_dict(self) -> _DeadCandidateJson: return _DeadCandidateJson( address=self.address, file_path=self.file_path, kind=self.kind, referenced=self.referenced, module_imported=self.module_imported, confidence=self.confidence, reason=self.reason, ) # ── Single-pass file analysis ───────────────────────────────────────────────── def _analyse_file( file_path: str, raw: bytes, kind_filter: str | None, max_file_bytes: int, ) -> _FileAnalysis: """Parse and extract symbols + references + imports from one file. Thread-safe: pure functions only, no shared mutable state. The caller is responsible for supplying the raw file bytes — either read from disk (working tree) or fetched from the object store (historical commit). """ lang = language_of(file_path) result = _FileAnalysis(file_path=file_path, lang=lang) if len(raw) > max_file_bytes: result.skipped = True return result suffix = pathlib.PurePosixPath(file_path).suffix.lower() if suffix not in SEMANTIC_EXTENSIONS: return result # ── Symbol extraction (all languages with AST support) ───────────────── try: tree = parse_symbols(raw, file_path) except Exception as exc: # noqa: BLE001 result.error = str(exc) return result for rec in tree.values(): if rec["kind"] == "import": result.imported_names.add(rec["qualified_name"]) if kind_filter: tree = {addr: rec for addr, rec in tree.items() if rec["kind"] == kind_filter} result.symbol_tree = tree # ── Reference + module-import extraction (Python only via stdlib ast) ───── # We walk ALL nodes once: # # ast.Name / ast.Attribute — broad reference tracking (fixes logger, # func=run keyword args, property access, isinstance args, annotations) # # ast.ImportFrom / ast.Import — extract the actual dotted module paths # ("from muse.core.store import X" → "muse.core.store"). The Muse # symbol tree stores imports as "import::symbolname" with no module # path, so we must supplement it here to make _module_is_imported work. if suffix in _PY_SUFFIXES: try: if len(raw) > MAX_AST_BYTES: return result py_tree = ast.parse(raw) except SyntaxError: return result for node in ast.walk(py_tree): if isinstance(node, ast.Name): result.ref_names.add(node.id) elif isinstance(node, ast.Attribute): result.ref_names.add(node.attr) elif isinstance(node, ast.ImportFrom): if node.module: result.imported_names.add(node.module) elif isinstance(node, ast.Import): for alias in node.names: result.imported_names.add(alias.name) return result # ── Module-import matching ──────────────────────────────────────────────────── def _module_is_imported(file_path: str, imported_names: set[str]) -> bool: """Return True if *file_path*'s module name appears anywhere in *imported_names*.""" stem = pathlib.PurePosixPath(file_path).stem module = pathlib.PurePosixPath(file_path).with_suffix("").as_posix().replace("/", ".") for imp in imported_names: if ( imp == stem or imp == module or imp.endswith(f".{stem}") or imp.endswith(f".{module}") or stem in imp.split(".") ): return True return False # ── Path filter ─────────────────────────────────────────────────────────────── def _matches_path_filter(file_path: str, pattern: str | None) -> bool: if pattern is None: return True return fnmatch.fnmatch(file_path, pattern) or fnmatch.fnmatch(file_path, f"**/{pattern}") # ── Symbol deletion (Python only) ───────────────────────────────────────────── def _find_symbol_span(source: bytes, bare_name: str, parent_class: str | None) -> tuple[int, int] | None: """Return (start_lineno, end_lineno) 1-indexed for the named symbol. Accounts for decorator lines (start is the first decorator's line). Returns None if the symbol cannot be located. """ try: if len(source) > MAX_AST_BYTES: return None tree = ast.parse(source) except SyntaxError: return None search_body: list[ast.stmt] = tree.body if parent_class: for node in ast.walk(tree): if isinstance(node, ast.ClassDef) and node.name == parent_class: search_body = list(node.body) break for node in search_body: node_name: str | None = None if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): node_name = node.name elif isinstance(node, ast.Assign): if len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): node_name = node.targets[0].id elif isinstance(node, ast.AnnAssign): if isinstance(node.target, ast.Name): node_name = node.target.id if node_name != bare_name: continue if not hasattr(node, "end_lineno") or node.end_lineno is None: return None end: int = node.end_lineno start: int = node.lineno if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): if node.decorator_list: start = node.decorator_list[0].lineno return (start, end) return None def _delete_symbol_lines(source_lines: list[str], start: int, end: int) -> list[str]: """Remove lines *start*–*end* (1-indexed, inclusive) and normalise blank lines.""" before = list(source_lines[: start - 1]) after = list(source_lines[end:]) # Strip trailing blank lines from the block before the deletion point. while before and not before[-1].strip(): before.pop() # Add one blank separator line if there is still content below. if after and any(line.strip() for line in after): return before + ["\n"] + after return before + after # ── Repo ID ─────────────────────────────────────────────────────────────────── # ── Allowlist ───────────────────────────────────────────────────────────────── def _load_allowlist(path: str | None) -> frozenset[str]: """Load a JSON list of symbol addresses that should be suppressed.""" if path is None: return frozenset() try: raw = pathlib.Path(path).read_text(encoding="utf-8") parsed = json.loads(raw) if not isinstance(parsed, list): logger.warning("dead-code allowlist must be a JSON array; ignoring %s", path) return frozenset() return frozenset(str(x) for x in parsed) except (OSError, json.JSONDecodeError) as exc: logger.warning("Could not load allowlist %s: %s", path, exc) return frozenset() # ── Shared scan pipeline ────────────────────────────────────────────────────── def _load_file_bytes( root: pathlib.Path, manifest: Manifest, from_disk: bool, ) -> _BlobMap: """Build the ``file_path → bytes`` map for the scan. When *from_disk* is True, read each file from the working tree. Files deleted from the working tree are excluded entirely — a deleted file has no symbols, so its symbols cannot be dead. When False, read exclusively from the object store (historical snapshot). """ result: _BlobMap = {} for fp, oid in manifest.items(): if from_disk: try: result[fp] = (root / fp).read_bytes() except OSError: pass # File deleted from working tree — exclude from scan. else: raw = read_object(root, oid) if raw is not None: result[fp] = raw return result def _scan_file_bytes( file_bytes: _BlobMap, kind_filter: str | None, max_file_bytes: int, workers: int, language_filter: str | None, path_filter: str | None, exclude_tests: bool, exclude_private: bool, high_confidence_only: bool, allowlist: frozenset[str], entry_point_addresses: frozenset[str] = frozenset(), ) -> tuple[list[_DeadCandidate], int, float, int, int]: """Full dead-code analysis pipeline. Args: file_bytes: Map of ``file_path → raw bytes`` to analyse. kind_filter: Restrict to symbols of this kind, or ``None``. max_file_bytes: Skip files larger than this many bytes. workers: Number of parallel parse threads. language_filter: Restrict to this language name, or ``None``. path_filter: Glob pattern for file path restriction. exclude_tests: When ``True``, skip test files. exclude_private: When ``True``, skip ``_private`` symbols. high_confidence_only: When ``True``, only return high-confidence hits. allowlist: Set of symbol addresses to suppress. entry_point_addresses: Addresses of framework-wired entry points. These are *never* reported as dead code because they are externally reachable via the framework even though no user code calls them explicitly. Returns: ``(candidates, scanned_symbols, elapsed_seconds, skipped, errors)``. """ t0 = time.monotonic() analyses: list[_FileAnalysis] = [] with ThreadPoolExecutor(max_workers=workers) as pool: futures = { pool.submit(_analyse_file, fp, raw, kind_filter, max_file_bytes): fp for fp, raw in file_bytes.items() } for future in as_completed(futures): analyses.append(future.result()) elapsed = time.monotonic() - t0 all_ref_names: set[str] = set() all_imported_names: set[str] = set() for a in analyses: all_ref_names.update(a.ref_names) all_imported_names.update(a.imported_names) candidates: list[_DeadCandidate] = [] scanned_symbols = 0 for analysis in sorted(analyses, key=lambda a: a.file_path): if analysis.skipped or analysis.error: continue if not _matches_path_filter(analysis.file_path, path_filter): continue if language_filter and analysis.lang != language_filter: continue if exclude_tests and _is_test_file(analysis.file_path): continue mod_imported = _module_is_imported(analysis.file_path, all_imported_names) for address, rec in sorted(analysis.symbol_tree.items()): if rec["kind"] == "import": continue scanned_symbols += 1 bare_name = rec["name"].split(".")[-1] if exclude_private and bare_name.startswith("_"): continue if address in allowlist: continue if bare_name in all_ref_names: continue if address in entry_point_addresses: continue cand = _DeadCandidate( address=address, file_path=analysis.file_path, kind=rec["kind"], referenced=False, module_imported=mod_imported, ) if high_confidence_only and cand.confidence != "high": continue candidates.append(cand) candidates.sort(key=lambda c: (c.confidence != "high", c.file_path, c.address)) skipped = sum(1 for a in analyses if a.skipped) errors = sum(1 for a in analyses if a.error) return candidates, scanned_symbols, elapsed, skipped, errors # ── CLI registration ────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the dead subcommand.""" parser = subparsers.add_parser( "dead", help="Find symbols with no references and no importers — dead code candidates.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--kind", "-k", default=None, metavar="KIND", dest="kind_filter", help="Restrict to symbols of this kind (function, class, method, async_function, …).", ) parser.add_argument( "--include-tests", action="store_true", dest="include_tests", help=( "Include test files (paths containing 'test' or 'spec') in the analysis. " "Tests are excluded by default because pytest discovers them by naming " "convention rather than by reference, which produces thousands of false positives." ), ) parser.add_argument( "--exclude-private", action="store_true", dest="exclude_private", help="Exclude symbols whose name starts with '_'.", ) parser.add_argument( "--high-confidence-only", action="store_true", dest="high_confidence_only", help="Show only HIGH confidence candidates (module not imported).", ) parser.add_argument( "--path", "-p", default=None, metavar="GLOB", dest="path_filter", help="Restrict to files matching this glob pattern (e.g. 'musehub/services/*').", ) parser.add_argument( "--language", "-l", default="Python", metavar="LANG", dest="language_filter", help=( "Restrict to files of a specific language (default: Python). " "Use --language all to scan every language including Markdown, TOML, etc. " "Markdown sections and variables are never Python references, so scanning " "them without this filter produces thousands of false positives." ), ) parser.add_argument( "--top", "-n", default=None, type=int, metavar="N", dest="top", help="Show only the top N candidates.", ) parser.add_argument( "--group-by-file", "-g", action="store_true", dest="group_by_file", help="Group output by file instead of a flat sorted list.", ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help=( "Analyse a historical committed snapshot instead of the working tree. " "Accepts a full commit ID, a short prefix, HEAD, or a branch name." ), ) parser.add_argument( "--workers", "-w", default=8, type=int, metavar="N", dest="workers", help="Number of parallel worker threads for parsing (default: 8).", ) parser.add_argument( "--max-file-bytes", default=_DEFAULT_MAX_FILE_BYTES, type=int, metavar="N", dest="max_file_bytes", help="Skip files larger than N bytes (default: 524288).", ) parser.add_argument( "--no-color", action="store_true", dest="no_color", help="Disable ANSI colour output.", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as JSON.", ) parser.add_argument( "--delete", action="store_true", dest="delete", help=( "Interactively delete dead symbols from the working tree (Python only). " "Prompts for each candidate unless --yes is also given." ), ) parser.add_argument( "--yes", "-y", action="store_true", dest="yes", help="Skip confirmation prompts when used with --delete.", ) parser.add_argument( "--allowlist", default=None, metavar="FILE", dest="allowlist", help=( "JSON file with a list of symbol addresses to suppress. " "Example: [\".muse/dead-allowlist.json\"]" ), ) parser.add_argument( "--compare", default=None, metavar="REF", dest="compare_ref", help=( "Diff dead-code results against this commit reference. " "Shows which symbols newly became dead and which were recovered." ), ) parser.add_argument( "--count", action="store_true", dest="count_only", help="Print only the total count of dead-code candidates (scriptable).", ) parser.add_argument( "--save-allowlist", default=None, metavar="FILE", dest="save_allowlist", help=( "Save all found dead-code candidate addresses to FILE as a JSON list. " "Use as input to --allowlist to permanently suppress known false positives." ), ) parser.set_defaults(func=run) # ── Main logic ──────────────────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Find symbols with no references and no importers — dead code candidates.""" kind_filter: str | None = args.kind_filter exclude_tests: bool = not args.include_tests exclude_private: bool = args.exclude_private high_confidence_only: bool = args.high_confidence_only path_filter: str | None = args.path_filter raw_lang: str = args.language_filter language_filter: str | None = None if raw_lang.lower() == "all" else raw_lang top: int | None = (clamp_int(args.top, 1, 100_000, 'top') if args.top is not None else None) group_by_file: bool = args.group_by_file ref: str | None = args.ref compare_ref: str | None = args.compare_ref workers: int = min(max(1, args.workers), _MAX_WORKERS) max_file_bytes: int = max(args.max_file_bytes, _MIN_FILE_BYTES) as_json: bool = args.as_json do_delete: bool = args.delete auto_yes: bool = args.yes allowlist_path: str | None = args.allowlist count_only: bool = args.count_only save_allowlist_path: str | None = args.save_allowlist use_color: bool = not args.no_color and sys.stdout.isatty() and not as_json and not do_delete if do_delete and compare_ref: print("❌ --delete and --compare are mutually exclusive.", file=sys.stderr) 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_dead_for_vault run_dead_for_vault(root, args) return except Exception: # noqa: BLE001 pass repo_id = read_repo_id(root) branch = read_current_branch(root) allowlist = _load_allowlist(allowlist_path) default_allowlist_path = root / ".muse" / "dead-allowlist.json" if default_allowlist_path.exists() and not allowlist_path: allowlist = allowlist | _load_allowlist(str(default_allowlist_path)) # ── Resolve file bytes ──────────────────────────────────────────────────── commit: CommitRecord | None source_label: str if ref is None: commit = resolve_commit_ref(root, repo_id, branch, None) if commit is None: _err("No commits found — repository may be empty.", use_color) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} file_bytes = _load_file_bytes(root, manifest, from_disk=True) source_label = "working tree" if not as_json and not do_delete and not count_only: _print_header_workdir(len(file_bytes), use_color) else: commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: _err(f"Commit '{ref}' not found.", use_color) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} file_bytes = _load_file_bytes(root, manifest, from_disk=False) source_label = f"{commit.commit_id[:8]} on {branch}" if not as_json and not do_delete and not count_only: _print_header(commit.commit_id, branch, len(file_bytes), use_color) # ── Build implicit entry-point graph ───────────────────────────────────── # Entry-point symbols are framework-wired (e.g. FastAPI route handlers). # They are externally reachable via the runtime even though no user code # calls them directly — they must never be flagged as dead. implicit_graph: ImplicitEdgeGraph = build_implicit_edge_graph(root, manifest) entry_point_addresses: frozenset[str] = frozenset(implicit_graph.keys()) # ── Common scan args ────────────────────────────────────────────────────── scan_kwargs = _ScanKwargs( kind_filter=kind_filter, max_file_bytes=max_file_bytes, workers=workers, language_filter=language_filter, path_filter=path_filter, exclude_tests=exclude_tests, exclude_private=exclude_private, high_confidence_only=high_confidence_only, allowlist=allowlist, ) candidates, scanned_symbols, elapsed, skipped_count, error_count = _scan_file_bytes( file_bytes, **scan_kwargs, entry_point_addresses=entry_point_addresses ) if top is not None: candidates = candidates[:top] # ── --save-allowlist ────────────────────────────────────────────────────── if save_allowlist_path: _save_allowlist(save_allowlist_path, candidates) # ── --compare diff ──────────────────────────────────────────────────────── compare_commit: CommitRecord | None = None new_dead: list[_DeadCandidate] = [] recovered: list[_DeadCandidate] = [] if compare_ref: compare_commit = resolve_commit_ref(root, repo_id, branch, compare_ref) if compare_commit is None: _err(f"--compare commit '{compare_ref}' not found.", use_color) raise SystemExit(ExitCode.USER_ERROR) compare_manifest = get_commit_snapshot_manifest(root, compare_commit.commit_id) or {} compare_file_bytes = _load_file_bytes(root, compare_manifest, from_disk=False) compare_candidates, _, _, _, _ = _scan_file_bytes( compare_file_bytes, **scan_kwargs ) current_addrs = {c.address for c in candidates} compare_addrs = {c.address for c in compare_candidates} new_dead = [c for c in candidates if c.address not in compare_addrs] recovered_addrs = compare_addrs - current_addrs recovered = [c for c in compare_candidates if c.address in recovered_addrs] # ── Stats ───────────────────────────────────────────────────────────────── high_count = sum(1 for c in candidates if c.confidence == "high") medium_count = sum(1 for c in candidates if c.confidence == "medium") by_kind: _KindCountMap = {} for c in candidates: by_kind[c.kind] = by_kind.get(c.kind, 0) + 1 files_with_dead: set[str] = {c.file_path for c in candidates} # ── Output ──────────────────────────────────────────────────────────────── if count_only and not as_json: print(len(candidates)) return if as_json: payload = _DeadPayload( source=source_label, total_files_scanned=len(file_bytes), total_symbols_scanned=scanned_symbols, elapsed_seconds=round(elapsed, 3), high_confidence_count=high_count, medium_confidence_count=medium_count, results=[c.to_dict() for c in candidates], ) if compare_commit is not None: payload["compare_commit_id"] = compare_commit.commit_id payload["new_dead"] = [c.to_dict() for c in new_dead] payload["recovered"] = [c.to_dict() for c in recovered] payload["net_change"] = len(new_dead) - len(recovered) print(json.dumps(payload, indent=2)) return if do_delete: _run_delete_mode(root, candidates, auto_yes) return if not candidates: print(f" {_c('✅ No dead code candidates found.', _GREEN, use_color=use_color)}") _print_footer_note(use_color) return if group_by_file: _print_grouped(candidates, use_color) else: _print_flat(candidates, use_color) _print_summary( candidates=candidates, high_count=high_count, medium_count=medium_count, by_kind=by_kind, files_with_dead=files_with_dead, scanned_symbols=scanned_symbols, total_files=len(file_bytes), skipped_count=skipped_count, error_count=error_count, elapsed=elapsed, top=top, use_color=use_color, ) if compare_commit is not None: _print_compare_diff(new_dead, recovered, compare_commit, use_color) # ── Delete mode ─────────────────────────────────────────────────────────────── def _run_delete_mode( root: pathlib.Path, candidates: list[_DeadCandidate], auto_yes: bool, ) -> None: """Interactively delete dead symbols from the working tree.""" py_candidates = [c for c in candidates if c.file_path.endswith((".py", ".pyi"))] skipped_non_py = len(candidates) - len(py_candidates) if not py_candidates: print(" No Python dead-code candidates to delete.") if skipped_non_py: print(f" ({skipped_non_py} non-Python candidate(s) skipped — delete is Python-only)") return print(f"\n{_BOLD}muse code dead --delete{_RESET} — {len(py_candidates)} Python candidate(s)") if skipped_non_py: print(f" {_GRAY}({skipped_non_py} non-Python candidate(s) not shown){_RESET}") print(_GRAY + "─" * 72 + _RESET) # Group by file so we process each file at most once and delete bottom-to-top. by_file: _DeadByFile = {} for c in py_candidates: by_file.setdefault(c.file_path, []).append(c) deleted_total = 0 skipped_total = 0 failed_total = 0 for file_path in sorted(by_file): file_candidates = by_file[file_path] print(f"\n {_CYAN}{_BOLD}{sanitize_display(file_path)}{_RESET} ({len(file_candidates)} candidate(s))") # Collect which symbols to delete (after user confirmation). to_delete: list[_DeadCandidate] = [] for c in file_candidates: bare = c.address.split("::")[-1] conf_label = ( f"{_RED}HIGH{_RESET}" if c.confidence == "high" else f"{_YELLOW}MED{_RESET}" ) kind_label = _BLUE + _kind_icon(c.kind) + _RESET print(f" {_RED}✗{_RESET} {_WHITE}{bare}{_RESET} {kind_label} [{conf_label}]") if auto_yes: to_delete.append(c) else: try: answer = input(" Delete? [y/N/q] ").strip().lower() except (EOFError, KeyboardInterrupt): print("\n Aborted.") return if answer == "q": print(" Aborted.") return if answer == "y": to_delete.append(c) else: skipped_total += 1 if not to_delete: continue # Find spans for all symbols we will delete, then remove bottom-to-top. abs_path = (root / file_path).resolve() # Path traversal guard: ensure the resolved path stays within root. try: abs_path.relative_to(root.resolve()) except ValueError: print(f" {_YELLOW}⚠ {sanitize_display(str(file_path))!r} escapes repo root — skipping{_RESET}") failed_total += len(to_delete) continue if not abs_path.exists(): print(f" {_YELLOW}⚠ file not in working tree — skipping{_RESET}") failed_total += len(to_delete) continue source = abs_path.read_bytes() spans: list[tuple[int, int, _DeadCandidate]] = [] for c in to_delete: parts = c.address.split("::") bare = parts[-1] parent_class = parts[-2] if len(parts) >= 3 else None span = _find_symbol_span(source, bare, parent_class) if span is None: print(f" {_YELLOW}⚠ could not locate {sanitize_display(bare)} in {sanitize_display(file_path)}{_RESET}") failed_total += 1 else: spans.append((*span, c)) if not spans: continue # Sort descending by start line so later deletions don't shift earlier lines. spans.sort(key=lambda x: -x[0]) lines = source.decode(errors="replace").splitlines(keepends=True) for start, end, c in spans: bare = c.address.split("::")[-1] lines = _delete_symbol_lines(lines, start, end) print(f" {_GREEN}✅ deleted {bare}{_RESET} (lines {start}–{end})") deleted_total += 1 if auto_yes: skipped_total = max(0, skipped_total) abs_path.write_text("".join(lines), encoding="utf-8") print(f"\n{_GRAY}{'─' * 72}{_RESET}") print(f" {_GREEN}Deleted:{_RESET} {deleted_total}") if skipped_total: print(f" {_GRAY}Skipped:{_RESET} {skipped_total}") if failed_total: print(f" {_YELLOW}Failed:{_RESET} {failed_total}") if deleted_total: print(f"\n Run {_CYAN}muse status{_RESET} to review, then {_CYAN}muse commit{_RESET} to record.") # ── Output helpers ──────────────────────────────────────────────────────────── def _save_allowlist(path: str, candidates: list[_DeadCandidate]) -> None: """Write candidate addresses to *path* as a JSON array.""" try: pathlib.Path(path).write_text( json.dumps(sorted(c.address for c in candidates), indent=2), encoding="utf-8", ) logger.info("✅ Saved %d address(es) to %s", len(candidates), path) except OSError as exc: logger.warning("⚠️ Could not write allowlist %s: %s", path, exc) def _is_test_file(file_path: str) -> bool: lower = file_path.lower() return "test" in lower or "spec" in lower def _err(msg: str, use_color: bool) -> None: print(_c(f"❌ {msg}", _RED, _BOLD, use_color=use_color), file=sys.stderr) def _print_header(commit_id: str, branch: str, total_files: int, use_color: bool) -> None: sha = _c(commit_id[:8], _CYAN, _BOLD, use_color=use_color) br = _c(branch, _MAGENTA, use_color=use_color) n = _c(str(total_files), _BOLD, use_color=use_color) print(f"\n{_c('Dead code candidates', _BOLD, use_color=use_color)} — commit {sha} on {br} — {n} files") print(_c("━" * 72, _GRAY, use_color=use_color)) def _print_header_workdir(total_files: int, use_color: bool) -> None: label = _c("working tree", _CYAN, _BOLD, use_color=use_color) n = _c(str(total_files), _BOLD, use_color=use_color) print(f"\n{_c('Dead code candidates', _BOLD, use_color=use_color)} — {label} — {n} files") print(_c("━" * 72, _GRAY, use_color=use_color)) def _kind_icon(kind: str) -> str: return { "function": "fn", "async_function": "async fn", "method": "method", "async_method": "async method", "class": "class", "variable": "var", "constant": "const", }.get(kind, kind) def _confidence_label(c: _DeadCandidate, use_color: bool) -> str: if c.confidence == "high": return _c("HIGH", _RED, _BOLD, use_color=use_color) return _c("MED ", _YELLOW, use_color=use_color) def _print_flat(candidates: list[_DeadCandidate], use_color: bool) -> None: max_addr = min(max(len(c.address) for c in candidates), 80) max_kind = max(len(_kind_icon(c.kind)) for c in candidates) prev_conf = "" for c in candidates: conf = c.confidence if conf != prev_conf: prev_conf = conf if conf == "high": label = _c(" ── HIGH CONFIDENCE — not referenced, module not imported", _RED, use_color=use_color) else: label = _c(" ── MEDIUM CONFIDENCE — not referenced, module is imported", _YELLOW, use_color=use_color) print(f"\n{label}") print(_c(" " + "─" * 68, _GRAY, use_color=use_color)) addr_str = _c(c.address[:max_addr], _WHITE if conf == "high" else _GRAY, use_color=use_color) kind_str = _c(_kind_icon(c.kind).ljust(max_kind), _BLUE, use_color=use_color) conf_str = _confidence_label(c, use_color) print(f" {addr_str:<{max_addr + 20}} {kind_str} [{conf_str}]") def _print_grouped(candidates: list[_DeadCandidate], use_color: bool) -> None: by_file: _DeadByFile = {} for c in candidates: by_file.setdefault(c.file_path, []).append(c) for file_path in sorted(by_file): group = by_file[file_path] high_n = sum(1 for c in group if c.confidence == "high") med_n = sum(1 for c in group if c.confidence == "medium") counts = [] if high_n: counts.append(_c(f"{high_n} high", _RED, use_color=use_color)) if med_n: counts.append(_c(f"{med_n} med", _YELLOW, use_color=use_color)) print(f"\n {_c(file_path, _CYAN, _BOLD, use_color=use_color)} {', '.join(counts)}") max_name = min(max(len(c.address.split('::')[-1]) for c in group), 60) for c in sorted(group, key=lambda x: (x.confidence != "high", x.address)): sym_name = c.address.split("::")[-1] if "::" in c.address else c.address kind_str = _c(_kind_icon(c.kind), _BLUE, use_color=use_color) if c.confidence == "high": sym_str = _c(sym_name.ljust(max_name), _WHITE, use_color=use_color) marker = _c("✗", _RED, _BOLD, use_color=use_color) else: sym_str = _c(sym_name.ljust(max_name), _GRAY, use_color=use_color) marker = _c("·", _YELLOW, use_color=use_color) print(f" {marker} {sym_str} {kind_str}") def _print_summary( candidates: list[_DeadCandidate], high_count: int, medium_count: int, by_kind: _KindCountMap, files_with_dead: set[str], scanned_symbols: int, total_files: int, skipped_count: int, error_count: int, elapsed: float, top: int | None, use_color: bool, ) -> None: print(f"\n{_c('━' * 72, _GRAY, use_color=use_color)}") print(f"{_c('Summary', _BOLD, use_color=use_color)}") print(f" {_c('High confidence', _RED, use_color=use_color):.<50} {high_count:>6}") print(f" {_c('Medium confidence', _YELLOW, use_color=use_color):.<50} {medium_count:>6}") print(f" {'Total candidates':.<42} {len(candidates):>6}") if top is not None: print(f" {_c(f'(showing top {top})', _GRAY, use_color=use_color)}") print(f" {'Symbols scanned':.<42} {scanned_symbols:>6,}") print(f" {'Files with dead symbols':.<42} {len(files_with_dead):>6}") print(f" {'Files scanned':.<42} {total_files:>6,}") if skipped_count: print(f" {_c('Files skipped (too large)', _GRAY, use_color=use_color):.<50} {skipped_count:>6}") if error_count: print(f" {_c('Files with parse errors', _YELLOW, use_color=use_color):.<50} {error_count:>6}") print(f" {'Elapsed':.<42} {elapsed:>5.1f}s") if by_kind: print(f"\n {_c('By kind:', _BOLD, use_color=use_color)}") for kind, count in sorted(by_kind.items(), key=lambda x: -x[1]): bar_len = min(count // max(1, max(by_kind.values()) // 20), 20) bar = _c("█" * bar_len, _BLUE, use_color=use_color) print(f" {_kind_icon(kind):<16} {bar} {count:>5,}") _print_footer_note(use_color) def _print_compare_diff( new_dead: list[_DeadCandidate], recovered: list[_DeadCandidate], compare_commit: CommitRecord, use_color: bool, ) -> None: """Render the dead-code diff section.""" print(f"\n{_c('━' * 72, _GRAY, use_color=use_color)}") sha = _c(compare_commit.commit_id[:8], _CYAN, _BOLD, use_color=use_color) print(f"{_c('Dead-code diff', _BOLD, use_color=use_color)} vs {sha}") net = len(new_dead) - len(recovered) sign = "+" if net >= 0 else "" colour = _RED if net > 0 else _GREEN if net < 0 else _GRAY print(f" Net change: {_c(f'{sign}{net}', colour, use_color=use_color)}") if new_dead: print(f"\n {_c(f'New dead ({len(new_dead)}):', _RED, use_color=use_color)}") for c in new_dead: print(f" + {c.address} [{_kind_icon(c.kind)}] [{c.confidence.upper()}]") if recovered: print(f"\n {_c(f'Recovered ({len(recovered)}):', _GREEN, use_color=use_color)}") for c in recovered: print(f" - {c.address} [{_kind_icon(c.kind)}]") def _print_footer_note(use_color: bool) -> None: note = ( "Dynamic dispatch, exported APIs, and entry points are not detected.\n" " Treat results as candidates — verify before deleting.\n" " Use --delete to interactively remove candidates from the working tree.\n" " Use --allowlist to suppress known false positives." ) print(f"\n{_c(note, _GRAY, use_color=use_color)}")