Merge pull request #846 from oraios/improve-regex-amiguity-detection

ReplaceContentTool: Detect ambiguous regex match and raise an exception

src/serena/tools/file_tools.py

@@ -9,6 +9,7 @@ File and file system-related tools, specifically for
 import os
 import re
 from collections import defaultdict
+from collections.abc import Callable
 from fnmatch import fnmatch
 from pathlib import Path
 from typing import Literal
@@ -202,6 +203,45 @@ class ReplaceContentTool(Tool, ToolMarkerCanEdit):
             relative_path, needle, repl, mode=mode, allow_multiple_occurrences=allow_multiple_occurrences, require_not_ignored=True
         )
 
+    @staticmethod
+    def _create_replacement_function(regex_pattern: str, repl_template: str, regex_flags: int) -> Callable[[re.Match], str]:
+        """
+        Creates a replacement function that validates for ambiguity and handles backreferences.
+
+        :param regex_pattern: The regex pattern being used for matching
+        :param repl_template: The replacement template with $!1, $!2, etc. for backreferences
+        :param regex_flags: The flags to use when searching (e.g., re.DOTALL | re.MULTILINE)
+        :return: A function suitable for use with re.sub() or re.subn()
+        """
+
+        def validate_and_replace(match: re.Match) -> str:
+            matched_text = match.group(0)
+
+            # For multi-line match, check if the same pattern matches again within the already-matched text,
+            # rendering the match ambiguous. Typical pattern in the code:
+            #    <start><other-stuff><start><stuff><end>
+            # When matching
+            #    <start>.*?<end>
+            # this will match the entire span above, while only the suffix may have been intended.
+            # (See test case for a practical example.)
+            # To detect this, we check if the same pattern matches again within the matched text,
+            if "\n" in matched_text and re.search(regex_pattern, matched_text[1:], flags=regex_flags):
+                raise ValueError(
+                    "Match is ambiguous: the search pattern matches multiple overlapping occurrences. "
+                    "Please revise the search pattern to be more specific to avoid ambiguity."
+                )
+
+            # Handle backreferences: replace $!1, $!2, etc. with actual matched groups
+            def expand_backreference(m: re.Match) -> str:
+                group_num = int(m.group(1))
+                group_value = match.group(group_num)
+                return group_value if group_value is not None else m.group(0)
+
+            result = re.sub(r"\$!(\d+)", expand_backreference, repl_template)
+            return result
+
+        return validate_and_replace
+
     def replace_content(
         self,
         relative_path: str,
@@ -226,19 +266,22 @@ class ReplaceContentTool(Tool, ToolMarkerCanEdit):
             else:
                 raise ValueError(f"Invalid mode: '{mode}', expected 'literal' or 'regex'.")
 
+            regex_flags = re.DOTALL | re.MULTILINE
+
             # escape backslashes in repl
             repl = repl.replace("\\", "\\\\")
-            # convert $!1, $!2, etc. to \1, \2, etc. for backreferences
-            repl = re.sub(r"\$!(\d+)", "\\\\1", repl)
+
+            # create replacement function with validation and backreference handling
+            repl_fn = self._create_replacement_function(regex, repl, regex_flags=regex_flags)
 
             # perform replacement
-            updated_content, n = re.subn(regex, repl, original_content, flags=re.DOTALL | re.MULTILINE)
+            updated_content, n = re.subn(regex, repl_fn, original_content, flags=regex_flags)
 
             if n == 0:
-                return f"Error: No matches of search expression found in file '{relative_path}'."
+                raise ValueError(f"Error: No matches of search expression found in file '{relative_path}'.")
             if not allow_multiple_occurrences and n > 1:
-                return (
-                    f"Error: Expression matches {n} occurrences in file '{relative_path}'. "
+                raise ValueError(
+                    f"Expression matches {n} occurrences in file '{relative_path}'. "
                     "Please revise the expression to be more specific or enable allow_multiple_occurrences if this is expected."
                 )
             context.set_updated_content(updated_content)

test/resources/repos/typescript/test_repo/ws_manager.js

@@ -0,0 +1,73 @@
+/**
+ * Dummy WebSocket manager class for testing ambiguous regex matching.
+ */
+class WebSocketManager {
+    constructor() {
+        this.ws = null;
+        this.statusElement = document.getElementById("status");
+    }
+
+    /**
+     * Connects to the WebSocket server.
+     */
+    connectToServer() {
+        if (this.ws?.readyState === WebSocket.OPEN) {
+            this.updateConnectionStatus("Already connected", true);
+            return;
+        }
+
+        try {
+            this.ws = new WebSocket("ws://localhost:4402");
+            this.updateConnectionStatus("Connecting...", false);
+
+            this.ws.onopen = () => {
+                console.log("Connected to server");
+                this.updateConnectionStatus("Connected", true);
+            };
+
+            this.ws.onmessage = (event) => {
+                console.log("Received message:", event.data);
+                try {
+                    const data = JSON.parse(event.data);
+                    this.handleMessage(data);
+                } catch (error) {
+                    console.error("Failed to parse message:", error);
+                    this.updateConnectionStatus("Parse error", false);
+                }
+            };
+
+            this.ws.onclose = (event) => {
+                console.log("Connection closed");
+                const message = event.reason || undefined;
+                this.updateConnectionStatus("Disconnected", false, message);
+                this.ws = null;
+            };
+
+            this.ws.onerror = (error) => {
+                console.error("WebSocket error:", error);
+                this.updateConnectionStatus("Connection error", false);
+            };
+        } catch (error) {
+            console.error("Failed to connect to server:", error);
+            this.updateConnectionStatus("Connection failed", false);
+        }
+    }
+
+    /**
+     * Updates the connection status display.
+     */
+    updateConnectionStatus(status, isConnected, message) {
+        if (this.statusElement) {
+            const text = message ? `${status}: ${message}` : status;
+            this.statusElement.textContent = text;
+            this.statusElement.style.color = isConnected ? "green" : "red";
+        }
+    }
+
+    /**
+     * Handles incoming messages.
+     */
+    handleMessage(data) {
+        console.log("Handling:", data);
+    }
+}

test/serena/test_serena_agent.py

@@ -3,13 +3,14 @@ import logging
 import os
 import time
 from collections.abc import Iterator
+from contextlib import contextmanager
 
 import pytest
 
 from serena.agent import SerenaAgent
 from serena.config.serena_config import ProjectConfig, RegisteredProject, SerenaConfig
 from serena.project import Project
-from serena.tools import FindReferencingSymbolsTool, FindSymbolTool, ReplaceSymbolBodyTool
+from serena.tools import SUCCESS_RESULT, FindReferencingSymbolsTool, FindSymbolTool, ReplaceContentTool, ReplaceSymbolBodyTool
 from solidlsp.ls_config import Language
 from test.conftest import get_repo_path, language_tests_enabled
 from test.solidlsp import clojure as clj
@@ -56,6 +57,24 @@ def serena_config():
     return config
 
 
+@contextmanager
+def project_file_modification_context(serena_agent: SerenaAgent, relative_path: str) -> Iterator[None]:
+    """Context manager to modify a project file and revert the changes after use."""
+    project = serena_agent.get_active_project()
+    file_path = os.path.join(project.project_root, relative_path)
+
+    # Read the original content
+    with open(file_path, encoding=project.project_config.encoding) as f:
+        original_content = f.read()
+
+    try:
+        yield
+    finally:
+        # Revert to the original content
+        with open(file_path, "w", encoding=project.project_config.encoding) as f:
+            f.write(original_content)
+
+
 @pytest.fixture
 def serena_agent(request: pytest.FixtureRequest, serena_config) -> Iterator[SerenaAgent]:
     language = Language(request.param)
@@ -70,7 +89,8 @@ def serena_agent(request: pytest.FixtureRequest, serena_config) -> Iterator[Sere
     agent.execute_task(lambda: None)
 
     yield agent
-    # explicitly delete to free resources
+
+    # explicitly shut down to free resources
     agent.shutdown(timeout=5)
 
 
@@ -355,3 +375,52 @@ class TestSerenaAgent:
         replace_symbol_body_tool = serena_agent.get_tool(ReplaceSymbolBodyTool)
         with pytest.raises(ValueError, match=match_text):
             replace_symbol_body_tool.apply(name_path=name_path, relative_path=relative_path, body="")
+
+    @pytest.mark.parametrize(
+        "serena_agent",
+        [
+            pytest.param(
+                Language.TYPESCRIPT,
+                marks=pytest.mark.typescript,
+            ),
+        ],
+        indirect=["serena_agent"],
+    )
+    def test_replace_content_regex(self, serena_agent: SerenaAgent):
+        """
+        Tests a regex-based content replacement that has a unique match
+        """
+        relative_path = "ws_manager.js"
+        with project_file_modification_context(serena_agent, relative_path):
+            replace_content_tool = serena_agent.get_tool(ReplaceContentTool)
+            result = replace_content_tool.apply(
+                needle=r'catch \(error\) \{\s*console.error\("Failed to connect.*?\}',
+                repl='catch(error) { console.log("Never mind"); }',
+                relative_path=relative_path,
+                mode="regex",
+            )
+            assert result == SUCCESS_RESULT
+
+    @pytest.mark.parametrize(
+        "serena_agent",
+        [
+            pytest.param(
+                Language.TYPESCRIPT,
+                marks=pytest.mark.typescript,
+            ),
+        ],
+        indirect=["serena_agent"],
+    )
+    def test_replace_content_regex_ambiguous(self, serena_agent: SerenaAgent):
+        """
+        Tests that an ambiguous replacement where there is a larger match that internally contains
+        a smaller match triggers an exception
+        """
+        replace_content_tool = serena_agent.get_tool(ReplaceContentTool)
+        with pytest.raises(ValueError, match="ambiguous"):
+            replace_content_tool.apply(
+                needle=r'catch \(error\) \{.*?this\.updateConnectionStatus\("Connection failed", false\);.*?\}',
+                repl='catch(error) { console.log("Never mind"); }',
+                relative_path="ws_manager.js",
+                mode="regex",
+            )