# src/codegraphcontext/tools/code_finder.py

from __future__ import annotations

import logging

from collections import Counter

from typing import Any, Dict, List, Literal, Optional, TYPE_CHECKING

from pathlib import Path

if TYPE_CHECKING:

from ..core.database import DatabaseManager

from ..utils.path_ignore import cypher_path_not_under_ignore_dirs

logger = logging.getLogger(__name__)

_MAX_TRAVERSAL_DEPTH = 20

def _sanitize_depth(depth, default: int = 3) -> int:

"""Coerce and clamp a traversal depth before interpolating it into Cypher.

The depth value ends up inside the query string (``[:CALLS*1..N]``), so it

must be a plain bounded integer to prevent Cypher injection.

"""

try:

depth = int(depth)

except (TypeError, ValueError):

return default

return max(1, min(depth, _MAX_TRAVERSAL_DEPTH))

def _levenshtein_distance(a: str, b: str) -> int:

"""Levenshtein distance for short identifiers (typo-tolerant name search)."""

if len(a) < len(b):

return _levenshtein_distance(b, a)

if not b:

return len(a)

prev = list(range(len(b) + 1))

for i, c1 in enumerate(a):

curr = [i + 1]

for j, c2 in enumerate(b):

curr.append(min(prev[j + 1] + 1, curr[j] + 1, prev[j] + (c1 != c2)))

prev = curr

return prev[-1]

def _normalize_identifier(s: str) -> str:

"""Lowercase and strip separator chars so camelCase / snake_case / spaces

all compare on equal footing.

Examples::

_normalize_identifier('myFunction') -> 'myfunction'

_normalize_identifier('my_function') -> 'myfunction'

_normalize_identifier('my function') -> 'myfunction'

_normalize_identifier('MyFunc tion') -> 'myfunction'

"""

return s.lower().replace('_', '').replace(' ', '')

def summarize_kotlin_call_ambiguity(

rows: List[Dict[str, Any]],

limit: int = 20,

) -> Dict[str, Any]:

"""Summarize multi-target Kotlin function CALLS edges by callsite/name group."""

groups: Dict[tuple, Dict[str, Any]] = {}

for row in rows:

key = (

row.get("caller_path"),

row.get("caller_line"),

row.get("caller_end_line"),

row.get("call_line"),

row.get("full_call_name"),

row.get("target_name"),

)

target = (

row.get("target_path"),

row.get("target_line"),

row.get("target_context"),

)

group = groups.setdefault(

key,

{

"caller_name": row.get("caller_name"),

"caller_path": row.get("caller_path"),

"caller_line": row.get("caller_line"),

"caller_end_line": row.get("caller_end_line"),

"call_line": row.get("call_line"),

"full_call_name": row.get("full_call_name"),

"target_name": row.get("target_name"),

"args": row.get("args"),

"targets": set(),

},

)

group["targets"].add(target)

ambiguous_groups = [

{

**{k: v for k, v in group.items() if k != "targets"},

"targets": [

{

"path": target_path,

"line_number": target_line,

"context": target_context,

}

for target_path, target_line, target_context in sorted(

group["targets"],

key=lambda target: (

str(target[0] or ""),

target[1] or 0,

str(target[2] or ""),

),

)

],

"target_count": len(group["targets"]),

}

for group in groups.values()

if len(group["targets"]) > 1

]

ambiguous_groups.sort(

key=lambda group: (

-group["target_count"],

str(group.get("caller_path") or ""),

group.get("call_line") or 0,

str(group.get("full_call_name") or ""),

)

)

top_names = Counter(

group.get("target_name")

for group in ambiguous_groups

if group.get("target_name")

)

return {

"kotlin_fn_to_fn_edges": len(rows),

"ambiguous_groups": len(ambiguous_groups),

"ambiguous_edges": sum(group["target_count"] for group in ambiguous_groups),

"top_names": [

{"name": name, "groups": count}

for name, count in top_names.most_common(limit)

],

"examples": ambiguous_groups[:limit],

}

class CodeFinder:

"""Module for finding relevant code snippets and analyzing relationships."""

def __init__(self, db_manager: DatabaseManager):

self.db_manager = db_manager

self.driver = self.db_manager.get_driver()

self._lacks_native_fulltext = getattr(db_manager, 'get_backend_type', lambda: 'neo4j')() != 'neo4j'

def audit_kotlin_call_ambiguity(

self,

repo_path: Optional[str] = None,

limit: int = 20,

) -> Dict[str, Any]:

"""Audit Kotlin function-to-function CALLS edges for multi-target callsites."""

repo_path = Path(repo_path).resolve().as_posix() if repo_path else None

repo_filter = "AND a.path STARTS WITH $repo_path" if repo_path else ""

