#!/usr/bin/env python3 """ Belief Map Search -- query tool for S-expression belief maps. Reads ``.belief_map.sexp`` and provides structured queries: module views, dependency trees, reverse dependency analysis, entity lookups, data flow tracing, and boundary discovery for minimal context loading. Output is S-expression facts (one per line, greppable). Usage:: belief_search.py analyze Full analysis in one step belief_search.py boundary File paths for change boundary belief_search.py quick Search + analyze in one step belief_search.py module Show module with entities and edges belief_search.py deps [depth] Outgoing dependency tree (default depth 2) belief_search.py rdeps [depth] Reverse dependency tree (default depth 2) belief_search.py entity Find entity: definition + all usages belief_search.py flow Trace data/call flow from entity belief_search.py search Regex search across all facts belief_search.py invariants [scope] Check naming-convention violations (default: all) belief_search.py find_function Find function/method definitions by name belief_search.py find_type Find type/class/interface/enum definitions by name belief_search.py find_callchain [depth] Trace call path between two functions belief_search.py find_callers [depth] Who calls this function (incoming calls) belief_search.py find_calls [depth] What does this function call (outgoing calls) belief_search.py grep_functions Search function bodies in source files belief_search.py diff_functions [ref] Identify changed functions in git diff (default: HEAD) belief_search.py query Evaluate a Scheme query expression belief_search.py repl Interactive Scheme REPL Scheme query examples:: belief_search.py query '(boundary "operatives.controller")' belief_search.py query '(files (intersect (deps "A" 2) (rdeps "B" 1)))' belief_search.py query '(filter (boundary "operatives.service") "dto")' belief_search.py query '(count (deps "operatives.service" 3))' CLI examples:: belief_search.py analyze drive/tools/drive_vsearch belief_search.py boundary operatives.service belief_search.py entity DriveDbCommitResponse belief_search.py deps drive/modules/drive_db/services/database_service 3 belief_search.py find_function createOrder belief_search.py find_type DriveDbCommitResponse belief_search.py find_callchain createOrder processPayment belief_search.py find_callers handleRequest 3 belief_search.py find_calls createOrder 2 belief_search.py grep_functions "validate.*input" belief_search.py diff_functions HEAD~3 """ from __future__ import annotations import os import re import sys from collections.abc import Callable from dataclasses import dataclass, field # --------------------------------------------------------------------------- # Data types # --------------------------------------------------------------------------- @dataclass class Node: nid: str lang: str purpose: str naming: str = "" # snake_case, camelCase, PascalCase, kebab-case, mixed pkg: str = "" # repository/package name @dataclass class EntityFact: module: str name: str kind: str # fn, cls, ifc, typ, enm line: int bases: list[str] = field(default_factory=list) deco: list[str] = field(default_factory=list) methods: list[str] = field(default_factory=list) @dataclass class Edge: etype: str # imports, calls-api, data-flow, refs, calls source: str target: str flags: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # S-expression line parser # --------------------------------------------------------------------------- def tokenize(line: str) -> list[str]: """Tokenize a single-line S-expression into a flat list of tokens.""" line = line.strip() if not line or line.startswith(";"): return [] if not line.startswith("(") or not line.endswith(")"): return [] inner = line[1:-1] tokens: list[str] = [] i = 0 while i < len(inner): ch = inner[i] if ch == " ": i += 1 elif ch == '"': j = i + 1 while j < len(inner): if inner[j] == "\\" and j + 1 < len(inner): j += 2 elif inner[j] == '"': break else: j += 1 tokens.append(inner[i + 1 : j].replace('\\"', '"').replace("\\\\", "\\")) i = j + 1 elif ch == "(": depth = 1 j = i + 1 while j < len(inner) and depth > 0: if inner[j] == "(": depth += 1 elif inner[j] == ")": depth -= 1 j += 1 tokens.append(inner[i:j]) i = j else: j = i while j < len(inner) and inner[j] not in (" ", ")"): j += 1 tokens.append(inner[i:j]) i = j return tokens def parse_sub_list(token: str) -> list[str]: """Parse a (:key a b c) token into ['a', 'b', 'c'].""" if token.startswith("(") and token.endswith(")"): parts = token[1:-1].split() return [p for p in parts if not p.startswith(":")] return [] # --------------------------------------------------------------------------- # S-expression output helpers # --------------------------------------------------------------------------- def _q(s: str) -> str: """Quote a string for S-expression output if it contains spaces.""" if " " in s or '"' in s: return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"' return s def _sexp_list(tag: str, items: list[str]) -> str: """Format a tagged list: (:tag a b c).""" if not items: return "" return f"(:{tag} {' '.join(items)})" # --------------------------------------------------------------------------- # No-match error helper # --------------------------------------------------------------------------- def _print_no_match(g: BeliefGraph, pattern: str) -> None: """Print a structured no-match error with fuzzy suggestions.""" suggestions = BeliefGraph._suggest_similar(pattern, g.nodes) if suggestions: sug_str = " ".join(suggestions) print(f"(error no-match {_q(pattern)} :suggestions {sug_str})") else: print(f"(error no-match {_q(pattern)})") # --------------------------------------------------------------------------- # File-path resolution helpers # --------------------------------------------------------------------------- _TS_EXTENSIONS = (".ts", ".tsx", ".js", ".jsx") _PY_EXTENSIONS = (".py",) def _module_id_to_file(module_id: str, root: str) -> str: """Convert a module ID back to a file path relative to root.""" base = os.path.join(root, module_id) for ext in _TS_EXTENSIONS + _PY_EXTENSIONS: candidate = base + ext if os.path.isfile(candidate): return candidate for idx in ("index.ts", "index.tsx", "index.js", "__init__.py"): candidate = os.path.join(base, idx) if os.path.isfile(candidate): return candidate return module_id # --------------------------------------------------------------------------- # Graph loader # --------------------------------------------------------------------------- class BeliefGraph: """In-memory belief map graph built from S-expression facts.""" def __init__(self) -> None: self.nodes: dict[str, Node] = {} self.entities: dict[str, list[EntityFact]] = {} # module -> entities self.entity_by_name: dict[str, list[EntityFact]] = {} # name -> entities self.edges_from: dict[str, list[Edge]] = {} # source module -> edges self.edges_to: dict[str, list[Edge]] = {} # target module -> edges self.edges_from_entity: dict[str, list[Edge]] = {} # entity-qualified source -> edges self.edges_by_entity_name: dict[str, list[Edge]] = {} # entity name -> edges self.all_edges: list[Edge] = [] self.pmap: dict[str, str] = {} self._root: str = "" def load(self, path: str) -> None: self._root = os.path.dirname(os.path.dirname(os.path.abspath(path))) pmap: dict[str, str] = {} in_paths = False path_stack: list[str] = [] with open(path, "r") as f: for line in f: stripped = line.strip() if not stripped or stripped.startswith(";"): continue if stripped == "(paths": in_paths = True path_stack = [] continue if in_paths: if stripped == ")": if path_stack: path_stack.pop() else: in_paths = False continue tokens = tokenize(stripped) if not tokens: inner = stripped.lstrip("(").rstrip() if inner: path_stack.append(inner) continue if len(tokens) == 2 and not tokens[0].startswith(":"): sid, leaf = tokens[0], tokens[1] if leaf == ":self": # Directory-as-module (index.ts stripped to dir name) full = "/".join(path_stack) if path_stack else sid pmap[sid] = full else: full = "/".join(path_stack + [leaf]) if path_stack else leaf pmap[sid] = full elif len(tokens) == 1: path_stack.append(tokens[0]) continue tokens = tokenize(stripped) if tokens and tokens[0] == "def" and len(tokens) >= 3: pmap[tokens[1]] = tokens[2] self.pmap = pmap def _resolve(sid: str) -> str: if "::" in sid: mod, ent = sid.split("::", 1) return f"{pmap.get(mod, mod)}::{ent}" return pmap.get(sid, sid) with open(path, "r") as f: for line in f: tokens = tokenize(line) if not tokens: continue kind = tokens[0] if kind in ("paths", "def", ")"): continue if kind == "node" and len(tokens) >= 4: nid = _resolve(tokens[1]) lang, purpose = tokens[2], tokens[3] naming, pkg = "", "" rest = tokens[4:] for i, t in enumerate(rest): if t == ":naming" and i + 1 < len(rest): naming = rest[i + 1] elif t == ":pkg" and i + 1 < len(rest): pkg = rest[i + 1] self.nodes[nid] = Node(nid, lang, purpose, naming, pkg) elif kind in ("fn", "cls", "ifc", "typ", "enm") and len(tokens) >= 4: module = _resolve(tokens[1]) name = tokens[2] line_num = int(tokens[3]) if tokens[3].isdigit() else 0 bases, deco, methods = [], [], [] for t in tokens[4:]: if t.startswith("(:bases"): bases = parse_sub_list(t) elif t.startswith("(:deco"): deco = parse_sub_list(t) elif t.startswith("(:methods"): methods = parse_sub_list(t) ent = EntityFact(module, name, kind, line_num, bases, deco, methods) self.entities.setdefault(module, []).append(ent) self.entity_by_name.setdefault(name, []).append(ent) elif kind in ("imports", "calls-api", "data-flow", "refs", "calls", "http-calls") and len(tokens) >= 3: source = _resolve(tokens[1]) target = _resolve(tokens[2]) flags = [t for t in tokens[3:] if t.startswith(":")] edge = Edge(kind, source, target, flags) self.all_edges.append(edge) src_mod = source.split("::")[0] tgt_mod = target.split("::")[0] self.edges_from.setdefault(src_mod, []).append(edge) self.edges_to.setdefault(tgt_mod, []).append(edge) self.edges_from_entity.setdefault(source, []).append(edge) for part in (source, target): if "::" in part: ent_name = part.split("::", 1)[1] self.edges_by_entity_name.setdefault(ent_name, []).append(edge) elif kind == "violation" and len(tokens) >= 5: source = _resolve(tokens[2]) target = _resolve(tokens[3]) edge = Edge("violation", source, target, [f":rule={tokens[1]}"]) self.all_edges.append(edge) def resolve_alias(self, module_id: str) -> str: """Resolve an internal path-map alias to a module ID when possible.""" if "::" in module_id: mod, ent = module_id.split("::", 1) return f"{self.pmap.get(mod, mod)}::{ent}" return self.pmap.get(module_id, module_id) def resolve_module(self, pattern: str) -> list[str]: """Find module IDs: exact > suffix > substring > case-insensitive substring. Returns an empty list when nothing matches (caller should use _print_no_match to report the failure with suggestions). """ resolved_pattern = self.resolve_alias(pattern) if resolved_pattern in self.nodes: return [resolved_pattern] if pattern in self.nodes: return [pattern] suffix_matches = [nid for nid in self.nodes if nid.endswith("/" + pattern) or nid.endswith(pattern)] if suffix_matches: return sorted(suffix_matches) sub_matches = [nid for nid in self.nodes if pattern in nid] if sub_matches: if len(sub_matches) > 10: print(f"; warning: '{pattern}' matches {len(sub_matches)} modules, showing first 10", file=sys.stderr) return sorted(sub_matches)[:10] return sorted(sub_matches) # Case-insensitive substring fallback lp = pattern.lower() ci_matches = [nid for nid in self.nodes if lp in nid.lower()] if ci_matches: if len(ci_matches) > 10: print(f"; warning: '{pattern}' (case-insensitive) matches {len(ci_matches)} modules, showing first 10", file=sys.stderr) return sorted(ci_matches)[:10] return sorted(ci_matches) return [] def search_modules(self, pattern: str) -> list[str]: """Find module IDs whose path or basename matches the regex pattern.""" regex = re.compile(pattern, re.IGNORECASE) matches = [ nid for nid in self.nodes if regex.search(nid) or regex.search(nid.rsplit("/", 1)[-1]) ] return sorted(matches) @staticmethod def _suggest_similar(pattern: str, nodes: "dict[str, Node]", max_results: int = 5) -> list[str]: """Return up to max_results node IDs whose path segments overlap with pattern. Splits pattern by '/', '.', '_', '-' and counts how many parts appear (case-insensitively) in each candidate node ID. Returns the top N nodes with at least one matching part, sorted by match count descending. """ import re as _re parts = [p for p in _re.split(r"[/._-]", pattern.lower()) if p] if not parts: return [] scored: list[tuple[int, str]] = [] for nid in nodes: nid_lower = nid.lower() count = sum(1 for p in parts if p in nid_lower) if count > 0: scored.append((count, nid)) scored.sort(key=lambda x: (-x[0], x[1])) return [nid for _, nid in scored[:max_results]] def module_to_file(self, module_id: str) -> str: """Resolve a module ID to a file path on disk.""" return _module_id_to_file(module_id, self._root) def module_purpose(self, module_id: str) -> str: """Return the purpose string of a module. Empty if unknown.""" node = self.nodes.get(module_id) return node.purpose if node else "" def module_lang(self, module_id: str) -> str: """Return the language of a module ('py', 'ts', 'tsx'). Empty if unknown.""" node = self.nodes.get(module_id) if not node: return "" # Normalize legacy 'ty' -> 'ts', 'tx' -> 'tsx' return {"ty": "ts", "tx": "tsx"}.get(node.lang, node.lang) # --------------------------------------------------------------------------- # Shared tree collectors # --------------------------------------------------------------------------- def _collect_dep_mods(g: BeliefGraph, nid: str) -> dict[str, str]: """Outgoing dependency modules (deduped): module -> edge_type.""" seen: dict[str, str] = {} for e in g.edges_from.get(nid, []): tgt_mod = e.target.split("::")[0] if tgt_mod != nid and tgt_mod not in seen: seen[tgt_mod] = e.etype return seen def _collect_rdep_mods(g: BeliefGraph, nid: str) -> dict[str, str]: """Incoming dependency modules (deduped): module -> edge_type.""" seen: dict[str, str] = {} for e in g.edges_to.get(nid, []): src_mod = e.source.split("::")[0] if src_mod != nid and src_mod not in seen: seen[src_mod] = e.etype return seen def _walk_deps( g: BeliefGraph, nid: str, visited: set[str], max_depth: int, depth: int, emit: Callable[[int, str, str, bool], None], ) -> None: if depth > max_depth: return for tgt, etype in sorted(_collect_dep_mods(g, nid).items()): cycle = tgt in visited emit(depth, etype, tgt, cycle) if not cycle: visited.add(tgt) _walk_deps(g, tgt, visited, max_depth, depth + 1, emit) def _walk_rdeps( g: BeliefGraph, nid: str, visited: set[str], max_depth: int, depth: int, emit: Callable[[int, str, str, bool], None], ) -> None: if depth > max_depth: return for src, etype in sorted(_collect_rdep_mods(g, nid).items()): cycle = src in visited emit(depth, etype, src, cycle) if not cycle: visited.add(src) _walk_rdeps(g, src, visited, max_depth, depth + 1, emit) # --------------------------------------------------------------------------- # Layer classifier + boundary rules # --------------------------------------------------------------------------- def _classify_layer_fallback(p: str) -> str: """Classify a module into an architecture layer from its path. Order matters: test/config checked first (override other patterns). NestJS conventions: guards/controllers are api, .module files are config, /interface/ folders are shared contracts (not api). """ # High-priority overrides (tests, config) if "/tests/" in p or "/test_" in p or ".test." in p or ".spec." in p or "/e2e/" in p: return "test" if "/config/" in p or "/env/" in p: return "config" # NestJS .module files are wiring/config, not service if ".module" in p and "/modules/" in p: return "config" # Interface folders are shared contracts, not api if "/interface/" in p or "/interfaces/" in p: return "shared" # Architecture layers if "/api/" in p or "/routes/" in p or "/controllers/" in p: return "api" if "/services/" in p or "/application/" in p: return "service" if "/domain/" in p or "/entities/" in p or "/schema" in p or "_schema" in p: return "domain" if "/repositories/" in p or "/db/" in p or "/infra/" in p or "/infrastructure/" in p or "rabbitmq" in p or "s3" in p: return "infra" if "/commons/" in p or "/common/" in p or "/utils/" in p or "/result" in p or "/types/" in p: return "shared" if "/components/" in p or "/pages/" in p or "/ui/" in p or "/hooks/" in p or "/stores/" in p: return "ui" # NestJS / backend patterns if "/dto/" in p or "/dtos/" in p: return "domain" if "/guards/" in p or "/guard." in p: return "api" if "/middleware/" in p or "/middlewares/" in p or "/interceptors/" in p or "/pipes/" in p: return "api" if "/adapters/" in p or "/adapter." in p: return "infra" if "/gateways/" in p or "/gateway." in p: return "infra" if "/migrations/" in p or "/seeds/" in p or "/seeders/" in p: return "infra" # Domain patterns if "/models/" in p or "/schemas/" in p or "/events/" in p or "/commands/" in p or "/queries/" in p: return "domain" if "/handlers/" in p: return "service" # Shared patterns if "/lib/" in p or "/helpers/" in p or "/constants/" in p or "/decorators/" in p or "/mappers/" in p: return "shared" # UI patterns (state management tied to UI) if ("/stores/" in p or "/store/" in p or "/contexts/" in p or "/providers/" in p) and "/ui/" in p: return "ui" # Test patterns if "/fixtures/" in p or "/mocks/" in p or "/__mocks__/" in p or "/factories/" in p: return "test" # Config / tooling patterns if "/scripts/" in p or "/tools/" in p or "/cli/" in p: return "config" return "other" _PURPOSE_TO_LAYER: dict[str, str] = { # Domain layer "domain logic": "domain", "domain entity": "domain", "data model": "domain", "data transfer object": "domain", "schema": "domain", "enum definitions": "domain", "policy": "domain", "value object": "domain", "aggregate root": "domain", "domain event": "domain", "Specification pattern": "domain", # Infrastructure layer "infrastructure adapter": "infra", "database migration": "infra", "database seed": "infra", "repository": "infra", "adapter": "infra", "gateway": "infra", "client": "infra", "connector": "infra", "scheduler": "infra", "worker": "infra", "Unit of Work": "infra", "Identity Map": "infra", "Proxy pattern": "infra", "SDK": "infra", "API client": "infra", # API layer "API layer": "api", "API controller": "api", "router": "api", "access guard": "api", "middleware": "api", "interceptor": "api", "validation pipe": "api", "resolver": "api", # Service layer "service": "service", "application service": "service", "event handler": "service", "handler": "service", "event listener": "service", "processor": "service", "provider": "service", "registry": "service", "agent": "service", "manager": "service", "subscriber": "service", "dispatcher": "service", "emitter": "service", "Facade pattern": "service", "Mediator pattern": "service", "Observer pattern": "service", "Command pattern": "service", "Chain of Responsibility": "service", "Memento pattern": "service", # Config layer "NestJS wiring module": "config", "NestJS injectable": "config", "configuration": "config", "application bootstrap": "config", "module entry point": "config", # UI layer "UI component": "ui", "page/route": "ui", "React hook": "ui", "state store": "ui", # Test layer "test": "test", "test fixture": "test", "test mock": "test", # Shared layer "shared utilities": "shared", "utility helpers": "shared", "constants": "shared", "type definitions": "shared", "interface definitions": "shared", "decorator": "shared", "data mapper": "shared", "barrel re-exports": "shared", "plugin": "shared", "Factory pattern": "shared", "Builder pattern": "shared", "Strategy pattern": "shared", "Singleton pattern": "shared", "Composite pattern": "shared", "Visitor pattern": "shared", "Iterator pattern": "shared", "Decorator pattern": "shared", } def _levenshtein(a: str, b: str) -> int: """Compute Levenshtein edit distance between two strings.""" if len(a) < len(b): return _levenshtein(b, a) if not b: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a): curr = [i + 1] for j, cb in enumerate(b): curr.append(min(prev[j + 1] + 1, curr[j] + 1, prev[j] + (ca != cb))) prev = curr return prev[-1] def _classify_layer_by_purpose(purpose: str) -> str: """Classify layer from the module's purpose string using fuzzy matching. Tries exact match, then substring, then Levenshtein distance on individual words from the purpose against the keyword map. """ if not purpose or purpose == "general module": return "other" pl = purpose.lower() # 1. Exact match for key, layer in _PURPOSE_TO_LAYER.items(): if key.lower() == pl: return layer # 2. Substring match for key, layer in _PURPOSE_TO_LAYER.items(): if key.lower() in pl: return layer # 3. Fuzzy word match: split purpose into words, compare each word # against keywords from the map using Levenshtein distance. # Threshold: distance <= 2 for words >= 5 chars, <= 1 for shorter. purpose_words = re.split(r"[\s;,/]+", pl) best_layer = "other" best_dist = 999 for key, layer in _PURPOSE_TO_LAYER.items(): key_words = key.lower().split() for kw in key_words: if len(kw) < 4: continue # skip short keywords (api, ui, dto) -- too ambiguous for fuzzy threshold = 2 if len(kw) >= 5 else 1 for pw in purpose_words: if len(pw) < 4: continue dist = _levenshtein(pw, kw) if dist <= threshold and dist < best_dist: best_dist = dist best_layer = layer return best_layer def _classify_layer(module_path: str, purpose: str = "") -> str: """Classify using path patterns first, fall back to purpose string.""" result = _classify_layer_fallback(module_path.lower()) if result == "other" and purpose: result = _classify_layer_by_purpose(purpose) return result # --------------------------------------------------------------------------- # Commands -- S-expression output (default) # --------------------------------------------------------------------------- def _entity_sexp(e: EntityFact, lang: str = "") -> str: """Format an entity as an S-expression fact with :lang.""" parts = [e.kind, e.module, e.name, str(e.line)] if lang: parts.append(f":lang {lang}") if e.bases: parts.append(_sexp_list("bases", e.bases)) if e.deco: parts.append(_sexp_list("deco", e.deco)) if e.methods: parts.append(_sexp_list("methods", e.methods)) return f"({' '.join(parts)})" def _edge_sexp(e: Edge) -> str: """Format an edge as an S-expression fact.""" flags = " " + " ".join(e.flags) if e.flags else "" return f"({e.etype} {e.source} {e.target}{flags})" def cmd_module(g: BeliefGraph, module_id: str) -> None: matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return for nid in matches: node = g.nodes[nid] fpath = g.module_to_file(nid) mod_lang = g.module_lang(nid) print(f"(node {nid} :lang {mod_lang} {_q(node.purpose)} :file {_q(fpath)})") for e in sorted(g.entities.get(nid, []), key=lambda x: x.line): print(_entity_sexp(e, mod_lang)) for e in sorted(g.edges_from.get(nid, []), key=lambda x: (x.etype, x.target)): print(_edge_sexp(e)) for e in sorted(g.edges_to.get(nid, []), key=lambda x: (x.etype, x.source)): if e.source.split("::")[0] != nid: print(_edge_sexp(e)) def cmd_deps(g: BeliefGraph, module_id: str, max_depth: int) -> None: matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return for start in matches: visited: set[str] = {start} print(f"(deps-root {start} :lang {g.module_lang(start)})") _walk_deps(g, start, visited, max_depth, 1, lambda d, et, tgt, cyc: print(f"(dep {start} {tgt} :lang {g.module_lang(tgt)} :via {et} :depth {d}{' :cycle' if cyc else ''})")) def cmd_rdeps(g: BeliefGraph, module_id: str, max_depth: int) -> None: matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return for start in matches: visited: set[str] = {start} print(f"(rdeps-root {start} :lang {g.module_lang(start)})") _walk_rdeps(g, start, visited, max_depth, 1, lambda d, et, src, cyc: print(f"(rdep {start} {src} :lang {g.module_lang(src)} :via {et} :depth {d}{' :cycle' if cyc else ''})")) def cmd_entity(g: BeliefGraph, name: str) -> None: # Always try exact match first, then regex fallback definitions: list[EntityFact] = [] if name in g.entity_by_name: definitions = g.entity_by_name[name] else: pattern = re.compile(name, re.IGNORECASE) for ename, ents in g.entity_by_name.items(): if pattern.search(ename): definitions.extend(ents) if not definitions: _print_no_match(g, name) return # Collect refs via index matched_names: set[str] = {d.name for d in definitions} seen_edge_ids: set[int] = set() unique_refs: list[Edge] = [] for ent_name in matched_names: for e in g.edges_by_entity_name.get(ent_name, []): eid = id(e) if eid not in seen_edge_ids: seen_edge_ids.add(eid) unique_refs.append(e) for defn in definitions: print(f"(entity-def {defn.kind} {defn.module} {defn.name} {defn.line} :lang {g.module_lang(defn.module)} :file {_q(g.module_to_file(defn.module))})") for e in sorted(unique_refs, key=lambda x: (x.etype, x.source)): print(_edge_sexp(e)) def cmd_flow(g: BeliefGraph, module_id: str, entity_name: str) -> None: matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return start_mod = matches[0] full_id = f"{start_mod}::{entity_name}" visited: set[str] = set() _trace_flow(g, full_id, visited, 0) def _trace_flow( g: BeliefGraph, entity_id: str, visited: set[str], depth: int, ) -> None: if entity_id in visited or depth > 5: return visited.add(entity_id) # Entity-level edges via index for edge in g.edges_from_entity.get(entity_id, []): tgt_lang = g.module_lang(edge.target.split("::")[0]) print(f"(flow {entity_id} {edge.target} :lang {tgt_lang} :via {edge.etype} :depth {depth})") _trace_flow(g, edge.target, visited, depth + 1) # Module-level edges for cross-module flow mod = entity_id.split("::")[0] if "::" in entity_id else entity_id if "::" in entity_id: for edge in g.edges_from.get(mod, []): if "::" in edge.source: continue tgt_mod = edge.target.split("::")[0] if tgt_mod not in visited: print(f"(flow {entity_id} {edge.target} :lang {g.module_lang(tgt_mod)} :via {edge.etype} :depth {depth} :module-level)") _trace_flow(g, edge.target, visited, depth + 1) def cmd_boundary(g: BeliefGraph, module_id: str, files_only: bool = False) -> None: """Output the minimal set of files for a change boundary. When files_only is True, emit one bare file path per line instead of sexp. """ matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return nid = matches[0] node = g.nodes.get(nid) if not node: print(f"(error no-node {_q(nid)})") return target_file = g.module_to_file(nid) import_mods = _collect_dep_mods(g, nid) dependent_mods = _collect_rdep_mods(g, nid) ents = g.entities.get(nid, []) if files_only: # Compact output: one file path per line, no sexp wrapping. seen: set[str] = set() for path in [target_file] + [g.module_to_file(m) for m in sorted(import_mods)] + [g.module_to_file(m) for m in sorted(dependent_mods)]: if path not in seen: seen.add(path) print(path) return lang = g.module_lang(nid) print(f"(boundary {nid} :lang {lang} :file {_q(target_file)} :purpose {_q(node.purpose)})") for e in sorted(ents, key=lambda x: x.line): print(_entity_sexp(e, lang)) for mod, etype in sorted(import_mods.items()): print(f"(boundary-dep {nid} {mod} :lang {g.module_lang(mod)} :file {_q(g.module_to_file(mod))} :relation {etype})") for mod, etype in sorted(dependent_mods.items()): print(f"(boundary-rdep {nid} {mod} :lang {g.module_lang(mod)} :file {_q(g.module_to_file(mod))} :relation {etype})") print(f"(boundary-summary {nid} :total {1 + len(import_mods) + len(dependent_mods)} :deps {len(import_mods)} :rdeps {len(dependent_mods)})") def cmd_analyze(g: BeliefGraph, module_id: str) -> None: """Full architectural analysis of a module.""" matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return nid = matches[0] node = g.nodes.get(nid) if not node: print(f"(error no-node {_q(nid)})") return target_file = g.module_to_file(nid) ents = g.entities.get(nid, []) out_edges = g.edges_from.get(nid, []) in_edges = g.edges_to.get(nid, []) # Classify outgoing edges import_targets: list[str] = [] ref_targets: list[tuple[str, str]] = [] data_flow_targets: list[str] = [] call_targets: list[tuple[str, str]] = [] for e in out_edges: tgt_mod = e.target.split("::")[0] if e.etype == "imports" and tgt_mod not in import_targets: import_targets.append(tgt_mod) elif e.etype == "refs": ref_targets.append((e.source, e.target)) elif e.etype == "data-flow" and tgt_mod not in data_flow_targets: data_flow_targets.append(tgt_mod) elif e.etype == "calls": call_targets.append((e.source, e.target)) # Classify incoming edges importers: set[str] = set() referencers: set[str] = set() for e in in_edges: src_mod = e.source.split("::")[0] if src_mod == nid: continue if e.etype == "imports": importers.add(src_mod) else: referencers.add(src_mod) # Key entities cross-repo usage (indexed) exported_types = [e for e in ents if e.kind in ("cls", "ifc", "typ") and not e.name.startswith("_")] key_entity_usage: list[tuple[EntityFact, int, set[str]]] = [] for et in exported_types: edges_for_name = g.edges_by_entity_name.get(et.name, []) ref_mods: set[str] = set() ref_count = 0 for e in edges_for_name: src_mod = e.source.split("::")[0] if src_mod != nid: ref_count += 1 ref_mods.add(src_mod) if ref_count > 0: key_entity_usage.append((et, ref_count, ref_mods)) # Boundary files boundary_files: list[str] = [target_file] for t in sorted(import_targets): f = g.module_to_file(t) if f not in boundary_files: boundary_files.append(f) for src in sorted(importers | referencers): f = g.module_to_file(src) if f not in boundary_files: boundary_files.append(f) _analyze_sexp(g, nid, node, target_file, ents, import_targets, ref_targets, data_flow_targets, call_targets, importers, referencers, key_entity_usage, boundary_files) def _analyze_sexp( g: BeliefGraph, nid: str, node: Node, target_file: str, ents: list[EntityFact], import_targets: list[str], ref_targets: list[tuple[str, str]], data_flow_targets: list[str], call_targets: list[tuple[str, str]], importers: set[str], referencers: set[str], key_entity_usage: list[tuple[EntityFact, int, set[str]]], boundary_files: list[str], ) -> None: # Module header lang = g.module_lang(nid) print(f"(analyze {nid} :lang {lang} {_q(node.purpose)} :file {_q(target_file)})") # Entities for e in sorted(ents, key=lambda x: x.line): print(_entity_sexp(e, lang)) # Imports for t in sorted(import_targets): print(f"(analyze-import {nid} {t} :lang {g.module_lang(t)} :file {_q(g.module_to_file(t))})") # Data flows for t in sorted(data_flow_targets): print(f"(analyze-dataflow {nid} {t} :lang {g.module_lang(t)} :validated)") # Entity refs for src, tgt in sorted(ref_targets): print(f"(refs {src} {tgt})") # Call edges for src, tgt in sorted(call_targets): print(f"(calls {src} {tgt})") # Dependents all_dependents = importers | referencers for src in sorted(all_dependents): kinds = [] if src in importers: kinds.append("imports") if src in referencers: kinds.append("refs") print(f"(analyze-dependent {nid} {src} :lang {g.module_lang(src)} :via {' '.join(kinds)} :file {_q(g.module_to_file(src))})") # Dependency tree (depth 2) visited: set[str] = {nid} _walk_deps(g, nid, visited, 2, 1, lambda d, et, tgt, cyc: print(f"(dep {nid} {tgt} :lang {g.module_lang(tgt)} :via {et} :depth {d}{' :cycle' if cyc else ''})")) # Reverse deps (depth 1) visited_r: set[str] = {nid} _walk_rdeps(g, nid, visited_r, 1, 1, lambda d, et, src, cyc: print(f"(rdep {nid} {src} :lang {g.module_lang(src)} :via {et} :depth {d}{' :cycle' if cyc else ''})")) # Key entities for et, ref_count, ref_mods in key_entity_usage: print(f"(key-entity {nid} {et.name} :lang {lang} :kind {et.kind} :line {et.line} :refs {ref_count} :modules {len(ref_mods)})") for rm in sorted(ref_mods): print(f"(key-entity-ref {et.name} {rm} :lang {g.module_lang(rm)})") # Boundary files for f in boundary_files: print(f"(boundary-file {nid} {_q(f)})") # Architecture layers for t in sorted(import_targets): layer = _classify_layer(t, g.module_purpose(t)) print(f"(layer {nid} {t.split('/')[-1]} :layer {layer})") # Summary print(f"(analyze-summary {nid} :entities {len(ents)} :imports {len(import_targets)} :dataflows {len(data_flow_targets)} :refs {len(ref_targets)} :dependents {len(all_dependents)} :boundary-files {len(boundary_files)})") def cmd_quick(g: BeliefGraph, keyword: str, sexp_path: str) -> None: """Search for keyword then run analyze for the first match. Resolution order: 1. resolve_module (exact/suffix/substring/case-insensitive) 2. Regex search across the raw sexp file, extract module IDs from matches Runs cmd_analyze on the first unique module found. """ matches = g.resolve_module(keyword) if matches: cmd_analyze(g, matches[0]) return # Fallback: grep the sexp file for the keyword and extract module IDs from lines regex = re.compile(re.escape(keyword), re.IGNORECASE) seen_modules: list[str] = [] seen_set: set[str] = set() with open(sexp_path, "r") as f: for line in f: if not regex.search(line): continue tokens = tokenize(line) if not tokens: continue # Heuristically grab module-like tokens and resolve legacy path-map aliases. for tok in tokens[1:3]: resolved = g.resolve_alias(tok) if resolved in g.nodes and resolved not in seen_set: seen_set.add(resolved) seen_modules.append(resolved) break if not seen_modules: _print_no_match(g, keyword) return cmd_analyze(g, seen_modules[0]) def cmd_search(_g: BeliefGraph, pattern: str, sexp_path: str) -> None: """Search module IDs first, then fall back to raw fact grep.""" module_matches = _g.search_modules(pattern) if module_matches: for nid in module_matches: print(f"(result {nid})") print(f"(result-count {len(module_matches)})") return regex = re.compile(pattern, re.IGNORECASE) with open(sexp_path, "r") as f: for line in f: if regex.search(line): print(line.rstrip()) def cmd_boundaries(g: BeliefGraph, module_id: str) -> None: """Check boundary violations inferred from architecture layer patterns. Uses path-based layer classification and well-known architecture rules: - domain must not import infra/api/ui - ui must not import infra - service must not import api - infra must not import api """ # Well-known deny rules (src_layer, tgt_layer, reason) deny_rules = [ ("domain", "infra", "domain imports infrastructure directly"), ("domain", "api", "domain imports API layer"), ("domain", "ui", "domain imports UI layer"), ("ui", "infra", "UI imports infrastructure directly"), ("service","api", "service imports API layer (inversion)"), ("infra", "api", "infrastructure imports API layer"), ] if module_id == "all": scope_mods = set(g.nodes.keys()) else: matches = g.resolve_module(module_id) if not matches: _print_no_match(g, module_id) return scope_mods = set(matches) violations: list[tuple[str, str, str, str, str, str]] = [] for src_mod in sorted(scope_mods): src_layer = _classify_layer(src_mod, g.module_purpose(src_mod)) for edge in g.edges_from.get(src_mod, []): if edge.etype not in ("imports", "data-flow"): continue tgt_mod = edge.target.split("::")[0] if tgt_mod == src_mod: continue tgt_layer = _classify_layer(tgt_mod, g.module_purpose(tgt_mod)) if src_layer == tgt_layer: continue for deny_src, deny_tgt, reason in deny_rules: if src_layer == deny_src and tgt_layer == deny_tgt: violations.append((src_mod, tgt_mod, src_layer, tgt_layer, edge.etype, reason)) break if not violations: print(f"(boundaries-ok :checked {len(scope_mods)})") return for src, tgt, sl, tl, etype, reason in sorted(violations): src_file = g.module_to_file(src) tgt_file = g.module_to_file(tgt) print(f"(violation {src} {tgt} :src-layer {sl} :tgt-layer {tl} :via {etype} :lang {g.module_lang(src)} :src-file {_q(src_file)} :tgt-file {_q(tgt_file)} {_q(reason)})") print(f"(boundaries-summary :violations {len(violations)} :checked {len(scope_mods)})") def cmd_layers(g: BeliefGraph) -> None: """Show layer classification for all modules, grouped by layer.""" by_layer: dict[str, list[str]] = {} for nid in sorted(g.nodes.keys()): layer = _classify_layer(nid, g.module_purpose(nid)) by_layer.setdefault(layer, []).append(nid) for layer_name in sorted(by_layer.keys()): mods = by_layer[layer_name] print(f"(layer-group {layer_name} :count {len(mods)})") for m in mods: print(f"(layer-member {layer_name} {m} :lang {g.module_lang(m)})") # --------------------------------------------------------------------------- # Naming convention invariant checker # --------------------------------------------------------------------------- # Allowed naming conventions per language. A module whose :naming value is # NOT in the allowed set is a violation. Languages absent from this map have # no strict rule and are always considered conforming. _NAMING_RULES: dict[str, set[str]] = { "py": {"snake_case", "mixed", "kebab-case"}, "tsx": {"PascalCase", "camelCase", "kebab-case", "mixed"}, "ts": {"camelCase", "PascalCase", "kebab-case", "mixed", "snake_case"}, } def cmd_invariants(g: BeliefGraph, scope: str) -> None: """Check naming-convention violations across modules. For each module that has a :naming annotation, validate the value against the allowed set for its language. Modules without a :naming annotation are silently skipped (no data, no opinion). Output format:: (invariant-violation :expected snake_case :actual PascalCase :lang py :pkg ) (invariants-summary :violations 12 :checked 3587) (invariants-ok :checked 3587) """ if scope == "all": candidates = list(g.nodes.values()) else: matches = g.resolve_module(scope) if not matches: _print_no_match(g, scope) return candidates = [g.nodes[nid] for nid in matches if nid in g.nodes] checked = 0 violations: list[tuple[str, str, str, str, str]] = [] # (nid, expected, actual, lang, pkg) for node in candidates: if not node.naming: continue checked += 1 lang = {"ty": "ts", "tx": "tsx"}.get(node.lang, node.lang) allowed = _NAMING_RULES.get(lang) if allowed is None: # No rule defined for this language -- always conforming. continue if node.naming not in allowed: expected_repr = " ".join(sorted(allowed)) violations.append((node.nid, expected_repr, node.naming, lang, node.pkg)) if not violations: print(f"(invariants-ok :checked {checked})") return for nid, expected, actual, lang, pkg in sorted(violations): pkg_part = f" :pkg {_q(pkg)}" if pkg else "" print( f"(invariant-violation {nid}" f" :expected {_q(expected)}" f" :actual {actual}" f" :lang {lang}" f"{pkg_part})" ) print(f"(invariants-summary :violations {len(violations)} :checked {checked})") # --------------------------------------------------------------------------- # Function / type search and call graph commands # --------------------------------------------------------------------------- def cmd_find_function(g: BeliefGraph, name: str) -> None: """Find function/method definitions matching a name pattern. Searches fn entities and cls methods. Exact match first, then regex. """ pattern = re.compile(name, re.IGNORECASE) results: list[tuple[EntityFact, str]] = [] for ent_name, ent_list in g.entity_by_name.items(): for ent in ent_list: if ent.kind == "fn": if ent_name == name or pattern.search(ent_name): results.append((ent, "")) # Also search methods inside classes for _mod, ent_list in g.entities.items(): for ent in ent_list: if ent.kind == "cls" and ent.methods: for method in ent.methods: if method == name or pattern.search(method): results.append((ent, method)) if not results: _print_no_match(g, name) return seen: set[str] = set() for ent, method in sorted(results, key=lambda x: (x[0].module, x[0].line)): lang = g.module_lang(ent.module) fpath = g.module_to_file(ent.module) if method: key = f"{ent.module}::{ent.name}.{method}" if key not in seen: seen.add(key) print(f"(fn-def {ent.module}::{ent.name}.{method} :line {ent.line} :lang {lang} :file {_q(fpath)} :kind method)") else: key = f"{ent.module}::{ent.name}" if key not in seen: seen.add(key) print(f"(fn-def {ent.module}::{ent.name} :line {ent.line} :lang {lang} :file {_q(fpath)} :kind function)") def cmd_find_type(g: BeliefGraph, name: str) -> None: """Find type/class/interface/enum definitions matching a name pattern. Searches cls, ifc, typ, enm entities. """ pattern = re.compile(name, re.IGNORECASE) results: list[EntityFact] = [] for ent_name, ent_list in g.entity_by_name.items(): for ent in ent_list: if ent.kind in ("cls", "ifc", "typ", "enm"): if ent_name == name or pattern.search(ent_name): results.append(ent) if not results: _print_no_match(g, name) return for ent in sorted(results, key=lambda x: (x.module, x.line)): lang = g.module_lang(ent.module) fpath = g.module_to_file(ent.module) extras = "" if ent.bases: extras += " " + _sexp_list("bases", ent.bases) if ent.methods: extras += " " + _sexp_list("methods", ent.methods) print(f"(type-def {ent.kind} {ent.module}::{ent.name} :line {ent.line} :lang {lang} :file {_q(fpath)}{extras})") def _resolve_fn_entity(g: BeliefGraph, fn_name: str) -> list[tuple[str, EntityFact]]: """Resolve a function name to (qualified_id, entity) pairs. Returns entity-qualified IDs like 'module::fnName'. Searches both standalone functions and class methods. """ pattern = re.compile(re.escape(fn_name), re.IGNORECASE) results: list[tuple[str, EntityFact]] = [] for ent_name, ent_list in g.entity_by_name.items(): for ent in ent_list: if ent.kind == "fn" and (ent_name == fn_name or pattern.search(ent_name)): results.append((f"{ent.module}::{ent_name}", ent)) # Also search class methods for _mod, ent_list in g.entities.items(): for ent in ent_list: if ent.kind == "cls" and ent.methods: for method in ent.methods: if method == fn_name or pattern.search(method): results.append((f"{ent.module}::{method}", ent)) return results def cmd_find_callers(g: BeliefGraph, fn_name: str, max_depth: int) -> None: """Find all callers of a function (incoming call graph). Uses 'calls' edges from the belief map. Falls back to refs edges when calls edges are unavailable. """ targets = _resolve_fn_entity(g, fn_name) if not targets: _print_no_match(g, fn_name) return target_mods: set[str] = set() target_qualified: set[str] = set() for qid, ent in targets: target_mods.add(ent.module) target_qualified.add(qid) lang = g.module_lang(ent.module) fpath = g.module_to_file(ent.module) print(f"(callers-target {qid} :line {ent.line} :lang {lang} :file {_q(fpath)})") # Collect direct callers via calls edges direct_callers: list[Edge] = [] for e in g.all_edges: if e.etype == "calls": tgt_fn = e.target.split("::")[-1] if "::" in e.target else "" if e.target in target_qualified or tgt_fn == fn_name: direct_callers.append(e) if direct_callers: for e in sorted(direct_callers, key=lambda x: (x.source, x.target)): src_mod = e.source.split("::")[0] print(f"(caller {e.source} -> {e.target} :lang {g.module_lang(src_mod)} :file {_q(g.module_to_file(src_mod))} {' '.join(e.flags)})") else: # Fallback: use module-level edges (imports/refs) pointing at target modules print("; no calls edges found, showing module-level callers via imports/refs") # Walk reverse deps to given depth for tgt_mod in target_mods: visited: set[str] = {tgt_mod} _walk_rdeps(g, tgt_mod, visited, max_depth, 1, lambda d, et, src, cyc: print( f"(caller-module {src} -> {tgt_mod} :via {et} :depth {d} " f":lang {g.module_lang(src)} :file {_q(g.module_to_file(src))}" f"{' :cycle' if cyc else ''})")) def cmd_find_calls(g: BeliefGraph, fn_name: str, max_depth: int) -> None: """Find all functions called by a given function (outgoing call graph). Uses 'calls' edges from the belief map. """ sources = _resolve_fn_entity(g, fn_name) if not sources: _print_no_match(g, fn_name) return source_mods: set[str] = set() source_qualified: set[str] = set() for qid, ent in sources: source_mods.add(ent.module) source_qualified.add(qid) lang = g.module_lang(ent.module) fpath = g.module_to_file(ent.module) print(f"(calls-source {qid} :line {ent.line} :lang {lang} :file {_q(fpath)})") # Collect direct calls via calls edges direct_calls: list[Edge] = [] for e in g.all_edges: if e.etype == "calls": src_fn = e.source.split("::")[-1] if "::" in e.source else "" if e.source in source_qualified or src_fn == fn_name: direct_calls.append(e) if direct_calls: for e in sorted(direct_calls, key=lambda x: (x.source, x.target)): tgt_mod = e.target.split("::")[0] print(f"(calls-target {e.source} -> {e.target} :lang {g.module_lang(tgt_mod)} :file {_q(g.module_to_file(tgt_mod))} {' '.join(e.flags)})") else: print("; no calls edges found, showing module-level deps via imports") # Walk forward deps to given depth for src_mod in source_mods: visited: set[str] = {src_mod} _walk_deps(g, src_mod, visited, max_depth, 1, lambda d, et, tgt, cyc: print( f"(calls-module {src_mod} -> {tgt} :via {et} :depth {d} " f":lang {g.module_lang(tgt)} :file {_q(g.module_to_file(tgt))}" f"{' :cycle' if cyc else ''})")) def cmd_find_callchain(g: BeliefGraph, src_fn: str, tgt_fn: str, max_depth: int) -> None: """Trace a call path between two functions via BFS on calls edges. Falls back to module-level path if no function-level calls edges exist. """ src_entities = _resolve_fn_entity(g, src_fn) tgt_entities = _resolve_fn_entity(g, tgt_fn) if not src_entities: print(f"(error no-match-source {_q(src_fn)})") return if not tgt_entities: print(f"(error no-match-target {_q(tgt_fn)})") return src_ids = {qid for qid, _ in src_entities} tgt_ids = {qid for qid, _ in tgt_entities} tgt_fn_names = {qid.split("::")[-1] for qid in tgt_ids} # Build adjacency from calls edges call_adj: dict[str, list[str]] = {} for e in g.all_edges: if e.etype == "calls": call_adj.setdefault(e.source, []).append(e.target) # BFS from source to target from collections import deque found_path: list[str] = [] for start in src_ids: queue: deque[list[str]] = deque([[start]]) visited: set[str] = {start} while queue: path = queue.popleft() if len(path) - 1 > max_depth: break current = path[-1] cur_fn = current.split("::")[-1] if "::" in current else "" if current in tgt_ids or cur_fn in tgt_fn_names: found_path = path break for nxt in call_adj.get(current, []): if nxt not in visited: visited.add(nxt) queue.append(path + [nxt]) if found_path: break if found_path: for i, node in enumerate(found_path): mod = node.split("::")[0] lang = g.module_lang(mod) fpath = g.module_to_file(mod) print(f"(callchain-step {i} {node} :lang {lang} :file {_q(fpath)})") print(f"(callchain-summary :length {len(found_path)} :hops {len(found_path) - 1})") return # Fallback: module-level BFS using both forward deps AND reverse deps src_mods = {ent.module for _, ent in src_entities} tgt_mods = {ent.module for _, ent in tgt_entities} print("; no function-level call path found, trying module-level path") # Build module adjacency (forward deps + shared entity refs for better connectivity) mod_adj: dict[str, set[str]] = {} for mod in g.nodes: neighbors = set(_collect_dep_mods(g, mod).keys()) # Also add reverse deps (bidirectional search for reachability) neighbors.update(_collect_rdep_mods(g, mod).keys()) mod_adj[mod] = neighbors for start_mod in src_mods: queue_m: deque[list[str]] = deque([[start_mod]]) visited_m: set[str] = {start_mod} mod_path: list[str] = [] while queue_m: path_m = queue_m.popleft() if len(path_m) - 1 > max_depth: break current_m = path_m[-1] if current_m in tgt_mods: mod_path = path_m break for nxt_m in sorted(mod_adj.get(current_m, set())): if nxt_m not in visited_m: visited_m.add(nxt_m) queue_m.append(path_m + [nxt_m]) if mod_path: for i, mod in enumerate(mod_path): lang = g.module_lang(mod) fpath = g.module_to_file(mod) print(f"(callchain-module-step {i} {mod} :lang {lang} :file {_q(fpath)})") print(f"(callchain-module-summary :length {len(mod_path)} :hops {len(mod_path) - 1})") return print(f"(callchain-none {_q(src_fn)} -> {_q(tgt_fn)} :max-depth {max_depth})") def cmd_grep_functions(g: BeliefGraph, pattern: str) -> None: """Search function bodies in source files for a regex pattern. For each function in the belief map, reads the source file and searches the function body (from definition line to next entity or EOF). """ regex = re.compile(pattern, re.IGNORECASE) results: list[tuple[str, str, int, str]] = [] # (module::fn, file, match_line, match_text) # Group entities by file for efficient reading file_entities: dict[str, list[tuple[str, EntityFact]]] = {} for mod, ent_list in g.entities.items(): fpath = g.module_to_file(mod) if fpath == mod: continue # file not found on disk for ent in ent_list: if ent.kind == "fn": file_entities.setdefault(fpath, []).append((mod, ent)) for fpath, fn_list in sorted(file_entities.items()): if not os.path.isfile(fpath): continue try: with open(fpath, "r", errors="replace") as f: lines = f.readlines() except OSError: continue # Sort functions by line number to determine body ranges sorted_fns = sorted(fn_list, key=lambda x: x[1].line) for idx, (mod, ent) in enumerate(sorted_fns): start = ent.line - 1 # 0-indexed if start < 0 or start >= len(lines): continue # End at next entity's line or +100 lines, whichever is smaller if idx + 1 < len(sorted_fns): end = min(sorted_fns[idx + 1][1].line - 1, start + 100, len(lines)) else: end = min(start + 100, len(lines)) for line_idx in range(start, end): line = lines[line_idx] if regex.search(line): results.append(( f"{mod}::{ent.name}", fpath, line_idx + 1, line.rstrip(), )) if not results: print(f"(grep-functions-empty {_q(pattern)})") return for qid, fpath, line_num, text in results: lang = g.module_lang(qid.split("::")[0]) print(f"(grep-fn {qid} :line {line_num} :lang {lang} :file {_q(fpath)} :match {_q(text)})") print(f"(grep-functions-summary :matches {len(results)} :pattern {_q(pattern)})") def cmd_diff_functions(g: BeliefGraph, ref: str) -> None: """Identify which functions were changed in a git diff. Parses 'git diff ' output to find changed line ranges, then maps those ranges to function definitions in the belief map. """ import subprocess # Use the belief map root as git working directory (not cwd) git_cwd = g._root or None try: result = subprocess.run( ["git", "diff", "--unified=0", ref], capture_output=True, text=True, timeout=30, cwd=git_cwd, ) if result.returncode != 0: # Try as a commit ref result = subprocess.run( ["git", "diff", "--unified=0", ref, "HEAD"], capture_output=True, text=True, timeout=30, cwd=git_cwd, ) if result.returncode != 0: print(f"(error git-diff-failed {_q(result.stderr.strip())})") return except FileNotFoundError: print("(error git-not-found)") return except subprocess.TimeoutExpired: print("(error git-diff-timeout)") return # Parse diff output to extract file -> changed line ranges changed_ranges: dict[str, list[tuple[int, int]]] = {} # file -> [(start, end)] current_file = "" for line in result.stdout.splitlines(): if line.startswith("+++ b/"): current_file = line[6:] elif line.startswith("@@ ") and current_file: # Parse hunk header: @@ -old,count +new,count @@ parts = line.split(" ") for part in parts: if part.startswith("+") and "," in part: nums = part[1:].split(",") start = int(nums[0]) count = int(nums[1]) if len(nums) > 1 else 1 if count > 0: changed_ranges.setdefault(current_file, []).append( (start, start + count - 1) ) elif part.startswith("+") and part[1:].isdigit(): start = int(part[1:]) changed_ranges.setdefault(current_file, []).append( (start, start) ) if not changed_ranges: print(f"(diff-functions-empty {_q(ref)})") return # Map changed lines to functions # Build file -> sorted entities index file_entities: dict[str, list[tuple[str, EntityFact]]] = {} for mod, ent_list in g.entities.items(): fpath = g.module_to_file(mod) for ent in ent_list: file_entities.setdefault(fpath, []).append((mod, ent)) # Normalize paths for matching (diff paths are relative to repo root) repo_root = g._root results: list[tuple[str, EntityFact, str]] = [] # (qualified_id, entity, file) for diff_file, ranges in changed_ranges.items(): # Try matching against belief map files abs_path = os.path.join(repo_root, diff_file) # Also try relative path candidates = [abs_path, diff_file] matched_ents: list[tuple[str, EntityFact]] = [] for candidate in candidates: if candidate in file_entities: matched_ents = file_entities[candidate] break if not matched_ents: # Try suffix matching for fpath, ent_list in file_entities.items(): if fpath.endswith(diff_file) or diff_file.endswith(fpath.lstrip("./")): matched_ents = ent_list break if not matched_ents: continue # Sort entities by line to determine body ranges sorted_ents = sorted(matched_ents, key=lambda x: x[1].line) for start, end in ranges: for idx, (mod, ent) in enumerate(sorted_ents): ent_start = ent.line if idx + 1 < len(sorted_ents): ent_end = sorted_ents[idx + 1][1].line - 1 else: ent_end = ent_start + 200 # approximate # Check overlap if start <= ent_end and end >= ent_start: qid = f"{mod}::{ent.name}" if not any(r[0] == qid for r in results): results.append((qid, ent, diff_file)) if not results: # Report changed files even if no function mapping for diff_file in sorted(changed_ranges.keys()): print(f"(diff-file {_q(diff_file)} :hunks {len(changed_ranges[diff_file])})") print(f"(diff-functions-no-match :files {len(changed_ranges)} :ref {_q(ref)})") return for qid, ent, diff_file in sorted(results, key=lambda x: (x[2], x[1].line)): lang = g.module_lang(ent.module) fpath = g.module_to_file(ent.module) print(f"(diff-fn {qid} :kind {ent.kind} :line {ent.line} :lang {lang} :file {_q(fpath)})") print(f"(diff-functions-summary :changed {len(results)} :files {len(changed_ranges)} :ref {_q(ref)})") # --------------------------------------------------------------------------- # Scheme query REPL -- composable S-expression queries on the graph # --------------------------------------------------------------------------- # # A tiny Scheme-dialect evaluator that exposes the belief graph as queryable # data. Every value is either a string, int, bool, list, or a "module-set" # (set of module IDs). Functions compose naturally: # # (deps "operatives.controller" 2) -> module-set of dependencies # (rdeps "operatives.controller" 1) -> module-set of reverse deps # (boundary "operatives.controller") -> module-set (target + deps + rdeps) # (entity "OperativesService") -> list of module IDs defining it # (imports "operatives.controller") -> module-set of direct imports # (dependents "operatives.controller") -> module-set of direct dependents # (intersect (deps "A" 2) (rdeps "B" 1)) -> set intersection # (union A B) -> set union # (diff A B) -> set difference # (filter SET "pattern") -> grep within set # (files SET) -> resolve module IDs to file paths # (count SET) -> cardinality # (analyze "id") -> full analysis (prints sexp) # (search "pattern") -> raw grep on sexp file _QUERY_HELP = """\ ; Scheme query language for .belief_map.sexp ; ; Primitives (return module-set or list): ; (resolve "pattern") Find module IDs matching pattern ; (deps "id" depth) Outgoing dependency set (default depth 2) ; (rdeps "id" depth) Reverse dependency set (default depth 1) ; (imports "id") Direct import targets ; (dependents "id") Direct dependents (who imports this) ; (boundary "id") Target + deps + rdeps (change boundary) ; (entity "Name") Modules defining entity ; (refs-to "Name") Modules referencing entity ; (search "pattern") Grep .belief_map.sexp ; ; Function & type search (return module-set): ; (find-function "name") Modules containing function definitions ; (find-type "name") Modules containing type/class/ifc/enum defs ; (find-callers "fn" depth) Modules that call a function (default depth 2) ; (find-calls "fn" depth) Modules that a function calls (default depth 2) ; (diff-functions "ref") Print changed functions (side-effect, returns empty) ; ; Set operations: ; (intersect A B) Set intersection ; (union A B ...) Set union ; (diff A B) A minus B ; (filter SET "pattern") Keep members matching regex ; ; Output: ; (files SET) Resolve to file paths ; (count SET) Count members ; (analyze "id") Full analysis output ; ; Composition: ; (files (intersect (deps "A" 2) (rdeps "B" 1))) ; (filter (boundary "X") "dto") ; (count (deps "operatives.service" 3)) ; (files (find-callers "createOrder" 3)) ; (intersect (find-function "handle") (deps "api" 2)) """ class _SchemeEval: """Tiny Scheme evaluator for belief graph queries.""" def __init__(self, g: BeliefGraph, sexp_path: str) -> None: self._g = g self._sexp_path = sexp_path def eval_str(self, expr_str: str) -> object: tokens = self._tokenize(expr_str) ast = self._parse(tokens) return self._eval(ast) # -- Tokenizer -- @staticmethod def _tokenize(s: str) -> list[str]: tokens: list[str] = [] i = 0 while i < len(s): ch = s[i] if ch in (" ", "\t", "\n", "\r"): i += 1 elif ch == ";": while i < len(s) and s[i] != "\n": i += 1 elif ch in ("(", ")"): tokens.append(ch) i += 1 elif ch == '"': j = i + 1 while j < len(s) and s[j] != '"': if s[j] == "\\": j += 1 j += 1 tokens.append(s[i + 1 : j]) i = j + 1 else: j = i while j < len(s) and s[j] not in (" ", "\t", "\n", "\r", "(", ")", '"', ";"): j += 1 tokens.append(s[i:j]) i = j return tokens # -- Parser: tokens -> nested lists -- @staticmethod def _parse(tokens: list[str]) -> object: if not tokens: return [] pos = [0] def _read() -> object: if pos[0] >= len(tokens): return None tok = tokens[pos[0]] if tok == "(": pos[0] += 1 lst: list[object] = [] while pos[0] < len(tokens) and tokens[pos[0]] != ")": lst.append(_read()) pos[0] += 1 # skip ) return lst if tok == ")": pos[0] += 1 return None pos[0] += 1 # Try int try: return int(tok) except ValueError: return tok result = _read() return result # -- Evaluator -- def _eval(self, ast: object) -> object: if isinstance(ast, (int, float)): return ast if isinstance(ast, str): return ast if not isinstance(ast, list) or len(ast) == 0: return set() head = ast[0] args = ast[1:] # Evaluate args (eager) evaled = [self._eval(a) for a in args] return self._apply(head, evaled) def _apply(self, fn: str, args: list[object]) -> object: g = self._g if fn == "resolve": pattern = str(args[0]) if args else "" return set(g.resolve_module(pattern)) if fn == "deps": pattern = str(args[0]) if args else "" depth = int(str(args[1])) if len(args) > 1 else 2 matches = g.resolve_module(pattern) result: set[str] = set() for m in matches: visited: set[str] = {m} self._collect_deps(m, visited, depth, 1, result) return result if fn == "rdeps": pattern = str(args[0]) if args else "" depth = int(str(args[1])) if len(args) > 1 else 1 matches = g.resolve_module(pattern) result = set() for m in matches: visited = {m} self._collect_rdeps(m, visited, depth, 1, result) return result if fn == "imports": pattern = str(args[0]) if args else "" matches = g.resolve_module(pattern) result = set() for m in matches: result.update(_collect_dep_mods(g, m).keys()) return result if fn == "dependents": pattern = str(args[0]) if args else "" matches = g.resolve_module(pattern) result = set() for m in matches: result.update(_collect_rdep_mods(g, m).keys()) return result if fn == "boundary": pattern = str(args[0]) if args else "" matches = g.resolve_module(pattern) result = set() for m in matches: result.add(m) result.update(_collect_dep_mods(g, m).keys()) result.update(_collect_rdep_mods(g, m).keys()) return result if fn == "entity": name = str(args[0]) if args else "" defs = g.entity_by_name.get(name, []) if not defs: pattern = re.compile(name, re.IGNORECASE) for ename, ents in g.entity_by_name.items(): if pattern.search(ename): defs.extend(ents) return set(d.module for d in defs) if fn == "refs-to": name = str(args[0]) if args else "" edges = g.edges_by_entity_name.get(name, []) return set(e.source.split("::")[0] for e in edges) if fn == "intersect": a = self._to_set(args[0]) b = self._to_set(args[1]) if len(args) > 1 else set() return a & b if fn == "union": result = set() for a in args: result |= self._to_set(a) return result if fn == "diff": a = self._to_set(args[0]) b = self._to_set(args[1]) if len(args) > 1 else set() return a - b if fn == "filter": s = self._to_set(args[0]) pattern = str(args[1]) if len(args) > 1 else "" regex = re.compile(pattern, re.IGNORECASE) return {m for m in s if regex.search(m)} if fn == "files": s = self._to_set(args[0]) return set(g.module_to_file(m) for m in s) if fn == "count": s = self._to_set(args[0]) return len(s) if fn == "analyze": pattern = str(args[0]) if args else "" cmd_analyze(g, pattern) return set() if fn == "search": pattern = str(args[0]) if args else "" cmd_search(g, pattern, self._sexp_path) return set() if fn == "layer": pattern = str(args[0]) if args else "" matches = g.resolve_module(pattern) if matches: return _classify_layer(matches[0], g.module_purpose(matches[0])) return "unknown" if fn == "violations": pattern = str(args[0]) if args else "all" scope = set(g.nodes.keys()) if pattern == "all" else set(g.resolve_module(pattern)) deny_rules = [ ("domain", "infra"), ("domain", "api"), ("domain", "ui"), ("ui", "infra"), ("service", "api"), ("infra", "api"), ] result: set[str] = set() for src_mod in scope: sl = _classify_layer(src_mod, g.module_purpose(src_mod)) for edge in g.edges_from.get(src_mod, []): if edge.etype not in ("imports", "data-flow"): continue tgt_mod = edge.target.split("::")[0] tl = _classify_layer(tgt_mod, g.module_purpose(tgt_mod)) if sl != tl and (sl, tl) in deny_rules: result.add(src_mod) return result if fn == "layer-of": s = self._to_set(args[0]) target_layer = str(args[1]) if len(args) > 1 else "" return {m for m in s if _classify_layer(m, g.module_purpose(m)) == target_layer} if fn == "find-function" or fn == "find_function": name = str(args[0]) if args else "" pattern = re.compile(name, re.IGNORECASE) result: set[str] = set() for ent_name, ent_list in g.entity_by_name.items(): for ent in ent_list: if ent.kind == "fn" and (ent_name == name or pattern.search(ent_name)): result.add(ent.module) return result if fn == "find-type" or fn == "find_type": name = str(args[0]) if args else "" pattern = re.compile(name, re.IGNORECASE) result = set() for ent_name, ent_list in g.entity_by_name.items(): for ent in ent_list: if ent.kind in ("cls", "ifc", "typ", "enm") and (ent_name == name or pattern.search(ent_name)): result.add(ent.module) return result if fn == "find-callers" or fn == "find_callers": fn_name = str(args[0]) if args else "" depth = int(str(args[1])) if len(args) > 1 else 2 targets = _resolve_fn_entity(g, fn_name) result = set() for _qid, ent in targets: result.add(ent.module) # Collect callers via calls edges tgt_qualified = {qid for qid, _ in targets} for e in g.all_edges: if e.etype == "calls" and e.target in tgt_qualified: result.add(e.source.split("::")[0]) # Also walk reverse deps for _qid, ent in targets: visited: set[str] = {ent.module} self._collect_rdeps(ent.module, visited, depth, 1, result) return result if fn == "find-calls" or fn == "find_calls": fn_name = str(args[0]) if args else "" depth = int(str(args[1])) if len(args) > 1 else 2 sources = _resolve_fn_entity(g, fn_name) result = set() src_qualified = {qid for qid, _ in sources} for e in g.all_edges: if e.etype == "calls" and e.source in src_qualified: result.add(e.target.split("::")[0]) for _qid, ent in sources: visited = {ent.module} self._collect_deps(ent.module, visited, depth, 1, result) return result if fn == "diff-functions" or fn == "diff_functions": ref = str(args[0]) if args else "HEAD" cmd_diff_functions(g, ref) return set() print(f"; unknown function: {fn}") return set() def _collect_deps( self, nid: str, visited: set[str], max_depth: int, depth: int, result: set[str], ) -> None: if depth > max_depth: return for tgt in _collect_dep_mods(self._g, nid): result.add(tgt) if tgt not in visited: visited.add(tgt) self._collect_deps(tgt, visited, max_depth, depth + 1, result) def _collect_rdeps( self, nid: str, visited: set[str], max_depth: int, depth: int, result: set[str], ) -> None: if depth > max_depth: return for src in _collect_rdep_mods(self._g, nid): result.add(src) if src not in visited: visited.add(src) self._collect_rdeps(src, visited, max_depth, depth + 1, result) @staticmethod def _to_set(val: object) -> set[str]: if isinstance(val, set): return val if isinstance(val, str): return {val} if isinstance(val, (list, tuple)): return set(str(v) for v in val) return set() def _format_result(val: object) -> None: """Print a Scheme query result as S-expression facts.""" if isinstance(val, set): for item in sorted(val): print(f"(result {_q(item)})") if not val: print("(result-empty)") elif isinstance(val, int): print(f"(result-count {val})") elif isinstance(val, str): print(f"(result {_q(val)})") elif isinstance(val, list): for item in val: print(f"(result {_q(str(item))})") else: print(f"(result {_q(str(val))})") def cmd_query(g: BeliefGraph, expr: str, sexp_path: str) -> None: """Evaluate a single Scheme query expression.""" evaluator = _SchemeEval(g, sexp_path) result = evaluator.eval_str(expr) _format_result(result) def cmd_repl(g: BeliefGraph, sexp_path: str) -> None: """Interactive Scheme REPL for belief map queries.""" evaluator = _SchemeEval(g, sexp_path) print("; belief-map query REPL") print("; type (help) for available functions, (quit) to exit") print() while True: try: expr = input("bm> ").strip() except (EOFError, KeyboardInterrupt): print() break if not expr or expr.startswith(";"): continue if expr in ("(quit)", "(exit)", "quit", "exit"): break if expr in ("(help)", "help"): print(_QUERY_HELP) continue result = evaluator.eval_str(expr) _format_result(result) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def find_sexp_file() -> str: """Find belief_map.sexp in current or parent dirs. Searches for (in priority order per directory): .belief_map.sexp, belief_map.sexp If only a JSON belief map is found, prints a migration hint. """ _SEXP_NAMES = (".belief_map.sexp", "belief_map.sexp") _JSON_NAMES = (".belief_map.json", "belief_map.json") cur = os.getcwd() json_found = "" while True: for name in _SEXP_NAMES: candidate = os.path.join(cur, name) if os.path.isfile(candidate): return candidate if not json_found: for name in _JSON_NAMES: candidate = os.path.join(cur, name) if os.path.isfile(candidate): json_found = candidate parent = os.path.dirname(cur) if parent == cur: break cur = parent if json_found: print(f"Error: found JSON belief map at {json_found} but this script requires sexp format.", file=sys.stderr) print("Rebuild with the sexp builder:", file=sys.stderr) print(f" python3 ~/.claude/skills/codespaces/scripts/build_belief_map.py", file=sys.stderr) else: print("Error: .belief_map.sexp not found. Run build_belief_map.py first.", file=sys.stderr) sys.exit(1) def main() -> None: args = sys.argv[1:] if not args or args[0] in ("-h", "--help", "help"): print(__doc__) sys.exit(0) sexp_path = find_sexp_file() g = BeliefGraph() g.load(sexp_path) cmd = args[0] if cmd == "analyze" and len(args) >= 2: cmd_analyze(g, args[1]) elif cmd == "boundary" and len(args) >= 2: files_only = "--files-only" in args module_arg = [a for a in args[1:] if a != "--files-only"][0] cmd_boundary(g, module_arg, files_only=files_only) elif cmd == "quick" and len(args) >= 2: cmd_quick(g, args[1], sexp_path) elif cmd == "module" and len(args) >= 2: cmd_module(g, args[1]) elif cmd == "deps" and len(args) >= 2: depth = int(args[2]) if len(args) >= 3 else 2 cmd_deps(g, args[1], depth) elif cmd == "rdeps" and len(args) >= 2: depth = int(args[2]) if len(args) >= 3 else 2 cmd_rdeps(g, args[1], depth) elif cmd == "entity" and len(args) >= 2: cmd_entity(g, args[1]) elif cmd == "flow" and len(args) >= 3: cmd_flow(g, args[1], args[2]) elif cmd == "search" and len(args) >= 2: cmd_search(g, args[1], sexp_path) elif cmd == "boundaries": target = args[1] if len(args) >= 2 else "all" cmd_boundaries(g, target) elif cmd == "layers": cmd_layers(g) elif cmd == "invariants": target = args[1] if len(args) >= 2 else "all" cmd_invariants(g, target) elif cmd == "find_function" and len(args) >= 2: cmd_find_function(g, args[1]) elif cmd == "find_type" and len(args) >= 2: cmd_find_type(g, args[1]) elif cmd == "find_callchain" and len(args) >= 3: depth = int(args[3]) if len(args) >= 4 else 5 cmd_find_callchain(g, args[1], args[2], depth) elif cmd == "find_callers" and len(args) >= 2: depth = int(args[2]) if len(args) >= 3 else 2 cmd_find_callers(g, args[1], depth) elif cmd == "find_calls" and len(args) >= 2: depth = int(args[2]) if len(args) >= 3 else 2 cmd_find_calls(g, args[1], depth) elif cmd == "grep_functions" and len(args) >= 2: cmd_grep_functions(g, args[1]) elif cmd == "diff_functions": ref = args[1] if len(args) >= 2 else "HEAD" cmd_diff_functions(g, ref) elif cmd == "query" and len(args) >= 2: cmd_query(g, " ".join(args[1:]), sexp_path) elif cmd == "repl": cmd_repl(g, sexp_path) else: print(f"(error unknown-command {_q(cmd if args else 'none')})") print("; Commands: analyze, boundary, quick, boundaries, layers, invariants, module, deps, rdeps, entity, flow, search,") print("; find_function, find_type, find_callchain, find_callers, find_calls, grep_functions, diff_functions,") print("; query, repl") sys.exit(1) if __name__ == "__main__": main()