"""Tools 2, 3, 5, 6, 9: query / search / stats helpers.""" from __future__ import annotations import logging from pathlib import Path from typing import Any from ..context_savings import attach_context_savings, estimate_file_tokens from ..embeddings import EmbeddingStore from ..graph import _sanitize_name, edge_to_dict, node_to_dict from ..hints import generate_hints, get_session from ..incremental import get_changed_files, get_db_path, get_staged_and_unstaged from ..search import hybrid_search from ._common import _BUILTIN_CALL_NAMES, _get_store, _resolve_graph_file_paths logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Tool 2: get_impact_radius # --------------------------------------------------------------------------- _QUERY_PATTERNS = { "callers_of": "Find all functions that call a given function", "callees_of": "Find all functions called by a given function", "imports_of": "Find all imports of a given file or module", "importers_of": "Find all files that import a given file or module", "children_of": "Find all nodes contained in a file or class", "tests_for": "Find all tests for a given function or class", "inheritors_of": "Find all classes that inherit from a given class", "file_summary": "Get a summary of all nodes in a file", } def get_impact_radius( changed_files: list[str] | None = None, max_depth: int = 2, max_results: int = 500, repo_root: str | None = None, base: str = "HEAD~1", detail_level: str = "standard", ) -> dict[str, Any]: """Analyze the blast radius of changed files. Args: changed_files: Explicit list of changed file paths (relative to repo root). If omitted, auto-detects from git diff. max_depth: How many hops to traverse in the graph (default: 2). max_results: Maximum impacted nodes to return (default: 500). repo_root: Repository root path. Auto-detected if omitted. base: Git ref for auto-detecting changes (default: HEAD~1). detail_level: "standard" (full output) or "minimal" (summary only). Returns: Changed nodes, impacted nodes, impacted files, connecting edges, plus ``truncated`` flag and ``total_impacted`` count. """ store, root = _get_store(repo_root) try: if changed_files is None: changed_files = get_changed_files(root, base) if not changed_files: changed_files = get_staged_and_unstaged(root) if not changed_files: return { "status": "ok", "summary": "No changed files detected.", "changed_nodes": [], "impacted_nodes": [], "impacted_files": [], "truncated": False, "total_impacted": 0, } # Resolve user-facing paths to the file paths stored in the graph. original_tokens = estimate_file_tokens(root, changed_files) abs_files = _resolve_graph_file_paths(store, root, changed_files) result = store.get_impact_radius( abs_files, max_depth=max_depth, max_nodes=max_results ) changed_dicts = [node_to_dict(n) for n in result["changed_nodes"]] impacted_dicts = [node_to_dict(n) for n in result["impacted_nodes"]] edge_dicts = [edge_to_dict(e) for e in result["edges"]] truncated = result["truncated"] total_impacted = result["total_impacted"] summary_parts = [ f"Blast radius for {len(changed_files)} changed file(s):", f" - {len(changed_dicts)} nodes directly changed", f" - {len(impacted_dicts)} nodes impacted (within {max_depth} hops)", f" - {len(result['impacted_files'])} additional files affected", ] if truncated: summary_parts.append( f" - Results truncated: showing {len(impacted_dicts)}" f" of {total_impacted} impacted nodes" ) if detail_level == "minimal": impacted_count = len(impacted_dicts) if impacted_count > 20: risk = "high" elif impacted_count > 5: risk = "medium" else: risk = "low" key_entities = [ n["name"] for n in impacted_dicts[:5] ] minimal_response = { "status": "ok", "summary": "\n".join(summary_parts), "risk": risk, "impacted_file_count": len(result["impacted_files"]), "key_entities": key_entities, "truncated": truncated, } attach_context_savings(minimal_response, original_tokens=original_tokens) return minimal_response response = { "status": "ok", "summary": "\n".join(summary_parts), "changed_files": changed_files, "changed_nodes": changed_dicts, "impacted_nodes": impacted_dicts, "impacted_files": result["impacted_files"], "edges": edge_dicts, "truncated": truncated, "total_impacted": total_impacted, } attach_context_savings(response, original_tokens=original_tokens) return response finally: store.close() # --------------------------------------------------------------------------- # Tool 3: query_graph # --------------------------------------------------------------------------- def query_graph( pattern: str, target: str, repo_root: str | None = None, detail_level: str = "standard", ) -> dict[str, Any]: """Run a predefined graph query. Args: pattern: Query pattern. One of: callers_of, callees_of, imports_of, importers_of, children_of, tests_for, inheritors_of, file_summary. target: The node name, qualified name, or file path to query about. repo_root: Repository root path. Auto-detected if omitted. detail_level: "standard" (full output) or "minimal" (summary only). Returns: Matching nodes and edges for the query. """ store, root = _get_store(repo_root) try: if pattern not in _QUERY_PATTERNS: return { "status": "error", "error": ( f"Unknown pattern '{pattern}'. " f"Available: {list(_QUERY_PATTERNS.keys())}" ), } results: list[dict] = [] edges_out: list[dict] = [] # For callers_of, skip common builtins early (bare names only) # "Who calls .map()?" returns hundreds of useless hits. # Qualified names (e.g. "utils.py::map") bypass this filter. if ( pattern == "callers_of" and target in _BUILTIN_CALL_NAMES and "::" not in target ): return { "status": "ok", "pattern": pattern, "target": target, "description": _QUERY_PATTERNS[pattern], "summary": ( f"'{target}' is a common builtin " "— callers_of skipped to avoid noise." ), "results": [], "edges": [], } # Resolve target - try as-is, then as absolute path, then search. # file_summary targets are paths, so skip broad node search. node = None if pattern != "file_summary": node = store.get_node(target) if not node: abs_target = str(root / target) node = store.get_node(abs_target) if not node: # Search by name candidates = store.search_nodes(target, limit=5) if len(candidates) == 1: node = candidates[0] target = node.qualified_name elif len(candidates) > 1: return { "status": "ambiguous", "summary": ( f"Multiple matches for '{target}'. " "Please use a qualified name." ), "candidates": [node_to_dict(c) for c in candidates], } if not node and pattern != "file_summary": return { "status": "not_found", "summary": f"No node found matching '{target}'.", } qn = node.qualified_name if node else target if pattern == "callers_of": seen_sources: set[str] = set() for e in store.get_edges_by_target(qn): if e.kind == "CALLS": if e.source_qualified not in seen_sources: seen_sources.add(e.source_qualified) caller = store.get_node(e.source_qualified) if caller: results.append(node_to_dict(caller)) edges_out.append(edge_to_dict(e)) # Fallback: CALLS edges store unqualified target names # (e.g. "generateTestCode") while qn is fully qualified # (e.g. "file.ts::generateTestCode"). Search by plain name too. if node: for e in store.search_edges_by_target_name(node.name): if e.source_qualified not in seen_sources: seen_sources.add(e.source_qualified) caller = store.get_node(e.source_qualified) if caller: results.append(node_to_dict(caller)) edges_out.append(edge_to_dict(e)) elif pattern == "callees_of": seen_targets: set[str] = set() for e in store.get_edges_by_source(qn): if e.kind == "CALLS": if e.target_qualified not in seen_targets: seen_targets.add(e.target_qualified) callee = store.get_node(e.target_qualified) if callee: results.append(node_to_dict(callee)) elif "::" not in e.target_qualified: results.append({ "kind": "Function", "name": e.target_qualified, "qualified_name": e.target_qualified, }) edges_out.append(edge_to_dict(e)) elif pattern == "imports_of": for e in store.get_edges_by_source(qn): if e.kind == "IMPORTS_FROM": results.append({"import_target": e.target_qualified}) edges_out.append(edge_to_dict(e)) elif pattern == "importers_of": # Find edges where target matches this file. # Use resolve() to canonicalize the path, matching how # _resolve_module_to_file stores edge targets. abs_target = ( str((root / target).resolve()) if node is None else node.file_path ) for e in store.get_edges_by_target(abs_target): if e.kind == "IMPORTS_FROM": results.append({ "importer": e.source_qualified, "file": e.file_path, }) edges_out.append(edge_to_dict(e)) elif pattern == "children_of": for e in store.get_edges_by_source(qn): if e.kind == "CONTAINS": child = store.get_node(e.target_qualified) if child: results.append(node_to_dict(child)) elif pattern == "tests_for": for e in store.get_edges_by_target(qn): if e.kind == "TESTED_BY": test = store.get_node(e.source_qualified) if test: results.append(node_to_dict(test)) # Also search by naming convention name = node.name if node else target test_nodes = store.search_nodes(f"test_{name}", limit=10) test_nodes += store.search_nodes(f"Test{name}", limit=10) seen = {r.get("qualified_name") for r in results} for t in test_nodes: if t.qualified_name not in seen and t.is_test: results.append(node_to_dict(t)) elif pattern == "inheritors_of": for e in store.get_edges_by_target(qn): if e.kind in ("INHERITS", "IMPLEMENTS"): child = store.get_node(e.source_qualified) if child: results.append(node_to_dict(child)) edges_out.append(edge_to_dict(e)) # Fallback: INHERITS/IMPLEMENTS edges store unqualified base names # (e.g. "Animal") while qn is fully qualified # (e.g. "sample.dart::Animal"). Search by plain name too. See: #87 if not results and node: for kind in ("INHERITS", "IMPLEMENTS"): for e in store.search_edges_by_target_name(node.name, kind=kind): child = store.get_node(e.source_qualified) if child: results.append(node_to_dict(child)) edges_out.append(edge_to_dict(e)) elif pattern == "file_summary": graph_paths = _resolve_graph_file_paths(store, root, [target]) for graph_path in graph_paths: for n in store.get_nodes_by_file(graph_path): results.append(node_to_dict(n)) summary = ( f"Found {len(results)} result(s) " f"for {pattern}('{target}')" ) if detail_level == "minimal": minimal_results = [ { k: r[k] for k in ("name", "kind", "file_path") if k in r } for r in results[:5] ] return { "status": "ok", "pattern": pattern, "target": target, "description": _QUERY_PATTERNS[pattern], "summary": summary, "result_count": len(results), "results": minimal_results, } return { "status": "ok", "pattern": pattern, "target": target, "description": _QUERY_PATTERNS[pattern], "summary": summary, "results": results, "edges": edges_out, } finally: store.close() # --------------------------------------------------------------------------- # Tool 5: semantic_search_nodes # --------------------------------------------------------------------------- def semantic_search_nodes( query: str, kind: str | None = None, limit: int = 20, repo_root: str | None = None, context_files: list[str] | None = None, model: str | None = None, provider: str | None = None, detail_level: str = "standard", ) -> dict[str, Any]: """Search for nodes by name, keyword, or semantic similarity. Uses hybrid search (FTS5 BM25 + vector embeddings merged via Reciprocal Rank Fusion) as the primary search path, with graceful fallback to keyword matching. Args: query: Search string to match against node names and qualified names. kind: Optional filter by node kind (File, Class, Function, Type, Test). limit: Maximum results to return (default: 20). repo_root: Repository root path. Auto-detected if omitted. context_files: Optional list of file paths. Nodes in these files receive a relevance boost. detail_level: "standard" (full output) or "minimal" (summary only). Returns: Ranked list of matching nodes. """ store, root = _get_store(repo_root) try: results = hybrid_search( store, query, kind=kind, limit=limit, context_files=context_files, model=model, provider=provider, ) search_mode = "hybrid" if not results: search_mode = "keyword" summary = f"Found {len(results)} node(s) matching '{query}'" + ( f" (kind={kind})" if kind else "" ) if detail_level == "minimal": minimal_results = [ { k: r[k] for k in ("name", "kind", "file_path", "score") if k in r } for r in results[:5] ] return { "status": "ok", "query": query, "search_mode": search_mode, "summary": summary, "results": minimal_results, } result: dict[str, object] = { "status": "ok", "query": query, "search_mode": search_mode, "summary": summary, "results": results, } result["_hints"] = generate_hints( "semantic_search_nodes", result, get_session() ) return result finally: store.close() # --------------------------------------------------------------------------- # Tool 6: list_graph_stats # --------------------------------------------------------------------------- def list_graph_stats(repo_root: str | None = None) -> dict[str, Any]: """Get aggregate statistics about the knowledge graph. Args: repo_root: Repository root path. Auto-detected if omitted. Returns: Total nodes, edges, breakdown by kind, languages, and last update time. """ store, root = _get_store(repo_root) try: stats = store.get_stats() summary_parts = [ f"Graph statistics for {root.name}:", f" Files: {stats.files_count}", f" Total nodes: {stats.total_nodes}", f" Total edges: {stats.total_edges}", f" Languages: {', '.join(stats.languages) if stats.languages else 'none'}", f" Last updated: {stats.last_updated or 'never'}", "", "Nodes by kind:", ] for kind, count in sorted(stats.nodes_by_kind.items()): summary_parts.append(f" {kind}: {count}") summary_parts.append("") summary_parts.append("Edges by kind:") for kind, count in sorted(stats.edges_by_kind.items()): summary_parts.append(f" {kind}: {count}") # Add embedding info if available emb_store = EmbeddingStore(get_db_path(root)) try: emb_count = emb_store.count() summary_parts.append("") summary_parts.append(f"Embeddings: {emb_count} nodes embedded") if not emb_store.available: summary_parts.append( " (install sentence-transformers for semantic search)" ) finally: emb_store.close() return { "status": "ok", "summary": "\n".join(summary_parts), "total_nodes": stats.total_nodes, "total_edges": stats.total_edges, "nodes_by_kind": stats.nodes_by_kind, "edges_by_kind": stats.edges_by_kind, "languages": stats.languages, "files_count": stats.files_count, "last_updated": stats.last_updated, "embeddings_count": emb_count, } finally: store.close() # --------------------------------------------------------------------------- # Tool 9: find_large_functions # --------------------------------------------------------------------------- def find_large_functions( min_lines: int = 50, kind: str | None = None, file_path_pattern: str | None = None, limit: int = 50, repo_root: str | None = None, ) -> dict[str, Any]: """Find functions, classes, or files exceeding a line-count threshold. Useful for identifying decomposition targets, code-quality audits, and enforcing size limits during code review. Args: min_lines: Minimum line count to flag (default: 50). kind: Filter by node kind: Function, Class, File, or Test. file_path_pattern: Filter by file path substring (e.g. "components/"). limit: Maximum results (default: 50). repo_root: Repository root path. Auto-detected if omitted. Returns: Oversized nodes with line counts, ordered largest first. """ store, root = _get_store(repo_root) try: nodes = store.get_nodes_by_size( min_lines=min_lines, kind=kind, file_path_pattern=file_path_pattern, limit=limit, ) results = [] for n in nodes: d = node_to_dict(n) d["line_count"] = ( (n.line_end - n.line_start + 1) if n.line_start and n.line_end else 0 ) # Make file_path relative for readability try: d["relative_path"] = str(Path(n.file_path).relative_to(root)) except ValueError: d["relative_path"] = n.file_path results.append(d) summary_parts = [ f"Found {len(results)} node(s) with >= {min_lines} lines" + (f" (kind={kind})" if kind else "") + (f" matching '{file_path_pattern}'" if file_path_pattern else "") + ":", ] for r in results[:10]: summary_parts.append( f" {r['line_count']:>4} lines | {r['kind']:>8} | " f"{r['name']} ({r['relative_path']}:{r['line_start']})" ) if len(results) > 10: summary_parts.append(f" ... and {len(results) - 10} more") return { "status": "ok", "summary": "\n".join(summary_parts), "total_found": len(results), "min_lines": min_lines, "results": results, } finally: store.close() # ------------------------------------------------------------------- # traverse_graph: free-form BFS / DFS traversal # ------------------------------------------------------------------- def traverse_graph_func( query: str, mode: str = "bfs", depth: int = 3, token_budget: int = 2000, repo_root: str | None = None, ) -> dict[str, Any]: """BFS/DFS traversal from best-matching node. Args: query: Search string to find the starting node. mode: "bfs" (breadth-first) or "dfs" (depth-first). depth: Max traversal depth (1-6). Default: 3. token_budget: Approximate token limit for results. repo_root: Repository root path. """ store, root = _get_store(repo_root) try: results = hybrid_search(store, query, limit=1) if not results: return { "error": f"No node matching '{query}'", "nodes": [], } start_qn = results[0]["qualified_name"] depth = max(1, min(depth, 6)) # BFS / DFS traversal visited: dict[str, int] = {} # qn -> depth queue: list[tuple[str, int]] = [ (start_qn, 0), ] traversal: list[dict] = [] approx_tokens = 0 while queue: if mode == "bfs": current_qn, cur_depth = queue.pop(0) else: current_qn, cur_depth = queue.pop() if current_qn in visited: continue if cur_depth > depth: continue visited[current_qn] = cur_depth node = store.get_node(current_qn) if not node: continue entry = { "name": _sanitize_name(node.name), "qualified_name": node.qualified_name, "kind": node.kind, "file": node.file_path, "depth": cur_depth, } approx_tokens += len(str(entry)) // 4 if approx_tokens > token_budget: break traversal.append(entry) # Get neighbours out_edges = store.get_edges_by_source( current_qn ) in_edges = store.get_edges_by_target( current_qn ) for e in out_edges: tgt = e.target_qualified if tgt not in visited: queue.append((tgt, cur_depth + 1)) for e in in_edges: src = e.source_qualified if src not in visited: queue.append((src, cur_depth + 1)) return { "start_node": start_qn, "mode": mode, "max_depth": depth, "nodes_visited": len(traversal), "traversal": traversal, "truncated": approx_tokens > token_budget, "next_tool_suggestions": [ "query_graph callers_of" " -- focused relationship query", "get_impact_radius" " -- blast radius analysis", ], } finally: store.close()