"""muse code cat — print the source of one or more symbols from HEAD or any commit. Address format:: muse code cat cache.py::LRUCache.get muse code cat cache.py::LRUCache.get --at abc123 muse code cat cache.py::LRUCache.get --at v0.1.4 # Multiple symbols in one call (useful for agents): muse code cat cache.py::LRUCache.get cache.py::LRUCache.set # All symbols in a file: muse code cat cache.py --all # Structured output for downstream processing: muse code cat cache.py::LRUCache.get --json muse code cat cache.py::LRUCache.get --format json The ``::`` separator is the same format used throughout Muse's symbol graph. The right-hand side is matched against the symbol's ``qualified_name`` first, then ``name`` (allowing short references like ``get`` when unambiguous). Without ``--at`` the working tree is read from disk, so uncommitted edits are immediately visible. With ``--at `` the specified committed snapshot is used instead. Exit codes ---------- 0 All requested symbols found and printed. 1 Address malformed, symbol not found, or file not tracked. 3 I/O error reading from the object store or disk. """ from __future__ import annotations import argparse import json import pathlib import sys import time 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.store import ( Manifest, get_commit_snapshot_manifest, get_head_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.core.validation import clamp_int, sanitize_display from muse.plugins.code.ast_parser import SymbolRecord, SymbolTree, adapter_for_path type _FileCache = dict[str, tuple[bytes, "SymbolTree"]] # ── Types ───────────────────────────────────────────────────────────────────── class CatResult(TypedDict): """One resolved symbol returned in --json mode.""" address: str file_path: str symbol: str kind: str lineno: int end_lineno: int source: str source_ref: str # "working tree" | "commit on " class CatError(TypedDict, total=False): """A failed lookup returned in --json mode. Fields ------ address The address that was requested. error Human-readable description of the failure. error_code Machine-parseable failure category (always present). hint Actionable recovery instruction for the caller. """ address: str error: str error_code: str hint: str class _FileError(Exception): """Raised by :func:`_get_file_bytes` instead of ``SystemExit`` so callers can map the failure to a precise ``error_code`` in JSON output.""" def __init__(self, message: str, code: str, hint: str = "") -> None: super().__init__(message) self.code = code self.hint = hint # ── Helpers ─────────────────────────────────────────────────────────────────── def _get_file_bytes( root: pathlib.Path, file_path: str, manifest: Manifest, source_is_workdir: bool, ) -> bytes: """Return raw bytes for *file_path*, reading from disk or the object store. When *source_is_workdir* is True we read from disk first (so uncommitted edits are visible) and fall back to the object store only if the file has been deleted. When False (historical commit) we always use the store. Raises :class:`_FileError` on all failure paths so that callers can decide how to surface the error (stderr + SystemExit for text mode; JSON errors list for ``--json`` mode). This avoids swallowing the precise error code in a bare ``except SystemExit`` handler. Security -------- Workdir reads include a symlink guard (symlinks are rejected) and a path containment check (resolved path must be inside the repo root) to prevent symlink-based directory traversal attacks where a tracked file is actually a symlink pointing outside the repository. """ if file_path not in manifest: raise _FileError( f"file not tracked in snapshot: {file_path}", code="FILE_NOT_TRACKED", hint="run `muse code add .` to stage all files, then `muse commit`", ) if source_is_workdir: disk = root / file_path # Symlink guard: refuse to follow symlinks, which could point outside # the repository and expose arbitrary files on the host filesystem. if disk.is_symlink(): raise _FileError( f"refusing to read symlink: {file_path}", code="SYMLINK_REJECTED", hint="dereference the symlink and commit the real file instead", ) # Path containment: resolved path must stay inside the repo root. try: disk.resolve().relative_to(root.resolve()) except ValueError: raise _FileError( f"path escapes repository root: {file_path}", code="PATH_TRAVERSAL", hint="file paths must be relative to the repository root", ) try: return disk.read_bytes() except OSError: pass # deleted in working tree — fall through to object store raw = read_object(root, manifest[file_path]) if raw is None: raise _FileError( f"blob not found in object store: {manifest[file_path][:12]}", code="BLOB_NOT_FOUND", hint="the object store may be corrupted; try `muse gc` to diagnose", ) return raw def _resolve_symbol( tree: SymbolTree, symbol_ref: str, file_path: str, ) -> tuple[SymbolRecord | None, str]: """Resolve *symbol_ref* against *tree*. Returns ``(record, "")`` on success or ``(None, error_message)`` on failure so callers can decide how to surface the error (stderr vs JSON errors list). Resolution order ---------------- 1. Exact ``qualified_name`` match (e.g. ``Invoice.compute_total``). 2. Bare ``name`` match when unambiguous (e.g. ``compute_total``). 3. Failure — returns ``None`` with a descriptive error message that lists available symbols (capped at :data:`_MAX_AVAIL_SHOWN`). """ # Exact qualified_name match first. match: SymbolRecord | None = next( (rec for rec in tree.values() if rec["qualified_name"] == symbol_ref), None, ) if match is not None: return match, "" # Fall back to bare name (unambiguous only). candidates = [rec for rec in tree.values() if rec["name"] == symbol_ref] if len(candidates) == 1: return candidates[0], "" if len(candidates) > 1: opts = ", ".join(rec["qualified_name"] for rec in candidates) return None, ( f"❌ Ambiguous symbol '{sanitize_display(symbol_ref)}' in " f"{sanitize_display(file_path)}. Qualify it:\n {opts}" ) # Not found — show a capped sample of available symbols. available = sorted( rec["qualified_name"] for rec in tree.values() if rec["kind"] != "import" ) return None, ( f"❌ Symbol '{sanitize_display(symbol_ref)}' not found in " f"{sanitize_display(file_path)}.\n" f" Available ({len(available)} total): " + ", ".join(available) ) def _extract_source(raw: bytes, lineno: int, end_lineno: int, context: int = 0) -> str: """Slice the source lines for a symbol, with optional surrounding context.""" text = raw.decode("utf-8", errors="replace") lines = text.splitlines() start = max(0, lineno - 1 - context) end = min(len(lines), end_lineno + context) return "\n".join(lines[start:end]) def _format_line_numbers(source: str, start_lineno: int, context: int = 0) -> str: """Prefix each line with its 1-based line number.""" first = max(1, start_lineno - context) lines = source.splitlines() width = len(str(first + len(lines) - 1)) return "\n".join(f"{first + i:{width}d} {line}" for i, line in enumerate(lines)) # ── CLI registration ────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the cat subcommand on *subparsers*.""" parser = subparsers.add_parser( "cat", help="Print the source code of one or more symbols.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "addresses", nargs="*", metavar="address", help=( "One or more symbol addresses: 'file.py::ClassName.method'. " "When --all is given, treat each argument as a file path instead. " "May be omitted when --file is provided." ), ) parser.add_argument( "--at", default=None, metavar="REF", help=( "Commit ref (SHA prefix, branch, tag, HEAD~N) to read from. " "Defaults to the working tree (disk content, uncommitted edits visible)." ), ) parser.add_argument( "--all", "-a", action="store_true", dest="all_symbols", help="Print every symbol in each file. Arguments are treated as file paths.", ) parser.add_argument( "--kind", "-k", default=None, metavar="KIND", dest="kind_filter", help="With --all: restrict to symbols of this kind (function, class, method, …).", ) parser.add_argument( "--line-numbers", "-n", action="store_true", dest="line_numbers", help="Prefix each output line with its 1-based line number.", ) parser.add_argument( "--context", "-C", default=0, type=int, metavar="N", dest="context", help="Include N lines of context before and after each symbol (default: 0).", ) parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.add_argument( "--file", default=None, metavar="PATH", dest="file_alias", help=( "Convenience alias: treat PATH as a file and print all its symbols. " "Equivalent to: muse code cat PATH --all. " "Canonical form: muse code symbols --file PATH" ), ) parser.set_defaults(func=run) # ── Main logic ──────────────────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Print the source code of one or more symbols. Resolves each address (``file.py::Symbol``) against the working tree or a historical commit (``--at REF``), extracts the exact source lines from the content-addressed object store or disk, and prints them. Performance ----------- File bytes and parsed symbol trees are cached per ``file_path`` within a single invocation. Batch lookups of 50 symbols from the same file perform one read + one parse, not 50. Security -------- Workdir reads reject symlinks and enforce path containment within the repository root to prevent directory traversal attacks. All user-supplied strings in error output are passed through :func:`~muse.core.validation.sanitize_display`. JSON output ----------- ``--json`` (or ``--format json``) emits a single object:: { "source_ref": "working tree" | "commit on ", "results": [CatResult, ...], "errors": [CatError, ...], "elapsed_seconds": float } Exits 0 when all addresses resolved; 1 when any error is present. """ t0 = time.monotonic() addresses: list[str] = args.addresses at: str | None = args.at all_symbols: bool = args.all_symbols kind_filter: str | None = args.kind_filter # --file PATH is a convenience alias for `muse code cat PATH --all`. # Agents and users reaching for --file (by analogy with muse code symbols) # get the same result without an argparse error. if args.file_alias is not None: addresses = [args.file_alias] + addresses all_symbols = True if not addresses: msg = "no address given — usage: muse code cat FILE::Symbol (or --file FILE)" if fmt == "json": print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) line_numbers: bool = args.line_numbers context: int = clamp_int(args.context, 0, 500, 'context') fmt: str = args.fmt as_json = fmt == "json" root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) # ── Resolve the manifest and source label ───────────────────────────────── source_is_workdir = at is None manifest: Manifest source_ref: str if source_is_workdir: manifest = get_head_snapshot_manifest(root, repo_id, branch) or {} source_ref = "working tree" else: resolved = resolve_commit_ref(root, repo_id, branch, at) if resolved is None: msg = f"Ref not found: {sanitize_display(at or '')}" if as_json: print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, resolved.commit_id) or {} source_ref = f"commit {resolved.commit_id[:8]} on {branch}" # ── Per-invocation file cache: path → (raw bytes, symbol tree) ──────────── # Avoids re-reading and re-parsing the same file for each address in a # batch request (e.g. 50 addresses all referencing billing.py). _file_cache: _FileCache = {} _file_failed: set[str] = set() # paths that errored; skip on repeat def _cached_file(file_path: str) -> tuple[bytes, SymbolTree] | _FileError: """Return cached (raw, tree) for *file_path*, or a _FileError.""" if file_path in _file_cache: return _file_cache[file_path] if file_path in _file_failed: return _FileError( f"file not tracked in snapshot: {file_path}", code="FILE_NOT_TRACKED", hint="run `muse code add .` to stage all files, then `muse commit`", ) try: raw = _get_file_bytes(root, file_path, manifest, source_is_workdir) except _FileError as exc: _file_failed.add(file_path) return exc adapter = adapter_for_path(file_path) tree = adapter.parse_symbols(raw, file_path) _file_cache[file_path] = (raw, tree) return (raw, tree) # ── Dispatch: --all mode (file paths) vs address mode ──────────────────── results: list[CatResult] = [] errors: list[CatError] = [] warnings: list[dict[str, object]] = [] any_ok = False if all_symbols: for file_path in addresses: cached = _cached_file(file_path) if isinstance(cached, _FileError): errors.append({ "address": file_path, "error": str(cached), "error_code": cached.code, "hint": cached.hint, }) if not as_json: print(f"❌ {sanitize_display(str(cached))}", file=sys.stderr) continue raw, tree = cached syms = [ rec for rec in tree.values() if rec["kind"] != "import" and (kind_filter is None or rec["kind"] == kind_filter) ] syms.sort(key=lambda r: r["lineno"]) for rec in syms: lineno = rec["lineno"] end_lineno = rec["end_lineno"] src = _extract_source(raw, lineno, end_lineno, context) if line_numbers: src = _format_line_numbers(src, lineno, context) result: CatResult = { "address": f"{file_path}::{rec['qualified_name']}", "file_path": file_path, "symbol": rec["qualified_name"], "kind": rec["kind"], "lineno": lineno, "end_lineno": end_lineno, "source": src, "source_ref": source_ref, } results.append(result) any_ok = True else: for address in addresses: if "::" not in address: hint = ( f"To read a symbol: muse code cat \"{address}::SymbolName\"\n" f"To list symbols: muse code symbols --file {address}\n" f"To read the file: use the Read tool or your editor" ) errors.append({ "address": address, "error": f"{address!r} is a file path, not a symbol address — add '::SymbolName'", "error_code": "INVALID_ADDRESS", "hint": hint, }) if not as_json: print( f"❌ {sanitize_display(address)!r} is a file path, not a symbol address.\n" f" To read a symbol: muse code cat \"{address}::SymbolName\"\n" f" To list symbols: muse code symbols --file {address}\n" f" To read the file: use the Read tool or your editor", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) continue file_path, _, symbol_ref = address.partition("::") cached = _cached_file(file_path) if isinstance(cached, _FileError): errors.append({ "address": address, "error": str(cached), "error_code": cached.code, "hint": cached.hint, }) if not as_json: print(f"❌ {sanitize_display(str(cached))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) continue raw, tree = cached if not tree: errors.append({ "address": address, "error": f"no symbols found in {file_path}", "error_code": "SYMBOL_PARSE_EMPTY", "hint": ( "symbol cache miss — run `muse code add .` to rebuild the index, " "or check that the file contains parseable Python/JS/TS" ), }) if not as_json: print( f"❌ no symbols found in {sanitize_display(file_path)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) continue # Filter out import pseudo-symbols before matching. code_tree: SymbolTree = { addr: rec for addr, rec in tree.items() if rec["kind"] != "import" } found, err_msg = _resolve_symbol(code_tree, symbol_ref, file_path) if found is None: # Global fallback: search all other tracked files for the symbol. # Covers the common agent mistake of specifying the wrong file — # e.g. `file.py::symbol` when the symbol is actually in `other.py`. fb_matches: list[tuple[str, SymbolRecord, bytes]] = [] for tracked_path in manifest: if tracked_path == file_path: continue fb_cached = _cached_file(tracked_path) if isinstance(fb_cached, _FileError): continue fb_raw, fb_tree = fb_cached fb_code_tree: SymbolTree = { a: r for a, r in fb_tree.items() if r["kind"] != "import" } fb_found, _ = _resolve_symbol(fb_code_tree, symbol_ref, tracked_path) if fb_found is not None: fb_matches.append((tracked_path, fb_found, fb_raw)) if len(fb_matches) == 1: # Unambiguous — cat from the actual file, note the redirect. actual_path, actual_rec, actual_raw = fb_matches[0] fb_lineno = actual_rec["lineno"] fb_end_lineno = actual_rec["end_lineno"] fb_src = _extract_source(actual_raw, fb_lineno, fb_end_lineno, context) if line_numbers: fb_src = _format_line_numbers(fb_src, fb_lineno, context) result = { "address": f"{actual_path}::{actual_rec['qualified_name']}", "file_path": actual_path, "symbol": actual_rec["qualified_name"], "kind": actual_rec["kind"], "lineno": fb_lineno, "end_lineno": fb_end_lineno, "source": fb_src, "source_ref": source_ref, } results.append(result) any_ok = True if not as_json: note = ( f"# note: '{sanitize_display(symbol_ref)}' not in " f"{sanitize_display(file_path)} — found in " f"{sanitize_display(actual_path)}" ) print(note) continue if len(fb_matches) > 1: locs_inline = ", ".join( f"{p}::{r['qualified_name']}" for p, r, _ in fb_matches ) locs_lines = "\n ".join( f"{p}::{r['qualified_name']}" for p, r, _ in fb_matches ) if not as_json: print( f"⚠️ '{sanitize_display(symbol_ref)}' not in " f"{sanitize_display(file_path)} — found in multiple files:\n" f" {locs_lines}\n" f" Specify the file explicitly.", file=sys.stderr, ) else: warnings.append({ "address": address, "warning": f"symbol not found in {file_path}; ambiguous across: {locs_inline}", "warning_code": "SYMBOL_AMBIGUOUS", "candidates": [ f"{p}::{r['qualified_name']}" for p, r, _ in fb_matches ], "hint": "specify the file explicitly", }) continue # Truly not found anywhere. errors.append({ "address": address, "error": f"symbol not found: {symbol_ref}", "error_code": "SYMBOL_NOT_FOUND", "hint": ( f"run `muse code symbols --file {file_path}` to list available symbols, " f"or `muse code grep \"{symbol_ref}\"` to search the full snapshot" ), }) if not as_json: print(err_msg, file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) continue lineno = found["lineno"] end_lineno = found["end_lineno"] src = _extract_source(raw, lineno, end_lineno, context) if line_numbers: src = _format_line_numbers(src, lineno, context) result = { "address": address, "file_path": file_path, "symbol": found["qualified_name"], "kind": found["kind"], "lineno": lineno, "end_lineno": end_lineno, "source": src, "source_ref": source_ref, } results.append(result) any_ok = True # ── Output ──────────────────────────────────────────────────────────────── elapsed = round(time.monotonic() - t0, 4) if as_json: out: dict[str, object] = { "source_ref": source_ref, "results": results, "errors": errors, "elapsed_seconds": elapsed, } if warnings: out["warnings"] = warnings print(json.dumps(out, indent=2)) raise SystemExit(0 if not errors else ExitCode.USER_ERROR) for result in results: header = ( f"# {result['file_path']}::{result['symbol']}" f" [{result['kind']}]" f" L{result['lineno']}–{result['end_lineno']}" f" ({result['source_ref']})" ) print(header) print(result["source"]) if len(results) > 1: print() # blank separator between multiple symbols if errors and not as_json: raise SystemExit(ExitCode.USER_ERROR)