"main",

"setup",

"run",

"application",

"entry",

"entrypoint",

"oncreate", "onstart", "onresume", "onpause", "onstop", "ondestroy",

"oncreateview", "onviewcreated", "ondestroyview", "onattach", "ondetach",

"onbind", "onunbind", "onrebind", "onstartcommand", "onreceive",

"onhandlework", "dowork", "onsaveinstancestate", "onrestoreinstancestate",

"onactivityresult", "onrequestpermissionsresult", "onnewintent",

"onconfigurationchanged", "onlowmemory", "ontrimmemory",

"onbindviewholder", "oncreateviewholder", "getitemcount",

"""(confidence, reason) for a zero-caller symbol (#1332).

name = (row.get("function_name") or "")

lname = name.lower()

path = (row.get("path") or "")

if name.startswith("__") and name.endswith("__"):

return "low", "dunder — called implicitly by the runtime"

if lname in _TEST_HOOK_NAMES or name.startswith(("test_", "_test")):

return "low", "test hook/prefix — invoked by the test runner"

if lname in {n.lower() for n in _ENTRY_POINT_NAMES}:

return "low", "entry-point name — typically called externally"

if lname in {n.lower() for n in _ANDROID_ENTRY_POINT_NAMES}:

return "low", "framework entry point"

if "override" in (row.get("modifiers") or []):

return "low", "override — dispatched via its base declaration"

if "/tests/" in path or "/test/" in path or path.endswith("conftest.py"):

return "low", "test-tree file — commonly runner-invoked"

if row.get("decorators"):

return "medium", "decorated — decorators often add implicit callers"

"Composable", "Preview",

"Test", "Before", "After", "BeforeEach", "AfterEach", "ParameterizedTest",

"Inject", "Provides", "Binds", "HiltViewModel", "AndroidEntryPoint",

"Dao", "Query", "Insert", "Update", "Delete", "Transaction", "TypeConverter",

"JvmStatic", "Keep", "Serializable",

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

try:

depth = int(depth)

except (TypeError, ValueError):

return default

"""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._lacks_native_fulltext = getattr(db_manager, 'get_backend_type', lambda: 'neo4j')() != 'neo4j'

self._active_graph = None

@property

def driver(self):

"""Returns driver for the active graph (set via graph_name params), or default."""

return self.db_manager.get_driver(self._active_graph)

def _normalize_repo_path_filter(self, repo_path: Optional[str]) -> Optional[str]:

"""Resolve a caller-supplied repo_path to the absolute path stored on indexed

if not isinstance(repo_path, str):

return repo_path

candidate = repo_path.strip().rstrip("/")

if not candidate:

return None

if Path(candidate).is_absolute():

return candidate

# Reuse the currently active graph so this lookup doesn't reset scoping.

repos = self.list_indexed_repositories(graph_name=self._active_graph)

norm = candidate.lower()

matches: List[str] = []

for repo in repos:

raw_path = str(repo.get("path", "")).strip().rstrip("/")

if not raw_path:

continue

repo_name = str(repo.get("name", "")).strip().lower()

base_name = Path(raw_path).name.lower()

path_lower = raw_path.lower()

if (

norm == repo_name

or norm == base_name

or norm == path_lower

or path_lower.endswith("/" + norm)

):

matches.append(raw_path)

unique_matches = sorted(set(matches))

if len(unique_matches) == 1:

return unique_matches[0]

# Fall back to resolving relative to cwd, in case the caller passed a path

# that's valid on the machine running this process (e.g. CLI usage).

try:

cwd_candidate = str((Path.cwd() / candidate).resolve()).rstrip("/")

except Exception:

cwd_candidate = None

if cwd_candidate and any(

str(repo.get("path", "")).strip().rstrip("/") == cwd_candidate

or str(repo.get("path", "")).strip().rstrip("/").startswith(cwd_candidate + "/")

for repo in repos

):

return cwd_candidate

# No confident match (zero or ambiguous multiple) — return unchanged rather

# than guess; the STARTS WITH filter will simply match nothing, same as today.

return candidate

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 = self._normalize_repo_path_filter(repo_path)

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, graph_name: str = None) -> Dict[str, Any]:

self._active_graph = graph_name

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

repo_path = self._normalize_repo_path_filter(repo_path)

# 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, limit: Optional[int] = None) -> List[Dict]:

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

path_filter = "AND f.path = $path" if path else ""

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

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"argument_name": argument_name, "repo_path": repo_path}

if path:

params["path"] = path

if limit is not None:

params["limit"] = limit

# A WHERE attached to an OPTIONAL MATCH only constrains the

# optional part — it can never filter f, so the original form

# returned EVERY function for any search term. Filter f itself

# after materializing the optional parameter match.

query = f"""

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

return result.data()

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

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

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

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"decorator_name": decorator_name, "repo_path": repo_path}

if limit is not None:

params["limit"] = limit

if path:

params["path"] = path

query = f"""

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

else:

query = f"""

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

return result.data()

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

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

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

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"function_name": function_name, "repo_path": repo_path}

if limit is not None:

params["limit"] = limit

if path:

params["path"] = path

result = session.run(f"""

""", **params)

results = result.data()

if not results:

params_no_path = {k: v for k, v in params.items() if k != "path"}

result = session.run(f"""

""", **params_no_path)

results = result.data()

else:

result = session.run(f"""

""", **params)

results = result.data()

return results

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

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"function_name": function_name, "repo_path": repo_path}

if limit is not None:

params["limit"] = limit

if path:

# Convert path to absolute path

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

params["absolute_file_path"] = absolute_file_path

result = session.run(f"""

""", **params)

else:

result = session.run(f"""

""", **params)

return result.data()

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

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

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

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"module_name": module_name, "repo_path": repo_path}

if limit is not None:

params["limit"] = limit

result = session.run(f"""

""", **params)

return result.data()

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

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

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

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"variable_name": variable_name, "repo_path": repo_path}

if limit is not None:

params["limit"] = limit

result = session.run(f"""

""", **params)

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

repo_path = self._normalize_repo_path_filter(repo_path)

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, limit: Optional[int] = None) -> List[Dict]:

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

repo_path = self._normalize_repo_path_filter(repo_path)

with self.driver.session() as session:

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

limit_clause = "LIMIT $limit" if limit is not None else ""

params = {"function_name": function_name, "repo_path": repo_path}

if limit is not None:

params["limit"] = limit

result = session.run(f"""

""", **params)

return result.data()

def find_dead_code(self, exclude_decorated_with: Optional[List[str]] = None, repo_path: Optional[str] = None, graph_name: str = None, limit: Optional[int] = None, include_low_confidence: bool = False) -> Dict[str, Any]:

self._active_graph = graph_name

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

repo_path = self._normalize_repo_path_filter(repo_path)

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

)