diff --git a/code_review_graph/analysis 2.py b/code_review_graph/analysis 2.py deleted file mode 100644 index 1668a1a3..00000000 --- a/code_review_graph/analysis 2.py +++ /dev/null @@ -1,410 +0,0 @@ -"""Graph analysis: hub detection, bridge nodes, knowledge gaps, -surprise scoring, suggested questions.""" - -from __future__ import annotations - -import logging -from collections import Counter, defaultdict - -from .graph import GraphStore, _sanitize_name - -logger = logging.getLogger(__name__) - - -def find_hub_nodes(store: GraphStore, top_n: int = 10) -> list[dict]: - """Find the most connected nodes (highest in+out degree), excluding File nodes. - - Returns list of dicts with: name, qualified_name, kind, file, - in_degree, out_degree, total_degree, community_id - """ - # Build degree counts from all edges - edges = store.get_all_edges() - in_degree: dict[str, int] = Counter() - out_degree: dict[str, int] = Counter() - for e in edges: - out_degree[e.source_qualified] += 1 - in_degree[e.target_qualified] += 1 - - # Get all non-File nodes - nodes = store.get_all_nodes(exclude_files=True) - community_map = store.get_all_community_ids() - - scored = [] - for n in nodes: - qn = n.qualified_name - ind = in_degree.get(qn, 0) - outd = out_degree.get(qn, 0) - total = ind + outd - if total == 0: - continue - scored.append({ - "name": _sanitize_name(n.name), - "qualified_name": n.qualified_name, - "kind": n.kind, - "file": n.file_path, - "in_degree": ind, - "out_degree": outd, - "total_degree": total, - "community_id": community_map.get(qn), - }) - - scored.sort( - key=lambda x: x.get("total_degree", 0), # type: ignore[arg-type,return-value] - reverse=True, - ) - return scored[:top_n] - - -def find_bridge_nodes( - store: GraphStore, top_n: int = 10 -) -> list[dict]: - """Find nodes with highest betweenness centrality. - - These are architectural chokepoints that sit on shortest paths - between many node pairs. If they break, multiple communities - lose connectivity. - - Returns list of dicts with: name, qualified_name, kind, file, - betweenness, community_id - """ - import networkx as nx - - # Build the graph — use cached version if available - nxg = store._build_networkx_graph() - - # Compute betweenness centrality (approximate for large graphs) - n_nodes = nxg.number_of_nodes() - if n_nodes > 5000: - # Sample-based approximation for large graphs - k = min(500, n_nodes) - bc = nx.betweenness_centrality(nxg, k=k, normalized=True) - elif n_nodes > 0: - bc = nx.betweenness_centrality(nxg, normalized=True) - else: - return [] - - community_map = store.get_all_community_ids() - node_map = { - n.qualified_name: n - for n in store.get_all_nodes(exclude_files=True) - } - - results = [] - for qn, score in bc.items(): - if score <= 0 or qn not in node_map: - continue - n = node_map[qn] - if n.kind == "File": - continue - results.append({ - "name": _sanitize_name(n.name), - "qualified_name": n.qualified_name, - "kind": n.kind, - "file": n.file_path, - "betweenness": round(score, 6), - "community_id": community_map.get(qn), - }) - - results.sort( - key=lambda x: float(x.get("betweenness", 0)), # type: ignore[arg-type,return-value] - reverse=True, - ) - return results[:top_n] - - -def find_knowledge_gaps(store: GraphStore) -> dict[str, list[dict]]: - """Identify structural weaknesses in the codebase graph. - - Returns dict with categories: - - isolated_nodes: degree <= 1, disconnected from graph - - thin_communities: fewer than 3 members - - untested_hotspots: high-degree nodes with no TESTED_BY edges - - single_file_communities: entire community in one file - """ - edges = store.get_all_edges() - nodes = store.get_all_nodes(exclude_files=True) - community_map = store.get_all_community_ids() - - # Build degree map - degree: dict[str, int] = Counter() - tested_nodes: set[str] = set() - for e in edges: - degree[e.source_qualified] += 1 - degree[e.target_qualified] += 1 - if e.kind == "TESTED_BY": - tested_nodes.add(e.source_qualified) - - # 1. Isolated nodes (degree <= 1, not File) - isolated = [] - for n in nodes: - d = degree.get(n.qualified_name, 0) - if d <= 1: - isolated.append({ - "name": _sanitize_name(n.name), - "qualified_name": n.qualified_name, - "kind": n.kind, - "file": n.file_path, - "degree": d, - }) - - # 2. Build community sizes and file maps from node data - comm_sizes: Counter[int] = Counter() - comm_files: dict[int, set[str]] = defaultdict(set) - for n in nodes: - cid = community_map.get(n.qualified_name) - if cid is not None: - comm_sizes[cid] += 1 - comm_files[cid].add(n.file_path) - - # Thin communities (< 3 members) - communities = store.get_communities_list() - thin = [] - for c in communities: - cid = int(c["id"]) - size = comm_sizes.get(cid, 0) - if size < 3: - thin.append({ - "community_id": cid, - "name": str(c["name"]), - "size": size, - }) - - # 3. Untested hotspots (degree >= 5, no TESTED_BY) - untested_hotspots = [] - for n in nodes: - d = degree.get(n.qualified_name, 0) - if (d >= 5 - and n.qualified_name not in tested_nodes - and not n.is_test): - untested_hotspots.append({ - "name": _sanitize_name(n.name), - "qualified_name": n.qualified_name, - "kind": n.kind, - "file": n.file_path, - "degree": d, - }) - untested_hotspots.sort( - key=lambda x: x.get("degree", 0), # type: ignore[arg-type,return-value] - reverse=True, - ) - - # 4. Single-file communities - single_file = [] - for c in communities: - cid = int(c["id"]) - files = comm_files.get(cid, set()) - size = comm_sizes.get(cid, 0) - if len(files) == 1 and size >= 3: - single_file.append({ - "community_id": cid, - "name": str(c["name"]), - "size": size, - "file": next(iter(files)), - }) - - return { - "isolated_nodes": isolated[:50], - "thin_communities": thin, - "untested_hotspots": untested_hotspots[:20], - "single_file_communities": single_file, - } - - -def find_surprising_connections( - store: GraphStore, top_n: int = 15 -) -> list[dict]: - """Find edges with high surprise scores. - - Detects unexpected architectural coupling based on: - - Cross-community: source and target in different communities - - Cross-language: different file languages - - Peripheral-to-hub: low-degree node to high-degree node - - Cross-file-type: test calling production or vice versa - - Non-standard edge kind for the node types - """ - edges = store.get_all_edges() - nodes = store.get_all_nodes(exclude_files=True) - community_map = store.get_all_community_ids() - - node_map = {n.qualified_name: n for n in nodes} - - # Build degree map - degree: dict[str, int] = Counter() - for e in edges: - degree[e.source_qualified] += 1 - degree[e.target_qualified] += 1 - - # Median degree for peripheral detection - degrees = [d for d in degree.values() if d > 0] - if not degrees: - return [] - median_deg = sorted(degrees)[len(degrees) // 2] - high_deg_threshold = max(median_deg * 3, 10) - - scored_edges = [] - for e in edges: - src = node_map.get(e.source_qualified) - tgt = node_map.get(e.target_qualified) - if not src or not tgt: - continue - if src.kind == "File" or tgt.kind == "File": - continue - - score = 0.0 - reasons = [] - - # Cross-community (+0.3) - src_cid = community_map.get(e.source_qualified) - tgt_cid = community_map.get(e.target_qualified) - if (src_cid is not None - and tgt_cid is not None - and src_cid != tgt_cid): - score += 0.3 - reasons.append("cross-community") - - # Cross-language (+0.2) - src_lang = ( - src.file_path.rsplit(".", 1)[-1] - if "." in src.file_path else "" - ) - tgt_lang = ( - tgt.file_path.rsplit(".", 1)[-1] - if "." in tgt.file_path else "" - ) - if src_lang and tgt_lang and src_lang != tgt_lang: - score += 0.2 - reasons.append("cross-language") - - # Peripheral-to-hub (+0.2) - src_deg = degree.get(e.source_qualified, 0) - tgt_deg = degree.get(e.target_qualified, 0) - if ((src_deg <= 2 and tgt_deg >= high_deg_threshold) - or (tgt_deg <= 2 - and src_deg >= high_deg_threshold)): - score += 0.2 - reasons.append("peripheral-to-hub") - - # Cross-file-type: test <-> non-test (+0.15) - if src.is_test != tgt.is_test and e.kind == "CALLS": - score += 0.15 - reasons.append("cross-test-boundary") - - # Non-standard edge kind (+0.15) - if e.kind == "CALLS" and src.kind == "Type": - score += 0.15 - reasons.append("unusual-edge-kind") - - if score > 0: - scored_edges.append({ - "source": _sanitize_name(src.name), - "source_qualified": e.source_qualified, - "target": _sanitize_name(tgt.name), - "target_qualified": e.target_qualified, - "edge_kind": e.kind, - "surprise_score": round(score, 2), - "reasons": reasons, - "source_community": src_cid, - "target_community": tgt_cid, - }) - - scored_edges.sort( - key=lambda x: float(x.get("surprise_score", 0)), # type: ignore[arg-type,return-value] - reverse=True, - ) - return scored_edges[:top_n] - - -def generate_suggested_questions( - store: GraphStore, -) -> list[dict]: - """Auto-generate review questions from graph analysis. - - Categories: - - bridge_node: Why does X connect communities A and B? - - isolated_node: Is X dead code or dynamically invoked? - - low_cohesion: Should community X be split? - - hub_risk: Does hub node X have adequate test coverage? - - surprising: Why does A call B across community boundary? - """ - questions = [] - - # Bridge node questions - bridges = find_bridge_nodes(store, top_n=3) - for b in bridges: - questions.append({ - "category": "bridge_node", - "question": ( - f"'{b['name']}' is a critical connector " - f"between multiple code regions. Is it " - f"adequately tested and documented?" - ), - "target": b["qualified_name"], - "priority": "high", - }) - - # Hub risk questions - hubs = find_hub_nodes(store, top_n=3) - edges = store.get_all_edges() - tested = { - e.source_qualified - for e in edges if e.kind == "TESTED_BY" - } - for h in hubs: - if h["qualified_name"] not in tested: - questions.append({ - "category": "hub_risk", - "question": ( - f"Hub node '{h['name']}' has " - f"{h['total_degree']} connections but no " - f"direct test coverage. Should it be " - f"tested?" - ), - "target": h["qualified_name"], - "priority": "high", - }) - - # Surprising connection questions - surprises = find_surprising_connections(store, top_n=3) - for s in surprises: - if "cross-community" in s["reasons"]: - questions.append({ - "category": "surprising_connection", - "question": ( - f"'{s['source']}' (community " - f"{s['source_community']}) calls " - f"'{s['target']}' (community " - f"{s['target_community']}). Is this " - f"coupling intentional?" - ), - "target": s["source_qualified"], - "priority": "medium", - }) - - # Knowledge gap questions - gaps = find_knowledge_gaps(store) - - for c in gaps["thin_communities"][:2]: - questions.append({ - "category": "thin_community", - "question": ( - f"Community '{c['name']}' has only " - f"{c['size']} member(s). Should it be " - f"merged with a neighbor?" - ), - "target": f"community:{c['community_id']}", - "priority": "low", - }) - - for h in gaps["untested_hotspots"][:2]: - questions.append({ - "category": "untested_hotspot", - "question": ( - f"'{h['name']}' has {h['degree']} " - f"connections but no test coverage. " - f"Is this a risk?" - ), - "target": h["qualified_name"], - "priority": "medium", - }) - - return questions diff --git a/code_review_graph/enrich 2.py b/code_review_graph/enrich 2.py deleted file mode 100644 index f95c334a..00000000 --- a/code_review_graph/enrich 2.py +++ /dev/null @@ -1,303 +0,0 @@ -"""PreToolUse search enrichment for Claude Code hooks. - -Intercepts Grep/Glob/Bash/Read tool calls and enriches them with -structural context from the code knowledge graph: callers, callees, -execution flows, community membership, and test coverage. -""" - -from __future__ import annotations - -import json -import logging -import os -import re -import sys -from pathlib import Path -from typing import Any - -logger = logging.getLogger(__name__) - -# Flags that consume the next token in grep/rg commands -_RG_FLAGS_WITH_VALUES = frozenset({ - "-e", "-f", "-m", "-A", "-B", "-C", "-g", "--glob", - "-t", "--type", "--include", "--exclude", "--max-count", - "--max-depth", "--max-filesize", "--color", "--colors", - "--context-separator", "--field-match-separator", - "--path-separator", "--replace", "--sort", "--sortr", -}) - - -def extract_pattern(tool_name: str, tool_input: dict[str, Any]) -> str | None: - """Extract a search pattern from a tool call's input. - - Returns None if no meaningful pattern can be extracted. - """ - if tool_name == "Grep": - return tool_input.get("pattern") - - if tool_name == "Glob": - raw = tool_input.get("pattern", "") - # Extract meaningful name from glob: "**/auth*.ts" -> "auth" - # Skip pure extension globs like "**/*.ts" - match = re.search(r"[*/]([a-zA-Z][a-zA-Z0-9_]{2,})", raw) - return match.group(1) if match else None - - if tool_name == "Bash": - cmd = tool_input.get("command", "") - if not re.search(r"\brg\b|\bgrep\b", cmd): - return None - tokens = cmd.split() - found_cmd = False - skip_next = False - for token in tokens: - if skip_next: - skip_next = False - continue - if not found_cmd: - if re.search(r"\brg$|\bgrep$", token): - found_cmd = True - continue - if token.startswith("-"): - if token in _RG_FLAGS_WITH_VALUES: - skip_next = True - continue - cleaned = token.strip("'\"") - return cleaned if len(cleaned) >= 3 else None - return None - - return None - - -def _make_relative(file_path: str, repo_root: str) -> str: - """Make a file path relative to repo_root for display.""" - try: - return str(Path(file_path).relative_to(repo_root)) - except ValueError: - return file_path - - -def _get_community_name(conn: Any, community_id: int) -> str: - """Fetch a community name by ID.""" - row = conn.execute( - "SELECT name FROM communities WHERE id = ?", (community_id,) - ).fetchone() - return row["name"] if row else "" - - -def _get_flow_names_for_node(conn: Any, node_id: int) -> list[str]: - """Fetch execution flow names that a node participates in (max 3).""" - rows = conn.execute( - "SELECT f.name FROM flow_memberships fm " - "JOIN flows f ON fm.flow_id = f.id " - "WHERE fm.node_id = ? LIMIT 3", - (node_id,), - ).fetchall() - return [r["name"] for r in rows] - - -def _format_node_context( - node: Any, - store: Any, - conn: Any, - repo_root: str, -) -> list[str]: - """Format a single node's structural context as plain text lines.""" - from .graph import GraphNode - assert isinstance(node, GraphNode) - - qn = node.qualified_name - loc = _make_relative(node.file_path, repo_root) - if node.line_start: - loc = f"{loc}:{node.line_start}" - - header = f"{node.name} ({loc})" - - # Community - if node.extra.get("community_id"): - cname = _get_community_name(conn, node.extra["community_id"]) - if cname: - header += f" [{cname}]" - else: - # Check via direct query - row = conn.execute( - "SELECT community_id FROM nodes WHERE id = ?", (node.id,) - ).fetchone() - if row and row["community_id"]: - cname = _get_community_name(conn, row["community_id"]) - if cname: - header += f" [{cname}]" - - lines = [header] - - # Callers (max 5, deduplicated) - callers: list[str] = [] - seen: set[str] = set() - for e in store.get_edges_by_target(qn): - if e.kind == "CALLS" and len(callers) < 5: - c = store.get_node(e.source_qualified) - if c and c.name not in seen: - seen.add(c.name) - callers.append(c.name) - if callers: - lines.append(f" Called by: {', '.join(callers)}") - - # Callees (max 5, deduplicated) - callees: list[str] = [] - seen.clear() - for e in store.get_edges_by_source(qn): - if e.kind == "CALLS" and len(callees) < 5: - c = store.get_node(e.target_qualified) - if c and c.name not in seen: - seen.add(c.name) - callees.append(c.name) - if callees: - lines.append(f" Calls: {', '.join(callees)}") - - # Execution flows - flow_names = _get_flow_names_for_node(conn, node.id) - if flow_names: - lines.append(f" Flows: {', '.join(flow_names)}") - - # Tests - tests: list[str] = [] - for e in store.get_edges_by_target(qn): - if e.kind == "TESTED_BY" and len(tests) < 3: - t = store.get_node(e.source_qualified) - if t: - tests.append(t.name) - if tests: - lines.append(f" Tests: {', '.join(tests)}") - - return lines - - -def enrich_search(pattern: str, repo_root: str) -> str: - """Search the graph for pattern and return enriched context.""" - from .graph import GraphStore - from .search import _fts_search - - db_path = Path(repo_root) / ".code-review-graph" / "graph.db" - if not db_path.exists(): - return "" - - store = GraphStore(db_path) - try: - conn = store._conn - - fts_results = _fts_search(conn, pattern, limit=8) - if not fts_results: - return "" - - all_lines: list[str] = [] - count = 0 - for node_id, _score in fts_results: - if count >= 5: - break - node = store.get_node_by_id(node_id) - if not node or node.is_test: - continue - node_lines = _format_node_context(node, store, conn, repo_root) - all_lines.extend(node_lines) - all_lines.append("") - count += 1 - - if not all_lines: - return "" - - header = f'[code-review-graph] {count} symbol(s) matching "{pattern}":\n' - return header + "\n".join(all_lines) - finally: - store.close() - - -def enrich_file_read(file_path: str, repo_root: str) -> str: - """Enrich a file read with structural context for functions in that file.""" - from .graph import GraphStore - - db_path = Path(repo_root) / ".code-review-graph" / "graph.db" - if not db_path.exists(): - return "" - - store = GraphStore(db_path) - try: - conn = store._conn - nodes = store.get_nodes_by_file(file_path) - if not nodes: - # Try with resolved path - try: - resolved = str(Path(file_path).resolve()) - nodes = store.get_nodes_by_file(resolved) - except (OSError, ValueError): - pass - if not nodes: - return "" - - # Filter to functions/classes/types (skip File nodes), limit to 10 - interesting = [ - n for n in nodes - if n.kind in ("Function", "Class", "Type", "Test") - ][:10] - - if not interesting: - return "" - - all_lines: list[str] = [] - for node in interesting: - node_lines = _format_node_context(node, store, conn, repo_root) - all_lines.extend(node_lines) - all_lines.append("") - - rel_path = _make_relative(file_path, repo_root) - header = ( - f"[code-review-graph] {len(interesting)} symbol(s) in {rel_path}:\n" - ) - return header + "\n".join(all_lines) - finally: - store.close() - - -def run_hook() -> None: - """Entry point for the enrich CLI subcommand. - - Reads Claude Code hook JSON from stdin, extracts the search pattern, - queries the graph, and outputs hookSpecificOutput JSON to stdout. - """ - try: - hook_input = json.load(sys.stdin) - except (json.JSONDecodeError, ValueError): - return - - tool_name = hook_input.get("tool_name", "") - tool_input = hook_input.get("tool_input", {}) - cwd = hook_input.get("cwd", os.getcwd()) - - # Find repo root by walking up from cwd - from .incremental import find_project_root - - repo_root = str(find_project_root(Path(cwd))) - db_path = Path(repo_root) / ".code-review-graph" / "graph.db" - if not db_path.exists(): - return - - # Dispatch - context = "" - if tool_name == "Read": - fp = tool_input.get("file_path", "") - if fp: - context = enrich_file_read(fp, repo_root) - else: - pattern = extract_pattern(tool_name, tool_input) - if not pattern or len(pattern) < 3: - return - context = enrich_search(pattern, repo_root) - - if not context: - return - - response = { - "hookSpecificOutput": { - "hookEventName": "PreToolUse", - "additionalContext": context, - } - } - json.dump(response, sys.stdout) diff --git a/code_review_graph/enrich 3.py b/code_review_graph/enrich 3.py deleted file mode 100644 index f95c334a..00000000 --- a/code_review_graph/enrich 3.py +++ /dev/null @@ -1,303 +0,0 @@ -"""PreToolUse search enrichment for Claude Code hooks. - -Intercepts Grep/Glob/Bash/Read tool calls and enriches them with -structural context from the code knowledge graph: callers, callees, -execution flows, community membership, and test coverage. -""" - -from __future__ import annotations - -import json -import logging -import os -import re -import sys -from pathlib import Path -from typing import Any - -logger = logging.getLogger(__name__) - -# Flags that consume the next token in grep/rg commands -_RG_FLAGS_WITH_VALUES = frozenset({ - "-e", "-f", "-m", "-A", "-B", "-C", "-g", "--glob", - "-t", "--type", "--include", "--exclude", "--max-count", - "--max-depth", "--max-filesize", "--color", "--colors", - "--context-separator", "--field-match-separator", - "--path-separator", "--replace", "--sort", "--sortr", -}) - - -def extract_pattern(tool_name: str, tool_input: dict[str, Any]) -> str | None: - """Extract a search pattern from a tool call's input. - - Returns None if no meaningful pattern can be extracted. - """ - if tool_name == "Grep": - return tool_input.get("pattern") - - if tool_name == "Glob": - raw = tool_input.get("pattern", "") - # Extract meaningful name from glob: "**/auth*.ts" -> "auth" - # Skip pure extension globs like "**/*.ts" - match = re.search(r"[*/]([a-zA-Z][a-zA-Z0-9_]{2,})", raw) - return match.group(1) if match else None - - if tool_name == "Bash": - cmd = tool_input.get("command", "") - if not re.search(r"\brg\b|\bgrep\b", cmd): - return None - tokens = cmd.split() - found_cmd = False - skip_next = False - for token in tokens: - if skip_next: - skip_next = False - continue - if not found_cmd: - if re.search(r"\brg$|\bgrep$", token): - found_cmd = True - continue - if token.startswith("-"): - if token in _RG_FLAGS_WITH_VALUES: - skip_next = True - continue - cleaned = token.strip("'\"") - return cleaned if len(cleaned) >= 3 else None - return None - - return None - - -def _make_relative(file_path: str, repo_root: str) -> str: - """Make a file path relative to repo_root for display.""" - try: - return str(Path(file_path).relative_to(repo_root)) - except ValueError: - return file_path - - -def _get_community_name(conn: Any, community_id: int) -> str: - """Fetch a community name by ID.""" - row = conn.execute( - "SELECT name FROM communities WHERE id = ?", (community_id,) - ).fetchone() - return row["name"] if row else "" - - -def _get_flow_names_for_node(conn: Any, node_id: int) -> list[str]: - """Fetch execution flow names that a node participates in (max 3).""" - rows = conn.execute( - "SELECT f.name FROM flow_memberships fm " - "JOIN flows f ON fm.flow_id = f.id " - "WHERE fm.node_id = ? LIMIT 3", - (node_id,), - ).fetchall() - return [r["name"] for r in rows] - - -def _format_node_context( - node: Any, - store: Any, - conn: Any, - repo_root: str, -) -> list[str]: - """Format a single node's structural context as plain text lines.""" - from .graph import GraphNode - assert isinstance(node, GraphNode) - - qn = node.qualified_name - loc = _make_relative(node.file_path, repo_root) - if node.line_start: - loc = f"{loc}:{node.line_start}" - - header = f"{node.name} ({loc})" - - # Community - if node.extra.get("community_id"): - cname = _get_community_name(conn, node.extra["community_id"]) - if cname: - header += f" [{cname}]" - else: - # Check via direct query - row = conn.execute( - "SELECT community_id FROM nodes WHERE id = ?", (node.id,) - ).fetchone() - if row and row["community_id"]: - cname = _get_community_name(conn, row["community_id"]) - if cname: - header += f" [{cname}]" - - lines = [header] - - # Callers (max 5, deduplicated) - callers: list[str] = [] - seen: set[str] = set() - for e in store.get_edges_by_target(qn): - if e.kind == "CALLS" and len(callers) < 5: - c = store.get_node(e.source_qualified) - if c and c.name not in seen: - seen.add(c.name) - callers.append(c.name) - if callers: - lines.append(f" Called by: {', '.join(callers)}") - - # Callees (max 5, deduplicated) - callees: list[str] = [] - seen.clear() - for e in store.get_edges_by_source(qn): - if e.kind == "CALLS" and len(callees) < 5: - c = store.get_node(e.target_qualified) - if c and c.name not in seen: - seen.add(c.name) - callees.append(c.name) - if callees: - lines.append(f" Calls: {', '.join(callees)}") - - # Execution flows - flow_names = _get_flow_names_for_node(conn, node.id) - if flow_names: - lines.append(f" Flows: {', '.join(flow_names)}") - - # Tests - tests: list[str] = [] - for e in store.get_edges_by_target(qn): - if e.kind == "TESTED_BY" and len(tests) < 3: - t = store.get_node(e.source_qualified) - if t: - tests.append(t.name) - if tests: - lines.append(f" Tests: {', '.join(tests)}") - - return lines - - -def enrich_search(pattern: str, repo_root: str) -> str: - """Search the graph for pattern and return enriched context.""" - from .graph import GraphStore - from .search import _fts_search - - db_path = Path(repo_root) / ".code-review-graph" / "graph.db" - if not db_path.exists(): - return "" - - store = GraphStore(db_path) - try: - conn = store._conn - - fts_results = _fts_search(conn, pattern, limit=8) - if not fts_results: - return "" - - all_lines: list[str] = [] - count = 0 - for node_id, _score in fts_results: - if count >= 5: - break - node = store.get_node_by_id(node_id) - if not node or node.is_test: - continue - node_lines = _format_node_context(node, store, conn, repo_root) - all_lines.extend(node_lines) - all_lines.append("") - count += 1 - - if not all_lines: - return "" - - header = f'[code-review-graph] {count} symbol(s) matching "{pattern}":\n' - return header + "\n".join(all_lines) - finally: - store.close() - - -def enrich_file_read(file_path: str, repo_root: str) -> str: - """Enrich a file read with structural context for functions in that file.""" - from .graph import GraphStore - - db_path = Path(repo_root) / ".code-review-graph" / "graph.db" - if not db_path.exists(): - return "" - - store = GraphStore(db_path) - try: - conn = store._conn - nodes = store.get_nodes_by_file(file_path) - if not nodes: - # Try with resolved path - try: - resolved = str(Path(file_path).resolve()) - nodes = store.get_nodes_by_file(resolved) - except (OSError, ValueError): - pass - if not nodes: - return "" - - # Filter to functions/classes/types (skip File nodes), limit to 10 - interesting = [ - n for n in nodes - if n.kind in ("Function", "Class", "Type", "Test") - ][:10] - - if not interesting: - return "" - - all_lines: list[str] = [] - for node in interesting: - node_lines = _format_node_context(node, store, conn, repo_root) - all_lines.extend(node_lines) - all_lines.append("") - - rel_path = _make_relative(file_path, repo_root) - header = ( - f"[code-review-graph] {len(interesting)} symbol(s) in {rel_path}:\n" - ) - return header + "\n".join(all_lines) - finally: - store.close() - - -def run_hook() -> None: - """Entry point for the enrich CLI subcommand. - - Reads Claude Code hook JSON from stdin, extracts the search pattern, - queries the graph, and outputs hookSpecificOutput JSON to stdout. - """ - try: - hook_input = json.load(sys.stdin) - except (json.JSONDecodeError, ValueError): - return - - tool_name = hook_input.get("tool_name", "") - tool_input = hook_input.get("tool_input", {}) - cwd = hook_input.get("cwd", os.getcwd()) - - # Find repo root by walking up from cwd - from .incremental import find_project_root - - repo_root = str(find_project_root(Path(cwd))) - db_path = Path(repo_root) / ".code-review-graph" / "graph.db" - if not db_path.exists(): - return - - # Dispatch - context = "" - if tool_name == "Read": - fp = tool_input.get("file_path", "") - if fp: - context = enrich_file_read(fp, repo_root) - else: - pattern = extract_pattern(tool_name, tool_input) - if not pattern or len(pattern) < 3: - return - context = enrich_search(pattern, repo_root) - - if not context: - return - - response = { - "hookSpecificOutput": { - "hookEventName": "PreToolUse", - "additionalContext": context, - } - } - json.dump(response, sys.stdout) diff --git a/code_review_graph/exports 2.py b/code_review_graph/exports 2.py deleted file mode 100644 index 92690997..00000000 --- a/code_review_graph/exports 2.py +++ /dev/null @@ -1,409 +0,0 @@ -"""Additional export formats: GraphML, Neo4j Cypher, Obsidian vault, SVG.""" - -from __future__ import annotations - -import html -import logging -import re -from pathlib import Path - -from .graph import GraphStore, _sanitize_name -from .visualization import export_graph_data - -logger = logging.getLogger(__name__) - - -# ------------------------------------------------------------------- -# GraphML export (for Gephi, yEd, Cytoscape) -# ------------------------------------------------------------------- - -def export_graphml(store: GraphStore, output_path: Path) -> Path: - """Export the graph as GraphML XML for Gephi/yEd/Cytoscape. - - Returns the path to the written file. - """ - data = export_graph_data(store) - nodes = data["nodes"] - edges = data["edges"] - - lines = [ - '', - '', - ' ', - ' ', - ' ', - ' ', - ' ', - ' ', - ] - - for n in nodes: - nid = html.escape(n["qualified_name"], quote=True) - lines.append(f' ') - lines.append(f' ' - f'{html.escape(n.get("kind", ""))}') - lines.append(f' ' - f'{html.escape(n.get("file_path", ""))}') - lang = n.get("language", "") or "" - lines.append(f' ' - f'{html.escape(lang)}') - cid = n.get("community_id") - if cid is not None: - lines.append(f' ' - f'{cid}') - lines.append(' ') - - for i, e in enumerate(edges): - src = html.escape(e["source"], quote=True) - tgt = html.escape(e["target"], quote=True) - kind = html.escape(e.get("kind", ""), quote=True) - lines.append( - f' ' - ) - lines.append(f' {kind}') - lines.append(' ') - - lines.append(' ') - lines.append('') - - output_path.write_text("\n".join(lines), encoding="utf-8") - logger.info("GraphML exported to %s (%d nodes, %d edges)", - output_path, len(nodes), len(edges)) - return output_path - - -# ------------------------------------------------------------------- -# Neo4j Cypher export -# ------------------------------------------------------------------- - -def export_neo4j_cypher(store: GraphStore, output_path: Path) -> Path: - """Export the graph as Neo4j Cypher CREATE statements. - - Returns the path to the written file. - """ - data = export_graph_data(store) - nodes = data["nodes"] - edges = data["edges"] - - lines = [ - "// Generated by code-review-graph", - "// Import: paste into Neo4j Browser or run via cypher-shell", - "", - ] - - # Create nodes - for n in nodes: - kind = n.get("kind", "Node") - props = { - "qualified_name": n["qualified_name"], - "name": n.get("name", ""), - "file_path": n.get("file_path", ""), - "language": n.get("language", "") or "", - } - cid = n.get("community_id") - if cid is not None: - props["community_id"] = cid - props_str = _cypher_props(props) - lines.append(f"CREATE (:{kind} {props_str});") - - lines.append("") - - # Create edges via MATCH - for e in edges: - kind = e.get("kind", "RELATES_TO") - src_qn = _cypher_escape(e["source"]) - tgt_qn = _cypher_escape(e["target"]) - lines.append( - f"MATCH (a {{qualified_name: '{src_qn}'}}), " - f"(b {{qualified_name: '{tgt_qn}'}}) " - f"CREATE (a)-[:{kind}]->(b);" - ) - - output_path.write_text("\n".join(lines), encoding="utf-8") - logger.info("Neo4j Cypher exported to %s (%d nodes, %d edges)", - output_path, len(nodes), len(edges)) - return output_path - - -def _cypher_escape(s: str) -> str: - """Escape a string for Cypher single-quoted literals.""" - return s.replace("\\", "\\\\").replace("'", "\\'") - - -def _cypher_props(d: dict) -> str: - """Format a dict as Cypher property map.""" - parts = [] - for k, v in d.items(): - if isinstance(v, str): - parts.append(f"{k}: '{_cypher_escape(v)}'") - elif isinstance(v, (int, float)): - parts.append(f"{k}: {v}") - elif isinstance(v, bool): - parts.append(f"{k}: {'true' if v else 'false'}") - return "{" + ", ".join(parts) + "}" - - -# ------------------------------------------------------------------- -# Obsidian vault export -# ------------------------------------------------------------------- - -def export_obsidian_vault( - store: GraphStore, output_dir: Path -) -> Path: - """Export the graph as an Obsidian vault with wikilinks. - - Creates: - - One .md per node with YAML frontmatter and [[wikilinks]] - - _COMMUNITY_*.md overview notes per community - - _INDEX.md with links to all nodes - - Returns the output directory path. - """ - data = export_graph_data(store) - nodes = data["nodes"] - edges = data["edges"] - communities = data.get("communities", []) - - output_dir.mkdir(parents=True, exist_ok=True) - - # Build adjacency for wikilinks - neighbors: dict[str, list[dict]] = {} - for e in edges: - src = e["source"] - tgt = e["target"] - kind = e.get("kind", "RELATES_TO") - neighbors.setdefault(src, []).append( - {"target": tgt, "kind": kind} - ) - neighbors.setdefault(tgt, []).append( - {"target": src, "kind": kind} - ) - - # Node name -> slug mapping - slugs: dict[str, str] = {} - for n in nodes: - slug = _obsidian_slug(n.get("name", n["qualified_name"])) - # Handle collisions - base_slug = slug - counter = 1 - while slug in slugs.values(): - slug = f"{base_slug}-{counter}" - counter += 1 - slugs[n["qualified_name"]] = slug - - # Write node pages - for n in nodes: - qn = n["qualified_name"] - slug = slugs[qn] - name = n.get("name", qn) - - frontmatter = { - "kind": n.get("kind", ""), - "file": n.get("file_path", ""), - "language": n.get("language", "") or "", - "community": n.get("community_id"), - "tags": [n.get("kind", "").lower()], - } - - lines = ["---"] - for k, v in frontmatter.items(): - if isinstance(v, list): - lines.append(f"{k}:") - for item in v: - lines.append(f" - {item}") - elif v is not None: - lines.append(f"{k}: {v}") - lines.append("---") - lines.append(f"# {_sanitize_name(name)}") - lines.append("") - lines.append(f"**Kind:** {n.get('kind', '')}") - lines.append(f"**File:** `{n.get('file_path', '')}`") - lines.append("") - - # Wikilinks to neighbors - nbrs = neighbors.get(qn, []) - if nbrs: - lines.append("## Connections") - lines.append("") - seen = set() - for nb in nbrs: - tgt_slug = slugs.get(nb["target"]) - if tgt_slug and tgt_slug not in seen: - seen.add(tgt_slug) - tgt_name = tgt_slug.replace("-", " ").title() - lines.append( - f"- {nb['kind']}: " - f"[[{tgt_slug}|{tgt_name}]]" - ) - - page_path = output_dir / f"{slug}.md" - page_path.write_text("\n".join(lines), encoding="utf-8") - - # Write community overview pages - community_map: dict[int, list[str]] = {} - for n in nodes: - cid = n.get("community_id") - if cid is not None: - community_map.setdefault(cid, []).append( - n["qualified_name"] - ) - - for c in communities: - cid = c.get("id") - cname = c.get("name", f"community-{cid}") - members = community_map.get(cid, []) - - lines = [f"# Community: {_sanitize_name(cname)}", ""] - lines.append(f"**Size:** {c.get('size', len(members))}") - lines.append(f"**Cohesion:** {c.get('cohesion', 0):.2f}") - lang = c.get("dominant_language", "") - if lang: - lines.append(f"**Language:** {lang}") - lines.append("") - lines.append("## Members") - lines.append("") - for qn in members[:50]: - slug = slugs.get(qn) - if slug: - lines.append(f"- [[{slug}]]") - - page_path = output_dir / f"_COMMUNITY_{cid}.md" - page_path.write_text("\n".join(lines), encoding="utf-8") - - # Write index - index_lines = ["# Code Graph Index", ""] - index_lines.append(f"**Nodes:** {len(nodes)}") - index_lines.append(f"**Edges:** {len(edges)}") - index_lines.append( - f"**Communities:** {len(communities)}" - ) - index_lines.append("") - index_lines.append("## All Nodes") - index_lines.append("") - for n in sorted(nodes, key=lambda x: x.get("name", "")): - slug = slugs.get(n["qualified_name"]) - if slug: - index_lines.append( - f"- [[{slug}]] ({n.get('kind', '')})" - ) - - (output_dir / "_INDEX.md").write_text( - "\n".join(index_lines), encoding="utf-8" - ) - - logger.info( - "Obsidian vault exported to %s (%d pages)", - output_dir, len(nodes) - ) - return output_dir - - -def _obsidian_slug(name: str) -> str: - """Convert a name to an Obsidian-friendly filename slug.""" - slug = re.sub(r"[^\w\s-]", "", name.lower()) - slug = re.sub(r"[\s_]+", "-", slug).strip("-") - return slug[:100] or "unnamed" - - -# ------------------------------------------------------------------- -# SVG export (matplotlib-based) -# ------------------------------------------------------------------- - -def export_svg(store: GraphStore, output_path: Path) -> Path: - """Export a static SVG graph visualization. - - Requires matplotlib (optional dependency). - Returns the path to the written file. - """ - try: - import matplotlib - matplotlib.use("Agg") - import matplotlib.pyplot as plt - except ImportError: - raise ImportError( - "matplotlib is required for SVG export. " - "Install with: pip install matplotlib" - ) - - import networkx as nx - - data = export_graph_data(store) - nodes_data = data["nodes"] - edges_data = data["edges"] - - nxg: nx.DiGraph = nx.DiGraph() # type: ignore[type-arg] - for n in nodes_data: - nxg.add_node( - n["qualified_name"], - label=n.get("name", ""), - kind=n.get("kind", ""), - ) - for e in edges_data: - if e["source"] in nxg and e["target"] in nxg: - nxg.add_edge(e["source"], e["target"]) - - if nxg.number_of_nodes() == 0: - raise ValueError("Graph is empty, nothing to export") - - # Color by kind - kind_colors = { - "File": "#6c757d", - "Class": "#0d6efd", - "Function": "#198754", - "Type": "#ffc107", - "Test": "#dc3545", - } - colors = [ - kind_colors.get( - nxg.nodes[n].get("kind", ""), "#adb5bd" - ) - for n in nxg.nodes() - ] - - fig, ax = plt.subplots(1, 1, figsize=(16, 12)) - pos = nx.spring_layout( - nxg, k=2 / (nxg.number_of_nodes() ** 0.5), - iterations=50, seed=42 - ) - - # Limit labels to avoid clutter - labels = {} - if nxg.number_of_nodes() <= 100: - labels = { - n: nxg.nodes[n].get("label", n.split("::")[-1]) - for n in nxg.nodes() - } - - nx.draw_networkx_nodes( - nxg, pos, ax=ax, node_color=colors, - node_size=30, alpha=0.8 - ) - nx.draw_networkx_edges( - nxg, pos, ax=ax, alpha=0.2, - arrows=True, arrowsize=5 - ) - if labels: - nx.draw_networkx_labels( - nxg, pos, labels=labels, ax=ax, - font_size=6 - ) - - ax.set_title("Code Review Graph", fontsize=14) - ax.axis("off") - - fig.savefig( - str(output_path), format="svg", - bbox_inches="tight", dpi=150 - ) - plt.close(fig) - - logger.info("SVG exported to %s (%d nodes)", - output_path, nxg.number_of_nodes()) - return output_path diff --git a/code_review_graph/exports 3.py b/code_review_graph/exports 3.py deleted file mode 100644 index 92690997..00000000 --- a/code_review_graph/exports 3.py +++ /dev/null @@ -1,409 +0,0 @@ -"""Additional export formats: GraphML, Neo4j Cypher, Obsidian vault, SVG.""" - -from __future__ import annotations - -import html -import logging -import re -from pathlib import Path - -from .graph import GraphStore, _sanitize_name -from .visualization import export_graph_data - -logger = logging.getLogger(__name__) - - -# ------------------------------------------------------------------- -# GraphML export (for Gephi, yEd, Cytoscape) -# ------------------------------------------------------------------- - -def export_graphml(store: GraphStore, output_path: Path) -> Path: - """Export the graph as GraphML XML for Gephi/yEd/Cytoscape. - - Returns the path to the written file. - """ - data = export_graph_data(store) - nodes = data["nodes"] - edges = data["edges"] - - lines = [ - '', - '', - ' ', - ' ', - ' ', - ' ', - ' ', - ' ', - ] - - for n in nodes: - nid = html.escape(n["qualified_name"], quote=True) - lines.append(f' ') - lines.append(f' ' - f'{html.escape(n.get("kind", ""))}') - lines.append(f' ' - f'{html.escape(n.get("file_path", ""))}') - lang = n.get("language", "") or "" - lines.append(f' ' - f'{html.escape(lang)}') - cid = n.get("community_id") - if cid is not None: - lines.append(f' ' - f'{cid}') - lines.append(' ') - - for i, e in enumerate(edges): - src = html.escape(e["source"], quote=True) - tgt = html.escape(e["target"], quote=True) - kind = html.escape(e.get("kind", ""), quote=True) - lines.append( - f' ' - ) - lines.append(f' {kind}') - lines.append(' ') - - lines.append(' ') - lines.append('') - - output_path.write_text("\n".join(lines), encoding="utf-8") - logger.info("GraphML exported to %s (%d nodes, %d edges)", - output_path, len(nodes), len(edges)) - return output_path - - -# ------------------------------------------------------------------- -# Neo4j Cypher export -# ------------------------------------------------------------------- - -def export_neo4j_cypher(store: GraphStore, output_path: Path) -> Path: - """Export the graph as Neo4j Cypher CREATE statements. - - Returns the path to the written file. - """ - data = export_graph_data(store) - nodes = data["nodes"] - edges = data["edges"] - - lines = [ - "// Generated by code-review-graph", - "// Import: paste into Neo4j Browser or run via cypher-shell", - "", - ] - - # Create nodes - for n in nodes: - kind = n.get("kind", "Node") - props = { - "qualified_name": n["qualified_name"], - "name": n.get("name", ""), - "file_path": n.get("file_path", ""), - "language": n.get("language", "") or "", - } - cid = n.get("community_id") - if cid is not None: - props["community_id"] = cid - props_str = _cypher_props(props) - lines.append(f"CREATE (:{kind} {props_str});") - - lines.append("") - - # Create edges via MATCH - for e in edges: - kind = e.get("kind", "RELATES_TO") - src_qn = _cypher_escape(e["source"]) - tgt_qn = _cypher_escape(e["target"]) - lines.append( - f"MATCH (a {{qualified_name: '{src_qn}'}}), " - f"(b {{qualified_name: '{tgt_qn}'}}) " - f"CREATE (a)-[:{kind}]->(b);" - ) - - output_path.write_text("\n".join(lines), encoding="utf-8") - logger.info("Neo4j Cypher exported to %s (%d nodes, %d edges)", - output_path, len(nodes), len(edges)) - return output_path - - -def _cypher_escape(s: str) -> str: - """Escape a string for Cypher single-quoted literals.""" - return s.replace("\\", "\\\\").replace("'", "\\'") - - -def _cypher_props(d: dict) -> str: - """Format a dict as Cypher property map.""" - parts = [] - for k, v in d.items(): - if isinstance(v, str): - parts.append(f"{k}: '{_cypher_escape(v)}'") - elif isinstance(v, (int, float)): - parts.append(f"{k}: {v}") - elif isinstance(v, bool): - parts.append(f"{k}: {'true' if v else 'false'}") - return "{" + ", ".join(parts) + "}" - - -# ------------------------------------------------------------------- -# Obsidian vault export -# ------------------------------------------------------------------- - -def export_obsidian_vault( - store: GraphStore, output_dir: Path -) -> Path: - """Export the graph as an Obsidian vault with wikilinks. - - Creates: - - One .md per node with YAML frontmatter and [[wikilinks]] - - _COMMUNITY_*.md overview notes per community - - _INDEX.md with links to all nodes - - Returns the output directory path. - """ - data = export_graph_data(store) - nodes = data["nodes"] - edges = data["edges"] - communities = data.get("communities", []) - - output_dir.mkdir(parents=True, exist_ok=True) - - # Build adjacency for wikilinks - neighbors: dict[str, list[dict]] = {} - for e in edges: - src = e["source"] - tgt = e["target"] - kind = e.get("kind", "RELATES_TO") - neighbors.setdefault(src, []).append( - {"target": tgt, "kind": kind} - ) - neighbors.setdefault(tgt, []).append( - {"target": src, "kind": kind} - ) - - # Node name -> slug mapping - slugs: dict[str, str] = {} - for n in nodes: - slug = _obsidian_slug(n.get("name", n["qualified_name"])) - # Handle collisions - base_slug = slug - counter = 1 - while slug in slugs.values(): - slug = f"{base_slug}-{counter}" - counter += 1 - slugs[n["qualified_name"]] = slug - - # Write node pages - for n in nodes: - qn = n["qualified_name"] - slug = slugs[qn] - name = n.get("name", qn) - - frontmatter = { - "kind": n.get("kind", ""), - "file": n.get("file_path", ""), - "language": n.get("language", "") or "", - "community": n.get("community_id"), - "tags": [n.get("kind", "").lower()], - } - - lines = ["---"] - for k, v in frontmatter.items(): - if isinstance(v, list): - lines.append(f"{k}:") - for item in v: - lines.append(f" - {item}") - elif v is not None: - lines.append(f"{k}: {v}") - lines.append("---") - lines.append(f"# {_sanitize_name(name)}") - lines.append("") - lines.append(f"**Kind:** {n.get('kind', '')}") - lines.append(f"**File:** `{n.get('file_path', '')}`") - lines.append("") - - # Wikilinks to neighbors - nbrs = neighbors.get(qn, []) - if nbrs: - lines.append("## Connections") - lines.append("") - seen = set() - for nb in nbrs: - tgt_slug = slugs.get(nb["target"]) - if tgt_slug and tgt_slug not in seen: - seen.add(tgt_slug) - tgt_name = tgt_slug.replace("-", " ").title() - lines.append( - f"- {nb['kind']}: " - f"[[{tgt_slug}|{tgt_name}]]" - ) - - page_path = output_dir / f"{slug}.md" - page_path.write_text("\n".join(lines), encoding="utf-8") - - # Write community overview pages - community_map: dict[int, list[str]] = {} - for n in nodes: - cid = n.get("community_id") - if cid is not None: - community_map.setdefault(cid, []).append( - n["qualified_name"] - ) - - for c in communities: - cid = c.get("id") - cname = c.get("name", f"community-{cid}") - members = community_map.get(cid, []) - - lines = [f"# Community: {_sanitize_name(cname)}", ""] - lines.append(f"**Size:** {c.get('size', len(members))}") - lines.append(f"**Cohesion:** {c.get('cohesion', 0):.2f}") - lang = c.get("dominant_language", "") - if lang: - lines.append(f"**Language:** {lang}") - lines.append("") - lines.append("## Members") - lines.append("") - for qn in members[:50]: - slug = slugs.get(qn) - if slug: - lines.append(f"- [[{slug}]]") - - page_path = output_dir / f"_COMMUNITY_{cid}.md" - page_path.write_text("\n".join(lines), encoding="utf-8") - - # Write index - index_lines = ["# Code Graph Index", ""] - index_lines.append(f"**Nodes:** {len(nodes)}") - index_lines.append(f"**Edges:** {len(edges)}") - index_lines.append( - f"**Communities:** {len(communities)}" - ) - index_lines.append("") - index_lines.append("## All Nodes") - index_lines.append("") - for n in sorted(nodes, key=lambda x: x.get("name", "")): - slug = slugs.get(n["qualified_name"]) - if slug: - index_lines.append( - f"- [[{slug}]] ({n.get('kind', '')})" - ) - - (output_dir / "_INDEX.md").write_text( - "\n".join(index_lines), encoding="utf-8" - ) - - logger.info( - "Obsidian vault exported to %s (%d pages)", - output_dir, len(nodes) - ) - return output_dir - - -def _obsidian_slug(name: str) -> str: - """Convert a name to an Obsidian-friendly filename slug.""" - slug = re.sub(r"[^\w\s-]", "", name.lower()) - slug = re.sub(r"[\s_]+", "-", slug).strip("-") - return slug[:100] or "unnamed" - - -# ------------------------------------------------------------------- -# SVG export (matplotlib-based) -# ------------------------------------------------------------------- - -def export_svg(store: GraphStore, output_path: Path) -> Path: - """Export a static SVG graph visualization. - - Requires matplotlib (optional dependency). - Returns the path to the written file. - """ - try: - import matplotlib - matplotlib.use("Agg") - import matplotlib.pyplot as plt - except ImportError: - raise ImportError( - "matplotlib is required for SVG export. " - "Install with: pip install matplotlib" - ) - - import networkx as nx - - data = export_graph_data(store) - nodes_data = data["nodes"] - edges_data = data["edges"] - - nxg: nx.DiGraph = nx.DiGraph() # type: ignore[type-arg] - for n in nodes_data: - nxg.add_node( - n["qualified_name"], - label=n.get("name", ""), - kind=n.get("kind", ""), - ) - for e in edges_data: - if e["source"] in nxg and e["target"] in nxg: - nxg.add_edge(e["source"], e["target"]) - - if nxg.number_of_nodes() == 0: - raise ValueError("Graph is empty, nothing to export") - - # Color by kind - kind_colors = { - "File": "#6c757d", - "Class": "#0d6efd", - "Function": "#198754", - "Type": "#ffc107", - "Test": "#dc3545", - } - colors = [ - kind_colors.get( - nxg.nodes[n].get("kind", ""), "#adb5bd" - ) - for n in nxg.nodes() - ] - - fig, ax = plt.subplots(1, 1, figsize=(16, 12)) - pos = nx.spring_layout( - nxg, k=2 / (nxg.number_of_nodes() ** 0.5), - iterations=50, seed=42 - ) - - # Limit labels to avoid clutter - labels = {} - if nxg.number_of_nodes() <= 100: - labels = { - n: nxg.nodes[n].get("label", n.split("::")[-1]) - for n in nxg.nodes() - } - - nx.draw_networkx_nodes( - nxg, pos, ax=ax, node_color=colors, - node_size=30, alpha=0.8 - ) - nx.draw_networkx_edges( - nxg, pos, ax=ax, alpha=0.2, - arrows=True, arrowsize=5 - ) - if labels: - nx.draw_networkx_labels( - nxg, pos, labels=labels, ax=ax, - font_size=6 - ) - - ax.set_title("Code Review Graph", fontsize=14) - ax.axis("off") - - fig.savefig( - str(output_path), format="svg", - bbox_inches="tight", dpi=150 - ) - plt.close(fig) - - logger.info("SVG exported to %s (%d nodes)", - output_path, nxg.number_of_nodes()) - return output_path diff --git a/code_review_graph/graph_diff 2.py b/code_review_graph/graph_diff 2.py deleted file mode 100644 index 0c9e0338..00000000 --- a/code_review_graph/graph_diff 2.py +++ /dev/null @@ -1,122 +0,0 @@ -"""Graph snapshot diffing -- compare graph state over time.""" - -from __future__ import annotations - -import json -import logging -from pathlib import Path -from typing import Any - -from .graph import GraphStore - -logger = logging.getLogger(__name__) - - -def take_snapshot(store: GraphStore) -> dict[str, Any]: - """Take a snapshot of the current graph state. - - Returns a dict with node and edge counts, qualified names, - and community assignments for later diffing. - """ - stats = store.get_stats() - nodes = store.get_all_nodes(exclude_files=False) - community_map = store.get_all_community_ids() - - return { - "node_count": stats.total_nodes, - "edge_count": stats.total_edges, - "nodes": { - n.qualified_name: { - "kind": n.kind, - "file": n.file_path, - "community_id": community_map.get( - n.qualified_name - ), - } - for n in nodes - }, - "edges": { - f"{e.source_qualified}->" - f"{e.target_qualified}:{e.kind}" - for e in store.get_all_edges() - }, - } - - -def save_snapshot(snapshot: dict, path: Path) -> None: - """Save a snapshot to a JSON file.""" - data = dict(snapshot) - if isinstance(data.get("edges"), set): - data["edges"] = sorted(data["edges"]) - path.write_text( - json.dumps(data, indent=2), encoding="utf-8" - ) - - -def load_snapshot(path: Path) -> dict: - """Load a snapshot from a JSON file.""" - data = json.loads(path.read_text(encoding="utf-8")) - if isinstance(data.get("edges"), list): - data["edges"] = set(data["edges"]) - return data - - -def diff_snapshots( - before: dict, after: dict, -) -> dict[str, Any]: - """Compare two graph snapshots. - - Returns: - Dict with new_nodes, removed_nodes, new_edges, - removed_edges, community_changes, and summary - statistics. - """ - before_nodes = set(before.get("nodes", {}).keys()) - after_nodes = set(after.get("nodes", {}).keys()) - before_edges = before.get("edges", set()) - after_edges = after.get("edges", set()) - - new_nodes = after_nodes - before_nodes - removed_nodes = before_nodes - after_nodes - new_edges = after_edges - before_edges - removed_edges = before_edges - after_edges - - # Community changes for nodes that exist in both - community_changes = [] - for qn in before_nodes & after_nodes: - before_cid = before["nodes"][qn].get( - "community_id" - ) - after_cid = after["nodes"][qn].get( - "community_id" - ) - if before_cid != after_cid: - community_changes.append({ - "node": qn, - "before_community": before_cid, - "after_community": after_cid, - }) - - return { - "new_nodes": [ - {"qualified_name": qn, **after["nodes"][qn]} - for qn in sorted(new_nodes) - ][:100], - "removed_nodes": sorted(removed_nodes)[:100], - "new_edges": sorted(new_edges)[:100], - "removed_edges": sorted(removed_edges)[:100], - "community_changes": community_changes[:50], - "summary": { - "nodes_added": len(new_nodes), - "nodes_removed": len(removed_nodes), - "edges_added": len(new_edges), - "edges_removed": len(removed_edges), - "community_moves": len(community_changes), - "before_total": before.get( - "node_count", 0 - ), - "after_total": after.get( - "node_count", 0 - ), - }, - } diff --git a/code_review_graph/jedi_resolver 2.py b/code_review_graph/jedi_resolver 2.py deleted file mode 100644 index 8ec007e5..00000000 --- a/code_review_graph/jedi_resolver 2.py +++ /dev/null @@ -1,303 +0,0 @@ -"""Post-build Jedi enrichment for Python call resolution. - -After tree-sitter parsing, many method calls on lowercase-receiver variables -are dropped (e.g. ``svc.authenticate()`` where ``svc = factory()``). Jedi -can resolve these by tracing return types across files. - -This module runs as a post-build step: it re-walks Python ASTs to find -dropped calls, uses ``jedi.Script.goto()`` to resolve them, and adds the -resulting CALLS edges to the graph database. -""" - -from __future__ import annotations - -import logging -import os -from pathlib import Path -from typing import Optional - -from .parser import CodeParser, EdgeInfo -from .parser import _is_test_file as _parser_is_test_file - -logger = logging.getLogger(__name__) - -_SELF_NAMES = frozenset({"self", "cls", "super"}) - - -def enrich_jedi_calls(store, repo_root: Path) -> dict: - """Resolve untracked Python method calls via Jedi. - - Walks Python files, finds ``receiver.method()`` calls that tree-sitter - dropped (lowercase receiver, not self/cls), resolves them with Jedi, - and inserts new CALLS edges. - - Returns stats dict with ``resolved`` count. - """ - try: - import jedi - except ImportError: - logger.info("Jedi not installed, skipping Python enrichment") - return {"skipped": True, "reason": "jedi not installed"} - - repo_root = Path(repo_root).resolve() - - # Get Python files from the graph — skip early if none - all_files = store.get_all_files() - py_files = [f for f in all_files if f.endswith(".py")] - - if not py_files: - return {"resolved": 0, "files": 0} - - # Scope the Jedi project to Python-only directories to avoid scanning - # non-Python files (e.g. node_modules, TS sources). This matters for - # polyglot monorepos where jedi.Project(path=repo_root) would scan - # thousands of irrelevant files during initialization. - py_dirs = sorted({str(Path(f).parent) for f in py_files}) - common_py_root = Path(os.path.commonpath(py_dirs)) if py_dirs else repo_root - if not str(common_py_root).startswith(str(repo_root)): - common_py_root = repo_root - project = jedi.Project( - path=str(common_py_root), - added_sys_path=[str(repo_root)], - smart_sys_path=False, - ) - - # Pre-parse all Python files to find which ones have pending method calls. - # This avoids expensive Jedi Script creation for files with nothing to resolve. - parser = CodeParser() - ts_parser = parser._get_parser("python") - if not ts_parser: - return {"resolved": 0, "files": 0} - - # Build set of method names that actually exist in project code. - # No point asking Jedi to resolve `logger.getLogger()` if no project - # file defines a function called `getLogger`. - project_func_names = { - r["name"] - for r in store._conn.execute( - "SELECT DISTINCT name FROM nodes WHERE kind IN ('Function', 'Test')" - ).fetchall() - } - - files_with_pending: list[tuple[str, bytes, list]] = [] - total_skipped = 0 - for file_path in py_files: - try: - source = Path(file_path).read_bytes() - except (OSError, PermissionError): - continue - tree = ts_parser.parse(source) - is_test = _parser_is_test_file(file_path) - pending = _find_untracked_method_calls(tree.root_node, is_test) - if pending: - # Only keep calls whose method name exists in project code - filtered = [p for p in pending if p[2] in project_func_names] - total_skipped += len(pending) - len(filtered) - if filtered: - files_with_pending.append((file_path, source, filtered)) - - if not files_with_pending: - return {"resolved": 0, "files": 0} - - logger.debug( - "Jedi: %d/%d Python files have pending calls (%d calls skipped — no project target)", - len(files_with_pending), len(py_files), total_skipped, - ) - - resolved_count = 0 - files_enriched = 0 - errors = 0 - - for file_path, source, pending in files_with_pending: - source_text = source.decode("utf-8", errors="replace") - - # Get existing CALLS edges for this file to skip duplicates - existing = set() - for edge in _get_file_call_edges(store, file_path): - existing.add((edge.source_qualified, edge.line)) - - # Get function nodes from DB for enclosing-function lookup - func_nodes = [ - n for n in store.get_nodes_by_file(file_path) - if n.kind in ("Function", "Test") - ] - - # Create Jedi script once per file - try: - script = jedi.Script(source_text, path=file_path, project=project) - except Exception as e: - logger.debug("Jedi failed to load %s: %s", file_path, e) - errors += 1 - continue - - file_resolved = 0 - for jedi_line, col, _method_name, _enclosing_name in pending: - # Find enclosing function qualified name - enclosing = _find_enclosing(func_nodes, jedi_line) - if not enclosing: - enclosing = file_path # module-level - - # Skip if we already have a CALLS edge from this source at this line - if (enclosing, jedi_line) in existing: - continue - - # Ask Jedi to resolve - try: - names = script.goto(jedi_line, col) - except Exception: # nosec B112 - Jedi may fail on malformed code - continue - - if not names: - continue - - name = names[0] - if not name.module_path: - continue - - module_path = Path(name.module_path).resolve() - - # Only emit edges for project-internal definitions - try: - module_path.relative_to(repo_root) - except ValueError: - continue - - # Build qualified target: file_path::Class.method or file_path::func - target_file = str(module_path) - parent = name.parent() - if parent and parent.type == "class": - target = f"{target_file}::{parent.name}.{name.name}" - else: - target = f"{target_file}::{name.name}" - - store.upsert_edge(EdgeInfo( - kind="CALLS", - source=enclosing, - target=target, - file_path=file_path, - line=jedi_line, - )) - existing.add((enclosing, jedi_line)) - file_resolved += 1 - - if file_resolved: - files_enriched += 1 - resolved_count += file_resolved - - if resolved_count: - store.commit() - logger.info( - "Jedi enrichment: resolved %d calls in %d files", - resolved_count, files_enriched, - ) - - return { - "resolved": resolved_count, - "files": files_enriched, - "errors": errors, - } - - -def _get_file_call_edges(store, file_path: str): - """Get all CALLS edges originating from a file.""" - conn = store._conn - rows = conn.execute( - "SELECT * FROM edges WHERE file_path = ? AND kind = 'CALLS'", - (file_path,), - ).fetchall() - from .graph import GraphEdge - return [ - GraphEdge( - id=r["id"], kind=r["kind"], - source_qualified=r["source_qualified"], - target_qualified=r["target_qualified"], - file_path=r["file_path"], line=r["line"], - extra={}, - ) - for r in rows - ] - - -def _find_enclosing(func_nodes, line: int) -> Optional[str]: - """Find the qualified name of the function enclosing a given line.""" - best = None - best_span = float("inf") - for node in func_nodes: - if node.line_start <= line <= node.line_end: - span = node.line_end - node.line_start - if span < best_span: - best = node.qualified_name - best_span = span - return best - - -def _find_untracked_method_calls(root, is_test_file: bool = False): - """Walk Python AST to find method calls the parser would have dropped. - - Returns list of (jedi_line, col, method_name, enclosing_func_name) tuples. - Jedi_line is 1-indexed, col is 0-indexed. - """ - results: list[tuple[int, int, str, Optional[str]]] = [] - _walk_calls(root, results, is_test_file, enclosing_func=None) - return results - - -def _walk_calls(node, results, is_test_file, enclosing_func): - """Recursively walk AST collecting dropped method calls.""" - # Track enclosing function scope - if node.type == "function_definition": - name = None - for child in node.children: - if child.type == "identifier": - name = child.text.decode("utf-8", errors="replace") - break - for child in node.children: - _walk_calls(child, results, is_test_file, name or enclosing_func) - return - - if node.type == "decorated_definition": - for child in node.children: - _walk_calls(child, results, is_test_file, enclosing_func) - return - - # Check for call expressions with attribute access - if node.type == "call": - first = node.children[0] if node.children else None - if first and first.type == "attribute": - _check_dropped_call(first, results, is_test_file, enclosing_func) - - for child in node.children: - _walk_calls(child, results, is_test_file, enclosing_func) - - -def _check_dropped_call(attr_node, results, is_test_file, enclosing_func): - """Check if an attribute-based call was dropped by the parser.""" - children = attr_node.children - if len(children) < 2: - return - - receiver = children[0] - # Only handle simple identifier receivers - if receiver.type != "identifier": - return - - receiver_text = receiver.text.decode("utf-8", errors="replace") - - # The parser keeps: self/cls/super calls and uppercase-receiver calls - # The parser keeps: calls handled by typed-var enrichment (but those are - # separate edges -- we check for duplicates via existing-edge set) - if receiver_text in _SELF_NAMES: - return - if receiver_text[:1].isupper(): - return - if is_test_file: - return # test files already track all calls - - # Find the method name identifier - method_node = children[-1] - if method_node.type != "identifier": - return - - row, col = method_node.start_point # 0-indexed - method_name = method_node.text.decode("utf-8", errors="replace") - results.append((row + 1, col, method_name, enclosing_func)) diff --git a/code_review_graph/main.py b/code_review_graph/main.py index 733336e1..bbcb3828 100644 --- a/code_review_graph/main.py +++ b/code_review_graph/main.py @@ -43,7 +43,6 @@ get_suggested_questions_func, get_surprising_connections_func, get_wiki_page_func, - traverse_graph_func, list_communities_func, list_flows, list_graph_stats, @@ -52,6 +51,7 @@ refactor_func, run_postprocess, semantic_search_nodes, + traverse_graph_func, ) # NOTE: Thread-safe for stdio MCP (single-threaded). If adding HTTP/SSE @@ -942,7 +942,22 @@ def _apply_tool_filter(tools: str | None = None) -> None: allowed = {t.strip() for t in raw.split(",") if t.strip()} if not allowed: return - registered = list(mcp._tool_manager._tools.keys()) + import asyncio as _asyncio + import concurrent.futures as _futures + + async def _get_names() -> list[str]: + return [t.name for t in await mcp.list_tools()] + + try: + # Inside a running event loop (e.g. pytest-asyncio) – use a thread + # to avoid the "loop already running" error from asyncio.run(). + _asyncio.get_running_loop() + with _futures.ThreadPoolExecutor(max_workers=1) as _pool: + registered = _pool.submit(_asyncio.run, _get_names()).result() + except RuntimeError: + # No running event loop; safe to use asyncio.run() directly. + registered = _asyncio.run(_get_names()) + for name in registered: if name not in allowed: mcp.remove_tool(name) diff --git a/code_review_graph/memory 2.py b/code_review_graph/memory 2.py deleted file mode 100644 index 81088eda..00000000 --- a/code_review_graph/memory 2.py +++ /dev/null @@ -1,142 +0,0 @@ -"""Memory/feedback loop -- persist Q&A results for graph enrichment.""" - -from __future__ import annotations - -import logging -import re -import time -from pathlib import Path -from typing import Any - -logger = logging.getLogger(__name__) - - -def save_result( - question: str, - answer: str, - nodes: list[str] | None = None, - result_type: str = "query", - memory_dir: Path | None = None, - repo_root: Path | None = None, -) -> Path: - """Save a Q&A result as markdown for re-ingestion. - - Args: - question: The question that was asked. - answer: The answer/result. - nodes: Related node qualified names. - result_type: Type of result (query, review, debug). - memory_dir: Directory to save to. Defaults to - /.code-review-graph/memory/ - repo_root: Repository root for default memory_dir. - - Returns: - Path to the saved file. - """ - if memory_dir is None: - if repo_root is None: - raise ValueError( - "Either memory_dir or repo_root required" - ) - memory_dir = ( - repo_root / ".code-review-graph" / "memory" - ) - - memory_dir.mkdir(parents=True, exist_ok=True) - - # Generate filename from question - slug = re.sub(r"[^\w\s-]", "", question.lower()) - slug = re.sub(r"[\s_]+", "-", slug).strip("-")[:60] - timestamp = int(time.time()) - filename = f"{slug}-{timestamp}.md" - - # Build markdown with YAML frontmatter - lines = [ - "---", - f"type: {result_type}", - f"timestamp: {timestamp}", - ] - if nodes: - lines.append("nodes:") - for n in nodes[:20]: - lines.append(f" - {n}") - lines.extend([ - "---", - "", - f"# {question}", - "", - answer, - ]) - - path = memory_dir / filename - path.write_text("\n".join(lines), encoding="utf-8") - logger.info("Saved result to %s", path) - return path - - -def list_memories( - memory_dir: Path | None = None, - repo_root: Path | None = None, -) -> list[dict[str, Any]]: - """List all saved memory files. - - Returns list of dicts with: path, question, type, timestamp. - """ - if memory_dir is None: - if repo_root is None: - return [] - memory_dir = ( - repo_root / ".code-review-graph" / "memory" - ) - - if not memory_dir.exists(): - return [] - - results = [] - for f in sorted(memory_dir.glob("*.md")): - try: - text = f.read_text(encoding="utf-8") - # Parse frontmatter - meta: dict[str, Any] = {"path": str(f)} - if text.startswith("---"): - parts = text.split("---", 2) - if len(parts) >= 3: - fm_lines = parts[1].strip().split("\n") - for line in fm_lines: - if ": " in line and not line.startswith(" "): - k, v = line.split(": ", 1) - meta[k.strip()] = v.strip() - # Extract question from first heading - for line in text.split("\n"): - if line.startswith("# "): - meta["question"] = line[2:].strip() - break - results.append(meta) - except OSError: - continue - - return results - - -def clear_memories( - memory_dir: Path | None = None, - repo_root: Path | None = None, -) -> int: - """Delete all memory files. Returns count deleted.""" - if memory_dir is None: - if repo_root is None: - return 0 - memory_dir = ( - repo_root / ".code-review-graph" / "memory" - ) - - if not memory_dir.exists(): - return 0 - - count = 0 - for f in memory_dir.glob("*.md"): - f.unlink() - count += 1 - - logger.info("Cleared %d memory files", count) - return count diff --git a/code_review_graph/memory 3.py b/code_review_graph/memory 3.py deleted file mode 100644 index 81088eda..00000000 --- a/code_review_graph/memory 3.py +++ /dev/null @@ -1,142 +0,0 @@ -"""Memory/feedback loop -- persist Q&A results for graph enrichment.""" - -from __future__ import annotations - -import logging -import re -import time -from pathlib import Path -from typing import Any - -logger = logging.getLogger(__name__) - - -def save_result( - question: str, - answer: str, - nodes: list[str] | None = None, - result_type: str = "query", - memory_dir: Path | None = None, - repo_root: Path | None = None, -) -> Path: - """Save a Q&A result as markdown for re-ingestion. - - Args: - question: The question that was asked. - answer: The answer/result. - nodes: Related node qualified names. - result_type: Type of result (query, review, debug). - memory_dir: Directory to save to. Defaults to - /.code-review-graph/memory/ - repo_root: Repository root for default memory_dir. - - Returns: - Path to the saved file. - """ - if memory_dir is None: - if repo_root is None: - raise ValueError( - "Either memory_dir or repo_root required" - ) - memory_dir = ( - repo_root / ".code-review-graph" / "memory" - ) - - memory_dir.mkdir(parents=True, exist_ok=True) - - # Generate filename from question - slug = re.sub(r"[^\w\s-]", "", question.lower()) - slug = re.sub(r"[\s_]+", "-", slug).strip("-")[:60] - timestamp = int(time.time()) - filename = f"{slug}-{timestamp}.md" - - # Build markdown with YAML frontmatter - lines = [ - "---", - f"type: {result_type}", - f"timestamp: {timestamp}", - ] - if nodes: - lines.append("nodes:") - for n in nodes[:20]: - lines.append(f" - {n}") - lines.extend([ - "---", - "", - f"# {question}", - "", - answer, - ]) - - path = memory_dir / filename - path.write_text("\n".join(lines), encoding="utf-8") - logger.info("Saved result to %s", path) - return path - - -def list_memories( - memory_dir: Path | None = None, - repo_root: Path | None = None, -) -> list[dict[str, Any]]: - """List all saved memory files. - - Returns list of dicts with: path, question, type, timestamp. - """ - if memory_dir is None: - if repo_root is None: - return [] - memory_dir = ( - repo_root / ".code-review-graph" / "memory" - ) - - if not memory_dir.exists(): - return [] - - results = [] - for f in sorted(memory_dir.glob("*.md")): - try: - text = f.read_text(encoding="utf-8") - # Parse frontmatter - meta: dict[str, Any] = {"path": str(f)} - if text.startswith("---"): - parts = text.split("---", 2) - if len(parts) >= 3: - fm_lines = parts[1].strip().split("\n") - for line in fm_lines: - if ": " in line and not line.startswith(" "): - k, v = line.split(": ", 1) - meta[k.strip()] = v.strip() - # Extract question from first heading - for line in text.split("\n"): - if line.startswith("# "): - meta["question"] = line[2:].strip() - break - results.append(meta) - except OSError: - continue - - return results - - -def clear_memories( - memory_dir: Path | None = None, - repo_root: Path | None = None, -) -> int: - """Delete all memory files. Returns count deleted.""" - if memory_dir is None: - if repo_root is None: - return 0 - memory_dir = ( - repo_root / ".code-review-graph" / "memory" - ) - - if not memory_dir.exists(): - return 0 - - count = 0 - for f in memory_dir.glob("*.md"): - f.unlink() - count += 1 - - logger.info("Cleared %d memory files", count) - return count diff --git a/code_review_graph/parser.py b/code_review_graph/parser.py index f681263a..a7ec77d7 100644 --- a/code_review_graph/parser.py +++ b/code_review_graph/parser.py @@ -639,8 +639,75 @@ def _get_parser(self, language: str): # type: ignore[arg-type] return None return self._parsers[language] + # Map shebang interpreter name → language label. + _SHEBANG_LANGUAGE: dict[str, str] = { + "bash": "bash", + "sh": "bash", + "python": "python", + "python2": "python", + "python3": "python", + "node": "javascript", + "nodejs": "javascript", + "ruby": "ruby", + "perl": "perl", + "php": "php", + "lua": "lua", + "Rscript": "r", + } + def detect_language(self, path: Path) -> Optional[str]: - return EXTENSION_TO_LANGUAGE.get(path.suffix.lower()) + lang = EXTENSION_TO_LANGUAGE.get(path.suffix.lower()) + if lang is not None: + return lang + # For extension-less files (or unknown extensions), try shebang. + if not path.suffix: + return self._detect_language_from_shebang(path) + return None + + def _detect_language_from_shebang(self, path: Path) -> Optional[str]: + """Read the first line of *path* and map the shebang interpreter to a language. + + Handles ``#!/bin/bash``, ``#!/usr/bin/env python3``, + ``#!/usr/bin/env -S node --flags``, and similar forms. + Binary/non-UTF-8 content is ignored safely. + """ + try: + with path.open("rb") as fh: + first_line = fh.readline(512) + except (OSError, PermissionError): + return None + + if not first_line.startswith(b"#!"): + return None + + try: + line = first_line[2:].decode("utf-8", errors="replace").strip() + except Exception: + return None + + # Split on whitespace; e.g. "/usr/bin/env -S python3 --flag" → tokens + tokens = line.split() + if not tokens: + return None + + interp_path = tokens[0] + + # /usr/bin/env [opts] interpreter [flags] + if interp_path.endswith("env"): + # Skip flags like "-S", "-u", etc. + interpreter = next( + (t for t in tokens[1:] if not t.startswith("-")), + None, + ) + else: + # /bin/bash, /usr/bin/python3 — basename only, strip version suffix + interpreter = interp_path.rsplit("/", 1)[-1] + + if not interpreter: + return None + + # Strip trailing version digits: "python3.11" → "python3", "node20" → "node" + return self._SHEBANG_LANGUAGE.get(interpreter) def parse_file(self, path: Path) -> tuple[list[NodeInfo], list[EdgeInfo]]: """Parse a single file and return extracted nodes and edges.""" @@ -672,9 +739,10 @@ def parse_bytes(self, path: Path, source: bytes) -> tuple[list[NodeInfo], list[E if language == "notebook": return self._parse_notebook(path, source) - # Databricks .py notebook exports - if language == "python" and source.startswith( - b"# Databricks notebook source\n", + # Databricks .py notebook exports (handle both LF and CRLF line endings) + if language == "python" and ( + source.startswith(b"# Databricks notebook source\n") + or source.startswith(b"# Databricks notebook source\r\n") ): return self._parse_databricks_py_notebook(path, source) @@ -2095,26 +2163,27 @@ def _extract_elixir_constructs( return True # ---- Everything else = a regular function/method call ---------- - # Emit a CALLS edge when we're inside a function (same rule as - # the generic _extract_calls path). - if enclosing_func: - # For dotted calls like `IO.puts(msg)`, prefer the dotted - # identifier; for bare calls use the first identifier. - call_name = ident - caller = self._qualify( - enclosing_func, file_path, enclosing_class, - ) - target = self._resolve_call_target( - call_name, file_path, language, - import_map or {}, defined_names or set(), - ) - edges.append(EdgeInfo( - kind="CALLS", - source=caller, - target=target, - file_path=file_path, - line=node.start_point[0] + 1, - )) + # Emit a CALLS edge from the enclosing function or, at module + # scope (enclosing_func is None), from the file path. This + # mirrors the generic _extract_calls path so that top-level + # Elixir scripts (IO.puts, etc.) are attributed to the file. + call_name = ident + caller = ( + self._qualify(enclosing_func, file_path, enclosing_class) + if enclosing_func + else file_path + ) + target = self._resolve_call_target( + call_name, file_path, language, + import_map or {}, defined_names or set(), + ) + edges.append(EdgeInfo( + kind="CALLS", + source=caller, + target=target, + file_path=file_path, + line=node.start_point[0] + 1, + )) # Recurse into arguments + do_block so nested calls are caught. for sub in node.children: if sub.type in ("arguments", "do_block"): @@ -2882,6 +2951,18 @@ def _extract_functions( line=child.start_point[0] + 1, )) + # Julia: function Base.show(...) → REFERENCES edge to "Base" + if language == "julia": + julia_module = self._get_julia_module_qualifier(child) + if julia_module: + edges.append(EdgeInfo( + kind="REFERENCES", + source=qualified, + target=julia_module, + file_path=file_path, + line=child.start_point[0] + 1, + )) + # Solidity: modifier invocations on functions -> CALLS edges if language == "solidity": for sub in child.children: @@ -3021,10 +3102,17 @@ def _extract_calls( ) return True - if call_name and enclosing_func: - caller = self._qualify( - enclosing_func, file_path, enclosing_class, - ) + if call_name: + # Emit CALLS even at module scope (enclosing_func is None). + # Module-scope calls use the file path as the source so that + # CLI entrypoints / notebook-helper functions are not flagged + # as dead code. See: TestModuleScopeCalls. + if enclosing_func: + caller = self._qualify( + enclosing_func, file_path, enclosing_class, + ) + else: + caller = file_path target = self._resolve_call_target( call_name, file_path, language, import_map or {}, defined_names or set(), @@ -3057,16 +3145,19 @@ def _extract_jsx_component_call( edges so caller/impact queries can cross the JSX boundary. Intrinsic DOM tags (``
``) are ignored. """ - if not enclosing_func: - return - target = self._resolve_jsx_component_target( child, language, file_path, import_map or {}, defined_names or set(), ) if not target: return - caller = self._qualify(enclosing_func, file_path, enclosing_class) + # At module scope (no enclosing function) attribute to the file, + # e.g. top-level in an index.tsx entry point. + caller = ( + self._qualify(enclosing_func, file_path, enclosing_class) + if enclosing_func + else file_path + ) edges.append(EdgeInfo( kind="CALLS", source=caller, @@ -3995,6 +4086,17 @@ def _get_name(self, node, language: str, kind: str) -> Optional[str]: for child in node.children: if child.type == "word": return child.text.decode("utf-8", errors="replace") + # Julia: function Base.show(io, d) — qualified name via field_expression. + # Extract only the final identifier ("show") as the function name; the + # module qualifier ("Base") is emitted as a REFERENCES edge separately. + if language == "julia" and node.type in ( + "function_definition", "short_function_definition", + ): + for child in node.children: + if child.type == "field_expression": + for sub in reversed(list(child.children)): + if sub.type == "identifier": + return sub.text.decode("utf-8", errors="replace") # Go methods: tree-sitter-go uses field_identifier for the name # (e.g. func (s *T) MethodName(...) { }). Must run before the generic # loop, which would match the result type's type_identifier (e.g. int64). @@ -4024,6 +4126,20 @@ def _get_name(self, node, language: str, kind: str) -> Optional[str]: return self._get_name(child, language, kind) return None + def _get_julia_module_qualifier(self, node) -> Optional[str]: + """Return the module name for a Julia qualified function definition. + + For ``function Base.show(io::IO, d::Dog)`` the ``function_definition`` + node contains a ``field_expression`` whose first ``identifier`` child + is the module name (``"Base"``). Returns ``None`` for plain functions. + """ + for child in node.children: + if child.type == "field_expression": + for sub in child.children: + if sub.type == "identifier": + return sub.text.decode("utf-8", errors="replace") + return None + def _get_go_receiver_type(self, node) -> Optional[str]: """Extract the receiver type from a Go method_declaration. @@ -4644,21 +4760,26 @@ def _handle_r_call( import_map, defined_names, ) - if enclosing_func: - call_name = self._get_call_name(node, language, source) - if call_name: - caller = self._qualify(enclosing_func, file_path, enclosing_class) - target = self._resolve_call_target( - call_name, file_path, language, - import_map or {}, defined_names or set(), - ) - edges.append(EdgeInfo( - kind="CALLS", - source=caller, - target=target, - file_path=file_path, - line=node.start_point[0] + 1, - )) + call_name = self._get_call_name(node, language, source) + if call_name: + # Emit CALLS even at module scope; R scripts overwhelmingly + # use top-level calls (worker(), plot(), etc.). + caller = ( + self._qualify(enclosing_func, file_path, enclosing_class) + if enclosing_func + else file_path + ) + target = self._resolve_call_target( + call_name, file_path, language, + import_map or {}, defined_names or set(), + ) + edges.append(EdgeInfo( + kind="CALLS", + source=caller, + target=target, + file_path=file_path, + line=node.start_point[0] + 1, + )) self._extract_from_tree( node, source, language, file_path, nodes, edges, diff --git a/code_review_graph/skills.py b/code_review_graph/skills.py index 66265534..e75ad93b 100644 --- a/code_review_graph/skills.py +++ b/code_review_graph/skills.py @@ -204,10 +204,18 @@ def _detect_serve_command() -> tuple[str, list[str]]: return (sys.executable, ["-m", "code_review_graph", "serve"]) -def _build_server_entry(plat: dict[str, Any], key: str = "") -> dict[str, Any]: - """Build the MCP server entry for a platform.""" +def _build_server_entry(plat: dict[str, Any], repo_root: Path, key: str = "") -> dict[str, Any]: + """Build the MCP server entry for a platform. + + Args: + plat: Platform metadata dict from PLATFORMS. + repo_root: Absolute path to the project root. Written as ``cwd`` so + that MCP clients always launch ``code-review-graph serve`` from the + correct directory and can locate ``.code-review-graph/graph.db``. + key: Platform key (e.g. ``"opencode"``), used for platform-specific tweaks. + """ command, args = _detect_serve_command() - entry: dict[str, Any] = {"command": command, "args": args} + entry: dict[str, Any] = {"command": command, "args": args, "cwd": str(repo_root)} if plat["needs_type"]: entry["type"] = "stdio" if key == "opencode": @@ -290,7 +298,7 @@ def install_platform_configs( for key, plat in platforms_to_install.items(): config_path: Path = plat["config_path"](repo_root) server_key = plat["key"] - server_entry = _build_server_entry(plat, key=key) + server_entry = _build_server_entry(plat, repo_root, key=key) if plat["format"] == "toml": changed = _merge_toml_mcp_server( diff --git a/code_review_graph/token_benchmark 2.py b/code_review_graph/token_benchmark 2.py deleted file mode 100644 index 7325f180..00000000 --- a/code_review_graph/token_benchmark 2.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Token reduction benchmark -- measures graph query efficiency vs naive file reading.""" - -from __future__ import annotations - -import logging -from pathlib import Path -from typing import Any - -from .graph import GraphStore -from .search import hybrid_search - -logger = logging.getLogger(__name__) - -# Sample questions for benchmarking -_SAMPLE_QUESTIONS = [ - "how does authentication work", - "what is the main entry point", - "how are database connections managed", - "what error handling patterns are used", - "how do tests verify core functionality", -] - - -def estimate_tokens(text: str) -> int: - """Rough token estimate: ~4 chars per token.""" - return max(1, len(text) // 4) - - -def compute_naive_tokens(repo_root: Path) -> int: - """Count tokens in all parseable source files.""" - total = 0 - exts = ( - ".py", ".js", ".ts", ".go", ".rs", ".java", - ".c", ".cpp", ".rb", ".php", ".swift", ".kt", - ) - for ext in exts: - for f in repo_root.rglob(f"*{ext}"): - try: - total += estimate_tokens( - f.read_text(errors="replace") - ) - except OSError: - continue - return total - - -def run_token_benchmark( - store: GraphStore, - repo_root: Path, - questions: list[str] | None = None, -) -> dict[str, Any]: - """Run token reduction benchmark. - - Compares naive full-corpus token cost vs graph query token - cost for a set of sample questions. - """ - if questions is None: - questions = _SAMPLE_QUESTIONS - - naive_total = compute_naive_tokens(repo_root) - - results = [] - for q in questions: - search_results = hybrid_search(store, q, limit=5) - # Simulate graph context: search results + neighbors - graph_tokens = 0 - for r in search_results: - graph_tokens += estimate_tokens(str(r)) - # Add approximate neighbor context - qn = r.get("qualified_name", "") - edges = store.get_edges_by_source(qn)[:5] - for e in edges: - graph_tokens += estimate_tokens(str(e)) - - if graph_tokens > 0: - ratio = naive_total / graph_tokens - else: - ratio = 0 - results.append({ - "question": q, - "naive_tokens": naive_total, - "graph_tokens": graph_tokens, - "reduction_ratio": round(ratio, 1), - }) - - if results: - total = sum( - r["reduction_ratio"] for r in results # type: ignore[misc] - ) - avg_ratio = float(total) / len(results) # type: ignore[arg-type] - else: - avg_ratio = 0.0 - - return { - "naive_corpus_tokens": naive_total, - "per_question": results, - "average_reduction_ratio": round(avg_ratio, 1), - "summary": ( - f"Graph queries use ~{avg_ratio:.0f}x fewer tokens " - f"than reading all source files" - ), - } diff --git a/code_review_graph/tools/review.py b/code_review_graph/tools/review.py index 772614f2..143b82b5 100644 --- a/code_review_graph/tools/review.py +++ b/code_review_graph/tools/review.py @@ -6,7 +6,7 @@ from pathlib import Path from typing import Any -from ..changes import analyze_changes, parse_diff_ranges, parse_git_diff_ranges +from ..changes import analyze_changes, parse_diff_ranges from ..flows import get_affected_flows as _get_affected_flows from ..graph import edge_to_dict, node_to_dict from ..hints import generate_hints, get_session diff --git a/tests/test_main.py b/tests/test_main.py index 61fbbf49..47bc94e9 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -222,16 +222,19 @@ class TestApplyToolFilter: """ @pytest.fixture(autouse=True) - def _restore_tools(self): + async def _restore_tools(self): """Snapshot registered tools before test, restore after. _apply_tool_filter calls ``mcp.remove_tool()`` which is - permanent. We restore by re-adding from the saved snapshot. + permanent. We restore by re-adding from the saved snapshot + using the public FastMCP API. """ - original = dict(crg_main.mcp._tool_manager._tools) + original_names = {t.name for t in await crg_main.mcp.list_tools()} yield - crg_main.mcp._tool_manager._tools.clear() - crg_main.mcp._tool_manager._tools.update(original) + current_names = {t.name for t in await crg_main.mcp.list_tools()} + for name in original_names: + if name not in current_names: + crg_main.mcp.add_tool(getattr(crg_main, name)) @pytest.fixture(autouse=True) def _clean_env(self, monkeypatch): @@ -241,9 +244,9 @@ def _clean_env(self, monkeypatch): @pytest.mark.asyncio async def test_no_filter_keeps_all_tools(self): """When neither --tools nor CRG_TOOLS is set, all tools remain.""" - before = set((await crg_main.mcp.get_tools()).keys()) + before = {t.name for t in await crg_main.mcp.list_tools()} crg_main._apply_tool_filter(None) - after = set((await crg_main.mcp.get_tools()).keys()) + after = {t.name for t in await crg_main.mcp.list_tools()} assert before == after @pytest.mark.asyncio @@ -251,7 +254,7 @@ async def test_filter_via_argument(self): """The ``tools`` argument keeps only the listed tools.""" keep = "query_graph_tool,semantic_search_nodes_tool" crg_main._apply_tool_filter(keep) - remaining = set((await crg_main.mcp.get_tools()).keys()) + remaining = {t.name for t in await crg_main.mcp.list_tools()} assert remaining == {"query_graph_tool", "semantic_search_nodes_tool"} @pytest.mark.asyncio @@ -259,7 +262,7 @@ async def test_filter_via_env_var(self, monkeypatch): """The ``CRG_TOOLS`` env var works as fallback.""" monkeypatch.setenv("CRG_TOOLS", "query_graph_tool") crg_main._apply_tool_filter(None) - remaining = set((await crg_main.mcp.get_tools()).keys()) + remaining = {t.name for t in await crg_main.mcp.list_tools()} assert remaining == {"query_graph_tool"} @pytest.mark.asyncio @@ -267,21 +270,21 @@ async def test_argument_takes_precedence_over_env(self, monkeypatch): """CLI --tools wins over CRG_TOOLS env var.""" monkeypatch.setenv("CRG_TOOLS", "list_repos_tool") crg_main._apply_tool_filter("query_graph_tool") - remaining = set((await crg_main.mcp.get_tools()).keys()) + remaining = {t.name for t in await crg_main.mcp.list_tools()} assert remaining == {"query_graph_tool"} @pytest.mark.asyncio async def test_empty_string_is_noop(self): """An empty string should not remove all tools.""" - before = set((await crg_main.mcp.get_tools()).keys()) + before = {t.name for t in await crg_main.mcp.list_tools()} crg_main._apply_tool_filter("") - after = set((await crg_main.mcp.get_tools()).keys()) + after = {t.name for t in await crg_main.mcp.list_tools()} assert before == after @pytest.mark.asyncio async def test_whitespace_handling(self): """Spaces around tool names are stripped.""" crg_main._apply_tool_filter(" query_graph_tool , semantic_search_nodes_tool ") - remaining = set((await crg_main.mcp.get_tools()).keys()) + remaining = {t.name for t in await crg_main.mcp.list_tools()} assert remaining == {"query_graph_tool", "semantic_search_nodes_tool"} diff --git a/tests/test_skills.py b/tests/test_skills.py index 873ed2f8..78b2319c 100644 --- a/tests/test_skills.py +++ b/tests/test_skills.py @@ -391,7 +391,10 @@ def test_install_codex_config(self, tmp_path): data = tomllib.loads(codex_config.read_text()) entry = data["mcp_servers"]["code-review-graph"] assert entry["type"] == "stdio" - assert "serve" in entry["args"] + expected_cmd, expected_args = _detect_serve_command() + assert entry["command"] == expected_cmd + assert entry["args"] == expected_args + assert entry["cwd"] == str(tmp_path) @_needs_tomllib def test_install_codex_preserves_existing_toml(self, tmp_path): @@ -417,6 +420,7 @@ def test_install_codex_preserves_existing_toml(self, tmp_path): assert data["mcp_servers"]["other"]["command"] == "other" expected_cmd, _ = _detect_serve_command() assert data["mcp_servers"]["code-review-graph"]["command"] == expected_cmd + assert data["mcp_servers"]["code-review-graph"]["cwd"] == str(tmp_path) def test_install_codex_no_duplicate(self, tmp_path): codex_config = tmp_path / ".codex" / "config.toml" @@ -459,7 +463,9 @@ def test_install_cursor_config(self, tmp_path): assert config_path.exists() data = json.loads(config_path.read_text()) assert "code-review-graph" in data["mcpServers"] - assert data["mcpServers"]["code-review-graph"]["type"] == "stdio" + entry = data["mcpServers"]["code-review-graph"] + assert entry["type"] == "stdio" + assert entry["cwd"] == str(tmp_path) def test_install_windsurf_config(self, tmp_path): windsurf_dir = tmp_path / ".codeium" / "windsurf" @@ -482,6 +488,7 @@ def test_install_windsurf_config(self, tmp_path): assert "type" not in entry expected_cmd, _ = _detect_serve_command() assert entry["command"] == expected_cmd + assert entry["cwd"] == str(tmp_path) def test_install_zed_config(self, tmp_path): zed_settings = tmp_path / "zed" / "settings.json" @@ -520,8 +527,10 @@ def test_install_continue_config(self, tmp_path): assert "Continue" in configured data = json.loads(config_path.read_text()) assert isinstance(data["mcpServers"], list) - assert data["mcpServers"][0]["name"] == "code-review-graph" - assert data["mcpServers"][0]["type"] == "stdio" + arr_entry = data["mcpServers"][0] + assert arr_entry["name"] == "code-review-graph" + assert arr_entry["type"] == "stdio" + assert arr_entry["cwd"] == str(tmp_path) def test_install_opencode_config(self, tmp_path): configured = install_platform_configs(tmp_path, target="opencode") @@ -531,6 +540,7 @@ def test_install_opencode_config(self, tmp_path): entry = data["mcpServers"]["code-review-graph"] assert entry["type"] == "stdio" assert entry["env"] == [] + assert entry["cwd"] == str(tmp_path) def test_install_qwen_config(self, tmp_path): """Qwen Code uses ~/.qwen/settings.json with mcpServers (see #83).""" @@ -551,6 +561,7 @@ def test_install_qwen_config(self, tmp_path): entry = data["mcpServers"]["code-review-graph"] assert entry["type"] == "stdio" assert entry["args"][-1] == "serve" + assert entry["cwd"] == str(tmp_path) def test_install_qwen_preserves_existing_servers(self, tmp_path): """Adding qwen should merge with, not clobber, existing mcpServers.""" @@ -662,9 +673,10 @@ def test_install_qoder_config(self, tmp_path): assert "mcpServers" in data assert "code-review-graph" in data["mcpServers"] assert data["mcpServers"]["code-review-graph"]["type"] == "stdio" - import shutil - expected_cmd = "uvx" if shutil.which("uvx") else "code-review-graph" + expected_cmd, expected_args = _detect_serve_command() assert data["mcpServers"]["code-review-graph"]["command"] == expected_cmd + assert data["mcpServers"]["code-review-graph"]["args"] == expected_args + assert data["mcpServers"]["code-review-graph"]["cwd"] == str(tmp_path) class TestCursorHooksConfig: @@ -854,6 +866,7 @@ def test_install_kiro_config(self, tmp_path): assert "code-review-graph" in data["mcpServers"] entry = data["mcpServers"]["code-review-graph"] assert entry["type"] == "stdio" + assert entry["cwd"] == str(tmp_path) def test_install_kiro_preserves_existing_servers(self, tmp_path): """Existing mcpServers entries are preserved when adding code-review-graph."""