diff --git a/codegen-examples/examples/swebench_agent_run/.env.template b/codegen-examples/examples/swebench_agent_run/.env.template index 3e799557e..6112a08f5 100644 --- a/codegen-examples/examples/swebench_agent_run/.env.template +++ b/codegen-examples/examples/swebench_agent_run/.env.template @@ -2,3 +2,5 @@ OPENAI_API_KEY= # Your OpenAI API key ANTHROPIC_API_KEY= # Your Anthropic API key LANGSMITH_API_KEY= # Your Langsmith API key LANGCHAIN_TRACING_V2= # `true` for tracing, `false` for no tracing +LANGCHAIN_PROJECT= # Your Langchain project +RELACE_API= # Your Relace API key diff --git a/codegen-examples/examples/swebench_agent_run/entry_point.py b/codegen-examples/examples/swebench_agent_run/entry_point.py index 0d5007419..bc5a2f757 100644 --- a/codegen-examples/examples/swebench_agent_run/entry_point.py +++ b/codegen-examples/examples/swebench_agent_run/entry_point.py @@ -4,7 +4,7 @@ image = ( modal.Image.debian_slim(python_version="3.13") - .apt_install("git") + .apt_install(["git", "ripgrep"]) .pip_install("fastapi[standard]") .copy_local_dir("../../../", "/root/codegen", ignore=[".venv", "**/.venv", "tests", "**/tests"]) .run_commands("pip install -e /root/codegen") diff --git a/src/codegen/extensions/langchain/tools.py b/src/codegen/extensions/langchain/tools.py index 0a8aac79a..849d2968a 100644 --- a/src/codegen/extensions/langchain/tools.py +++ b/src/codegen/extensions/langchain/tools.py @@ -111,23 +111,30 @@ def _run(self, dirpath: str = "./", depth: int = 1) -> str: class SearchInput(BaseModel): """Input for searching the codebase.""" - query: str = Field(..., description="The search query, passed into python's re.match()") + query: str = Field( + ..., + description="The search query to find in the codebase. When ripgrep is available, this will be passed as a ripgrep pattern. For regex searches, set use_regex=True. Ripgrep is the preferred method.", + ) target_directories: Optional[list[str]] = Field(default=None, description="Optional list of directories to search in") + file_extensions: Optional[list[str]] = Field(default=None, description="Optional list of file extensions to search (e.g. ['.py', '.ts'])") + page: int = Field(default=1, description="Page number to return (1-based, default: 1)") + files_per_page: int = Field(default=10, description="Number of files to return per page (default: 10)") + use_regex: bool = Field(default=False, description="Whether to treat query as a regex pattern (default: False)") class SearchTool(BaseTool): """Tool for searching the codebase.""" name: ClassVar[str] = "search" - description: ClassVar[str] = "Search the codebase using text search" + description: ClassVar[str] = "Search the codebase using text search or regex pattern matching" args_schema: ClassVar[type[BaseModel]] = SearchInput codebase: Codebase = Field(exclude=True) def __init__(self, codebase: Codebase) -> None: super().__init__(codebase=codebase) - def _run(self, query: str, target_directories: Optional[list[str]] = None) -> str: - result = search(self.codebase, query, target_directories) + def _run(self, query: str, target_directories: Optional[list[str]] = None, file_extensions: Optional[list[str]] = None, page: int = 1, files_per_page: int = 10, use_regex: bool = False) -> str: + result = search(self.codebase, query, target_directories=target_directories, file_extensions=file_extensions, page=page, files_per_page=files_per_page, use_regex=use_regex) return result.render() diff --git a/src/codegen/extensions/mcp/codebase_tools.py b/src/codegen/extensions/mcp/codebase_tools.py index 744430d47..52a25b1d6 100644 --- a/src/codegen/extensions/mcp/codebase_tools.py +++ b/src/codegen/extensions/mcp/codebase_tools.py @@ -37,16 +37,19 @@ def reveal_symbol_tool( return json.dumps(result, indent=2) -@mcp.tool(name="search_codebase", description="Search the codebase using text search or regex pattern matching") +@mcp.tool(name="search_codebase", description="The search query to find in the codebase. When ripgrep is available, this will be passed as a ripgrep pattern. For regex searches, set use_regex=True") def search_codebase_tool( - query: str, - target_directories: Annotated[Optional[list[str]], "list of directories to search within"], + query: Annotated[str, "The search query to find in the codebase. When ripgrep is available, this will be passed as a ripgrep pattern. For regex searches, set use_regex=True."], codebase_dir: Annotated[str, "The root directory of your codebase"], codebase_language: Annotated[ProgrammingLanguage, "The language the codebase is written in"], - use_regex: Annotated[bool, "use regex for the search query"], + target_directories: Annotated[Optional[list[str]], "list of directories to search within"] = None, + file_extensions: Annotated[Optional[list[str]], "list of file extensions to search (e.g. ['.py', '.ts'])"] = None, + page: Annotated[int, "page number to return (1-based)"] = 1, + files_per_page: Annotated[int, "number of files to return per page"] = 10, + use_regex: Annotated[bool, "use regex for the search query"] = False, ): codebase = Codebase(repo_path=codebase_dir, language=codebase_language) - result = search(codebase, query, target_directories, use_regex=use_regex) + result = search(codebase, query, target_directories=target_directories, file_extensions=file_extensions, page=page, files_per_page=files_per_page, use_regex=use_regex) return json.dumps(result, indent=2) diff --git a/src/codegen/extensions/tools/search.py b/src/codegen/extensions/tools/search.py index 8bdb4d214..4bcdfb74e 100644 --- a/src/codegen/extensions/tools/search.py +++ b/src/codegen/extensions/tools/search.py @@ -5,7 +5,9 @@ Results are paginated with a default of 10 files per page. """ +import os import re +import subprocess from typing import ClassVar, Optional from pydantic import Field @@ -109,7 +111,7 @@ def render(self) -> str: return "\n".join(lines) -def search( +def _search_with_ripgrep( codebase: Codebase, query: str, target_directories: Optional[list[str]] = None, @@ -118,25 +120,159 @@ def search( files_per_page: int = 10, use_regex: bool = False, ) -> SearchObservation: - """Search the codebase using text search or regex pattern matching. + """Search the codebase using ripgrep. - If use_regex is True, performs a regex pattern match on each line. - Otherwise, performs a case-insensitive text search. - Returns matching lines with their line numbers, grouped by file. - Results are paginated by files, with a default of 10 files per page. + This is faster than the Python implementation, especially for large codebases. + """ + # Build ripgrep command + cmd = ["rg", "--line-number"] + + # Add case insensitivity if not using regex + if not use_regex: + cmd.append("--fixed-strings") + cmd.append("--ignore-case") + + # Add file extensions if specified + if file_extensions: + for ext in file_extensions: + # Remove leading dot if present + ext = ext[1:] if ext.startswith(".") else ext + cmd.extend(["--type-add", f"custom:{ext}", "--type", "custom"]) + + # Add target directories if specified + search_path = codebase.repo_path + if target_directories: + # We'll handle target directories by filtering results later + pass + + # Add the query and path + cmd.append(query) + cmd.append(search_path) + + # Run ripgrep + try: + # Use text mode and UTF-8 encoding + result = subprocess.run( + cmd, + capture_output=True, + text=True, + encoding="utf-8", + check=False, # Don't raise exception on non-zero exit code (no matches) + ) + + # Parse the output + all_results: dict[str, list[SearchMatch]] = {} + + # ripgrep returns non-zero exit code when no matches are found + if result.returncode != 0 and result.returncode != 1: + # Real error occurred + return SearchObservation( + status="error", + error=f"ripgrep error: {result.stderr}", + query=query, + page=page, + total_pages=0, + total_files=0, + files_per_page=files_per_page, + results=[], + ) - Args: - codebase: The codebase to operate on - query: The text to search for or regex pattern to match - target_directories: Optional list of directories to search in - file_extensions: Optional list of file extensions to search (e.g. ['.py', '.ts']). - If None, searches all files ('*') - page: Page number to return (1-based, default: 1) - files_per_page: Number of files to return per page (default: 10) - use_regex: Whether to treat query as a regex pattern (default: False) + # Parse output lines + for line in result.stdout.splitlines(): + # ripgrep output format: file:line:content + parts = line.split(":", 2) + if len(parts) < 3: + continue + + filepath, line_number_str, content = parts + + # Convert to relative path within the codebase + rel_path = os.path.relpath(filepath, codebase.repo_path) + + # Skip if not in target directories + if target_directories and not any(rel_path.startswith(d) for d in target_directories): + continue + + try: + line_number = int(line_number_str) + + # Find the actual match text + match_text = query + if use_regex: + # For regex, we need to find what actually matched + # This is a simplification - ideally we'd use ripgrep's --json option + # to get the exact match positions + pattern = re.compile(query) + match_obj = pattern.search(content) + if match_obj: + match_text = match_obj.group(0) + + # Create or append to file results + if rel_path not in all_results: + all_results[rel_path] = [] + + all_results[rel_path].append( + SearchMatch( + status="success", + line_number=line_number, + line=content.strip(), + match=match_text, + ) + ) + except ValueError: + # Skip lines with invalid line numbers + continue + + # Convert to SearchFileResult objects + file_results = [] + for filepath, matches in all_results.items(): + file_results.append( + SearchFileResult( + status="success", + filepath=filepath, + matches=sorted(matches, key=lambda x: x.line_number), + ) + ) - Returns: - SearchObservation containing search results with matches and their sources + # Sort results by filepath + file_results.sort(key=lambda x: x.filepath) + + # Calculate pagination + total_files = len(file_results) + total_pages = (total_files + files_per_page - 1) // files_per_page + start_idx = (page - 1) * files_per_page + end_idx = start_idx + files_per_page + + # Get the current page of results + paginated_results = file_results[start_idx:end_idx] + + return SearchObservation( + status="success", + query=query, + page=page, + total_pages=total_pages, + total_files=total_files, + files_per_page=files_per_page, + results=paginated_results, + ) + + except (subprocess.SubprocessError, FileNotFoundError) as e: + # Let the caller handle this by falling back to Python implementation + raise + + +def _search_with_python( + codebase: Codebase, + query: str, + target_directories: Optional[list[str]] = None, + file_extensions: Optional[list[str]] = None, + page: int = 1, + files_per_page: int = 10, + use_regex: bool = False, +) -> SearchObservation: + """Search the codebase using Python's regex engine. + + This is a fallback for when ripgrep is not available. """ # Validate pagination parameters if page < 1: @@ -225,3 +361,41 @@ def search( files_per_page=files_per_page, results=paginated_results, ) + + +def search( + codebase: Codebase, + query: str, + target_directories: Optional[list[str]] = None, + file_extensions: Optional[list[str]] = None, + page: int = 1, + files_per_page: int = 10, + use_regex: bool = False, +) -> SearchObservation: + """Search the codebase using text search or regex pattern matching. + + Uses ripgrep for performance when available, with fallback to Python's regex engine. + If use_regex is True, performs a regex pattern match on each line. + Otherwise, performs a case-insensitive text search. + Returns matching lines with their line numbers, grouped by file. + Results are paginated by files, with a default of 10 files per page. + + Args: + codebase: The codebase to operate on + query: The text to search for or regex pattern to match + target_directories: Optional list of directories to search in + file_extensions: Optional list of file extensions to search (e.g. ['.py', '.ts']). + If None, searches all files ('*') + page: Page number to return (1-based, default: 1) + files_per_page: Number of files to return per page (default: 10) + use_regex: Whether to treat query as a regex pattern (default: False) + + Returns: + SearchObservation containing search results with matches and their sources + """ + # Try to use ripgrep first + try: + return _search_with_ripgrep(codebase, query, target_directories, file_extensions, page, files_per_page, use_regex) + except (FileNotFoundError, subprocess.SubprocessError): + # Fall back to Python implementation if ripgrep fails or isn't available + return _search_with_python(codebase, query, target_directories, file_extensions, page, files_per_page, use_regex) diff --git a/tests/unit/codegen/extensions/test_tools.py b/tests/unit/codegen/extensions/test_tools.py index 9d2b6fdb5..ec394312e 100644 --- a/tests/unit/codegen/extensions/test_tools.py +++ b/tests/unit/codegen/extensions/test_tools.py @@ -1,5 +1,7 @@ """Tests for codebase tools.""" +import subprocess + import pytest from codegen.extensions.tools import ( @@ -236,6 +238,222 @@ def test_search(codebase): assert result.status == "success" assert len(result.results) > 0 + # Check that we found the right content + assert any("hello" in match.match.lower() for file_result in result.results for match in file_result.matches) + + # Check pagination info + assert result.page == 1 + assert result.total_pages >= 1 + assert result.files_per_page == 10 + + +def test_search_regex(codebase): + """Test searching with regex.""" + # Search for function definitions + result = search(codebase, r"def\s+\w+", use_regex=True) + assert result.status == "success" + assert len(result.results) > 0 + + # Should find both 'def hello' and 'def greet' + matches = [match.line for file_result in result.results for match in file_result.matches] + assert any("def hello" in match for match in matches) + assert any("def greet" in match for match in matches) + + +def test_search_target_directories(codebase): + """Test searching with target directory filtering.""" + # First search without filter to ensure we have results + result_all = search(codebase, "hello") + assert result_all.status == "success" + assert len(result_all.results) > 0 + + # Now search with correct target directory + result_filtered = search(codebase, "hello", target_directories=["src"]) + assert result_filtered.status == "success" + assert len(result_filtered.results) > 0 + + # Search with non-existent directory + result_none = search(codebase, "hello", target_directories=["nonexistent"]) + assert result_none.status == "success" + assert len(result_none.results) == 0 + + +def test_search_file_extensions(codebase, tmpdir): + """Test searching with file extension filtering.""" + # Add a non-Python file + js_content = "function hello() { console.log('Hello from JS!'); }" + js_file = tmpdir / "src" / "script.js" + js_file.write_text(js_content, encoding="utf-8") + + # Search all files + result_all = search(codebase, "hello") + assert result_all.status == "success" + assert len(result_all.results) > 0 + + # Search only Python files + result_py = search(codebase, "hello", file_extensions=[".py"]) + assert result_py.status == "success" + assert all(file_result.filepath.endswith(".py") for file_result in result_py.results) + + # Search only JS files + result_js = search(codebase, "hello", file_extensions=[".js"]) + assert result_js.status == "success" + if len(result_js.results) > 0: # Only if JS file was properly added to codebase + assert all(file_result.filepath.endswith(".js") for file_result in result_js.results) + + +def test_search_pagination(codebase, tmpdir): + """Test search pagination.""" + # Create multiple files to test pagination + files_dict = {} + for i in range(15): # Create enough files to span multiple pages + content = f"def function_{i}():\n print('Hello from function {i}!')" + files_dict[f"src/file_{i}.py"] = content + + # Create a new codebase with all the files + with get_codebase_session(tmpdir=tmpdir, files=files_dict) as pagination_codebase: + # Search with default pagination (page 1) + result_page1 = search(pagination_codebase, "Hello", files_per_page=5) + assert result_page1.status == "success" + assert result_page1.page == 1 + assert len(result_page1.results) <= 5 + + # If we have enough results for multiple pages + if result_page1.total_pages > 1: + # Get page 2 + result_page2 = search(pagination_codebase, "Hello", page=2, files_per_page=5) + assert result_page2.status == "success" + assert result_page2.page == 2 + assert len(result_page2.results) <= 5 + + # Ensure different files on different pages + page1_files = {r.filepath for r in result_page1.results} + page2_files = {r.filepath for r in result_page2.results} + assert not page1_files.intersection(page2_files) + + +def test_search_invalid_regex(codebase): + """Test search with invalid regex pattern.""" + result = search(codebase, "(unclosed", use_regex=True) + assert result.status == "error" + # Check for either Python's error message or ripgrep's error message + assert any( + error_msg in result.error + for error_msg in [ + "Invalid regex pattern", # Python error message + "regex parse error", # ripgrep error message + "unclosed group", # Common error description + ] + ) + + +def test_search_fallback(codebase, monkeypatch): + """Test fallback to Python implementation when ripgrep fails.""" + + # Mock subprocess.run to simulate ripgrep failure + def mock_subprocess_run(*args, **kwargs): + msg = "Simulated ripgrep failure" + raise subprocess.SubprocessError(msg) + + # Apply the mock + monkeypatch.setattr(subprocess, "run", mock_subprocess_run) + + # Search should still work using Python fallback + result = search(codebase, "hello") + assert result.status == "success" + assert len(result.results) > 0 + + +def test_search_ripgrep_not_found(codebase, monkeypatch): + """Test fallback to Python implementation when ripgrep is not installed.""" + + # Mock subprocess.run to simulate ripgrep not found + def mock_subprocess_run(*args, **kwargs): + msg = "Simulated ripgrep not found" + raise FileNotFoundError(msg) + + # Apply the mock + monkeypatch.setattr(subprocess, "run", mock_subprocess_run) + + # Search should still work using Python fallback + result = search(codebase, "hello") + assert result.status == "success" + assert len(result.results) > 0 + + +def test_search_uses_ripgrep(codebase, monkeypatch): + """Test that ripgrep is used when available.""" + # Track if ripgrep was called + ripgrep_called = False + + # Store original subprocess.run + original_run = subprocess.run + + # Mock subprocess.run to track calls and then call the original + def mock_subprocess_run(*args, **kwargs): + nonlocal ripgrep_called + # Check if this is a ripgrep call + if args and args[0] and isinstance(args[0], list) and args[0][0] == "rg": + ripgrep_called = True + # Call the original implementation + return original_run(*args, **kwargs) + + # Apply the mock + monkeypatch.setattr(subprocess, "run", mock_subprocess_run) + + # Perform a search + result = search(codebase, "hello") + assert result.status == "success" + + # Verify ripgrep was called + assert ripgrep_called, "Ripgrep was not used for the search" + + +def test_search_implementation_consistency(codebase, monkeypatch): + """Test that ripgrep and Python implementations produce consistent results.""" + from codegen.extensions.tools.search import _search_with_python, _search_with_ripgrep + + # Skip test if ripgrep is not available + try: + subprocess.run(["rg", "--version"], capture_output=True, check=False) + except FileNotFoundError: + pytest.skip("Ripgrep not available, skipping consistency test") + + # Simple search that should work in both implementations + query = "hello" + + # Get results from both implementations + ripgrep_result = _search_with_ripgrep(codebase, query) + python_result = _search_with_python(codebase, query) + + # Compare basic metadata + assert ripgrep_result.status == python_result.status + assert ripgrep_result.query == python_result.query + + # Compare file paths found (order might differ) + ripgrep_files = {r.filepath for r in ripgrep_result.results} + python_files = {r.filepath for r in python_result.results} + + # There might be slight differences in which files are found due to how ripgrep handles + # certain files, so we'll check for substantial overlap rather than exact equality + common_files = ripgrep_files.intersection(python_files) + assert len(common_files) > 0, "No common files found between ripgrep and Python implementations" + + # For common files, compare the line numbers found + for filepath in common_files: + # Find the corresponding file results + ripgrep_file_result = next(r for r in ripgrep_result.results if r.filepath == filepath) + python_file_result = next(r for r in python_result.results if r.filepath == filepath) + + # Compare line numbers - there might be slight differences in how matches are found + ripgrep_lines = {m.line_number for m in ripgrep_file_result.matches} + python_lines = {m.line_number for m in python_file_result.matches} + + # Check for substantial overlap in line numbers + common_lines = ripgrep_lines.intersection(python_lines) + if ripgrep_lines and python_lines: # Only check if both found matches + assert len(common_lines) > 0, f"No common line matches found in {filepath}" + def test_edit_file(codebase): """Test editing a file."""