query = f"""

MATCH (a:Function)-[r:CALLS]->(b:Function)

WHERE a.path ENDS WITH '.kt'

AND b.path ENDS WITH '.kt'

{repo_filter}

RETURN a.name as caller_name,

a.path as caller_path,

a.line_number as caller_line,

a.end_line as caller_end_line,

r.line_number as call_line,

r.full_call_name as full_call_name,

b.name as target_name,

b.path as target_path,

b.line_number as target_line,

b.context as target_context,

r.args as args

"""

with self.driver.session() as session:

rows = session.run(query, repo_path=repo_path).data()

return summarize_kotlin_call_ambiguity(rows, limit=limit)

def format_query(self, find_by: Literal["Class", "Function"], fuzzy_search:bool, repo_path: Optional[str] = None) -> str:

"""Format the search query based on the search type and fuzzy search settings."""

repo_filter = "AND node.path STARTS WITH $repo_path" if repo_path else ""

if self._lacks_native_fulltext:

# FalkorDB does not support CALL db.idx.fulltext.queryNodes.

# Fall back to a pure Cypher CONTAINS/toLower match on node name.

name_filter = "toLower(node.name) CONTAINS toLower($search_term)"

return f"""

MATCH (node:{find_by})

WHERE {name_filter} {repo_filter}

RETURN node.name as name, node.path as path, node.line_number as line_number,

node.source as source, node.docstring as docstring, node.is_dependency as is_dependency

ORDER BY node.is_dependency ASC, node.name

LIMIT 20

"""

return f"""

CALL db.index.fulltext.queryNodes("code_search_index", $search_term) YIELD node, score

WITH node, score

WHERE node:{find_by} {'AND node.name CONTAINS $search_term' if not fuzzy_search else ''} {repo_filter}

RETURN node.name as name, node.path as path, node.line_number as line_number,

node.source as source, node.docstring as docstring, node.is_dependency as is_dependency

ORDER BY score DESC

LIMIT 20

"""

def _find_by_name_fuzzy_portable(

self,

label: Literal["Function", "Class"],

search_term: str,

edit_distance: int,

repo_path: Optional[str],

) -> List[Dict]:

"""Fuzzy name match for backends without Lucene fuzzy syntax (Kùzu, FalkorDB, …).

Compares both the raw query and its identifier-normalised form against each

candidate name, taking the minimum distance. This lets camelCase queries

match snake_case stored names and vice-versa without inflating the distance.

"""

if not search_term.strip():

return []

where_clause = "WHERE node.path STARTS WITH $repo_path" if repo_path else ""

# Without a repo filter we must cap the candidate scan. 20 000 is enough to

# cover any realistic single-repo codebase while keeping latency acceptable.

limit_tail = "" if repo_path else " LIMIT 20000"

params: Dict[str, Any] = {}

if repo_path:

params["repo_path"] = repo_path

query = f"""

MATCH (node:{label})

{where_clause}

RETURN node.name as name, node.path as path, node.line_number as line_number,

node.source as source, node.docstring as docstring, node.is_dependency as is_dependency

{limit_tail}

"""

with self.driver.session() as session:

rows = session.run(query, **params).data()

# Two query forms:

# q_raw – lowercased original (e.g. "myFuncton" → "myfuncton")

# q_norm – separator-stripped (e.g. "my_functon" → "myfuncton")

# Using the minimum of both distances avoids the space-inflation bug where

# the handler's replace('_', ' ') turns "my_functon" into "my functon",

# which compares poorly against camelCase stored names.

q_raw = search_term.lower()

q_norm = _normalize_identifier(search_term)

scored: List[tuple[int, Dict]] = []

for row in rows:

nm = row.get("name")

if not isinstance(nm, str):

continue

nm_lower = nm.lower()

nm_norm = _normalize_identifier(nm)

d = min(

_levenshtein_distance(q_raw, nm_lower),

_levenshtein_distance(q_norm, nm_norm),

)

if d <= edit_distance:

scored.append((d, row))

scored.sort(key=lambda x: x[0])

return [r for _, r in scored[:20]]

def find_by_function_name(

self,

search_term: str,

fuzzy_search: bool,

repo_path: Optional[str] = None,

edit_distance: int = 2,

) -> List[Dict]:

"""Find functions by name matching."""

if not fuzzy_search:

with self.driver.session() as session:

result = session.run(f"""

MATCH (node:Function {{name: $name}})

{"WHERE node.path STARTS WITH $repo_path" if repo_path else ""}

RETURN node.name as name, node.path as path, node.line_number as line_number,

node.source as source, node.docstring as docstring, node.is_dependency as is_dependency

LIMIT 20

""", name=search_term, repo_path=repo_path)

return result.data()

if self._lacks_native_fulltext:

return self._find_by_name_fuzzy_portable(

"Function", search_term, edit_distance, repo_path

)

formatted_search_term = f"name:{search_term}"

with self.driver.session() as session:

result = session.run(

self.format_query("Function", fuzzy_search, repo_path),

search_term=formatted_search_term,

repo_path=repo_path,

)

return result.data()

def find_by_class_name(

self,

search_term: str,

fuzzy_search: bool,

repo_path: Optional[str] = None,

edit_distance: int = 2,

) -> List[Dict]:

"""Find classes by name matching."""

if not fuzzy_search:

with self.driver.session() as session:

result = session.run(f"""

MATCH (node:Class {{name: $name}})

{"WHERE node.path STARTS WITH $repo_path" if repo_path else ""}

RETURN node.name as name, node.path as path, node.line_number as line_number,

node.source as source, node.docstring as docstring, node.is_dependency as is_dependency

LIMIT 20

""", name=search_term, repo_path=repo_path)

return result.data()

if self._lacks_native_fulltext:

return self._find_by_name_fuzzy_portable(

"Class", search_term, edit_distance, repo_path

)

formatted_search_term = f"name:{search_term}"

with self.driver.session() as session:

result = session.run(

self.format_query("Class", fuzzy_search, repo_path),

search_term=formatted_search_term,

repo_path=repo_path,

)

return result.data()

def find_by_variable_name(self, search_term: str, repo_path: Optional[str] = None) -> List[Dict]:

"""Find variables by name matching"""

with self.driver.session() as session:

result = session.run(f"""

MATCH (v:Variable)

WHERE v.name CONTAINS $search_term {"AND v.path STARTS WITH $repo_path" if repo_path else ""}

RETURN v.name as name, v.path as path, v.line_number as line_number,

v.value as value, v.context as context, v.is_dependency as is_dependency

ORDER BY v.is_dependency ASC, v.name

LIMIT 20

""", search_term=search_term, repo_path=repo_path)

return result.data()

def find_by_content(self, search_term: str, repo_path: Optional[str] = None) -> List[Dict]:

"""Find code by content matching in source or docstrings using the full-text index."""

if self._lacks_native_fulltext:

return self._find_by_content_falkordb(search_term, repo_path)

with self.driver.session() as session:

result = session.run(f"""

CALL db.index.fulltext.queryNodes("code_search_index", $search_term) YIELD node, score

WITH node, score

WHERE (node:Function OR node:Class OR node:Variable) {"AND node.path STARTS WITH $repo_path" if repo_path else ""}

MATCH (node)<-[:CONTAINS]-(f:File)

RETURN

CASE

WHEN node:Function THEN 'function'

WHEN node:Class THEN 'class'

ELSE 'variable'

END as type,

node.name as name, f.path as path,

node.line_number as line_number, node.source as source,

node.docstring as docstring, node.is_dependency as is_dependency

ORDER BY score DESC

LIMIT 20

""", search_term=search_term, repo_path=repo_path)

return result.data()

def _find_by_content_falkordb(self, search_term: str, repo_path: Optional[str] = None) -> List[Dict]:

"""FalkorDB-compatible content search using pure Cypher CONTAINS matching.

FalkorDB does not support CALL db.idx.fulltext.queryNodes, so we fall back

to substring matching on name, source, and docstring fields."""

all_results = []

with self.driver.session() as session:

repo_filter = "AND node.path STARTS WITH $repo_path" if repo_path else ""

for label, type_name in [('Function', 'function'), ('Class', 'class')]:

try:

result = session.run(f"""

MATCH (node:{label})

WHERE (toLower(node.name) CONTAINS toLower($search_term)

OR (node.source IS NOT NULL AND toLower(node.source) CONTAINS toLower($search_term))

OR (node.docstring IS NOT NULL AND toLower(node.docstring) CONTAINS toLower($search_term)))

{repo_filter}

RETURN

'{type_name}' as type,

node.name as name, node.path as path,

node.line_number as line_number, node.source as source,

node.docstring as docstring, node.is_dependency as is_dependency

ORDER BY node.is_dependency ASC, node.name

LIMIT 20

""", search_term=search_term, repo_path=repo_path)

all_results.extend(result.data())

except Exception:

logger.debug(f"FalkorDB content query failed for label {label}", exc_info=True)

return all_results[:20]

def find_by_module_name(self, search_term: str) -> List[Dict]:

"""Find modules by name matching"""

with self.driver.session() as session:

result = session.run("""

MATCH (m:Module)

WHERE m.name CONTAINS $search_term

RETURN m.name as name, m.lang as lang

ORDER BY m.name

LIMIT 20

""", search_term=search_term)

return result.data()

def find_imports(self, search_term: str) -> List[Dict]:

"""Find imported symbols (aliases or original names)."""

with self.driver.session() as session:

result = session.run("""

MATCH (f:File)-[r:IMPORTS]->(m:Module)

WHERE r.alias = $search_term OR r.imported_name = $search_term

RETURN

r.alias as alias,

r.imported_name as imported_name,

m.name as module_name,

f.path as path,

r.line_number as line_number

ORDER BY f.path

LIMIT 20

""", search_term=search_term)

return result.data()

def find_related_code(self, user_query: str, fuzzy_search: bool, edit_distance: int, repo_path: Optional[str] = None) -> Dict[str, Any]:

"""Find code related to a query using multiple search strategies"""

# For Lucene backends: split snake_case/underscore tokens so Lucene sees

# individual words, then append the fuzzy modifier.

# For portable backends: keep user_query verbatim — _find_by_name_fuzzy_portable

# handles normalisation via _normalize_identifier.

if fuzzy_search and not self._lacks_native_fulltext:

lucene_base = user_query.replace("_", " ").strip()

lucene_fuzzy_query = " ".join(f"{t}~{edit_distance}" for t in lucene_base.split())

else:

lucene_fuzzy_query = user_query

# For portable backends, always pass the *original* query to the fuzzy name

# matcher — _find_by_name_fuzzy_portable applies its own normalisation.

# For Lucene-capable backends, use the Lucene fuzzy token form.

if self._lacks_native_fulltext:

name_lookup_q = user_query

else:

name_lookup_q = lucene_fuzzy_query if fuzzy_search else user_query

content_lookup_q = lucene_fuzzy_query if (fuzzy_search and not self._lacks_native_fulltext) else user_query

results: Dict[str, Any] = {

"query": lucene_fuzzy_query if fuzzy_search else user_query,

"functions_by_name": self.find_by_function_name(

name_lookup_q, fuzzy_search, repo_path, edit_distance

),

"classes_by_name": self.find_by_class_name(

name_lookup_q, fuzzy_search, repo_path, edit_distance

),

"variables_by_name": self.find_by_variable_name(user_query, repo_path), # no fuzzy for variables as they are not using full-text index

"content_matches": self.find_by_content(content_lookup_q, repo_path),

}

all_results: List[Dict[str, Any]] = []

for func in results["functions_by_name"]:

func["search_type"] = "function_name"

func["relevance_score"] = 0.9 if not func["is_dependency"] else 0.7

all_results.append(func)

for cls in results["classes_by_name"]:

cls["search_type"] = "class_name"

cls["relevance_score"] = 0.8 if not cls["is_dependency"] else 0.6

all_results.append(cls)

for var in results["variables_by_name"]:

var["search_type"] = "variable_name"

var["relevance_score"] = 0.7 if not var["is_dependency"] else 0.5

all_results.append(var)

for content in results["content_matches"]:

content["search_type"] = "content"

content["relevance_score"] = 0.6 if not content["is_dependency"] else 0.4

all_results.append(content)

all_results.sort(key=lambda x: x["relevance_score"], reverse=True)

results["ranked_results"] = all_results[:15]

results["total_matches"] = len(all_results)

return results

def find_functions_by_argument(self, argument_name: str, path: Optional[str] = None, repo_path: Optional[str] = None) -> List[Dict]:

"""Find functions that take a specific argument name."""

with self.driver.session() as session:

repo_filter = "AND f.path STARTS WITH $repo_path" if repo_path else ""

if path:

query = f"""

MATCH (f:Function)-[:HAS_PARAMETER]->(p:Parameter)

WHERE p.name = $argument_name AND f.path = $path {repo_filter}

RETURN f.name AS function_name, f.path AS path, f.line_number AS line_number,

f.docstring AS docstring, f.is_dependency AS is_dependency

ORDER BY f.is_dependency ASC, f.path, f.line_number

LIMIT 20

"""

result = session.run(query, argument_name=argument_name, path=path, repo_path=repo_path)

else:

query = f"""

MATCH (f:Function)-[:HAS_PARAMETER]->(p:Parameter)

WHERE p.name = $argument_name {repo_filter}

RETURN f.name AS function_name, f.path AS path, f.line_number AS line_number,

f.docstring AS docstring, f.is_dependency AS is_dependency

ORDER BY f.is_dependency ASC, f.path, f.line_number

LIMIT 20

"""

result = session.run(query, argument_name=argument_name, repo_path=repo_path)

return result.data()

def find_functions_by_decorator(self, decorator_name: str, path: Optional[str] = None, repo_path: Optional[str] = None) -> List[Dict]:

"""Find functions that have a specific decorator applied to them."""

with self.driver.session() as session:

repo_filter = "AND f.path STARTS WITH $repo_path" if repo_path else ""

if path:

query = f"""

MATCH (f:Function)

WHERE f.path = $path AND $decorator_name IN f.decorators {repo_filter}

RETURN f.name AS function_name, f.path AS path, f.line_number AS line_number,

f.docstring AS docstring, f.is_dependency AS is_dependency, f.decorators AS decorators

ORDER BY f.is_dependency ASC, f.path, f.line_number

LIMIT 20

"""

