diff --git a/src/codegraphcontext/cli/main.py b/src/codegraphcontext/cli/main.py index 840a2c85..692ed515 100644 --- a/src/codegraphcontext/cli/main.py +++ b/src/codegraphcontext/cli/main.py @@ -46,6 +46,7 @@ setup_scip_helper, ) from .hook_manager import HookError, get_hook_status, install_hooks, uninstall_hooks +from codegraphcontext.utils.tool_limits import get_tool_result_limit # Set the log level for the noisy neo4j, asyncio, and urllib3 loggers to keep the output clean. # Get the log level from config, defaulting to WARNING @@ -2274,7 +2275,11 @@ def find_by_decorator_search( db_manager, graph_builder, code_finder = services[:3] try: - results = code_finder.find_functions_by_decorator(decorator, file) + req_limit = get_tool_result_limit("find_functions_by_decorator") + results = code_finder.find_functions_by_decorator(decorator, file, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(results) > req_limit) + if truncated: + results = results[:req_limit] if not results: console.print(f"[yellow]No functions found with decorator '@{decorator}'[/yellow]") @@ -2299,6 +2304,8 @@ def find_by_decorator_search( console.print(f"[cyan]Found {len(results)} function(s) with decorator '@{decorator}':[/cyan]") console.print(table) + if truncated: + console.print(f"[dim]... truncated ({req_limit} shown), more exist[/dim]") finally: db_manager.close_driver() @@ -2322,7 +2329,11 @@ def find_by_argument_search( db_manager, graph_builder, code_finder = services[:3] try: - results = code_finder.find_functions_by_argument(argument, file) + req_limit = get_tool_result_limit("find_functions_by_argument") + results = code_finder.find_functions_by_argument(argument, file, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(results) > req_limit) + if truncated: + results = results[:req_limit] if not results: console.print(f"[yellow]No functions found with argument '{argument}'[/yellow]") @@ -2344,6 +2355,8 @@ def find_by_argument_search( console.print(f"[cyan]Found {len(results)} function(s) with argument '{argument}':[/cyan]") console.print(table) + if truncated: + console.print(f"[dim]... truncated ({req_limit} shown), more exist[/dim]") finally: db_manager.close_driver() @@ -2378,7 +2391,11 @@ def analyze_calls( db_manager, graph_builder, code_finder = services[:3] try: - results = code_finder.what_does_function_call(function, file) + req_limit = get_tool_result_limit("find_callees") + results = code_finder.what_does_function_call(function, file, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(results) > req_limit) + if truncated: + results = results[:req_limit] if not results: console.print(f"[yellow]No function calls found for '{function}'[/yellow]") @@ -2407,7 +2424,10 @@ def analyze_calls( console.print(f"\n[bold cyan]Function '{function}' calls:[/bold cyan]") console.print(table) - console.print(f"\n[dim]Total: {len(results)} function(s)[/dim]") + if truncated: + console.print(f"\n[dim]Total: {len(results)} function(s) (truncated, {req_limit}+ exist)[/dim]") + else: + console.print(f"\n[dim]Total: {len(results)} function(s)[/dim]") finally: db_manager.close_driver() @@ -2434,7 +2454,11 @@ def analyze_callers( db_manager, graph_builder, code_finder = services[:3] try: - results = code_finder.who_calls_function(function, file) + req_limit = get_tool_result_limit("find_callers") + results = code_finder.who_calls_function(function, file, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(results) > req_limit) + if truncated: + results = results[:req_limit] if not results: console.print(f"[yellow]No callers found for '{function}'[/yellow]") @@ -2465,7 +2489,10 @@ def analyze_callers( console.print(f"\n[bold cyan]Functions that call '{function}':[/bold cyan]") console.print(table) - console.print(f"\n[dim]Total: {len(results)} caller(s)[/dim]") + if truncated: + console.print(f"\n[dim]Total: {len(results)} caller(s) (truncated, {req_limit}+ exist)[/dim]") + else: + console.print(f"\n[dim]Total: {len(results)} caller(s)[/dim]") finally: db_manager.close_driver() @@ -2495,7 +2522,11 @@ def analyze_chain( db_manager, graph_builder, code_finder = services[:3] try: - results = code_finder.find_function_call_chain(from_func, to_func, max_depth, from_file, to_file) + req_limit = get_tool_result_limit("call_chain") + results = code_finder.find_function_call_chain(from_func, to_func, max_depth, from_file, to_file, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(results) > req_limit) + if truncated: + results = results[:req_limit] if not results: console.print(f"[yellow]No call chain found between '{from_func}' and '{to_func}' within depth {max_depth}[/yellow]") @@ -2908,7 +2939,11 @@ def analyze_overrides( db_manager, graph_builder, code_finder = services[:3] try: - results = code_finder.find_function_overrides(function_name) + req_limit = get_tool_result_limit("overrides") + results = code_finder.find_function_overrides(function_name, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(results) > req_limit) + if truncated: + results = results[:req_limit] if not results: console.print(f"[yellow]No implementations found for function '{function_name}'[/yellow]") @@ -2937,6 +2972,8 @@ def analyze_overrides( console.print(f"\n[bold cyan]Found {len(results)} implementation(s) of '{function_name}':[/bold cyan]") console.print(table) + if truncated: + console.print(f"[dim]... truncated ({req_limit} shown), more exist[/dim]") finally: db_manager.close_driver() diff --git a/src/codegraphcontext/tools/code_finder.py b/src/codegraphcontext/tools/code_finder.py index 6771aa0f..7aee8bb3 100644 --- a/src/codegraphcontext/tools/code_finder.py +++ b/src/codegraphcontext/tools/code_finder.py @@ -9,6 +9,7 @@ if TYPE_CHECKING: from ..core.database import DatabaseManager from ..utils.path_ignore import cypher_path_not_under_ignore_dirs +from ..utils.tool_limits import get_tool_result_limit logger = logging.getLogger(__name__) @@ -506,20 +507,25 @@ def find_related_code(self, user_query: str, fuzzy_search: bool, edit_distance: return results - def find_functions_by_argument(self, argument_name: str, path: Optional[str] = None, repo_path: Optional[str] = None) -> List[Dict]: + 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.""" 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 = {"argument_name": argument_name, "repo_path": repo_path} + if limit is not None: + params["limit"] = limit if path: + params["path"] = 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 + {limit_clause} """ - result = session.run(query, argument_name=argument_name, path=path, repo_path=repo_path) + result = session.run(query, **params) else: query = f""" MATCH (f:Function)-[:HAS_PARAMETER]->(p:Parameter) @@ -527,25 +533,30 @@ def find_functions_by_argument(self, argument_name: str, path: Optional[str] = N 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 + {limit_clause} """ - result = session.run(query, argument_name=argument_name, repo_path=repo_path) + 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) -> List[Dict]: + 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.""" 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""" 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 + {limit_clause} """ - result = session.run(query, decorator_name=decorator_name, path=path, repo_path=repo_path) + result = session.run(query, **params) else: query = f""" MATCH (f:Function) @@ -553,16 +564,21 @@ def find_functions_by_decorator(self, decorator_name: str, path: Optional[str] = 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 + {limit_clause} """ - result = session.run(query, decorator_name=decorator_name, repo_path=repo_path) + 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) -> List[Dict]: + 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""" 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""" MATCH (caller)-[call:CALLS|HEURISTIC_CALLS]->(target:Function {{name: $function_name, path: $path}}) WHERE (caller:Function OR caller:Class OR caller:File) {repo_filter} @@ -576,11 +592,12 @@ def who_calls_function(self, function_name: str, path: Optional[str] = None, rep caller.is_dependency as caller_is_dependency, 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) + {limit_clause} + """, **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""" MATCH (caller)-[call:CALLS|HEURISTIC_CALLS]->(target:Function {{name: $function_name}}) WHERE (caller:Function OR caller:Class OR caller:File) {repo_filter} @@ -594,8 +611,8 @@ def who_calls_function(self, function_name: str, path: Optional[str] = None, rep caller.is_dependency as caller_is_dependency, 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) + {limit_clause} + """, **params_no_path) results = result.data() else: result = session.run(f""" @@ -611,18 +628,23 @@ def who_calls_function(self, function_name: str, path: Optional[str] = None, rep caller.is_dependency as caller_is_dependency, 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) + {limit_clause} + """, **params) 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]: + 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""" 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""" MATCH (caller:Function {{name: $function_name, path: $absolute_file_path}}) MATCH (caller)-[call:CALLS|HEURISTIC_CALLS]->(called:Function) @@ -638,8 +660,8 @@ def what_does_function_call(self, function_name: str, path: Optional[str] = None 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) + {limit_clause} + """, **params) else: result = session.run(f""" MATCH (caller:Function {{name: $function_name}})-[call:CALLS|HEURISTIC_CALLS]->(called:Function) @@ -655,15 +677,19 @@ def what_does_function_call(self, function_name: str, path: Optional[str] = None 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) + {limit_clause} + """, **params) return result.data() - def who_imports_module(self, module_name: str, repo_path: Optional[str] = None) -> List[Dict]: + 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""" 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""" MATCH (file:File)-[imp:IMPORTS]->(module:Module) WHERE (module.name = $module_name OR module.full_import_name CONTAINS $module_name) {repo_filter} @@ -681,15 +707,19 @@ def who_imports_module(self, module_name: str, repo_path: Optional[str] = None) repo.name AS repository_name, imports ORDER BY file_is_dependency ASC, path - LIMIT 20 - """, module_name=module_name, repo_path=repo_path) + {limit_clause} + """, **params) return result.data() - def who_modifies_variable(self, variable_name: str, repo_path: Optional[str] = None) -> List[Dict]: + 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""" 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""" MATCH (var:Variable {{name: $variable_name}}) MATCH (container)-[:CONTAINS]->(var) @@ -713,8 +743,8 @@ def who_modifies_variable(self, variable_name: str, repo_path: Optional[str] = N 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) + {limit_clause} + """, **params) return result.data() @@ -781,10 +811,14 @@ def find_class_hierarchy(self, class_name: str, path: Optional[str] = None, repo "methods": methods_result.data() } - def find_function_overrides(self, function_name: str, repo_path: Optional[str] = None) -> List[Dict]: + 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""" 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""" MATCH (class:Class)-[:CONTAINS]->(func:Function {{name: $function_name}}) WHERE 1=1 {repo_filter} @@ -799,8 +833,8 @@ def find_function_overrides(self, function_name: str, repo_path: Optional[str] = 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) + {limit_clause} + """, **params) return result.data() @@ -939,7 +973,7 @@ def find_all_callees(self, function_name: str, path: Optional[str] = None, repo_ 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]: + 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, limit: Optional[int] = None) -> List[Dict]: """Find call chains between two functions""" with self.driver.session() as session: # Build match clauses based on whether files are specified @@ -948,6 +982,7 @@ def find_function_call_chain(self, start_function: str, end_function: str, max_d # KùzuDB-compatible: Use anonymous end node and filter repo_filter = "WHERE 1=1 AND (start.path IS NULL OR start.path STARTS WITH $repo_path) AND (end_target.path IS NULL OR end_target.path STARTS WITH $repo_path)" if repo_path else "" + limit_clause = "LIMIT $limit" if limit is not None else "" query = f""" MATCH (start:Function {start_props}), (end_target:Function {end_props}) {repo_filter} @@ -958,7 +993,7 @@ def find_function_call_chain(self, start_function: str, end_function: str, max_d WHERE path_end.name = end_target.name AND (end_target.path IS NULL OR path_end.path = end_target.path) RETURN func_nodes as function_nodes, call_rels as call_nodes, size(call_rels) as chain_length ORDER BY chain_length ASC - LIMIT 20 + {limit_clause} """ # Prepare parameters @@ -969,6 +1004,8 @@ def find_function_call_chain(self, start_function: str, end_function: str, max_d "end_file": end_file, "repo_path": repo_path } + if limit is not None: + params["limit"] = limit result = session.run(query, **params) @@ -1223,25 +1260,25 @@ def analyze_code_relationships(self, query_type: str, target: str, context: Opti try: if query_type == "find_callers": - results = self.who_calls_function(target, context, repo_path=repo_path) - # target_file_path is only constant when `context` pinned a single - # definition. Without it, who_calls_function matches - # Function {name: $function_name} across every file, so rows - # legitimately carry different targets — hoisting the first row's - # value and stripping the rest reported every caller as calling - # whichever definition happened to sort first. - # - # Hoist only when the rows agree; otherwise leave the field on each - # row so the ambiguity is visible to the caller. + req_limit = get_tool_result_limit("find_callers") + raw_results = self.who_calls_function(target, context, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + target_paths = {r.get("target_file_path") for r in results} target_file_path = target_paths.pop() if len(target_paths) == 1 else None if target_file_path is not None: for r in results: r.pop("target_file_path", None) + + summary = f"Found {len(results)} functions that call '{target}'" + if truncated: + summary += " (truncated — more exist)" + envelope = { "query_type": "find_callers", "target": target, "context": context, "target_file_path": target_file_path, "results": results, - "summary": f"Found {len(results)} functions that call '{target}'" + "summary": summary, "truncated": truncated, "result_limit": req_limit } if target_file_path is None and len(target_paths) > 1: envelope["note"] = ( @@ -1251,38 +1288,78 @@ def analyze_code_relationships(self, query_type: str, target: str, context: Opti return envelope elif query_type == "find_callees": - results = self.what_does_function_call(target, context, repo_path=repo_path) + req_limit = get_tool_result_limit("find_callees") + raw_results = self.what_does_function_call(target, context, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Function '{target}' calls {len(results)} other functions" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "find_callees", "target": target, "context": context, "results": results, - "summary": f"Function '{target}' calls {len(results)} other functions" + "summary": summary, "truncated": truncated, "result_limit": req_limit } elif query_type == "find_importers": - results = self.who_imports_module(target, repo_path=repo_path) + req_limit = get_tool_result_limit("find_importers") + raw_results = self.who_imports_module(target, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Found {len(results)} files that import '{target}'" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "find_importers", "target": target, "results": results, - "summary": f"Found {len(results)} files that import '{target}'" + "summary": summary, "truncated": truncated, "result_limit": req_limit } elif query_type == "find_functions_by_argument": - results = self.find_functions_by_argument(target, context, repo_path=repo_path) + req_limit = get_tool_result_limit("find_functions_by_argument") + raw_results = self.find_functions_by_argument(target, context, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Found {len(results)} functions that take '{target}' as an argument" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "find_functions_by_argument", "target": target, "context": context, "results": results, - "summary": f"Found {len(results)} functions that take '{target}' as an argument" + "summary": summary, "truncated": truncated, "result_limit": req_limit } elif query_type == "find_functions_by_decorator": - results = self.find_functions_by_decorator(target, context, repo_path=repo_path) + req_limit = get_tool_result_limit("find_functions_by_decorator") + raw_results = self.find_functions_by_decorator(target, context, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Found {len(results)} functions decorated with '{target}'" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "find_functions_by_decorator", "target": target, "context": context, "results": results, - "summary": f"Found {len(results)} functions decorated with '{target}'" + "summary": summary, "truncated": truncated, "result_limit": req_limit } elif query_type in ["who_modifies", "modifies", "mutations", "changes", "variable_usage"]: - results = self.who_modifies_variable(target, repo_path=repo_path) + req_limit = get_tool_result_limit("who_modifies") + raw_results = self.who_modifies_variable(target, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Found {len(results)} containers that hold variable '{target}'" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "who_modifies", "target": target, "results": results, - "summary": f"Found {len(results)} containers that hold variable '{target}'" + "summary": summary, "truncated": truncated, "result_limit": req_limit } elif query_type in ["class_hierarchy", "inheritance", "extends"]: @@ -1293,10 +1370,18 @@ def analyze_code_relationships(self, query_type: str, target: str, context: Opti } elif query_type in ["overrides", "implementations", "polymorphism"]: - results = self.find_function_overrides(target, repo_path=repo_path) + req_limit = get_tool_result_limit("overrides") + raw_results = self.find_function_overrides(target, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Found {len(results)} implementations of function '{target}'" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "overrides", "target": target, "results": results, - "summary": f"Found {len(results)} implementations of function '{target}'" + "summary": summary, "truncated": truncated, "result_limit": req_limit } elif query_type in ["dead_code", "unused", "unreachable"]: @@ -1336,10 +1421,18 @@ def analyze_code_relationships(self, query_type: str, target: str, context: Opti if '->' in target: start_func, end_func = target.split('->', 1) max_depth = int(context) if context and str(context).isdigit() else 5 - results = self.find_function_call_chain(start_func.strip(), end_func.strip(), max_depth, repo_path=repo_path) + req_limit = get_tool_result_limit("call_chain") + raw_results = self.find_function_call_chain(start_func.strip(), end_func.strip(), max_depth, repo_path=repo_path, limit=req_limit + 1 if req_limit is not None else None) + truncated = bool(req_limit and len(raw_results) > req_limit) + results = raw_results[:req_limit] if truncated else raw_results + + summary = f"Found {len(results)} call chains from '{start_func.strip()}' to '{end_func.strip()}' (max depth: {max_depth})" + if truncated: + summary += " (truncated — more exist)" + return { "query_type": "call_chain", "target": target, "results": results, - "summary": f"Found {len(results)} call chains from '{start_func.strip()}' to '{end_func.strip()}' (max depth: {max_depth})" + "summary": summary, "truncated": truncated, "result_limit": req_limit } else: return { diff --git a/src/codegraphcontext/tools/handlers/analysis_handlers.py b/src/codegraphcontext/tools/handlers/analysis_handlers.py index d5e2fa6e..fd6b0630 100644 --- a/src/codegraphcontext/tools/handlers/analysis_handlers.py +++ b/src/codegraphcontext/tools/handlers/analysis_handlers.py @@ -103,20 +103,13 @@ def analyze_code_relationships(code_finder: CodeFinder, **args) -> Dict[str, Any debug_log(f"Analyzing relationships: {query_type} for {target}, repo_path={repo_path}, depth={depth}") results = code_finder.analyze_code_relationships(query_type, target, context, repo_path=repo_path, depth=depth, graph_name=graph_name) - # Apply per-query-type limit (falls back to tool-level limit) - limit = get_tool_result_limit(query_type) or get_tool_result_limit("analyze_code_relationships") - truncated = False - if limit and isinstance(results, list) and len(results) > limit: - results = results[:limit] - truncated = True - response = { "success": True, "query_type": query_type, "target": target, "context": context, "results": results, } - if truncated: - response["result_limit"] = limit - response["truncated"] = True + if isinstance(results, dict) and "truncated" in results: + response["truncated"] = results["truncated"] + response["result_limit"] = results.get("result_limit") return response except Exception as e: diff --git a/tests/unit/tools/test_relationship_tool_limits.py b/tests/unit/tools/test_relationship_tool_limits.py new file mode 100644 index 00000000..88c4d43e --- /dev/null +++ b/tests/unit/tools/test_relationship_tool_limits.py @@ -0,0 +1,170 @@ +"""Unit tests for relationship query tool limits and truncation flags (Issue #1542).""" +import json +from pathlib import Path +import pytest + +from codegraphcontext.core.database_kuzu import KuzuDBManager +from codegraphcontext.tools.code_finder import CodeFinder +from codegraphcontext.tools.handlers import analysis_handlers + +kuzu = pytest.importorskip("kuzu") + + +class _KuzuDBAdapter: + def __init__(self, driver): + self._driver = driver + + def get_driver(self, graph_name=None): + return self._driver + + def get_backend_type(self) -> str: + return "kuzudb" + + +def test_find_callers_truncation(tmp_path, monkeypatch): + manager = KuzuDBManager(str(tmp_path / "db")) + driver = manager.get_driver() + try: + with driver.session() as session: + session.run( + "CREATE (:Function {uid: 'target:lib.py:1', name: 'target_fn', path: 'lib.py', " + "line_number: 1, is_dependency: false})" + ) + for i in range(25): + caller_name = f"caller_{i}" + session.run( + "CREATE (:Function {uid: $uid, name: $name, path: 'app.py', " + "line_number: $line, is_dependency: false})", + uid=f"{caller_name}:app.py:{i+10}", + name=caller_name, + line=i + 10, + ) + session.run( + "MATCH (a:Function {name: $caller}), (b:Function {name: 'target_fn'}) " + "CREATE (a)-[:CALLS {line_number: $line}]->(b)", + caller=caller_name, + line=i + 10, + ) + + finder = CodeFinder(_KuzuDBAdapter(driver)) + + # Default limit 20 + res = finder.analyze_code_relationships("find_callers", "target_fn") + assert len(res["results"]) == 20 + assert res["truncated"] is True + assert res["result_limit"] == 20 + assert "(truncated — more exist)" in res["summary"] + + # Handler forwarding + handler_res = analysis_handlers.analyze_code_relationships(finder, query_type="find_callers", target="target_fn") + assert handler_res["truncated"] is True + assert handler_res["result_limit"] == 20 + + # Custom config limit 50 via TOOL_RESULT_LIMITS + monkeypatch.setenv("TOOL_RESULT_LIMITS", json.dumps({"find_callers": 50})) + res_custom = finder.analyze_code_relationships("find_callers", "target_fn") + assert len(res_custom["results"]) == 25 + assert res_custom["truncated"] is False + assert res_custom["result_limit"] == 50 + assert "(truncated" not in res_custom["summary"] + finally: + manager.close_driver() + + +def test_find_callees_truncation(tmp_path): + manager = KuzuDBManager(str(tmp_path / "db")) + driver = manager.get_driver() + try: + with driver.session() as session: + session.run( + "CREATE (:Function {uid: 'caller:app.py:1', name: 'caller_fn', path: 'app.py', " + "line_number: 1, is_dependency: false})" + ) + for i in range(30): + callee_name = f"callee_{i}" + session.run( + "CREATE (:Function {uid: $uid, name: $name, path: 'lib.py', " + "line_number: $line, is_dependency: false})", + uid=f"{callee_name}:lib.py:{i+10}", + name=callee_name, + line=i + 10, + ) + session.run( + "MATCH (a:Function {name: 'caller_fn'}), (b:Function {name: $callee}) " + "CREATE (a)-[:CALLS {line_number: $line}]->(b)", + callee=callee_name, + line=i + 10, + ) + + finder = CodeFinder(_KuzuDBAdapter(driver)) + res = finder.analyze_code_relationships("find_callees", "caller_fn") + assert len(res["results"]) == 20 + assert res["truncated"] is True + assert res["result_limit"] == 20 + assert "(truncated — more exist)" in res["summary"] + finally: + manager.close_driver() + + +def test_find_importers_truncation(tmp_path): + manager = KuzuDBManager(str(tmp_path / "db")) + driver = manager.get_driver() + try: + with driver.session() as session: + session.run("CREATE (:Module {name: 'mod_a', full_import_name: 'mod_a'})") + for i in range(25): + file_path = f"src/file_{i}.py" + session.run( + "CREATE (:File {name: $name, path: $path, relative_path: $path, is_dependency: false})", + name=f"file_{i}.py", + path=file_path, + ) + session.run( + "MATCH (f:File {path: $path}), (m:Module {name: 'mod_a'}) " + "CREATE (f)-[:IMPORTS {alias: NULL}]->(m)", + path=file_path, + ) + + finder = CodeFinder(_KuzuDBAdapter(driver)) + res = finder.analyze_code_relationships("find_importers", "mod_a") + assert len(res["results"]) == 20 + assert res["truncated"] is True + assert res["result_limit"] == 20 + assert "(truncated — more exist)" in res["summary"] + finally: + manager.close_driver() + + +def test_regression_untruncated_results(tmp_path): + manager = KuzuDBManager(str(tmp_path / "db")) + driver = manager.get_driver() + try: + with driver.session() as session: + session.run( + "CREATE (:Function {uid: 'target:lib.py:1', name: 'target_fn', path: 'lib.py', " + "line_number: 1, is_dependency: false})" + ) + for i in range(5): + caller_name = f"caller_{i}" + session.run( + "CREATE (:Function {uid: $uid, name: $name, path: 'app.py', " + "line_number: $line, is_dependency: false})", + uid=f"{caller_name}:app.py:{i+10}", + name=caller_name, + line=i + 10, + ) + session.run( + "MATCH (a:Function {name: $caller}), (b:Function {name: 'target_fn'}) " + "CREATE (a)-[:CALLS {line_number: $line}]->(b)", + caller=caller_name, + line=i + 10, + ) + + finder = CodeFinder(_KuzuDBAdapter(driver)) + res = finder.analyze_code_relationships("find_callers", "target_fn") + assert len(res["results"]) == 5 + assert res["truncated"] is False + assert res["result_limit"] == 20 + assert "(truncated" not in res["summary"] + finally: + manager.close_driver()