Add search strategy implementations for ugrep, ripgrep, ag, and grep

This commit introduces multiple search strategies to enhance the code indexer's search capabilities. Each strategy is encapsulated in its own class, implementing a common interface for searching code. The new strategies include:

- UgrepStrategy: Utilizes the ugrep command-line tool, supporting fuzzy search.
- RipgrepStrategy: Implements the ripgrep tool for fast searching.
- AgStrategy: Integrates The Silver Searcher (ag) for efficient searching.
- GrepStrategy: Provides a fallback using the standard grep tool.
- BasicSearchStrategy: A pure-Python implementation for basic searching when no external tools are available.

Additionally, the project settings have been updated to manage available search strategies dynamically, improving the overall search functionality and user experience.

Author: johnhuang316

src/code_index_mcp/project_settings.py

@@ -16,6 +16,39 @@
 from .constants import (
     SETTINGS_DIR, CONFIG_FILE, INDEX_FILE, CACHE_FILE
 )
+from .search.base import SearchStrategy
+from .search.ugrep import UgrepStrategy
+from .search.ripgrep import RipgrepStrategy
+from .search.ag import AgStrategy
+from .search.grep import GrepStrategy
+from .search.basic import BasicSearchStrategy
+
+
+# Prioritized list of search strategies
+SEARCH_STRATEGY_CLASSES = [
+    UgrepStrategy,
+    RipgrepStrategy,
+    AgStrategy,
+    GrepStrategy,
+    BasicSearchStrategy,
+]
+
+
+def _get_available_strategies() -> list[SearchStrategy]:
+    """
+    Detect and return a list of available search strategy instances,
+    ordered by preference.
+    """
+    available = []
+    for strategy_class in SEARCH_STRATEGY_CLASSES:
+        try:
+            strategy = strategy_class()
+            if strategy.is_available():
+                available.append(strategy)
+        except Exception as e:
+            print(f"Error initializing strategy {strategy_class.__name__}: {e}")
+    return available
+
 
 class ProjectSettings:
     """Class for managing project settings and index data"""
@@ -29,7 +62,8 @@ def __init__(self, base_path, skip_load=False):
         """
         self.base_path = base_path
         self.skip_load = skip_load
-        self._search_tools_cache = None  # Lazy-loaded search tools configuration
+        self.available_strategies: list[SearchStrategy] = []
+        self.refresh_available_strategies()
 
         # Ensure the base path of the temporary directory exists
         try:
@@ -478,84 +512,31 @@ def get_stats(self):
             }
 
     def get_search_tools_config(self):
-        """Get search tools configuration with lazy loading.
-        
-        Returns:
-            dict: Search tools configuration with preferred tool and available tools
-        """
-        if self._search_tools_cache is None:
-            print("Detecting available search tools...")
-            self._search_tools_cache = self._detect_search_tools()
-            print(f"Search tools detected. Preferred: {self._search_tools_cache.get('preferred_tool', 'basic')}")
-        
-        return self._search_tools_cache
+        """Get the configuration of available search tools.
 
-    def get_preferred_search_tool(self):
-        """Get the preferred search tool name.
-        
         Returns:
-            str: Name of preferred search tool ('ripgrep', 'ag', 'grep', or 'basic')
+            dict: A dictionary containing the list of available tool names.
         """
-        config = self.get_search_tools_config()
-        return config.get('preferred_tool', 'basic')
+        return {
+            "available_tools": [s.name for s in self.available_strategies],
+            "preferred_tool": self.get_preferred_search_tool().name if self.available_strategies else None
+        }
+
+    def get_preferred_search_tool(self) -> SearchStrategy | None:
+        """Get the preferred search tool based on availability and priority.
 
-    def _detect_search_tools(self):
-        """Detect available search tools on the system.
-        
         Returns:
-            dict: Configuration with available tools and preferred tool
+            SearchStrategy: An instance of the preferred search strategy, or None.
         """
-        tools_info = {
-            'detected_at': self._get_timestamp(),
-            'available_tools': {},
-            'preferred_tool': 'basic'
-        }
-        
-        # Check tools in priority order: ugrep > ripgrep > ag > grep
-        search_tools = [
-            ('ugrep', 'ug'),
-            ('ripgrep', 'rg'),
-            ('ag', 'ag'), 
-            ('grep', 'grep')
-        ]
+        if not self.available_strategies:
+            self.refresh_available_strategies()
 
-        for tool_name, command in search_tools:
-            is_available = self._is_tool_available(command)
-            tools_info['available_tools'][tool_name] = is_available
-            
-            # Set the first available tool as preferred
-            if is_available and tools_info['preferred_tool'] == 'basic':
-                tools_info['preferred_tool'] = tool_name
-        
-        return tools_info
+        return self.available_strategies[0] if self.available_strategies else None
 
-    def _is_tool_available(self, command):
-        """Check if a search tool is available on the system.
-        
-        Args:
-            command (str): Command to check (e.g., 'rg', 'ag', 'grep')
-            
-        Returns:
-            bool: True if tool is available, False otherwise
+    def refresh_available_strategies(self):
         """
-        try:
-            result = subprocess.run(
-                [command, '--version'], 
-                capture_output=True, 
-                timeout=3,
-                check=False
-            )
-            return result.returncode == 0
-        except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
-            return False
-
-    def refresh_search_tools(self):
-        """Manually refresh search tools detection.
-        
-        Returns:
-            dict: Updated search tools configuration
+        Force a refresh of the available search tools list.
         """
-        print("Refreshing search tools detection...")
-        self._search_tools_cache = self._detect_search_tools()
-        print(f"Search tools refreshed. Preferred: {self._search_tools_cache.get('preferred_tool', 'basic')}")
-        return self._search_tools_cache
+        print("Refreshing available search strategies...")
+        self.available_strategies = _get_available_strategies()
+        print(f"Available strategies found: {[s.name for s in self.available_strategies]}")

src/code_index_mcp/search/__init__.py

@@ -0,0 +1 @@
+"""Search strategies package."""

src/code_index_mcp/search/ag.py

@@ -0,0 +1,89 @@
+"""
+Search Strategy for The Silver Searcher (ag)
+"""
+import shutil
+import subprocess
+from typing import Dict, List, Optional, Tuple
+
+from .base import SearchStrategy, parse_search_output, create_safe_fuzzy_pattern
+
+class AgStrategy(SearchStrategy):
+    """Search strategy using 'The Silver Searcher' (ag) command-line tool."""
+
+    @property
+    def name(self) -> str:
+        """The name of the search tool."""
+        return 'ag'
+
+    def is_available(self) -> bool:
+        """Check if 'ag' command is available on the system."""
+        return shutil.which('ag') is not None
+
+    def search(
+        self,
+        pattern: str,
+        base_path: str,
+        case_sensitive: bool = True,
+        context_lines: int = 0,
+        file_pattern: Optional[str] = None,
+        fuzzy: bool = False
+    ) -> Dict[str, List[Tuple[int, str]]]:
+        """
+        Execute a search using The Silver Searcher (ag).
+
+        Note: ag does not support native fuzzy searching. When fuzzy=True, a
+        safe fuzzy pattern with word boundaries is used for regex search.
+        When fuzzy=False, a literal string search is performed.
+        """
+        # ag prints line numbers and groups by file by default, which is good.
+        # --noheading is used to be consistent with other tools' output format.
+        cmd = ['ag', '--noheading']
+
+        if not case_sensitive:
+            cmd.append('--ignore-case')
+
+        # Prepare search pattern
+        search_pattern = pattern
+        if fuzzy:
+            # Use safe fuzzy pattern for regex search
+            search_pattern = create_safe_fuzzy_pattern(pattern)
+        else:
+            cmd.append('--literal') # or -Q
+
+        if context_lines > 0:
+            cmd.extend(['--before', str(context_lines)])
+            cmd.extend(['--after', str(context_lines)])
+            
+        if file_pattern:
+            # Use -G to filter files by regex pattern
+            cmd.extend(['-G', file_pattern])
+
+        # Add -- to treat pattern as a literal argument, preventing injection
+        cmd.append('--')
+        cmd.append(search_pattern)
+        cmd.append(base_path)
+        
+        try:
+            # ag exits with 1 if no matches are found, which is not an error.
+            # It exits with 0 on success (match found). Other codes are errors.
+            process = subprocess.run(
+                cmd, 
+                capture_output=True, 
+                text=True, 
+                encoding='utf-8',
+                errors='replace',
+                check=False  # Do not raise CalledProcessError on non-zero exit
+            )
+            # We don't check returncode > 1 because ag's exit code behavior
+            # is less standardized than rg/ug. 0 for match, 1 for no match.
+            # Any actual error will likely raise an exception or be in stderr.
+            if process.returncode > 1:
+                 raise RuntimeError(f"ag failed with exit code {process.returncode}: {process.stderr}")
+
+            return parse_search_output(process.stdout, base_path)
+        
+        except FileNotFoundError:
+            raise RuntimeError("'ag' (The Silver Searcher) not found. Please install it and ensure it's in your PATH.")
+        except Exception as e:
+            # Re-raise other potential exceptions like permission errors
+            raise RuntimeError(f"An error occurred while running ag: {e}") 

src/code_index_mcp/search/base.py

@@ -0,0 +1,141 @@
+"""
+Search Strategies for Code Indexer
+
+This module defines the abstract base class for search strategies and will contain
+concrete implementations for different search tools like ugrep, ripgrep, etc.
+"""
+import os
+import re
+import shutil
+import subprocess
+import sys
+from abc import ABC, abstractmethod
+from typing import Dict, List, Optional, Tuple, Any
+
+def parse_search_output(output: str, base_path: str) -> Dict[str, List[Tuple[int, str]]]:
+    """
+    Parse the output of command-line search tools (grep, ag, rg).
+
+    Args:
+        output: The raw output from the command-line tool.
+        base_path: The base path of the project to make file paths relative.
+
+    Returns:
+        A dictionary where keys are file paths and values are lists of (line_number, line_content) tuples.
+    """
+    results = {}
+    # Normalize base_path to ensure consistent path separation
+    normalized_base_path = os.path.normpath(base_path)
+
+    for line in output.strip().split('\n'):
+        if not line.strip():
+            continue
+        try:
+            # Handle Windows paths which might have a drive letter, e.g., C:
+            parts = line.split(':', 2)
+            if sys.platform == "win32" and len(parts[0]) == 1 and parts[1].startswith('\\'):
+                # Re-join drive letter with the rest of the path
+                file_path_abs = f"{parts[0]}:{parts[1]}"
+                line_number_str = parts[2].split(':', 1)[0]
+                content = parts[2].split(':', 1)[1]
+            else:
+                file_path_abs = parts[0]
+                line_number_str = parts[1]
+                content = parts[2]
+            
+            line_number = int(line_number_str)
+
+            # Make the file path relative to the base_path
+            relative_path = os.path.relpath(file_path_abs, normalized_base_path)
+            
+            # Normalize path separators for consistency
+            relative_path = relative_path.replace('\\', '/')
+
+            if relative_path not in results:
+                results[relative_path] = []
+            results[relative_path].append((line_number, content))
+        except (ValueError, IndexError):
+            # Silently ignore lines that don't match the expected format
+            # This can happen with summary lines or other tool-specific output
+            pass
+
+    return results
+
+
+def create_safe_fuzzy_pattern(pattern: str) -> str:
+    """
+    Create safe fuzzy search patterns that are more permissive than exact match
+    but still safe from regex injection attacks.
+    
+    Args:
+        pattern: Original search pattern
+        
+    Returns:
+        Safe fuzzy pattern for extended regex
+    """
+    # Escape any regex special characters to make them literal
+    escaped = re.escape(pattern)
+    
+    # Create fuzzy pattern that matches:
+    # 1. Word at start of word boundary (e.g., "test" in "testing")
+    # 2. Word at end of word boundary (e.g., "test" in "mytest") 
+    # 3. Whole word (e.g., "test" as standalone word)
+    if len(pattern) >= 3:  # Only for patterns of reasonable length
+        # This pattern allows partial matches at word boundaries
+        fuzzy_pattern = f"\\b{escaped}|{escaped}\\b"
+    else:
+        # For short patterns, require full word boundaries to avoid too many matches
+        fuzzy_pattern = f"\\b{escaped}\\b"
+    
+    return fuzzy_pattern
+
+
+class SearchStrategy(ABC):
+    """
+    Abstract base class for a search strategy.
+    
+    Each strategy is responsible for searching code using a specific tool or method.
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """The name of the search tool (e.g., 'ugrep', 'ripgrep')."""
+        pass
+
+    @abstractmethod
+    def is_available(self) -> bool:
+        """
+        Check if the search tool for this strategy is available on the system.
+        
+        Returns:
+            True if the tool is available, False otherwise.
+        """
+        pass
+
+    @abstractmethod
+    def search(
+        self,
+        pattern: str,
+        base_path: str,
+        case_sensitive: bool = True,
+        context_lines: int = 0,
+        file_pattern: Optional[str] = None,
+        fuzzy: bool = False
+    ) -> Dict[str, List[Tuple[int, str]]]:
+        """
+        Execute a search using the specific strategy.
+
+        Args:
+            pattern: The search pattern (string or regex).
+            base_path: The root directory to search in.
+            case_sensitive: Whether the search is case-sensitive.
+            context_lines: Number of context lines to show around each match.
+            file_pattern: Glob pattern to filter files (e.g., "*.py").
+            fuzzy: Whether to enable fuzzy search.
+
+        Returns:
+            A dictionary mapping filenames to lists of (line_number, line_content) tuples.
+        """
+        pass
+