result = session.run(query, decorator_name=decorator_name, path=path, repo_path=repo_path)

else:

query = f"""

MATCH (f:Function)

WHERE $decorator_name IN f.decorators {repo_filter}

RETURN f.name AS function_name, f.path AS path, f.line_number AS line_number,

f.docstring AS docstring, f.is_dependency AS is_dependency, f.decorators AS decorators

ORDER BY f.is_dependency ASC, f.path, f.line_number

LIMIT 20

"""

result = session.run(query, decorator_name=decorator_name, repo_path=repo_path)

return result.data()

def who_calls_function(self, function_name: str, path: Optional[str] = None, repo_path: Optional[str] = None) -> List[Dict]:

"""Find what functions call a specific function using CALLS relationships with improved matching"""

with self.driver.session() as session:

repo_filter = "AND caller.path STARTS WITH $repo_path" if repo_path else ""

if path:

result = session.run(f"""

MATCH (caller)-[call:CALLS]->(target:Function {{name: $function_name, path: $path}})

WHERE (caller:Function OR caller:Class OR caller:File) {repo_filter}

OPTIONAL MATCH (caller_file:File)-[:CONTAINS]->(caller)

RETURN DISTINCT

caller.name as caller_function,

COALESCE(caller.path, caller_file.path) as caller_file_path,

caller.line_number as caller_line_number,

caller.docstring as caller_docstring,

caller.is_dependency as caller_is_dependency,

call.line_number as call_line_number,

call.args as call_args,

call.full_call_name as full_call_name,

target.path as target_file_path

ORDER BY caller_is_dependency ASC, caller_file_path, caller_line_number

LIMIT 20

""", function_name=function_name, path=path, repo_path=repo_path)

results = result.data()

if not results:

result = session.run(f"""

MATCH (caller)-[call:CALLS]->(target:Function {{name: $function_name}})

WHERE (caller:Function OR caller:Class OR caller:File) {repo_filter}

OPTIONAL MATCH (caller_file:File)-[:CONTAINS]->(caller)

RETURN DISTINCT

caller.name as caller_function,

COALESCE(caller.path, caller_file.path) as caller_file_path,

caller.line_number as caller_line_number,

caller.docstring as caller_docstring,

caller.is_dependency as caller_is_dependency,

call.line_number as call_line_number,

call.args as call_args,

call.full_call_name as full_call_name,

target.path as target_file_path

ORDER BY caller_is_dependency ASC, caller_file_path, caller_line_number

LIMIT 20

""", function_name=function_name, repo_path=repo_path)

results = result.data()

else:

result = session.run(f"""

MATCH (caller:Function)-[call:CALLS]->(target:Function {{name: $function_name}})

WHERE 1=1 {repo_filter}

OPTIONAL MATCH (caller_file:File)-[:CONTAINS]->(caller)

RETURN DISTINCT

caller.name as caller_function,

caller.path as caller_file_path,

caller.line_number as caller_line_number,

caller.docstring as caller_docstring,

caller.is_dependency as caller_is_dependency,

call.line_number as call_line_number,

call.args as call_args,

call.full_call_name as full_call_name,

target.path as target_file_path

ORDER BY caller_is_dependency ASC, caller_file_path, caller_line_number

LIMIT 20

""", function_name=function_name, repo_path=repo_path)

results = result.data()

return results

def what_does_function_call(self, function_name: str, path: Optional[str] = None, repo_path: Optional[str] = None) -> List[Dict]:

"""Find what functions a specific function calls using CALLS relationships"""

with self.driver.session() as session:

if path:

# Convert path to absolute path

absolute_file_path = str(Path(path).resolve())

result = session.run(f"""

MATCH (caller:Function {{name: $function_name, path: $absolute_file_path}})

MATCH (caller)-[call:CALLS]->(called:Function)

WHERE called.path STARTS WITH $repo_path OR $repo_path IS NULL

OPTIONAL MATCH (called_file:File)-[:CONTAINS]->(called)

RETURN DISTINCT

called.name as called_function,

called.path as called_file_path,

called.line_number as called_line_number,

called.docstring as called_docstring,

called.is_dependency as called_is_dependency,

call.line_number as call_line_number,

call.args as call_args,

call.full_call_name as full_call_name

ORDER BY called_is_dependency ASC, called_function

LIMIT 20

""", function_name=function_name, absolute_file_path=absolute_file_path, repo_path=repo_path)

else:

