aaronrene / gabriel-muse public
dead.py python
1,197 lines 47.2 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """muse code dead — dead code detection.
2
3 Finds symbols that are **never referenced** and whose containing module is
4 **never imported** by anything else in the codebase.
5
6 A symbol is a dead-code candidate when two independent conditions hold:
7
8 1. **No reference**: its bare name does not appear as any ``ast.Name`` id
9 or ``ast.Attribute`` attr anywhere in the codebase. This is broader
10 than call-site detection — it catches attribute accesses, keyword
11 argument values, type annotations, ``isinstance`` checks, and every
12 other form of name usage, not just direct calls.
13
14 2. **No import**: its containing file's module name does not appear in
15 any ``import``-kind symbol in any other file.
16
17 Both conditions must hold simultaneously. A function that is never
18 referenced but lives in a module that *is* imported is still reachable —
19 it may be part of an exported API even if it's not used internally.
20
21 Performance
22 -----------
23 All files are processed in a **single parallel pass** and AST-parsed
24 exactly once. Without ``--commit`` the working tree is read from disk,
25 so uncommitted changes are immediately visible. With ``--commit`` the
26 specified historical snapshot is read from the object store.
27 Imports, references, and symbol trees are all extracted in the same
28 pass, then combined. ``--workers`` controls the thread-pool size.
29
30 Security
31 --------
32 ``ast.parse`` never executes code. Files exceeding ``--max-file-bytes``
33 (default 512 KB) are skipped to prevent stalls on generated or minified
34 files. ``--delete`` validates every file path inside the repo root before
35 touching the working tree.
36
37 Known limitations
38 -----------------
39 - Symbols whose names are extremely common (e.g. ``run``, ``name``) may
40 appear as false negatives because a matching name exists somewhere else.
41 - Exported APIs: symbols accessed from outside the repo (library code)
42 appear dead because the callers are not in the snapshot.
43 - Entry points: ``main()``, CLI callbacks, and test functions appear dead
44 by design. Use ``--exclude-tests`` to hide test file symbols.
45 - tree-sitter languages: reference extraction is Python-only. Symbols in
46 Go/Rust/TypeScript files are checked for import-graph reachability only.
47 - ``--delete`` is Python-only (requires AST line-range information).
48
49 Usage::
50
51 muse code dead
52 muse code dead --kind function
53 muse code dead --exclude-tests
54 muse code dead --exclude-private
55 muse code dead --high-confidence-only
56 muse code dead --path "musehub/services/*"
57 muse code dead --language Python
58 muse code dead --top 50
59 muse code dead --group-by-file
60 muse code dead --commit HEAD~10
61 muse code dead --workers 8
62 muse code dead --json
63 muse code dead --delete
64 muse code dead --delete --yes
65 muse code dead --allowlist .muse/dead-allowlist.json
66
67 Confidence levels::
68
69 HIGH — not referenced AND module not imported → almost certainly dead
70 MEDIUM — not referenced, but module IS imported → may be exported API surface
71
72 Flags:
73
74 ``--kind KIND, -k KIND``
75 Restrict to symbols of a specific kind (function, class, method, …).
76
77 ``--exclude-tests``
78 Exclude symbols in files whose path contains ``test`` or ``spec``.
79
80 ``--exclude-private``
81 Exclude symbols whose bare name starts with ``_``.
82
83 ``--high-confidence-only``
84 Show only HIGH confidence candidates (module not imported).
85
86 ``--path GLOB, -p GLOB``
87 Restrict to files matching this glob pattern (e.g. ``"musehub/services/*"``).
88
89 ``--language LANG, -l LANG``
90 Restrict to files of a specific language (e.g. ``Python``, ``TypeScript``).
91
92 ``--top N``
93 Show only the top N candidates.
94
95 ``--group-by-file, -g``
96 Group output by file instead of a flat sorted list.
97
98 ``--commit REF, -c REF``
99 Analyse a historical snapshot instead of HEAD.
100
101 ``--workers N, -w N``
102 Number of parallel worker threads for file parsing (default: 8).
103
104 ``--max-file-bytes N``
105 Skip files larger than N bytes (default: 524288 = 512 KB).
106
107 ``--no-color``
108 Disable ANSI color output.
109
110 ``--json``
111 Emit results as JSON.
112
113 ``--delete``
114 Interactively delete dead symbols from the working tree (Python only).
115 Prompts for each candidate unless ``--yes`` is also given.
116
117 ``--yes, -y``
118 Skip confirmation prompts when used with ``--delete``.
119
120 ``--allowlist FILE``
121 JSON file containing a list of symbol addresses to suppress from output.
122 Addresses are matched as exact strings against the ``address`` field.
123 Example file: ``[\"muse/cli/config.py::MuseConfig\"]``
124 """
125
126 from __future__ import annotations
127
128 import argparse
129 import ast
130 import fnmatch
131 import json
132 import logging
133 import os
134 import pathlib
135 import sys
136 import time
137 from concurrent.futures import ThreadPoolExecutor, as_completed
138 from dataclasses import dataclass, field
139 from typing import TypedDict
140
141 from muse.core.errors import ExitCode
142 from muse.core.object_store import read_object
143 from muse.core.repo import read_repo_id, require_repo
144 from muse.core._types import Manifest
145 from muse.core.store import (
146 CommitRecord,
147 get_commit_snapshot_manifest,
148 read_current_branch,
149 resolve_commit_ref,
150 )
151 from muse.plugins.code._framework import ImplicitEdgeGraph, build_implicit_edge_graph
152 from muse.plugins.code._query import language_of
153 from muse.plugins.code.ast_parser import SEMANTIC_EXTENSIONS, SymbolTree, parse_symbols
154
155 type _BlobMap = dict[str, bytes]
156 type _KindCountMap = dict[str, int]
157 type _DeadByFile = dict[str, list["_DeadCandidate"]]
158 from muse.core.validation import MAX_AST_BYTES, clamp_int, sanitize_display
159
160 logger = logging.getLogger(__name__)
161
162
163 class _DeadCandidateJson(TypedDict):
164 """JSON-serialisable representation of one dead-code candidate."""
165
166 address: str
167 file_path: str
168 kind: str
169 referenced: bool
170 module_imported: bool
171 confidence: str
172 reason: str
173
174
175 class _DeadPayload(TypedDict, total=False):
176 """JSON payload for ``muse code dead`` output."""
177
178 source: str
179 total_files_scanned: int
180 total_symbols_scanned: int
181 elapsed_seconds: float
182 high_confidence_count: int
183 medium_confidence_count: int
184 results: list[_DeadCandidateJson]
185 compare_commit_id: str
186 new_dead: list[_DeadCandidateJson]
187 recovered: list[_DeadCandidateJson]
188 net_change: int
189
190
191 class _ScanKwargs(TypedDict):
192 """Keyword arguments forwarded to every :func:`_scan_file_bytes` call.
193
194 Collected into a TypedDict so the ``**scan_kwargs`` spread is type-safe
195 without a ``# type: ignore`` and the common args are defined once.
196 """
197
198 kind_filter: str | None
199 max_file_bytes: int
200 workers: int
201 language_filter: str | None
202 path_filter: str | None
203 exclude_tests: bool
204 exclude_private: bool
205 high_confidence_only: bool
206 allowlist: frozenset[str]
207
208
209 _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"})
210 _MAX_WORKERS: int = 64
211 _MIN_FILE_BYTES: int = 4_096
212
213 # Maximum file size we'll parse (512 KB). Prevents stalling on generated files.
214 _DEFAULT_MAX_FILE_BYTES: int = 524_288
215
216 # ── ANSI colours ──────────────────────────────────────────────────────────────
217
218 _RESET = "\033[0m"
219 _BOLD = "\033[1m"
220 _DIM = "\033[2m"
221 _RED = "\033[31m"
222 _YELLOW = "\033[33m"
223 _CYAN = "\033[36m"
224 _GREEN = "\033[32m"
225 _BLUE = "\033[34m"
226 _MAGENTA = "\033[35m"
227 _WHITE = "\033[37m"
228 _GRAY = "\033[90m"
229
230
231 def _c(text: str, *codes: str, use_color: bool = True) -> str:
232 """Wrap *text* with ANSI escape codes if *use_color* is True."""
233 if not use_color:
234 return text
235 return "".join(codes) + text + _RESET
236
237
238 # ── Data structures ───────────────────────────────────────────────────────────
239
240 @dataclass
241 class _FileAnalysis:
242 """Everything extracted from a single file in one pass."""
243 file_path: str
244 lang: str
245 symbol_tree: SymbolTree = field(default_factory=dict)
246 # Every name referenced anywhere in the file (ast.Name ids + ast.Attribute attrs).
247 # This is broader than call-sites: catches attribute access, keyword args,
248 # type annotations, isinstance checks, decorator names, etc.
249 ref_names: set[str] = field(default_factory=set)
250 # Imported module/name strings (from import-kind symbols)
251 imported_names: set[str] = field(default_factory=set)
252 skipped: bool = False
253 error: str | None = None
254
255
256 @dataclass
257 class _DeadCandidate:
258 address: str
259 file_path: str
260 kind: str
261 referenced: bool
262 module_imported: bool
263
264 @property
265 def confidence(self) -> str:
266 return "high" if not self.module_imported else "medium"
267
268 @property
269 def reason(self) -> str:
270 if not self.referenced and not self.module_imported:
271 return "not referenced, module not imported"
272 return "not referenced (module imported — may be exported API)"
273
274 def to_dict(self) -> _DeadCandidateJson:
275 return _DeadCandidateJson(
276 address=self.address,
277 file_path=self.file_path,
278 kind=self.kind,
279 referenced=self.referenced,
280 module_imported=self.module_imported,
281 confidence=self.confidence,
282 reason=self.reason,
283 )
284
285
286 # ── Single-pass file analysis ─────────────────────────────────────────────────
287
288 def _analyse_file(
289 file_path: str,
290 raw: bytes,
291 kind_filter: str | None,
292 max_file_bytes: int,
293 ) -> _FileAnalysis:
294 """Parse and extract symbols + references + imports from one file.
295
296 Thread-safe: pure functions only, no shared mutable state.
297 The caller is responsible for supplying the raw file bytes — either read
298 from disk (working tree) or fetched from the object store (historical commit).
299 """
300 lang = language_of(file_path)
301 result = _FileAnalysis(file_path=file_path, lang=lang)
302
303 if len(raw) > max_file_bytes:
304 result.skipped = True
305 return result
306
307 suffix = pathlib.PurePosixPath(file_path).suffix.lower()
308 if suffix not in SEMANTIC_EXTENSIONS:
309 return result
310
311 # ── Symbol extraction (all languages with AST support) ─────────────────
312 try:
313 tree = parse_symbols(raw, file_path)
314 except Exception as exc: # noqa: BLE001
315 result.error = str(exc)
316 return result
317
318 for rec in tree.values():
319 if rec["kind"] == "import":
320 result.imported_names.add(rec["qualified_name"])
321
322 if kind_filter:
323 tree = {addr: rec for addr, rec in tree.items() if rec["kind"] == kind_filter}
324 result.symbol_tree = tree
325
326 # ── Reference + module-import extraction (Python only via stdlib ast) ─────
327 # We walk ALL nodes once:
328 #
329 # ast.Name / ast.Attribute — broad reference tracking (fixes logger,
330 # func=run keyword args, property access, isinstance args, annotations)
331 #
332 # ast.ImportFrom / ast.Import — extract the actual dotted module paths
333 # ("from muse.core.store import X" → "muse.core.store"). The Muse
334 # symbol tree stores imports as "import::symbolname" with no module
335 # path, so we must supplement it here to make _module_is_imported work.
336 if suffix in _PY_SUFFIXES:
337 try:
338 if len(raw) > MAX_AST_BYTES:
339 return result
340 py_tree = ast.parse(raw)
341 except SyntaxError:
342 return result
343 for node in ast.walk(py_tree):
344 if isinstance(node, ast.Name):
345 result.ref_names.add(node.id)
346 elif isinstance(node, ast.Attribute):
347 result.ref_names.add(node.attr)
348 elif isinstance(node, ast.ImportFrom):
349 if node.module:
350 result.imported_names.add(node.module)
351 elif isinstance(node, ast.Import):
352 for alias in node.names:
353 result.imported_names.add(alias.name)
354
355 return result
356
357
358 # ── Module-import matching ────────────────────────────────────────────────────
359
360 def _module_is_imported(file_path: str, imported_names: set[str]) -> bool:
361 """Return True if *file_path*'s module name appears anywhere in *imported_names*."""
362 stem = pathlib.PurePosixPath(file_path).stem
363 module = pathlib.PurePosixPath(file_path).with_suffix("").as_posix().replace("/", ".")
364 for imp in imported_names:
365 if (
366 imp == stem
367 or imp == module
368 or imp.endswith(f".{stem}")
369 or imp.endswith(f".{module}")
370 or stem in imp.split(".")
371 ):
372 return True
373 return False
374
375
376 # ── Path filter ───────────────────────────────────────────────────────────────
377
378 def _matches_path_filter(file_path: str, pattern: str | None) -> bool:
379 if pattern is None:
380 return True
381 return fnmatch.fnmatch(file_path, pattern) or fnmatch.fnmatch(file_path, f"**/{pattern}")
382
383
384 # ── Symbol deletion (Python only) ─────────────────────────────────────────────
385
386 def _find_symbol_span(source: bytes, bare_name: str, parent_class: str | None) -> tuple[int, int] | None:
387 """Return (start_lineno, end_lineno) 1-indexed for the named symbol.
388
389 Accounts for decorator lines (start is the first decorator's line).
390 Returns None if the symbol cannot be located.
391 """
392 try:
393 if len(source) > MAX_AST_BYTES:
394 return None
395 tree = ast.parse(source)
396 except SyntaxError:
397 return None
398
399 search_body: list[ast.stmt] = tree.body
400 if parent_class:
401 for node in ast.walk(tree):
402 if isinstance(node, ast.ClassDef) and node.name == parent_class:
403 search_body = list(node.body)
404 break
405
406 for node in search_body:
407 node_name: str | None = None
408 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
409 node_name = node.name
410 elif isinstance(node, ast.Assign):
411 if len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
412 node_name = node.targets[0].id
413 elif isinstance(node, ast.AnnAssign):
414 if isinstance(node.target, ast.Name):
415 node_name = node.target.id
416
417 if node_name != bare_name:
418 continue
419
420 if not hasattr(node, "end_lineno") or node.end_lineno is None:
421 return None
422 end: int = node.end_lineno
423 start: int = node.lineno
424 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
425 if node.decorator_list:
426 start = node.decorator_list[0].lineno
427 return (start, end)
428
429 return None
430
431
432 def _delete_symbol_lines(source_lines: list[str], start: int, end: int) -> list[str]:
433 """Remove lines *start*–*end* (1-indexed, inclusive) and normalise blank lines."""
434 before = list(source_lines[: start - 1])
435 after = list(source_lines[end:])
436
437 # Strip trailing blank lines from the block before the deletion point.
438 while before and not before[-1].strip():
439 before.pop()
440
441 # Add one blank separator line if there is still content below.
442 if after and any(line.strip() for line in after):
443 return before + ["\n"] + after
444 return before + after
445
446
447
448 # ── Repo ID ───────────────────────────────────────────────────────────────────
449
450
451
452 # ── Allowlist ─────────────────────────────────────────────────────────────────
453
454 def _load_allowlist(path: str | None) -> frozenset[str]:
455 """Load a JSON list of symbol addresses that should be suppressed."""
456 if path is None:
457 return frozenset()
458 try:
459 raw = pathlib.Path(path).read_text(encoding="utf-8")
460 parsed = json.loads(raw)
461 if not isinstance(parsed, list):
462 logger.warning("dead-code allowlist must be a JSON array; ignoring %s", path)
463 return frozenset()
464 return frozenset(str(x) for x in parsed)
465 except (OSError, json.JSONDecodeError) as exc:
466 logger.warning("Could not load allowlist %s: %s", path, exc)
467 return frozenset()
468
469
470 # ── Shared scan pipeline ──────────────────────────────────────────────────────
471
472
473 def _load_file_bytes(
474 root: pathlib.Path,
475 manifest: Manifest,
476 from_disk: bool,
477 ) -> _BlobMap:
478 """Build the ``file_path → bytes`` map for the scan.
479
480 When *from_disk* is True, read each file from the working tree. Files
481 deleted from the working tree are excluded entirely — a deleted file has
482 no symbols, so its symbols cannot be dead. When False, read exclusively
483 from the object store (historical snapshot).
484 """
485 result: _BlobMap = {}
486 for fp, oid in manifest.items():
487 if from_disk:
488 try:
489 result[fp] = (root / fp).read_bytes()
490 except OSError:
491 pass # File deleted from working tree — exclude from scan.
492 else:
493 raw = read_object(root, oid)
494 if raw is not None:
495 result[fp] = raw
496 return result
497
498
499 def _scan_file_bytes(
500 file_bytes: _BlobMap,
501 kind_filter: str | None,
502 max_file_bytes: int,
503 workers: int,
504 language_filter: str | None,
505 path_filter: str | None,
506 exclude_tests: bool,
507 exclude_private: bool,
508 high_confidence_only: bool,
509 allowlist: frozenset[str],
510 entry_point_addresses: frozenset[str] = frozenset(),
511 ) -> tuple[list[_DeadCandidate], int, float, int, int]:
512 """Full dead-code analysis pipeline.
513
514 Args:
515 file_bytes: Map of ``file_path → raw bytes`` to analyse.
516 kind_filter: Restrict to symbols of this kind, or ``None``.
517 max_file_bytes: Skip files larger than this many bytes.
518 workers: Number of parallel parse threads.
519 language_filter: Restrict to this language name, or ``None``.
520 path_filter: Glob pattern for file path restriction.
521 exclude_tests: When ``True``, skip test files.
522 exclude_private: When ``True``, skip ``_private`` symbols.
523 high_confidence_only: When ``True``, only return high-confidence hits.
524 allowlist: Set of symbol addresses to suppress.
525 entry_point_addresses: Addresses of framework-wired entry points.
526 These are *never* reported as dead code because
527 they are externally reachable via the framework
528 even though no user code calls them explicitly.
529
530 Returns:
531 ``(candidates, scanned_symbols, elapsed_seconds, skipped, errors)``.
532 """
533 t0 = time.monotonic()
534 analyses: list[_FileAnalysis] = []
535
536 with ThreadPoolExecutor(max_workers=workers) as pool:
537 futures = {
538 pool.submit(_analyse_file, fp, raw, kind_filter, max_file_bytes): fp
539 for fp, raw in file_bytes.items()
540 }
541 for future in as_completed(futures):
542 analyses.append(future.result())
543
544 elapsed = time.monotonic() - t0
545
546 all_ref_names: set[str] = set()
547 all_imported_names: set[str] = set()
548 for a in analyses:
549 all_ref_names.update(a.ref_names)
550 all_imported_names.update(a.imported_names)
551
552 candidates: list[_DeadCandidate] = []
553 scanned_symbols = 0
554
555 for analysis in sorted(analyses, key=lambda a: a.file_path):
556 if analysis.skipped or analysis.error:
557 continue
558 if not _matches_path_filter(analysis.file_path, path_filter):
559 continue
560 if language_filter and analysis.lang != language_filter:
561 continue
562 if exclude_tests and _is_test_file(analysis.file_path):
563 continue
564
565 mod_imported = _module_is_imported(analysis.file_path, all_imported_names)
566
567 for address, rec in sorted(analysis.symbol_tree.items()):
568 if rec["kind"] == "import":
569 continue
570 scanned_symbols += 1
571 bare_name = rec["name"].split(".")[-1]
572 if exclude_private and bare_name.startswith("_"):
573 continue
574 if address in allowlist:
575 continue
576 if bare_name in all_ref_names:
577 continue
578 if address in entry_point_addresses:
579 continue
580 cand = _DeadCandidate(
581 address=address,
582 file_path=analysis.file_path,
583 kind=rec["kind"],
584 referenced=False,
585 module_imported=mod_imported,
586 )
587 if high_confidence_only and cand.confidence != "high":
588 continue
589 candidates.append(cand)
590
591 candidates.sort(key=lambda c: (c.confidence != "high", c.file_path, c.address))
592
593 skipped = sum(1 for a in analyses if a.skipped)
594 errors = sum(1 for a in analyses if a.error)
595
596 return candidates, scanned_symbols, elapsed, skipped, errors
597
598
599 # ── CLI registration ──────────────────────────────────────────────────────────
600
601 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
602 """Register the dead subcommand."""
603 parser = subparsers.add_parser(
604 "dead",
605 help="Find symbols with no references and no importers — dead code candidates.",
606 description=__doc__,
607 formatter_class=argparse.RawDescriptionHelpFormatter,
608 )
609 parser.add_argument(
610 "--kind", "-k", default=None, metavar="KIND", dest="kind_filter",
611 help="Restrict to symbols of this kind (function, class, method, async_function, …).",
612 )
613 parser.add_argument(
614 "--include-tests", action="store_true", dest="include_tests",
615 help=(
616 "Include test files (paths containing 'test' or 'spec') in the analysis. "
617 "Tests are excluded by default because pytest discovers them by naming "
618 "convention rather than by reference, which produces thousands of false positives."
619 ),
620 )
621 parser.add_argument(
622 "--exclude-private", action="store_true", dest="exclude_private",
623 help="Exclude symbols whose name starts with '_'.",
624 )
625 parser.add_argument(
626 "--high-confidence-only", action="store_true", dest="high_confidence_only",
627 help="Show only HIGH confidence candidates (module not imported).",
628 )
629 parser.add_argument(
630 "--path", "-p", default=None, metavar="GLOB", dest="path_filter",
631 help="Restrict to files matching this glob pattern (e.g. 'musehub/services/*').",
632 )
633 parser.add_argument(
634 "--language", "-l", default="Python", metavar="LANG", dest="language_filter",
635 help=(
636 "Restrict to files of a specific language (default: Python). "
637 "Use --language all to scan every language including Markdown, TOML, etc. "
638 "Markdown sections and variables are never Python references, so scanning "
639 "them without this filter produces thousands of false positives."
640 ),
641 )
642 parser.add_argument(
643 "--top", "-n", default=None, type=int, metavar="N", dest="top",
644 help="Show only the top N candidates.",
645 )
646 parser.add_argument(
647 "--group-by-file", "-g", action="store_true", dest="group_by_file",
648 help="Group output by file instead of a flat sorted list.",
649 )
650 parser.add_argument(
651 "--commit", "-c", default=None, metavar="REF", dest="ref",
652 help=(
653 "Analyse a historical committed snapshot instead of the working tree. "
654 "Accepts a full commit ID, a short prefix, HEAD, or a branch name."
655 ),
656 )
657 parser.add_argument(
658 "--workers", "-w", default=8, type=int, metavar="N", dest="workers",
659 help="Number of parallel worker threads for parsing (default: 8).",
660 )
661 parser.add_argument(
662 "--max-file-bytes", default=_DEFAULT_MAX_FILE_BYTES, type=int,
663 metavar="N", dest="max_file_bytes",
664 help="Skip files larger than N bytes (default: 524288).",
665 )
666 parser.add_argument(
667 "--no-color", action="store_true", dest="no_color",
668 help="Disable ANSI colour output.",
669 )
670 parser.add_argument(
671 "--json", action="store_true", dest="as_json",
672 help="Emit results as JSON.",
673 )
674 parser.add_argument(
675 "--delete", action="store_true", dest="delete",
676 help=(
677 "Interactively delete dead symbols from the working tree (Python only). "
678 "Prompts for each candidate unless --yes is also given."
679 ),
680 )
681 parser.add_argument(
682 "--yes", "-y", action="store_true", dest="yes",
683 help="Skip confirmation prompts when used with --delete.",
684 )
685 parser.add_argument(
686 "--allowlist", default=None, metavar="FILE", dest="allowlist",
687 help=(
688 "JSON file with a list of symbol addresses to suppress. "
689 "Example: [\".muse/dead-allowlist.json\"]"
690 ),
691 )
692 parser.add_argument(
693 "--compare", default=None, metavar="REF", dest="compare_ref",
694 help=(
695 "Diff dead-code results against this commit reference. "
696 "Shows which symbols newly became dead and which were recovered."
697 ),
698 )
699 parser.add_argument(
700 "--count", action="store_true", dest="count_only",
701 help="Print only the total count of dead-code candidates (scriptable).",
702 )
703 parser.add_argument(
704 "--save-allowlist", default=None, metavar="FILE", dest="save_allowlist",
705 help=(
706 "Save all found dead-code candidate addresses to FILE as a JSON list. "
707 "Use as input to --allowlist to permanently suppress known false positives."
708 ),
709 )
710 parser.set_defaults(func=run)
711
712
713 # ── Main logic ────────────────────────────────────────────────────────────────
714
715 def run(args: argparse.Namespace) -> None:
716 """Find symbols with no references and no importers — dead code candidates."""
717 kind_filter: str | None = args.kind_filter
718 exclude_tests: bool = not args.include_tests
719 exclude_private: bool = args.exclude_private
720 high_confidence_only: bool = args.high_confidence_only
721 path_filter: str | None = args.path_filter
722 raw_lang: str = args.language_filter
723 language_filter: str | None = None if raw_lang.lower() == "all" else raw_lang
724 top: int | None = (clamp_int(args.top, 1, 100_000, 'top') if args.top is not None else None)
725 group_by_file: bool = args.group_by_file
726 ref: str | None = args.ref
727 compare_ref: str | None = args.compare_ref
728 workers: int = min(max(1, args.workers), _MAX_WORKERS)
729 max_file_bytes: int = max(args.max_file_bytes, _MIN_FILE_BYTES)
730 as_json: bool = args.as_json
731 do_delete: bool = args.delete
732 auto_yes: bool = args.yes
733 allowlist_path: str | None = args.allowlist
734 count_only: bool = args.count_only
735 save_allowlist_path: str | None = args.save_allowlist
736 use_color: bool = not args.no_color and sys.stdout.isatty() and not as_json and not do_delete
737
738 if do_delete and compare_ref:
739 print("❌ --delete and --compare are mutually exclusive.", file=sys.stderr)
740 raise SystemExit(ExitCode.USER_ERROR)
741
742 root = require_repo()
743
744 from muse.plugins.registry import read_domain
745 try:
746 if read_domain(root) == "knowtation":
747 from muse.plugins.knowtation.cli_hooks import run_dead_for_vault
748 run_dead_for_vault(root, args)
749 return
750 except Exception: # noqa: BLE001
751 pass
752
753 repo_id = read_repo_id(root)
754 branch = read_current_branch(root)
755
756 allowlist = _load_allowlist(allowlist_path)
757 default_allowlist_path = root / ".muse" / "dead-allowlist.json"
758 if default_allowlist_path.exists() and not allowlist_path:
759 allowlist = allowlist | _load_allowlist(str(default_allowlist_path))
760
761 # ── Resolve file bytes ────────────────────────────────────────────────────
762 commit: CommitRecord | None
763 source_label: str
764
765 if ref is None:
766 commit = resolve_commit_ref(root, repo_id, branch, None)
767 if commit is None:
768 _err("No commits found — repository may be empty.", use_color)
769 raise SystemExit(ExitCode.USER_ERROR)
770 manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
771 file_bytes = _load_file_bytes(root, manifest, from_disk=True)
772 source_label = "working tree"
773 if not as_json and not do_delete and not count_only:
774 _print_header_workdir(len(file_bytes), use_color)
775 else:
776 commit = resolve_commit_ref(root, repo_id, branch, ref)
777 if commit is None:
778 _err(f"Commit '{ref}' not found.", use_color)
779 raise SystemExit(ExitCode.USER_ERROR)
780 manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
781 file_bytes = _load_file_bytes(root, manifest, from_disk=False)
782 source_label = f"{commit.commit_id[:8]} on {branch}"
783 if not as_json and not do_delete and not count_only:
784 _print_header(commit.commit_id, branch, len(file_bytes), use_color)
785
786 # ── Build implicit entry-point graph ─────────────────────────────────────
787 # Entry-point symbols are framework-wired (e.g. FastAPI route handlers).
788 # They are externally reachable via the runtime even though no user code
789 # calls them directly — they must never be flagged as dead.
790 implicit_graph: ImplicitEdgeGraph = build_implicit_edge_graph(root, manifest)
791 entry_point_addresses: frozenset[str] = frozenset(implicit_graph.keys())
792
793 # ── Common scan args ──────────────────────────────────────────────────────
794 scan_kwargs = _ScanKwargs(
795 kind_filter=kind_filter,
796 max_file_bytes=max_file_bytes,
797 workers=workers,
798 language_filter=language_filter,
799 path_filter=path_filter,
800 exclude_tests=exclude_tests,
801 exclude_private=exclude_private,
802 high_confidence_only=high_confidence_only,
803 allowlist=allowlist,
804 )
805
806 candidates, scanned_symbols, elapsed, skipped_count, error_count = _scan_file_bytes(
807 file_bytes, **scan_kwargs, entry_point_addresses=entry_point_addresses
808 )
809
810 if top is not None:
811 candidates = candidates[:top]
812
813 # ── --save-allowlist ──────────────────────────────────────────────────────
814 if save_allowlist_path:
815 _save_allowlist(save_allowlist_path, candidates)
816
817 # ── --compare diff ────────────────────────────────────────────────────────
818 compare_commit: CommitRecord | None = None
819 new_dead: list[_DeadCandidate] = []
820 recovered: list[_DeadCandidate] = []
821
822 if compare_ref:
823 compare_commit = resolve_commit_ref(root, repo_id, branch, compare_ref)
824 if compare_commit is None:
825 _err(f"--compare commit '{compare_ref}' not found.", use_color)
826 raise SystemExit(ExitCode.USER_ERROR)
827 compare_manifest = get_commit_snapshot_manifest(root, compare_commit.commit_id) or {}
828 compare_file_bytes = _load_file_bytes(root, compare_manifest, from_disk=False)
829 compare_candidates, _, _, _, _ = _scan_file_bytes(
830 compare_file_bytes, **scan_kwargs
831 )
832 current_addrs = {c.address for c in candidates}
833 compare_addrs = {c.address for c in compare_candidates}
834 new_dead = [c for c in candidates if c.address not in compare_addrs]
835 recovered_addrs = compare_addrs - current_addrs
836 recovered = [c for c in compare_candidates if c.address in recovered_addrs]
837
838 # ── Stats ─────────────────────────────────────────────────────────────────
839 high_count = sum(1 for c in candidates if c.confidence == "high")
840 medium_count = sum(1 for c in candidates if c.confidence == "medium")
841 by_kind: _KindCountMap = {}
842 for c in candidates:
843 by_kind[c.kind] = by_kind.get(c.kind, 0) + 1
844 files_with_dead: set[str] = {c.file_path for c in candidates}
845
846 # ── Output ────────────────────────────────────────────────────────────────
847 if count_only and not as_json:
848 print(len(candidates))
849 return
850
851 if as_json:
852 payload = _DeadPayload(
853 source=source_label,
854 total_files_scanned=len(file_bytes),
855 total_symbols_scanned=scanned_symbols,
856 elapsed_seconds=round(elapsed, 3),
857 high_confidence_count=high_count,
858 medium_confidence_count=medium_count,
859 results=[c.to_dict() for c in candidates],
860 )
861 if compare_commit is not None:
862 payload["compare_commit_id"] = compare_commit.commit_id
863 payload["new_dead"] = [c.to_dict() for c in new_dead]
864 payload["recovered"] = [c.to_dict() for c in recovered]
865 payload["net_change"] = len(new_dead) - len(recovered)
866 print(json.dumps(payload, indent=2))
867 return
868
869 if do_delete:
870 _run_delete_mode(root, candidates, auto_yes)
871 return
872
873 if not candidates:
874 print(f" {_c('✅ No dead code candidates found.', _GREEN, use_color=use_color)}")
875 _print_footer_note(use_color)
876 return
877
878 if group_by_file:
879 _print_grouped(candidates, use_color)
880 else:
881 _print_flat(candidates, use_color)
882
883 _print_summary(
884 candidates=candidates,
885 high_count=high_count,
886 medium_count=medium_count,
887 by_kind=by_kind,
888 files_with_dead=files_with_dead,
889 scanned_symbols=scanned_symbols,
890 total_files=len(file_bytes),
891 skipped_count=skipped_count,
892 error_count=error_count,
893 elapsed=elapsed,
894 top=top,
895 use_color=use_color,
896 )
897
898 if compare_commit is not None:
899 _print_compare_diff(new_dead, recovered, compare_commit, use_color)
900
901
902 # ── Delete mode ───────────────────────────────────────────────────────────────
903
904 def _run_delete_mode(
905 root: pathlib.Path,
906 candidates: list[_DeadCandidate],
907 auto_yes: bool,
908 ) -> None:
909 """Interactively delete dead symbols from the working tree."""
910 py_candidates = [c for c in candidates if c.file_path.endswith((".py", ".pyi"))]
911 skipped_non_py = len(candidates) - len(py_candidates)
912
913 if not py_candidates:
914 print(" No Python dead-code candidates to delete.")
915 if skipped_non_py:
916 print(f" ({skipped_non_py} non-Python candidate(s) skipped — delete is Python-only)")
917 return
918
919 print(f"\n{_BOLD}muse code dead --delete{_RESET} — {len(py_candidates)} Python candidate(s)")
920 if skipped_non_py:
921 print(f" {_GRAY}({skipped_non_py} non-Python candidate(s) not shown){_RESET}")
922 print(_GRAY + "─" * 72 + _RESET)
923
924 # Group by file so we process each file at most once and delete bottom-to-top.
925 by_file: _DeadByFile = {}
926 for c in py_candidates:
927 by_file.setdefault(c.file_path, []).append(c)
928
929 deleted_total = 0
930 skipped_total = 0
931 failed_total = 0
932
933 for file_path in sorted(by_file):
934 file_candidates = by_file[file_path]
935
936 print(f"\n {_CYAN}{_BOLD}{sanitize_display(file_path)}{_RESET} ({len(file_candidates)} candidate(s))")
937
938 # Collect which symbols to delete (after user confirmation).
939 to_delete: list[_DeadCandidate] = []
940 for c in file_candidates:
941 bare = c.address.split("::")[-1]
942 conf_label = (
943 f"{_RED}HIGH{_RESET}" if c.confidence == "high" else f"{_YELLOW}MED{_RESET}"
944 )
945 kind_label = _BLUE + _kind_icon(c.kind) + _RESET
946 print(f" {_RED}✗{_RESET} {_WHITE}{bare}{_RESET} {kind_label} [{conf_label}]")
947
948 if auto_yes:
949 to_delete.append(c)
950 else:
951 try:
952 answer = input(" Delete? [y/N/q] ").strip().lower()
953 except (EOFError, KeyboardInterrupt):
954 print("\n Aborted.")
955 return
956 if answer == "q":
957 print(" Aborted.")
958 return
959 if answer == "y":
960 to_delete.append(c)
961 else:
962 skipped_total += 1
963
964 if not to_delete:
965 continue
966
967 # Find spans for all symbols we will delete, then remove bottom-to-top.
968 abs_path = (root / file_path).resolve()
969 # Path traversal guard: ensure the resolved path stays within root.
970 try:
971 abs_path.relative_to(root.resolve())
972 except ValueError:
973 print(f" {_YELLOW}⚠ {sanitize_display(str(file_path))!r} escapes repo root — skipping{_RESET}")
974 failed_total += len(to_delete)
975 continue
976 if not abs_path.exists():
977 print(f" {_YELLOW}⚠ file not in working tree — skipping{_RESET}")
978 failed_total += len(to_delete)
979 continue
980
981 source = abs_path.read_bytes()
982 spans: list[tuple[int, int, _DeadCandidate]] = []
983 for c in to_delete:
984 parts = c.address.split("::")
985 bare = parts[-1]
986 parent_class = parts[-2] if len(parts) >= 3 else None
987 span = _find_symbol_span(source, bare, parent_class)
988 if span is None:
989 print(f" {_YELLOW}⚠ could not locate {sanitize_display(bare)} in {sanitize_display(file_path)}{_RESET}")
990 failed_total += 1
991 else:
992 spans.append((*span, c))
993
994 if not spans:
995 continue
996
997 # Sort descending by start line so later deletions don't shift earlier lines.
998 spans.sort(key=lambda x: -x[0])
999
1000 lines = source.decode(errors="replace").splitlines(keepends=True)
1001 for start, end, c in spans:
1002 bare = c.address.split("::")[-1]
1003 lines = _delete_symbol_lines(lines, start, end)
1004 print(f" {_GREEN}✅ deleted {bare}{_RESET} (lines {start}–{end})")
1005 deleted_total += 1
1006 if auto_yes:
1007 skipped_total = max(0, skipped_total)
1008
1009 abs_path.write_text("".join(lines), encoding="utf-8")
1010
1011 print(f"\n{_GRAY}{'─' * 72}{_RESET}")
1012 print(f" {_GREEN}Deleted:{_RESET} {deleted_total}")
1013 if skipped_total:
1014 print(f" {_GRAY}Skipped:{_RESET} {skipped_total}")
1015 if failed_total:
1016 print(f" {_YELLOW}Failed:{_RESET} {failed_total}")
1017 if deleted_total:
1018 print(f"\n Run {_CYAN}muse status{_RESET} to review, then {_CYAN}muse commit{_RESET} to record.")
1019
1020
1021 # ── Output helpers ────────────────────────────────────────────────────────────
1022
1023
1024 def _save_allowlist(path: str, candidates: list[_DeadCandidate]) -> None:
1025 """Write candidate addresses to *path* as a JSON array."""
1026 try:
1027 pathlib.Path(path).write_text(
1028 json.dumps(sorted(c.address for c in candidates), indent=2),
1029 encoding="utf-8",
1030 )
1031 logger.info("✅ Saved %d address(es) to %s", len(candidates), path)
1032 except OSError as exc:
1033 logger.warning("⚠️ Could not write allowlist %s: %s", path, exc)
1034
1035
1036 def _is_test_file(file_path: str) -> bool:
1037 lower = file_path.lower()
1038 return "test" in lower or "spec" in lower
1039
1040
1041 def _err(msg: str, use_color: bool) -> None:
1042 print(_c(f"❌ {msg}", _RED, _BOLD, use_color=use_color), file=sys.stderr)
1043
1044
1045 def _print_header(commit_id: str, branch: str, total_files: int, use_color: bool) -> None:
1046 sha = _c(commit_id[:8], _CYAN, _BOLD, use_color=use_color)
1047 br = _c(branch, _MAGENTA, use_color=use_color)
1048 n = _c(str(total_files), _BOLD, use_color=use_color)
1049 print(f"\n{_c('Dead code candidates', _BOLD, use_color=use_color)} — commit {sha} on {br} — {n} files")
1050 print(_c("━" * 72, _GRAY, use_color=use_color))
1051
1052
1053 def _print_header_workdir(total_files: int, use_color: bool) -> None:
1054 label = _c("working tree", _CYAN, _BOLD, use_color=use_color)
1055 n = _c(str(total_files), _BOLD, use_color=use_color)
1056 print(f"\n{_c('Dead code candidates', _BOLD, use_color=use_color)} — {label} — {n} files")
1057 print(_c("━" * 72, _GRAY, use_color=use_color))
1058
1059
1060 def _kind_icon(kind: str) -> str:
1061 return {
1062 "function": "fn",
1063 "async_function": "async fn",
1064 "method": "method",
1065 "async_method": "async method",
1066 "class": "class",
1067 "variable": "var",
1068 "constant": "const",
1069 }.get(kind, kind)
1070
1071
1072 def _confidence_label(c: _DeadCandidate, use_color: bool) -> str:
1073 if c.confidence == "high":
1074 return _c("HIGH", _RED, _BOLD, use_color=use_color)
1075 return _c("MED ", _YELLOW, use_color=use_color)
1076
1077
1078 def _print_flat(candidates: list[_DeadCandidate], use_color: bool) -> None:
1079 max_addr = min(max(len(c.address) for c in candidates), 80)
1080 max_kind = max(len(_kind_icon(c.kind)) for c in candidates)
1081 prev_conf = ""
1082 for c in candidates:
1083 conf = c.confidence
1084 if conf != prev_conf:
1085 prev_conf = conf
1086 if conf == "high":
1087 label = _c(" ── HIGH CONFIDENCE — not referenced, module not imported", _RED, use_color=use_color)
1088 else:
1089 label = _c(" ── MEDIUM CONFIDENCE — not referenced, module is imported", _YELLOW, use_color=use_color)
1090 print(f"\n{label}")
1091 print(_c(" " + "─" * 68, _GRAY, use_color=use_color))
1092 addr_str = _c(c.address[:max_addr], _WHITE if conf == "high" else _GRAY, use_color=use_color)
1093 kind_str = _c(_kind_icon(c.kind).ljust(max_kind), _BLUE, use_color=use_color)
1094 conf_str = _confidence_label(c, use_color)
1095 print(f" {addr_str:<{max_addr + 20}} {kind_str} [{conf_str}]")
1096
1097
1098 def _print_grouped(candidates: list[_DeadCandidate], use_color: bool) -> None:
1099 by_file: _DeadByFile = {}
1100 for c in candidates:
1101 by_file.setdefault(c.file_path, []).append(c)
1102
1103 for file_path in sorted(by_file):
1104 group = by_file[file_path]
1105 high_n = sum(1 for c in group if c.confidence == "high")
1106 med_n = sum(1 for c in group if c.confidence == "medium")
1107 counts = []
1108 if high_n:
1109 counts.append(_c(f"{high_n} high", _RED, use_color=use_color))
1110 if med_n:
1111 counts.append(_c(f"{med_n} med", _YELLOW, use_color=use_color))
1112 print(f"\n {_c(file_path, _CYAN, _BOLD, use_color=use_color)} {', '.join(counts)}")
1113 max_name = min(max(len(c.address.split('::')[-1]) for c in group), 60)
1114 for c in sorted(group, key=lambda x: (x.confidence != "high", x.address)):
1115 sym_name = c.address.split("::")[-1] if "::" in c.address else c.address
1116 kind_str = _c(_kind_icon(c.kind), _BLUE, use_color=use_color)
1117 if c.confidence == "high":
1118 sym_str = _c(sym_name.ljust(max_name), _WHITE, use_color=use_color)
1119 marker = _c("✗", _RED, _BOLD, use_color=use_color)
1120 else:
1121 sym_str = _c(sym_name.ljust(max_name), _GRAY, use_color=use_color)
1122 marker = _c("·", _YELLOW, use_color=use_color)
1123 print(f" {marker} {sym_str} {kind_str}")
1124
1125
1126 def _print_summary(
1127 candidates: list[_DeadCandidate],
1128 high_count: int,
1129 medium_count: int,
1130 by_kind: _KindCountMap,
1131 files_with_dead: set[str],
1132 scanned_symbols: int,
1133 total_files: int,
1134 skipped_count: int,
1135 error_count: int,
1136 elapsed: float,
1137 top: int | None,
1138 use_color: bool,
1139 ) -> None:
1140 print(f"\n{_c('━' * 72, _GRAY, use_color=use_color)}")
1141 print(f"{_c('Summary', _BOLD, use_color=use_color)}")
1142 print(f" {_c('High confidence', _RED, use_color=use_color):.<50} {high_count:>6}")
1143 print(f" {_c('Medium confidence', _YELLOW, use_color=use_color):.<50} {medium_count:>6}")
1144 print(f" {'Total candidates':.<42} {len(candidates):>6}")
1145 if top is not None:
1146 print(f" {_c(f'(showing top {top})', _GRAY, use_color=use_color)}")
1147 print(f" {'Symbols scanned':.<42} {scanned_symbols:>6,}")
1148 print(f" {'Files with dead symbols':.<42} {len(files_with_dead):>6}")
1149 print(f" {'Files scanned':.<42} {total_files:>6,}")
1150 if skipped_count:
1151 print(f" {_c('Files skipped (too large)', _GRAY, use_color=use_color):.<50} {skipped_count:>6}")
1152 if error_count:
1153 print(f" {_c('Files with parse errors', _YELLOW, use_color=use_color):.<50} {error_count:>6}")
1154 print(f" {'Elapsed':.<42} {elapsed:>5.1f}s")
1155
1156 if by_kind:
1157 print(f"\n {_c('By kind:', _BOLD, use_color=use_color)}")
1158 for kind, count in sorted(by_kind.items(), key=lambda x: -x[1]):
1159 bar_len = min(count // max(1, max(by_kind.values()) // 20), 20)
1160 bar = _c("█" * bar_len, _BLUE, use_color=use_color)
1161 print(f" {_kind_icon(kind):<16} {bar} {count:>5,}")
1162
1163 _print_footer_note(use_color)
1164
1165
1166 def _print_compare_diff(
1167 new_dead: list[_DeadCandidate],
1168 recovered: list[_DeadCandidate],
1169 compare_commit: CommitRecord,
1170 use_color: bool,
1171 ) -> None:
1172 """Render the dead-code diff section."""
1173 print(f"\n{_c('━' * 72, _GRAY, use_color=use_color)}")
1174 sha = _c(compare_commit.commit_id[:8], _CYAN, _BOLD, use_color=use_color)
1175 print(f"{_c('Dead-code diff', _BOLD, use_color=use_color)} vs {sha}")
1176 net = len(new_dead) - len(recovered)
1177 sign = "+" if net >= 0 else ""
1178 colour = _RED if net > 0 else _GREEN if net < 0 else _GRAY
1179 print(f" Net change: {_c(f'{sign}{net}', colour, use_color=use_color)}")
1180 if new_dead:
1181 print(f"\n {_c(f'New dead ({len(new_dead)}):', _RED, use_color=use_color)}")
1182 for c in new_dead:
1183 print(f" + {c.address} [{_kind_icon(c.kind)}] [{c.confidence.upper()}]")
1184 if recovered:
1185 print(f"\n {_c(f'Recovered ({len(recovered)}):', _GREEN, use_color=use_color)}")
1186 for c in recovered:
1187 print(f" - {c.address} [{_kind_icon(c.kind)}]")
1188
1189
1190 def _print_footer_note(use_color: bool) -> None:
1191 note = (
1192 "Dynamic dispatch, exported APIs, and entry points are not detected.\n"
1193 " Treat results as candidates — verify before deleting.\n"
1194 " Use --delete to interactively remove candidates from the working tree.\n"
1195 " Use --allowlist to suppress known false positives."
1196 )
1197 print(f"\n{_c(note, _GRAY, use_color=use_color)}")
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago