Refactor search in files/iterative file processing

* Introduce abstractions: FileCollection, FileProxy
* Unify handling of pattern search in project files
  (source code vs. all files)

This prepares for processing of external files,
which can implement FileProxy differently.

src/serena/project.py

@@ -14,7 +14,8 @@ from serena.config.serena_config import (
 )
 from serena.ls_manager import LanguageServerFactory, LanguageServerManager
 from serena.memories.memory_manager import MemoryManager
-from serena.util.file_system import GitignoreParser, match_path
+from serena.util.file_proxy import FileCollection
+from serena.util.file_system import GitignoreParser, match_path, scan_directory
 from serena.util.text_utils import MatchedConsecutiveLines, search_files
 from solidlsp import SolidLanguageServer
 from solidlsp.ls_config import Language
@@ -344,7 +345,7 @@ class Project(ToStringMixin):
                         )
             return rel_file_paths
 
-    def search_source_files_for_pattern(
+    def search_project_files_for_pattern(
         self,
         pattern: str,
         relative_path: str = "",
@@ -353,6 +354,7 @@ class Project(ToStringMixin):
         paths_include_glob: str | None = None,
         paths_exclude_glob: str | None = None,
         multiline: bool = True,
+        code_files_only: bool = True,
     ) -> list[MatchedConsecutiveLines]:
         """
         Search for a pattern across all (non-ignored) source files
@@ -366,12 +368,26 @@ class Project(ToStringMixin):
         :param multiline: Whether to compile the regex with the DOTALL flag (``.`` matches newlines).
         :return: List of matched consecutive lines with context
         """
-        relative_file_paths = self.gather_source_files(relative_path=relative_path)
+        if code_files_only:
+            relative_file_paths = self.gather_source_files(relative_path=relative_path)
+            file_collection = FileCollection.from_local_project_paths(relative_file_paths, self)
+        else:
+            abs_path = os.path.join(self.project_root, relative_path)
+            if os.path.isfile(abs_path):
+                rel_paths_to_search = [relative_path]
+            else:
+                _dirs, rel_paths_to_search = scan_directory(
+                    path=abs_path,
+                    recursive=True,
+                    is_ignored_dir=self.is_ignored_path,
+                    is_ignored_file=self.is_ignored_path,
+                    relative_to=self.project_root,
+                )
+            file_collection = FileCollection.from_local_project_paths(rel_paths_to_search, self)
+
         return search_files(
-            relative_file_paths,
+            file_collection,
             pattern,
-            root_path=self.project_root,
-            file_reader=self.read_file,
             context_lines_before=context_lines_before,
             context_lines_after=context_lines_after,
             paths_include_glob=paths_include_glob,

src/serena/tools/file_tools.py

@@ -20,7 +20,6 @@ from serena.util.text_utils import (
     ReplacementOccurrence,
     expand_braces,
     glob_match,
-    search_files,
 )
 
 
@@ -599,40 +598,16 @@ class SearchForPatternTool(Tool):
         if not os.path.exists(abs_path):
             raise FileNotFoundError(f"Relative path {relative_path} does not exist.")
 
-        if restrict_search_to_code_files:
-            matches = self.project.search_source_files_for_pattern(
-                pattern=substring_pattern,
-                relative_path=relative_path,
-                context_lines_before=context_lines_before,
-                context_lines_after=context_lines_after,
-                paths_include_glob=paths_include_glob.strip(),
-                paths_exclude_glob=paths_exclude_glob.strip(),
-                multiline=multiline,
-            )
-        else:
-            if os.path.isfile(abs_path):
-                rel_paths_to_search = [relative_path]
-            else:
-                _dirs, rel_paths_to_search = scan_directory(
-                    path=abs_path,
-                    recursive=True,
-                    is_ignored_dir=self.project.is_ignored_path,
-                    is_ignored_file=self.project.is_ignored_path,
-                    relative_to=self.get_project_root(),
-                )
-            # TODO (maybe): not super efficient to walk through the files again and filter if glob patterns are provided
-            #   but it probably never matters and this version required no further refactoring
-            matches = search_files(
-                rel_paths_to_search,
-                substring_pattern,
-                context_lines_before=context_lines_before,
-                context_lines_after=context_lines_after,
-                file_reader=self.project.read_file,
-                root_path=self.get_project_root(),
-                paths_include_glob=paths_include_glob,
-                paths_exclude_glob=paths_exclude_glob,
-                multiline=multiline,
-            )
+        matches = self.project.search_project_files_for_pattern(
+            pattern=substring_pattern,
+            relative_path=relative_path,
+            context_lines_before=context_lines_before,
+            context_lines_after=context_lines_after,
+            paths_include_glob=paths_include_glob.strip(),
+            paths_exclude_glob=paths_exclude_glob.strip(),
+            multiline=multiline,
+            code_files_only=restrict_search_to_code_files,
+        )
 
         # group matches by file
         file_to_matches: dict[str, list[str]] = defaultdict(list)

src/serena/util/file_proxy.py

@@ -0,0 +1,83 @@
+import logging
+import os
+from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from typing import TYPE_CHECKING, Self
+
+if TYPE_CHECKING:
+    from serena.project import Project
+
+log = logging.getLogger(__name__)
+
+
+class FileProxy(ABC):
+    @abstractmethod
+    def get_contents(self) -> str:
+        """:return: the contents of the file as a string."""
+
+    @abstractmethod
+    def get_relative_path(self) -> str:
+        """:return: the relative path reported by Serena (actual relative path or encoded external path)"""
+
+
+class LocalProjectFileProxy(FileProxy):
+    def __init__(self, relative_path: str, project: "Project"):
+        self._relative_path = relative_path
+        self._project = project
+
+    def get_contents(self) -> str:
+        abs_path = os.path.join(self._project.project_root, self._relative_path)
+        with open(abs_path, encoding=self._project.project_config.encoding) as f:
+            return f.read()
+
+    def get_relative_path(self) -> str:
+        return self._relative_path
+
+
+class FileCollection:
+    def __init__(self, file_proxies: list[FileProxy]):
+        self._file_proxies = file_proxies
+
+    def __len__(self) -> int:
+        return len(self._file_proxies)
+
+    def __iter__(self) -> Iterator[FileProxy]:
+        return iter(self._file_proxies)
+
+    @classmethod
+    def from_local_project_paths(cls, relative_paths: list[str], project: "Project") -> Self:
+        return cls([LocalProjectFileProxy(path, project) for path in relative_paths])
+
+    def filter_glob(self, paths_include_glob: str | None = None, paths_exclude_glob: str | None = None) -> "FileCollection":
+        """
+        Filters the collection based on the given patterns.
+        Note: Filtering is applied only to local project files. Other files are always retained.
+
+        :param paths_include_glob: optional glob pattern to include files from the list
+        :param paths_exclude_glob: optional glob pattern to exclude files from the list
+        :return: the filtered collection
+        """
+        from serena.util.text_utils import GlobMatcher
+
+        if paths_include_glob is None and paths_exclude_glob is None:
+            return self
+
+        include_glob_matcher = GlobMatcher(paths_include_glob) if paths_include_glob else None
+        exclude_glob_matcher = GlobMatcher(paths_exclude_glob) if paths_exclude_glob else None
+
+        filtered_files = []
+        for f in self._file_proxies:
+            if isinstance(f, LocalProjectFileProxy):
+                local_file: LocalProjectFileProxy = f
+                path = local_file.get_relative_path()
+                if include_glob_matcher:
+                    if not include_glob_matcher.matches(path):
+                        log.debug(f"Skipping {path}: does not match include pattern {paths_include_glob}")
+                        continue
+                if exclude_glob_matcher:
+                    if exclude_glob_matcher.matches(path):
+                        log.debug(f"Skipping {path}: matches exclude pattern {paths_exclude_glob}")
+                        continue
+            filtered_files.append(f)
+
+        return FileCollection(filtered_files)

src/serena/util/text_utils.py

@@ -1,7 +1,6 @@
 import fnmatch
 import hashlib
 import logging
-import os
 import re
 from collections.abc import Callable
 from dataclasses import dataclass, field
@@ -11,7 +10,7 @@ from typing import Any, Literal, Self
 from bs4 import BeautifulSoup
 from joblib import Parallel, delayed
 
-from serena.constants import DEFAULT_SOURCE_FILE_ENCODING
+from serena.util.file_proxy import FileCollection, FileProxy
 from solidlsp.ls_utils import TextUtils
 
 log = logging.getLogger(__name__)
@@ -237,12 +236,6 @@ def search_text(
     return matches
 
 
-def default_file_reader(file_path: str) -> str:
-    """Reads using the default encoding."""
-    with open(file_path, encoding=DEFAULT_SOURCE_FILE_ENCODING) as f:
-        return f.read()
-
-
 def expand_braces(pattern: str) -> list[str]:
     """
     Expands brace patterns in a glob string.
@@ -319,11 +312,17 @@ def glob_match(pattern: str, path: str) -> bool:
         return fnmatch.fnmatch(path, pattern)
 
 
+class GlobMatcher:
+    def __init__(self, expr: str):
+        self._patterns = expand_braces(expr)
+
+    def matches(self, path: str) -> bool:
+        return any(glob_match(p, path) for p in self._patterns)
+
+
 def search_files(
-    relative_file_paths: list[str],
+    file_collection: FileCollection,
     pattern: str,
-    root_path: str = "",
-    file_reader: Callable[[str], str] = default_file_reader,
     context_lines_before: int = 0,
     context_lines_after: int = 0,
     paths_include_glob: str | None = None,
@@ -333,11 +332,8 @@ def search_files(
     """
     Search for a pattern in a list of files.
 
-    :param relative_file_paths: list of relative file paths in which to search
+    :param file_collection: the collection of files to search (will be optionally filtered by glob patterns)
     :param pattern: pattern to search for
-    :param root_path: root path to resolve relative paths against (by default, current working directory).
-    :param file_reader: function to read a file, by default will just use os.open.
-        All files that can't be read by it will be skipped.
     :param context_lines_before: number of context lines to include before matches
     :param context_lines_after: number of context lines to include after matches
     :param paths_include_glob: optional glob pattern to include files from the list
@@ -345,53 +341,36 @@ def search_files(
     :param multiline: whether to apply multi-line matching, enabling the flags re.DOTALL and re.MULTILINE (default: True)
     :return: list of MatchedConsecutiveLines objects
     """
-    # Pre-filter paths (done sequentially to avoid overhead)
-    # Use proper glob matching instead of gitignore patterns
-    include_patterns = expand_braces(paths_include_glob) if paths_include_glob else None
-    exclude_patterns = expand_braces(paths_exclude_glob) if paths_exclude_glob else None
+    # apply glob filter
+    file_collection = file_collection.filter_glob(paths_include_glob=paths_include_glob, paths_exclude_glob=paths_exclude_glob)
+    log.info(f"Processing {len(file_collection)} files.")
 
-    filtered_paths = []
-    for path in relative_file_paths:
-        if include_patterns:
-            if not any(glob_match(p, path) for p in include_patterns):
-                log.debug(f"Skipping {path}: does not match include pattern {paths_include_glob}")
-                continue
-
-        if exclude_patterns:
-            if any(glob_match(p, path) for p in exclude_patterns):
-                log.debug(f"Skipping {path}: matches exclude pattern {paths_exclude_glob}")
-                continue
-
-        filtered_paths.append(path)
-
-    log.info(f"Processing {len(filtered_paths)} files.")
-
-    def process_single_file(path: str) -> dict[str, Any]:
+    def process_single_file(file_proxy: FileProxy) -> dict[str, Any]:
         """Process a single file - this function will be parallelized."""
+        relative_path = file_proxy.get_relative_path()
         try:
-            abs_path = os.path.join(root_path, path)
-            file_content = file_reader(abs_path)
+            file_content = file_proxy.get_contents()
             search_results = search_text(
                 pattern,
                 content=file_content,
-                source_file_path=path,
+                source_file_path=relative_path,
                 allow_multiline_match=True,
                 context_lines_before=context_lines_before,
                 context_lines_after=context_lines_after,
                 multiline=multiline,
             )
             if len(search_results) > 0:
-                log.debug(f"Found {len(search_results)} matches in {path}")
-            return {"path": path, "results": search_results, "error": None}
+                log.debug(f"Found {len(search_results)} matches in {relative_path}")
+            return {"path": relative_path, "results": search_results, "error": None}
         except Exception as e:
-            log.debug(f"Error processing {path}: {e}")
-            return {"path": path, "results": [], "error": str(e)}
+            log.debug(f"Error processing {relative_path}: {e}")
+            return {"path": relative_path, "results": [], "error": str(e)}
 
     # Execute in parallel using joblib
     results = Parallel(
         n_jobs=-1,
         backend="threading",
-    )(delayed(process_single_file)(path) for path in filtered_paths)
+    )(delayed(process_single_file)(file_proxy) for file_proxy in file_collection)
 
     # Collect results and errors
     matches = []
@@ -406,7 +385,7 @@ def search_files(
     if skipped_file_error_tuples:
         log.debug(f"Failed to read {len(skipped_file_error_tuples)} files: {skipped_file_error_tuples}")
 
-    log.info(f"Found {len(matches)} total matches across {len(filtered_paths)} files")
+    log.info(f"Found {len(matches)} total matches across {len(file_collection)} files")
     return matches
 
 

test/serena/test_text_utils.py

@@ -1,7 +1,9 @@
 import re
+from collections.abc import Callable
 
 import pytest
 
+from serena.util.file_proxy import FileCollection, FileProxy
 from serena.util.text_utils import LineType, MultiFileContentReplacer, search_files, search_text
 
 
@@ -221,6 +223,23 @@ def mock_reader_always_match(file_path: str) -> str:
     return "This line contains a match."
 
 
+class MockFileProxy(FileProxy):
+    def __init__(self, relative_path: str, mock_reader: Callable[[str], str] = mock_reader_always_match):
+        self.relative_path = relative_path
+        self.mock_reader = mock_reader
+
+    def get_contents(self) -> str:
+        return self.mock_reader(self.relative_path)
+
+    def get_relative_path(self) -> str:
+        return self.relative_path
+
+
+class MockFileCollection(FileCollection):
+    def __init__(self, file_paths, mock_reader: Callable[[str], str] = mock_reader_always_match):
+        super().__init__([MockFileProxy(path, mock_reader) for path in file_paths])
+
+
 class TestSearchFiles:
     @pytest.mark.parametrize(
         "file_paths, pattern, paths_include_glob, paths_exclude_glob, expected_matched_files, description",
@@ -254,9 +273,8 @@ class TestSearchFiles:
         Test the include/exclude glob filtering logic in search_files using PathSpec patterns.
         """
         results = search_files(
-            relative_file_paths=file_paths,
+            MockFileCollection(file_paths),
             pattern=pattern,
-            file_reader=mock_reader_always_match,
             paths_include_glob=paths_include_glob,
             paths_exclude_glob=paths_exclude_glob,
             context_lines_before=0,  # No context needed for this test focus
@@ -333,9 +351,8 @@ class TestSearchFiles:
         Test glob patterns that were problematic with the previous gitignore-based implementation.
         """
         results = search_files(
-            relative_file_paths=file_paths,
+            MockFileCollection(file_paths),
             pattern=pattern,
-            file_reader=mock_reader_always_match,
             paths_include_glob=paths_include_glob,
             paths_exclude_glob=paths_exclude_glob,
             context_lines_before=0,
@@ -414,9 +431,8 @@ class TestSearchFiles:
     ):
         """Test search_files with glob patterns containing brace expansions."""
         results = search_files(
-            relative_file_paths=file_paths,
+            MockFileCollection(file_paths),
             pattern=pattern,
-            file_reader=mock_reader_always_match,
             paths_include_glob=paths_include_glob,
             paths_exclude_glob=paths_exclude_glob,
         )
@@ -429,9 +445,8 @@ class TestSearchFiles:
         file_paths = ["a.py", "b.txt"]
         pattern = "non_existent_pattern_in_mock_content"  # This won't match mock_reader_always_match content
         results = search_files(
-            relative_file_paths=file_paths,
+            MockFileCollection(file_paths),
             pattern=pattern,
-            file_reader=mock_reader_always_match,  # Content is "This line contains a match."
             paths_include_glob=None,  # Both files would pass filters
             paths_exclude_glob=None,
         )
@@ -454,9 +469,8 @@ class TestSearchFiles:
         pattern = r"value=(\d+)"
 
         results = search_files(
-            relative_file_paths=file_paths,
+            MockFileCollection(file_paths, specific_mock_reader),
             pattern=pattern,
-            file_reader=specific_mock_reader,
             paths_include_glob="*.py",  # Only include .py files
             paths_exclude_glob="b.*",  # Exclude files starting with b
         )
@@ -484,9 +498,8 @@ class TestSearchFiles:
         pattern = "MATCH HERE"
 
         results = search_files(
-            relative_file_paths=file_paths,
+            MockFileCollection(file_paths, context_mock_reader),
             pattern=pattern,
-            file_reader=context_mock_reader,
             paths_include_glob="*.txt",  # Only include .txt files
             paths_exclude_glob=None,
             context_lines_before=1,

test/solidlsp/clojure/test_clojure_basic.py

@@ -211,7 +211,7 @@ class TestProjectBasics:
 
     @pytest.mark.parametrize("project", [Language.CLOJURE], indirect=True)
     def test_search_files_for_pattern(self, project: Project) -> None:
-        result = project.search_source_files_for_pattern("defn.*greet")
+        result = project.search_project_files_for_pattern("defn.*greet")
 
         assert result is not None, "Pattern search should return results"
         assert len(result) > 0, "Should find at least one match for 'defn.*greet'"
@@ -219,7 +219,7 @@ class TestProjectBasics:
         core_matches = [match for match in result if match.source_file_path and "core.clj" in match.source_file_path]
         assert len(core_matches) > 0, "Should find greet function in core.clj"
 
-        result = project.search_source_files_for_pattern(":require")
+        result = project.search_project_files_for_pattern(":require")
 
         assert result is not None, "Should find require statements"
         utils_matches = [match for match in result if match.source_file_path and "utils.clj" in match.source_file_path]

test/solidlsp/elixir/test_elixir_integration.py

@@ -115,7 +115,7 @@ class TestElixirProject:
         """Test comprehensive symbol search across the entire project."""
         # Search for all function definitions
         function_pattern = r"def\s+\w+\s*[\(\s]"
-        function_matches = project.search_source_files_for_pattern(function_pattern)
+        function_matches = project.search_project_files_for_pattern(function_pattern)
 
         # Should find functions across multiple files
         if function_matches:
@@ -131,7 +131,7 @@ class TestElixirProject:
 
         # Search for struct definitions
         struct_pattern = r"defstruct\s+\["
-        struct_matches = project.search_source_files_for_pattern(struct_pattern)
+        struct_matches = project.search_project_files_for_pattern(struct_pattern)
 
         if struct_matches:
             # Should find structs primarily in models.ex
@@ -143,7 +143,7 @@ class TestElixirProject:
         """Test that the language server understands Elixir protocols and implementations."""
         # Search for protocol definitions
         protocol_pattern = r"defprotocol\s+\w+"
-        protocol_matches = project.search_source_files_for_pattern(protocol_pattern, paths_include_glob="**/models.ex")
+        protocol_matches = project.search_project_files_for_pattern(protocol_pattern, paths_include_glob="**/models.ex")
 
         if protocol_matches:
             # Should find the Serializable protocol
@@ -152,7 +152,7 @@ class TestElixirProject:
 
         # Search for protocol implementations
         impl_pattern = r"defimpl\s+\w+"
-        impl_matches = project.search_source_files_for_pattern(impl_pattern, paths_include_glob="**/models.ex")
+        impl_matches = project.search_project_files_for_pattern(impl_pattern, paths_include_glob="**/models.ex")
 
         if impl_matches:
             # Should find multiple implementations

test/solidlsp/python/test_python_basic.py

@@ -214,28 +214,28 @@ class TestProjectBasics:
         """Test search_files_for_pattern with various patterns and glob filters."""
         # Test 1: Search for class definitions across all files
         class_pattern = r"class\s+\w+\s*(?:\([^{]*\)|:)"
-        matches = project.search_source_files_for_pattern(class_pattern)
+        matches = project.search_project_files_for_pattern(class_pattern)
         assert len(matches) > 0
         # Should find multiple classes like User, Item, BaseModel, etc.
         assert len(matches) >= 5
 
         # Test 2: Search for specific class with include glob
         user_class_pattern = r"class\s+User\s*(?:\([^{]*\)|:)"
-        matches = project.search_source_files_for_pattern(user_class_pattern, paths_include_glob="**/models.py")
+        matches = project.search_project_files_for_pattern(user_class_pattern, paths_include_glob="**/models.py")
         assert len(matches) == 1  # Should only find User class in models.py
         assert matches[0].source_file_path is not None
         assert "models.py" in matches[0].source_file_path
 
         # Test 3: Search for method definitions with exclude glob
         method_pattern = r"def\s+\w+\s*\([^)]*\):"
-        matches = project.search_source_files_for_pattern(method_pattern, paths_exclude_glob="**/models.py")
+        matches = project.search_project_files_for_pattern(method_pattern, paths_exclude_glob="**/models.py")
         assert len(matches) > 0
         # Should find methods in services.py but not in models.py
         assert all(match.source_file_path is not None and "models.py" not in match.source_file_path for match in matches)
 
         # Test 4: Search for specific method with both include and exclude globs
         create_user_pattern = r"def\s+create_user\s*\([^)]*\)(?:\s*->[^:]+)?:"
-        matches = project.search_source_files_for_pattern(
+        matches = project.search_project_files_for_pattern(
             create_user_pattern, paths_include_glob="**/*.py", paths_exclude_glob="**/models.py"
         )
         assert len(matches) == 1  # Should only find create_user in services.py
@@ -244,7 +244,7 @@ class TestProjectBasics:
 
         # Test 5: Search for a pattern that should appear in multiple files
         init_pattern = r"def\s+__init__\s*\([^)]*\):"
-        matches = project.search_source_files_for_pattern(init_pattern)
+        matches = project.search_project_files_for_pattern(init_pattern)
         assert len(matches) > 1  # Should find __init__ in multiple classes
         # Should find __init__ in both models.py and services.py
         assert any(match.source_file_path is not None and "models.py" in match.source_file_path for match in matches)
@@ -252,7 +252,7 @@ class TestProjectBasics:
 
         # Test 6: Search with a pattern that should have no matches
         no_match_pattern = r"def\s+this_method_does_not_exist\s*\([^)]*\):"
-        matches = project.search_source_files_for_pattern(no_match_pattern)
+        matches = project.search_project_files_for_pattern(no_match_pattern)
         assert len(matches) == 0
 
     @pytest.mark.parametrize("language_server", PYTHON_BACKEND_LANGUAGES, indirect=True)

test/solidlsp/swift/test_swift_basic.py

@@ -198,7 +198,7 @@ class TestSwiftProjectBasics:
 
         # Scenario 3: Search for struct definitions
         struct_pattern = r"struct\s+\w+"
-        matches = project.search_source_files_for_pattern(struct_pattern)
+        matches = project.search_project_files_for_pattern(struct_pattern)
         assert len(matches) > 0, "Should find struct definitions"
         # Should find User struct
         user_matches = [m for m in matches if "User" in str(m)]
@@ -206,7 +206,7 @@ class TestSwiftProjectBasics:
 
         # Scenario 4: Search for class definitions
         class_pattern = r"class\s+\w+"
-        matches = project.search_source_files_for_pattern(class_pattern)
+        matches = project.search_project_files_for_pattern(class_pattern)
         assert len(matches) > 0, "Should find class definitions"
         # Should find Calculator and Circle classes
         calculator_matches = [m for m in matches if "Calculator" in str(m)]
@@ -216,7 +216,7 @@ class TestSwiftProjectBasics:
 
         # Scenario 5: Search for enum definitions
         enum_pattern = r"enum\s+\w+"
-        matches = project.search_source_files_for_pattern(enum_pattern)
+        matches = project.search_project_files_for_pattern(enum_pattern)
         assert len(matches) > 0, "Should find enum definitions"
         # Should find Status enum
         status_matches = [m for m in matches if "Status" in str(m)]