result = session.run(f"""

MATCH (caller:Function {{name: $function_name}})-[call:CALLS]->(called:Function)

WHERE called.path STARTS WITH $repo_path OR $repo_path IS NULL

OPTIONAL MATCH (called_file:File)-[:CONTAINS]->(called)

RETURN DISTINCT

called.name as called_function,

called.path as called_file_path,

called.line_number as called_line_number,

called.docstring as called_docstring,

called.is_dependency as called_is_dependency,

call.line_number as call_line_number,

call.args as call_args,

call.full_call_name as full_call_name

ORDER BY called_is_dependency ASC, called_function

LIMIT 20

""", function_name=function_name, repo_path=repo_path)

return result.data()

def who_imports_module(self, module_name: str, repo_path: Optional[str] = None) -> List[Dict]:

"""Find what files import a specific module using IMPORTS relationships"""

with self.driver.session() as session:

repo_filter = "AND file.path STARTS WITH $repo_path" if repo_path else ""

result = session.run(f"""

MATCH (file:File)-[imp:IMPORTS]->(module:Module)

WHERE (module.name = $module_name OR module.full_import_name CONTAINS $module_name) {repo_filter}

OPTIONAL MATCH (repo:Repository)-[:CONTAINS]->(file)

WITH file, repo, COLLECT({{

imported_module: module.name,

import_alias: module.alias,

full_import_name: module.full_import_name

}}) AS imports

RETURN

file.name AS file_name,

file.path AS path,

file.relative_path AS file_relative_path,

file.is_dependency AS file_is_dependency,

repo.name AS repository_name,

imports

ORDER BY file_is_dependency ASC, path

LIMIT 20

""", module_name=module_name, repo_path=repo_path)

return result.data()

def who_modifies_variable(self, variable_name: str, repo_path: Optional[str] = None) -> List[Dict]:

"""Find what functions contain or modify a specific variable"""

with self.driver.session() as session:

repo_filter = "AND container.path STARTS WITH $repo_path" if repo_path else ""

result = session.run(f"""

MATCH (var:Variable {{name: $variable_name}})

MATCH (container)-[:CONTAINS]->(var)

WHERE (container:Function OR container:Class OR container:File) {repo_filter}

OPTIONAL MATCH (file:File)-[:CONTAINS]->(container)

RETURN DISTINCT

CASE

WHEN container:Function THEN container.name

WHEN container:Class THEN container.name

ELSE 'file_level'

END as container_name,

CASE

WHEN container:Function THEN 'function'

WHEN container:Class THEN 'class'

ELSE 'file'

END as container_type,

COALESCE(container.path, file.path) as path,

container.line_number as container_line_number,

var.line_number as variable_line_number,

var.value as variable_value,

var.context as variable_context,

COALESCE(container.is_dependency, file.is_dependency, false) as is_dependency

ORDER BY is_dependency ASC, path, variable_line_number

LIMIT 20

""", variable_name=variable_name, repo_path=repo_path)

return result.data()

def find_class_hierarchy(self, class_name: str, path: Optional[str] = None, repo_path: Optional[str] = None) -> Dict[str, Any]:

"""Find class inheritance relationships using INHERITS relationships"""

with self.driver.session() as session:

repo_filter = "AND parent.path STARTS WITH $repo_path" if repo_path else ""

if path:

match_clause = "MATCH (child:Class {name: $class_name, path: $path})"

else:

match_clause = "MATCH (child:Class {name: $class_name})"

parents_query = f"""

{match_clause}

MATCH (child)-[:INHERITS]->(parent:Class)

WHERE 1=1 {repo_filter}

OPTIONAL MATCH (parent_file:File)-[:CONTAINS]->(parent)

RETURN DISTINCT

parent.name as parent_class,

parent.path as parent_file_path,

parent.line_number as parent_line_number,

parent.docstring as parent_docstring,

parent.is_dependency as parent_is_dependency

ORDER BY parent_is_dependency ASC, parent_class

"""

parents_result = session.run(parents_query, class_name=class_name, path=path, repo_path=repo_path)

repo_filter_child = "AND grandchild.path STARTS WITH $repo_path" if repo_path else ""

children_query = f"""

{match_clause}

MATCH (grandchild:Class)-[:INHERITS]->(child)

WHERE 1=1 {repo_filter_child}

OPTIONAL MATCH (child_file:File)-[:CONTAINS]->(grandchild)

RETURN DISTINCT

grandchild.name as child_class,

grandchild.path as child_file_path,

grandchild.line_number as child_line_number,

grandchild.docstring as child_docstring,

grandchild.is_dependency as child_is_dependency

ORDER BY child_is_dependency ASC, child_class

"""

children_result = session.run(children_query, class_name=class_name, path=path, repo_path=repo_path)

repo_filter_method = "WHERE method.path STARTS WITH $repo_path" if repo_path else ""

methods_query = f"""

{match_clause}

MATCH (child)-[:CONTAINS]->(method:Function)

{repo_filter_method}

RETURN DISTINCT

method.name as method_name,

method.path as method_file_path,

method.line_number as method_line_number,

method.args as method_args,

method.docstring as method_docstring,

method.is_dependency as method_is_dependency

ORDER BY method_is_dependency ASC, method_line_number

"""

