"""muse code symbols -- list every semantic symbol in a snapshot. Muse tracks the semantic interior of every source file -- the full symbol graph the code plugin builds at commit time -- giving each function, class, method, and variable a stable, content-addressed identity independent of line numbers or formatting. Output (default -- human-readable table):: src/utils.py fn calculate_total line 12 fn _validate_amount line 28 class Invoice line 45 method Invoice.to_dict line 52 method Invoice.from_dict line 61 src/models.py class User line 8 method User.__init__ line 10 method User.save line 19 12 symbols across 2 files (Python: 12) Flags:: --commit Inspect a specific commit instead of the working tree. Accepts a full or abbreviated commit SHA, a branch name, or HEAD~N. --kind Filter to a specific symbol kind: function, async_function, class, method, async_method, variable, import, section, rule. --file Show symbols from a single file. Accepts an exact path or a unique suffix (e.g. "billing.py" matches "src/billing.py"). --language Show symbols from files of this language only (case-insensitive, e.g. "python", "TypeScript", "go"). --count Print only the total symbol count and per-language breakdown. --hashes Include content hashes alongside each symbol. --json Emit a structured JSON object for tooling integration:: { "source_ref": "a1b2c3d4", "working_tree": true, "total_symbols": 12, "results": [ {"address": "src/utils.py::calculate_total", "file": "src/utils.py", "kind": "function", ...} ] } """ from __future__ import annotations import argparse import json import logging import pathlib import sys from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( Manifest, get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.plugins.code._query import language_of, symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolTree type _SymbolTreeMap = dict[str, SymbolTree] type _CounterMap = dict[str, int] type _KindDisplay = dict[str, tuple[str, list[str]]] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # ANSI colour helpers — only emitted when stdout is a TTY. # --------------------------------------------------------------------------- _RESET = "\033[0m" _BOLD = "\033[1m" _DIM = "\033[2m" _CYAN = "\033[36m" _YELLOW = "\033[33m" _BLUE = "\033[34m" _GREEN = "\033[32m" _MAGENTA = "\033[35m" def _c(text: str, *codes: str, tty: bool) -> str: """Wrap *text* in ANSI *codes* when *tty* is True.""" if not tty: return text return "".join(codes) + text + _RESET # Maps symbol kind → (short icon, ANSI colour codes). _KIND_DISPLAY: _KindDisplay = { "function": ("fn", [_BLUE]), "async_function": ("fn~", [_BLUE, _DIM]), "class": ("class", [_YELLOW, _BOLD]), "method": ("method", [_CYAN]), "async_method": ("method~", [_CYAN, _DIM]), "variable": ("var", [_DIM]), "import": ("import", [_DIM]), "section": ("section", [_GREEN]), "rule": ("rule", [_MAGENTA]), } _VALID_KINDS: frozenset[str] = frozenset(_KIND_DISPLAY) # --------------------------------------------------------------------------- # Language helpers # --------------------------------------------------------------------------- # Canonical map: lowercase language name → display name. # Built from _SUFFIX_LANG in _query.py to stay in sync. from muse.plugins.code._query import _SUFFIX_LANG # noqa: E402 (module-level import) from muse.core.validation import sanitize_display _LANG_CANONICAL: Manifest = {lang.lower(): lang for lang in set(_SUFFIX_LANG.values())} def _normalise_language(lang: str) -> str: """Return the canonical capitalisation for *lang*, or *lang* unchanged.""" return _LANG_CANONICAL.get(lang.strip().lower(), lang.strip()) # --------------------------------------------------------------------------- # File-filter helpers # --------------------------------------------------------------------------- def _file_matches(file_path: str, file_filter: str) -> bool: """Return True if *file_path* equals or uniquely ends with *file_filter*. Allows passing ``"billing.py"`` to match ``"src/billing.py"`` without requiring callers to know the full directory prefix. Uses a separator anchor (``/``) to prevent ``y.py`` matching ``billy.py``. """ if file_path == file_filter: return True normalized = file_filter.replace("\\", "/") return file_path.endswith("/" + normalized) def _resolve_file_filter( file_filter: str, manifest: Manifest, ) -> str | None: """Resolve *file_filter* to the exact manifest path, or ``None`` on ambiguity/miss. Prints a helpful message to stderr and raises ``SystemExit`` on ambiguity. Returns ``None`` when there is no match (caller emits "no symbols found"). """ matching = [p for p in sorted(manifest) if _file_matches(p, file_filter)] if len(matching) == 1: return matching[0] if len(matching) > 1: print( f"❌ '{file_filter}' is ambiguous — matches {len(matching)} files. " "Use a more specific path:", file=sys.stderr, ) for m in matching[:10]: print(f" {m}", file=sys.stderr) if len(matching) > 10: print(f" … and {len(matching) - 10} more", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return None # no match — caller handles the empty result # --------------------------------------------------------------------------- # Repository helpers # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Output helpers # --------------------------------------------------------------------------- def _lang_counts(symbol_map: _SymbolTreeMap) -> _CounterMap: """Return a language-name → symbol-count mapping for *symbol_map*.""" counts: _CounterMap = {} for file_path, tree in symbol_map.items(): lang = language_of(file_path) counts[lang] = counts.get(lang, 0) + len(tree) return counts def _print_human( symbol_map: _SymbolTreeMap, show_hashes: bool, tty: bool, ) -> None: """Render symbol_map as a human-readable, optionally coloured table.""" if not symbol_map: print(" (no semantic symbols found)") return total = 0 for file_path, tree in symbol_map.items(): total += len(tree) print(f"\n{_c(sanitize_display(file_path), _BOLD, tty=tty)}") for _addr, rec in sorted(tree.items(), key=lambda kv: kv[1]["lineno"]): kind = rec["kind"] icon, colour_codes = _KIND_DISPLAY.get(kind, (kind, [])) name = rec["qualified_name"] lineno = rec["lineno"] icon_str = _c(f"{icon:<10}", *colour_codes, tty=tty) name_str = f"{name:<40}" line_str = _c(f"line {lineno:>4}", _DIM, tty=tty) hash_suffix = ( _c(f" {rec['content_id'][:8]}..", _DIM, tty=tty) if show_hashes else "" ) print(f" {icon_str} {name_str} {line_str}{hash_suffix}") counts = _lang_counts(symbol_map) lang_str = ", ".join(f"{lang}: {count:,}" for lang, count in sorted(counts.items())) sym_word = "symbol" if total == 1 else "symbols" file_word = "file" if len(symbol_map) == 1 else "files" print( f"\n{_c(f'{total:,}', _BOLD, tty=tty)} {sym_word} across " f"{len(symbol_map):,} {file_word} ({lang_str})" ) def _emit_json( symbol_map: _SymbolTreeMap, source_ref: str, working_tree: bool, ) -> None: """Emit the symbol map as a structured JSON object. Schema:: { "source_ref": "", // or "working-tree" "working_tree": true | false, "total_symbols": , "results": [ { "address": "::", "kind": "function" | "class" | ..., "name": "", "qualified_name": "", "file": "", "lineno": , "end_lineno": , "content_id": "", "body_hash": "", "signature_id": "" } ] } """ results: list[dict[str, str | int]] = [] for file_path, tree in symbol_map.items(): for addr, rec in sorted(tree.items(), key=lambda kv: kv[1]["lineno"]): results.append({ "address": addr, "kind": rec["kind"], "name": rec["name"], "qualified_name": rec["qualified_name"], "file": file_path, "lineno": rec["lineno"], "end_lineno": rec["end_lineno"], "content_id": rec["content_id"], "body_hash": rec["body_hash"], "signature_id": rec["signature_id"], }) print(json.dumps({ "source_ref": source_ref, "working_tree": working_tree, "total_symbols": len(results), "results": results, }, indent=2)) # --------------------------------------------------------------------------- # Argument parser registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the symbols subcommand.""" parser = subparsers.add_parser( "symbols", help="List every semantic symbol (function, class, method…) in a snapshot.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--commit", "-c", dest="ref", default=None, metavar="REF", help="Commit ID or branch to inspect (default: working tree).", ) parser.add_argument( "--kind", "-k", dest="kind_filter", default=None, metavar="KIND", help=( "Filter to symbols of a specific kind " "(function, async_function, class, method, async_method, " "variable, import, section, rule)." ), ) parser.add_argument( "--file", "-f", dest="file_filter", default=None, metavar="PATH", help=( "Show symbols from a single file. Accepts an exact path or a " "unique path suffix (e.g. 'billing.py' matches 'src/billing.py')." ), ) parser.add_argument( "--language", "-l", dest="language_filter", default=None, metavar="LANG", help="Show symbols from files of this language only (case-insensitive).", ) parser.add_argument( "--hashes", dest="show_hashes", action="store_true", help="Include content hashes in the output.", ) output_group = parser.add_mutually_exclusive_group() output_group.add_argument( "--count", dest="count_only", action="store_true", help="Print only the total symbol count and language breakdown.", ) output_group.add_argument( "--json", dest="as_json", action="store_true", help="Emit the full symbol table as JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command entry point # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """List every semantic symbol (function, class, method…) in a snapshot. Unlike ``git grep`` or ``ctags``, ``muse code symbols`` reads the semantic symbol graph produced by the domain plugin's AST analysis — stable, content-addressed identities for every symbol, independent of line numbers or formatting. When ``--commit`` is omitted the command reads directly from the working tree, reflecting edits that have not yet been committed. Pass ``--commit HEAD`` (or any ref) to inspect a historical snapshot instead. Use ``--kind`` and ``--file`` to narrow the output. Use ``--json`` for tooling and agent integration. """ ref: str | None = args.ref kind_filter: str | None = args.kind_filter file_filter: str | None = args.file_filter language_filter: str | None = args.language_filter count_only: bool = args.count_only show_hashes: bool = args.show_hashes as_json: bool = args.as_json tty: bool = sys.stdout.isatty() # ── Input validation ────────────────────────────────────────────────────── if kind_filter is not None and kind_filter not in _VALID_KINDS: valid = ", ".join(sorted(_VALID_KINDS)) print(f"❌ Unknown kind '{kind_filter}'. Valid kinds: {valid}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if language_filter is not None: language_filter = _normalise_language(language_filter) # ── Repo / commit resolution ────────────────────────────────────────────── root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: label = ref or "HEAD" print(f"❌ Commit '{label}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} if not manifest: print( f"❌ Snapshot for commit {commit.commit_id[:8]} has no files.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # ── Working-tree vs object-store mode / file-filter resolution ─────────── working_tree = ref is None # True when no --commit was given workdir = root if working_tree else None resolved_file_filter = file_filter if file_filter is not None: found = _resolve_file_filter(file_filter, manifest) if found is not None: resolved_file_filter = found elif working_tree: # File not in HEAD manifest — may be a new uncommitted file. # Inject a synthetic manifest entry so symbols_for_snapshot can # parse it directly from the working directory. candidate = root / file_filter if candidate.is_file(): manifest = {file_filter: ""} resolved_file_filter = file_filter # ── Symbol extraction ───────────────────────────────────────────────────── symbol_map = symbols_for_snapshot( root, manifest, kind_filter=kind_filter, file_filter=resolved_file_filter, language_filter=language_filter, workdir=workdir, ) # ── Source reference label ──────────────────────────────────────────────── if working_tree: source_ref = "working-tree" else: source_ref = commit.commit_id[:8] # ── Output ──────────────────────────────────────────────────────────────── if count_only: total = sum(len(t) for t in symbol_map.values()) counts = _lang_counts(symbol_map) lang_str = ", ".join(f"{lang}: {count:,}" for lang, count in sorted(counts.items())) sym_word = "symbol" if total == 1 else "symbols" print(f"{total:,} {sym_word} ({lang_str})") return if as_json: _emit_json(symbol_map, source_ref=source_ref, working_tree=working_tree) return if working_tree: header = ( f'working tree ' f'(HEAD {commit.commit_id[:8]} "{sanitize_display(commit.message)}")' ) else: header = f'commit {commit.commit_id[:8]} "{sanitize_display(commit.message)}"' print(_c(header, _DIM, tty=tty)) _print_human(symbol_map, show_hashes, tty)