"""muse code deps — import graph and call-graph analysis. Answers questions that Git cannot answer structurally: **File mode** (``muse code deps src/billing.py``): What does this file import, and what files in the repo import it? **Symbol mode** (``muse code deps "src/billing.py::compute_invoice_total"``): What does this function call? (Python only; uses stdlib ``ast``.) With ``--reverse``: what symbols in the repo call this function? With ``--depth N``: walk the call chain transitively N hops deep. With ``--transitive``: unlimited-depth BFS through the full call graph. These relationships are *structural impossibilities* in Git: Git stores files as blobs of text with no concept of imports or call-sites. Muse reads the typed symbol graph and the AST to answer these questions in milliseconds. Usage:: muse code deps src/billing.py # what does billing.py import? muse code deps src/billing.py --reverse # who imports billing.py? muse code deps src/billing.py --reverse --filter tests/ # only importers in tests/ muse code deps "src/billing.py::compute_total" # direct callees (Python) muse code deps "src/billing.py::compute_total" --reverse # direct callers muse code deps "src/billing.py::compute_total" --reverse --depth 3 # 3-hop callers muse code deps "src/billing.py::compute_total" --transitive # full blast radius muse code deps "src/billing.py::compute_total" --count # caller count only Flags: ``--commit, -c REF`` Inspect a historical snapshot instead of HEAD. ``--reverse`` Invert the query: show callers (symbol mode) or importers (file mode). ``--depth N`` With ``--reverse``: walk transitive callers up to N hops. Without ``--reverse``: walk transitive callees up to N hops. ``0`` means unlimited (same as ``--transitive``). ``--transitive`` Walk the full call graph without a depth cap (equivalent to ``--depth 0``). ``--filter PATTERN`` Restrict results to addresses or file paths containing PATTERN. ``--count`` Emit only the total count of results. ``--json`` Emit results as JSON. """ from __future__ import annotations import argparse import fnmatch 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 get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref from muse.core.symbol_cache import load_symbol_cache from muse.plugins.registry import read_domain from muse.plugins.code._callgraph import ( build_forward_graph, build_reverse_graph, callees_for_symbol, transitive_callees, transitive_callers, ) from muse.plugins.code._query import language_of, symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolTree from muse.core.validation import clamp_int, sanitize_display from muse.core._types import Manifest logger = logging.getLogger(__name__) def _validate_file_rel(file_rel: str) -> None: """Exit with USER_ERROR if *file_rel* looks like a path traversal attempt.""" if not file_rel: print("❌ Target file path cannot be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) p = pathlib.PurePosixPath(file_rel) if p.is_absolute() or ".." in p.parts: print( f"❌ Target path '{file_rel}' must be a relative path" " with no '..' components.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # --------------------------------------------------------------------------- # Import graph helpers # --------------------------------------------------------------------------- def _imports_in_tree(tree: SymbolTree) -> list[str]: """Return the list of module/symbol names imported by symbols in *tree*.""" return sorted( rec["qualified_name"] for rec in tree.values() if rec["kind"] == "import" ) def _file_imports( root: pathlib.Path, manifest: Manifest, target_file: str, *, workdir: pathlib.Path | None = None, ) -> list[str]: """Return import names declared in *target_file* within *manifest*.""" cache = load_symbol_cache(root) sym_map = symbols_for_snapshot( root, manifest, file_filter=target_file, workdir=workdir, cache=cache ) cache.save() tree = sym_map.get(target_file, {}) return _imports_in_tree(tree) def _reverse_imports( root: pathlib.Path, manifest: Manifest, target_file: str, file_filter: str | None = None, ) -> list[str]: """Return files in *manifest* that import a name matching *target_file*. The heuristic: the target file's stem (e.g. ``billing`` for ``src/billing.py``) is matched against each other file's import names. This catches ``import billing``, ``from billing import X``, and fully- qualified paths like ``src.billing``. Args: root: Repository root. manifest: Snapshot manifest. target_file: File path to find importers of. file_filter: Optional glob pattern — only files matching this pattern are included in the importer list. """ target_stem = pathlib.PurePosixPath(target_file).stem target_module = ( pathlib.PurePosixPath(target_file) .with_suffix("") .as_posix() .replace("/", ".") ) # Use the shared cache — one load for the entire manifest scan. cache = load_symbol_cache(root) sym_map = symbols_for_snapshot(root, manifest, cache=cache) cache.save() importers: list[str] = [] for file_path, tree in sym_map.items(): if file_path == target_file: continue if file_filter and not fnmatch.fnmatch(file_path, f"*{file_filter}*"): continue for imp_name in _imports_in_tree(tree): if ( imp_name == target_stem or imp_name == target_module or imp_name.endswith(f".{target_stem}") or imp_name.endswith(f".{target_module}") or target_stem in imp_name.split(".") ): importers.append(file_path) break return sorted(importers) # --------------------------------------------------------------------------- # Knowtation domain path # --------------------------------------------------------------------------- def _run_knowtation_deps( *, root: pathlib.Path, target: str, reverse: bool, as_json: bool, count_only: bool, file_filter: str | None, ) -> None: """Knowtation-domain path for ``muse code deps``. Treats *target* as a vault-relative note path (``.md`` / ``.markdown``). Builds a fresh :class:`~muse.plugins.knowtation.link_index.LinkIndex` for the working tree and returns outgoing / incoming note edges. Args: root: Repository root. target: Vault-relative path to the note. reverse: When ``True``, return notes that link **to** *target*. as_json: Emit JSON instead of text. count_only: Emit only the integer count of edges. file_filter: Restrict results to paths containing this substring. """ from muse.plugins.knowtation.code_intel import note_deps from muse.plugins.knowtation.link_index import build_link_index from muse.plugins.knowtation._query import is_note _validate_file_rel(target) if not is_note(target): print( f"❌ '{target}' is not a Markdown note. " "Knowtation deps requires a .md / .markdown / .mdx path.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) candidate = root / target if not candidate.is_file(): print(f"❌ Note '{target}' not found in vault.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) try: index = build_link_index(root) except ValueError as exc: print(f"❌ Could not index vault: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) result = note_deps(index, target, reverse=reverse) links = result["links"] if file_filter: links = [ link for link in links if (link["target"] and file_filter in link["target"]) or (file_filter in link["source"]) ] result["links"] = links result["count"] = len(links) if count_only: print(result["count"]) return if as_json: result["domain"] = "knowtation" print(json.dumps(result, indent=2)) return direction_label = "Notes that link to" if reverse else "Notes linked from" print(f"\n{direction_label} {sanitize_display(target)}") print("─" * 62) if not links: msg = "no incoming links" if reverse else "no outgoing links" print(f" ({msg})") else: for link in links: other = link["source"] if reverse else (link["target"] or "(broken)") kind = link["kind"] fragment = f"#{link['fragment']}" if link["fragment"] else "" print(f" {sanitize_display(other)}{fragment} [{kind}]") print(f"\n{result['count']} link(s)") # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the deps subcommand.""" parser = subparsers.add_parser( "deps", help="Show the import graph or call graph for a file or symbol.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "target", metavar="TARGET", help=( 'File path (e.g. "src/billing.py") for import graph, or ' 'symbol address (e.g. "src/billing.py::compute_invoice_total")' " for call graph." ), ) parser.add_argument( "--reverse", "-r", action="store_true", help="Show importers (file mode) or callers (symbol mode) instead.", ) parser.add_argument( "--depth", type=int, default=1, metavar="N", help=( "Walk transitive callers/callees up to N hops" " (symbol mode only; 0 = unlimited)." ), ) parser.add_argument( "--transitive", action="store_true", help="Walk the full call graph without a depth cap.", ) parser.add_argument( "--filter", dest="file_filter", default=None, metavar="PATTERN", help="Restrict results to addresses/paths containing PATTERN.", ) parser.add_argument( "--count", action="store_true", help="Emit only the total count of results.", ) parser.add_argument( "--commit", "-c", default=None, metavar="REF", dest="ref", help="Inspect a historical commit instead of HEAD.", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit results as JSON.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Run # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Show the import graph or call graph for a file or symbol. **File mode** — pass a file path:: muse code deps src/billing.py # what does billing.py import? muse code deps src/billing.py --reverse # what files import billing.py? **Symbol mode** — pass a symbol address (Python call-graph):: muse code deps "src/billing.py::compute_total" muse code deps "src/billing.py::compute_total" --reverse --depth 3 muse code deps "src/billing.py::compute_total" --transitive Call-graph analysis uses the committed snapshot (or ``--commit`` to pin). """ target: str = args.target reverse: bool = args.reverse ref: str | None = args.ref as_json: bool = args.as_json depth: int = clamp_int(args.depth, 1, 50, "depth") transitive: bool = args.transitive file_filter: str | None = args.file_filter count_only: bool = args.count # --transitive overrides --depth to unlimited (0 signals no BFS cap internally). effective_depth = 0 if transitive else depth root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) try: active_domain = read_domain(root) except Exception: active_domain = "code" if active_domain == "knowtation": _run_knowtation_deps( root=root, target=target, reverse=reverse, as_json=as_json, count_only=count_only, file_filter=file_filter, ) return is_symbol_mode = "::" in target # ── Symbol mode: call-graph (Python only) ───────────────────────────────── if is_symbol_mode: file_rel, sym_qualified = target.split("::", 1) _validate_file_rel(file_rel) lang = language_of(file_rel) if lang != "Python": print( f"❌ Call-graph analysis is currently Python-only." f" '{file_rel}' is {lang}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr ) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} shared_cache = load_symbol_cache(root) if not reverse: # ── Forward: what does this symbol call? ────────────────────── if effective_depth == 1: # Single hop — fast path using source bytes. obj_id = manifest.get(file_rel) if obj_id is None: print( f"❌ '{file_rel}' not found in snapshot" f" '{commit.commit_id[:8]}'.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) from muse.core.object_store import read_object # noqa: PLC0415 raw = read_object(root, obj_id) if raw is None: print( f"❌ Object for '{file_rel}' missing from store.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) callees = callees_for_symbol(raw, target) if file_filter: callees = [c for c in callees if file_filter in c] shared_cache.save() if count_only: print(len(callees)) return if as_json: print( json.dumps( {"address": target, "depth": 1, "calls": callees}, indent=2, ) ) return print(f"\nDirect callees of {sanitize_display(target)}") print("─" * 62) if not callees: print(" (no function calls detected)") else: for name in callees: print(f" {name}") print(f"\n{len(callees)} callee(s)") else: # Multi-hop: build full forward graph and BFS. forward = build_forward_graph(root, manifest, cache=shared_cache) shared_cache.save() by_depth = transitive_callees(target, forward, effective_depth) flat = [ name for d in sorted(by_depth) for name in sorted(by_depth[d]) ] if file_filter: flat = [c for c in flat if file_filter in c] total = sum(len(v) for v in by_depth.values()) if count_only: print(total) return if as_json: print( json.dumps( { "address": target, "depth": effective_depth, "transitive": True, "by_depth": { str(d): sorted(by_depth[d]) for d in sorted(by_depth) }, }, indent=2, ) ) return depth_label = ( "∞" if effective_depth == 0 else str(effective_depth) ) print( f"\nTransitive callees of {target}" f" (depth ≤ {depth_label})" ) print("─" * 62) if not by_depth: print(" (no callees found)") else: for d in sorted(by_depth): names = sorted(by_depth[d]) print(f"\n depth {d}:") for name in names: print(f" {sanitize_display(name)}") print(f"\n{total} callee(s) across {len(by_depth)} depth(s)") else: # ── Reverse: who calls this symbol? ─────────────────────────── target_name = sym_qualified.split(".")[-1] reverse_graph = build_reverse_graph(root, manifest, cache=shared_cache) shared_cache.save() if effective_depth == 1: # Direct callers only. callers = reverse_graph.get(target_name, []) if file_filter: callers = [c for c in callers if file_filter in c] if count_only: print(len(callers)) return if as_json: print( json.dumps( { "address": target, "target_name": target_name, "depth": 1, "called_by": callers, }, indent=2, ) ) return print(f"\nDirect callers of {sanitize_display(target)}") print("─" * 62) if not callers: print(" (no callers found in snapshot)") else: for addr in callers: print(f" {sanitize_display(addr)}") print(f"\n{len(callers)} caller(s)") else: # Transitive callers via BFS. by_depth = transitive_callers(target_name, reverse_graph, effective_depth) if file_filter: by_depth = { d: [c for c in callers if file_filter in c] for d, callers in by_depth.items() } by_depth = {d: v for d, v in by_depth.items() if v} total = sum(len(v) for v in by_depth.values()) if count_only: print(total) return if as_json: print( json.dumps( { "address": target, "target_name": target_name, "depth": effective_depth, "transitive": True, "by_depth": { str(d): sorted(by_depth[d]) for d in sorted(by_depth) }, }, indent=2, ) ) return depth_label = ( "∞" if effective_depth == 0 else str(effective_depth) ) print( f"\nTransitive callers of {target}" f" (depth ≤ {depth_label})" ) print(f" (matching bare name: {target_name!r})") print("─" * 62) if not by_depth: print(" (no callers found)") else: for d in sorted(by_depth): addrs = sorted(by_depth[d]) print(f"\n depth {d}:") for addr in addrs: print(f" {addr}") print(f"\n{total} caller(s) across {len(by_depth)} depth(s)") return # ── File mode: import graph ──────────────────────────────────────────────── _validate_file_rel(target) commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} # When working-tree mode and the target is not yet committed, inject a # synthetic manifest entry so symbols_for_snapshot can read it from disk. working_tree = ref is None if working_tree and target not in manifest: candidate = root / target if candidate.is_file(): manifest = dict(manifest) manifest[target] = "" if not reverse: imports = _file_imports(root, manifest, target, workdir=root if working_tree else None) if file_filter: imports = [i for i in imports if file_filter in i] if count_only: print(len(imports)) return if as_json: print( json.dumps({"file": target, "imports": imports}, indent=2) ) return print(f"\nImports declared in {sanitize_display(target)}") print("─" * 62) if not imports: print(" (no imports found)") else: for name in imports: print(f" {sanitize_display(name)}") print(f"\n{len(imports)} import(s)") else: importers = _reverse_imports(root, manifest, target, file_filter) if count_only: print(len(importers)) return if as_json: print( json.dumps( {"file": target, "imported_by": importers}, indent=2 ) ) return print(f"\nFiles that import {sanitize_display(target)}") if file_filter: print(f" (filtered to: *{file_filter}*)") print("─" * 62) if not importers: print(" (no files import this module in the committed snapshot)") else: for fp in importers: print(f" {sanitize_display(fp)}") print(f"\n{len(importers)} importer(s) found")