methods_result = session.run(methods_query, class_name=class_name, path=path, repo_path=repo_path)

return {

"class_name": class_name,

"parent_classes": parents_result.data(),

"child_classes": children_result.data(),

"methods": methods_result.data()

}

def find_function_overrides(self, function_name: str, repo_path: Optional[str] = None) -> List[Dict]:

"""Find all implementations of a function across different classes"""

with self.driver.session() as session:

repo_filter = "AND class.path STARTS WITH $repo_path" if repo_path else ""

result = session.run(f"""

MATCH (class:Class)-[:CONTAINS]->(func:Function {{name: $function_name}})

WHERE 1=1 {repo_filter}

OPTIONAL MATCH (file:File)-[:CONTAINS]->(class)

RETURN DISTINCT

class.name as class_name,

class.path as class_file_path,

func.name as function_name,

func.line_number as function_line_number,

func.args as function_args,

func.docstring as function_docstring,

func.is_dependency as is_dependency,

file.name as file_name

ORDER BY is_dependency ASC, class_name

LIMIT 20

""", function_name=function_name, repo_path=repo_path)

return result.data()

def find_dead_code(self, exclude_decorated_with: Optional[List[str]] = None, repo_path: Optional[str] = None) -> Dict[str, Any]:

"""Find potentially unused functions (not called by other functions in the project), optionally excluding those with specific decorators."""

if exclude_decorated_with is None:

exclude_decorated_with = []

with self.driver.session() as session:

repo_filter = "AND func.path STARTS WITH $repo_path" if repo_path else ""

decorator_filter = ""

if exclude_decorated_with:

any_conditions = " AND ".join(

f"NOT ANY(d IN func.decorators WHERE d CONTAINS '{p.replace(chr(39), chr(39)+chr(39))}')"

for p in exclude_decorated_with

)

decorator_filter = f"AND {any_conditions}"

func_ignore = cypher_path_not_under_ignore_dirs("func.path")

caller_ignore = cypher_path_not_under_ignore_dirs("caller.path")

query = f"""

MATCH (func:Function)

WHERE func.is_dependency = false {repo_filter} {func_ignore}

AND NOT func.name IN ['main', 'setup', 'run']

AND NOT (func.name STARTS WITH '__' AND func.name ENDS WITH '__')

AND NOT func.name STARTS WITH '_test'

AND NOT func.name STARTS WITH 'test_'

AND NOT func.name CONTAINS 'main'

AND NOT toLower(func.name) CONTAINS 'application'

AND NOT toLower(func.name) CONTAINS 'entry'

AND NOT toLower(func.name) CONTAINS 'entrypoint'

{decorator_filter}

WITH func

OPTIONAL MATCH (caller:Function)-[:CALLS]->(func)

WHERE caller.is_dependency = false {caller_ignore}

WITH func, count(caller) as caller_count

WHERE caller_count = 0

OPTIONAL MATCH (file:File)-[:CONTAINS]->(func)

RETURN

func.name as function_name,

func.path as path,

func.line_number as line_number,

func.docstring as docstring,

func.context as context,

file.name as file_name

ORDER BY func.path, func.line_number

LIMIT 50

"""

params = {}

if repo_path:

params["repo_path"] = repo_path

result = session.run(query, **params)

return {

"potentially_unused_functions": result.data(),

"note": "These functions might be unused, but could be entry points, callbacks, or called dynamically"

}

def find_all_callers(self, function_name: str, path: Optional[str] = None, repo_path: Optional[str] = None, depth: int = 3) -> List[Dict]:

"""Find all direct and indirect callers of a specific function, returning edges."""

depth = _sanitize_depth(depth)

with self.driver.session() as session:

repo_filter = "AND caller.path STARTS WITH $repo_path" if repo_path else ""

depth_str = f"1..{depth}" if depth > 1 else "1"

# KùzuDB-optimized: matching on the path end node via nodes(p) indexing

# ensures we avoid Binder exceptions for multi-labeled property lookups

# on the end node of variable-length paths.

if path:

query = f"""

MATCH p = (caller:Function)-[:CALLS*{depth_str}]->(target:Function)

WITH p, nodes(p) as path_nodes, relationships(p) as rels

WITH p, path_nodes, rels, path_nodes[size(path_nodes)-1] as last_node

WHERE last_node.name = $function_name AND last_node.path = $path

{repo_filter}

UNWIND rels as r

WITH startNode(r) as s, endNode(r) as e, r

RETURN DISTINCT s.name as caller_name, s.path as caller_path,

e.name as callee_name, e.path as callee_path,

r.line_number as line

LIMIT 100

"""

result = session.run(query, function_name=function_name, path=path, repo_path=repo_path)

