"""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

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

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],

"""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 = str(Path(repo_path).resolve()) if repo_path else None

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

query = f"""

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"""

return f"""

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, …).

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"""

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"""

""", 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"""

""", 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"""

""", 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"""

""", 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.

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"""

""", 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("""

""", 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("""

""", 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"""

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

else:

query = f"""

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"""

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

else:

query = f"""

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"""

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

results = result.data()

if not results:

result = session.run(f"""

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

results = result.data()

else:

result = session.run(f"""

""", 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"""

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

else:

result = session.run(f"""

""", 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"""

""", 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"""

""", 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"""

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"""

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"""

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"""

""", 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"""

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."""

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"""

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

else:

query = f"""

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."""

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"""

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

else:

query = f"""

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"""

# 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:

props = {}

call_details.append(

{

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

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

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

}

)

transformed.append(

{

"function_chain": function_chain,

"call_details": call_details,

"chain_length": chain_len,

}

)