else:

query = f"""

MATCH p = (caller:Function)-[:CALLS*{depth_str}]->(target:Function)

WITH p, nodes(p) as path_nodes, relationships(p) as rels

WITH p, path_nodes, rels, path_nodes[size(path_nodes)-1] as last_node

WHERE last_node.name = $function_name

{repo_filter}

UNWIND rels as r

WITH startNode(r) as s, endNode(r) as e, r

RETURN DISTINCT s.name as caller_name, s.path as caller_path,

e.name as callee_name, e.path as callee_path,

r.line_number as line

LIMIT 100

"""

result = session.run(query, function_name=function_name, repo_path=repo_path)

return result.data()

def find_all_callees(self, function_name: str, path: Optional[str] = None, repo_path: Optional[str] = None, depth: int = 3) -> List[Dict]:

"""Find all direct and indirect callees of a specific function, returning edges."""

depth = _sanitize_depth(depth)

with self.driver.session() as session:

repo_filter = "AND callee.path STARTS WITH $repo_path" if repo_path else ""

depth_str = f"1..{depth}" if depth > 1 else "1"

if path:

query = f"""

MATCH p = (caller:Function {{name: $function_name, path: $path}})-[:CALLS*{depth_str}]->(callee:Function)

WITH p, nodes(p) as path_nodes, relationships(p) as rels

WITH p, path_nodes, rels, path_nodes[size(path_nodes)-1] as last_node

WHERE 1=1 {repo_filter}

UNWIND rels as r

WITH startNode(r) as s, endNode(r) as e, r

RETURN DISTINCT s.name as caller_name, s.path as caller_path,

e.name as callee_name, e.path as callee_path,

r.line_number as line

LIMIT 100

"""

result = session.run(query, function_name=function_name, path=path, repo_path=repo_path)

else:

query = f"""

MATCH p = (caller:Function {{name: $function_name}})-[:CALLS*{depth_str}]->(callee:Function)

WITH p, nodes(p) as path_nodes, relationships(p) as rels

WITH p, path_nodes, rels, path_nodes[size(path_nodes)-1] as last_node

WHERE 1=1 {repo_filter}

UNWIND rels as r

WITH startNode(r) as s, endNode(r) as e, r

RETURN DISTINCT s.name as caller_name, s.path as caller_path,

e.name as callee_name, e.path as callee_path,

r.line_number as line

LIMIT 100

"""

result = session.run(query, function_name=function_name, repo_path=repo_path)

return result.data()

def find_function_call_chain(self, start_function: str, end_function: str, max_depth: int = 5, start_file: Optional[str] = None, end_file: Optional[str] = None, repo_path: Optional[str] = None) -> List[Dict]:

"""Find call chains between two functions"""

with self.driver.session() as session:

# Build match clauses based on whether files are specified

start_props = "{name: $start_function" + (", path: $start_file}" if start_file else "}")

end_props = "{name: $end_function" + (", path: $end_file}" if end_file else "}")

repo_clauses = []

if repo_path:

repo_clauses.append("start.path STARTS WITH $repo_path")

repo_clauses.append("end_target.path STARTS WITH $repo_path")

repo_where = ("WHERE " + " AND ".join(repo_clauses)) if repo_clauses else ""

query = f"""

MATCH (start:Function {start_props}), (end_target:Function {end_props})

{repo_where}

MATCH path = (start)-[:CALLS*1..{max_depth}]->(end_target)

RETURN nodes(path) AS function_nodes, relationships(path) AS call_nodes, length(path) AS chain_length

ORDER BY chain_length ASC

LIMIT 20

"""

# Prepare parameters

params = {

"start_function": start_function,

"end_function": end_function,

"start_file": start_file,

"end_file": end_file,

"repo_path": repo_path

}

result = session.run(query, **params)

# Post-process Node/Rel objects into plain dicts so CLI output stays stable

rows = result.data()

transformed: List[Dict[str, Any]] = []

for row in rows:

func_nodes = row.get("function_nodes") or []

rel_nodes = row.get("call_nodes") or []

chain_len = row.get("chain_length", 0)

function_chain = []

for n in func_nodes:

# Depending on KùzuDB + driver wrapping, list elements can arrive

# either as Node/Rel objects or already-materialized dicts.

if isinstance(n, dict):

props = n

else:

props = None

try:

props = n.get_properties()

except Exception:

props = getattr(n, "properties", None)

if props is None:

props = {}

function_chain.append(

{

"name": props.get("name"),

"path": props.get("path"),

"line_number": props.get("line_number"),

"is_dependency": props.get("is_dependency"),

}

)

call_details = []

for r in rel_nodes:

if isinstance(r, dict):

props = r

else:

props = None

try:

props = r.get_properties()

except Exception:

props = getattr(r, "properties", None)

if props is None: