Refactoring: Introduce base class EditingToolWithDiagnostics

src/serena/tools/file_tools.py

@@ -13,7 +13,7 @@ from pathlib import Path
 from typing import Literal
 
 from serena.code_editor import EditedFilePath
-from serena.tools import SUCCESS_RESULT, EditedFileContext, Tool, ToolMarkerCanEdit, ToolMarkerOptional
+from serena.tools import SUCCESS_RESULT, EditedFileContext, EditingToolWithDiagnostics, Tool, ToolMarkerOptional
 from serena.util.file_system import scan_directory
 from serena.util.text_utils import ContentReplacer, search_files
 
@@ -49,7 +49,7 @@ class ReadFileTool(Tool):
         return self._limit_length(result, max_answer_chars)
 
 
-class CreateTextFileTool(Tool, ToolMarkerCanEdit):
+class CreateTextFileTool(EditingToolWithDiagnostics):
     """
     Creates/overwrites a file in the project directory.
     """
@@ -164,13 +164,11 @@ class FindFileTool(Tool):
         return result
 
 
-class ReplaceContentTool(Tool, ToolMarkerCanEdit):
+class ReplaceContentTool(EditingToolWithDiagnostics):
     """
     Replaces content in a file (optionally using regular expressions).
     """
 
-    _CAPTURE_DIAGNOSTICS: bool = False
-
     def apply(
         self,
         relative_path: str,
@@ -237,7 +235,7 @@ class ReplaceContentTool(Tool, ToolMarkerCanEdit):
         return self._format_lsp_edit_result_with_new_diagnostics(SUCCESS_RESULT, edited_file_paths, diagnostics_snapshot)
 
 
-class DeleteLinesTool(Tool, ToolMarkerCanEdit, ToolMarkerOptional):
+class DeleteLinesTool(EditingToolWithDiagnostics, ToolMarkerOptional):
     """
     Deletes a range of lines within a file.
     """
@@ -268,7 +266,7 @@ class DeleteLinesTool(Tool, ToolMarkerCanEdit, ToolMarkerOptional):
         return self._format_lsp_edit_result_with_new_diagnostics(SUCCESS_RESULT, edited_file_paths, diagnostics_snapshot)
 
 
-class ReplaceLinesTool(Tool, ToolMarkerCanEdit, ToolMarkerOptional):
+class ReplaceLinesTool(EditingToolWithDiagnostics, ToolMarkerOptional):
     """
     Replaces a range of lines within a file with new content.
     """
@@ -306,7 +304,7 @@ class ReplaceLinesTool(Tool, ToolMarkerCanEdit, ToolMarkerOptional):
         return self._format_lsp_edit_result_with_new_diagnostics(SUCCESS_RESULT, edited_file_paths, diagnostics_snapshot)
 
 
-class InsertAtLineTool(Tool, ToolMarkerCanEdit, ToolMarkerOptional):
+class InsertAtLineTool(EditingToolWithDiagnostics, ToolMarkerOptional):
     """
     Inserts content at a given line in a file.
     """

src/serena/tools/symbol_tools.py

@@ -13,6 +13,7 @@ from serena.code_editor import EditedFilePath
 from serena.symbol import LanguageServerSymbol, LanguageServerSymbolDictGrouper
 from serena.tools import (
     SUCCESS_RESULT,
+    EditingToolWithDiagnostics,
     Tool,
     ToolMarkerSymbolicEdit,
     ToolMarkerSymbolicRead,
@@ -771,7 +772,7 @@ class GetDiagnosticsForSymbolTool(Tool, ToolMarkerSymbolicRead, ToolMarkerOption
         return self._limit_length(result, max_answer_chars)
 
 
-class ReplaceSymbolBodyTool(Tool, ToolMarkerSymbolicEdit):
+class ReplaceSymbolBodyTool(EditingToolWithDiagnostics):
     """
     Replaces the full definition of a symbol using the language server backend.
     """
@@ -810,7 +811,7 @@ class ReplaceSymbolBodyTool(Tool, ToolMarkerSymbolicEdit):
         return self._format_lsp_edit_result_with_new_diagnostics(SUCCESS_RESULT, edited_file_paths, diagnostics_snapshot)
 
 
-class InsertAfterSymbolTool(Tool, ToolMarkerSymbolicEdit):
+class InsertAfterSymbolTool(EditingToolWithDiagnostics):
     """
     Inserts content after the end of the definition of a given symbol.
     """
@@ -841,7 +842,7 @@ class InsertAfterSymbolTool(Tool, ToolMarkerSymbolicEdit):
         return self._format_lsp_edit_result_with_new_diagnostics(SUCCESS_RESULT, edited_file_paths, diagnostics_snapshot)
 
 
-class InsertBeforeSymbolTool(Tool, ToolMarkerSymbolicEdit):
+class InsertBeforeSymbolTool(EditingToolWithDiagnostics):
     """
     Inserts content before the beginning of the definition of a given symbol.
     """

src/serena/tools/tools_base.py

@@ -165,8 +165,6 @@ class Tool(Component):
     _last_tool_call_client_str: str | None = None
     """We can only get the client info from within a tool call. Each tool call will update this variable."""
 
-    _ENABLE_DIAGNOSTICS: bool = False
-
     def __init__(self, agent: "SerenaAgent"):
         super().__init__(agent)
 
@@ -326,6 +324,118 @@ class Tool(Component):
     def is_symbolic(self) -> bool:
         return issubclass(self.__class__, ToolMarkerSymbolicRead) or issubclass(self.__class__, ToolMarkerSymbolicEdit)
 
+    def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, mcp_ctx: Context | None = None, **kwargs) -> str:  # type: ignore
+        """
+        Applies the tool with logging and exception handling, using the given keyword arguments
+        """
+        session_id = "global"
+        if mcp_ctx is not None:
+            try:
+                session_id = "%x" % id(mcp_ctx.session)
+                client_params = mcp_ctx.session.client_params
+                if client_params is not None:
+                    client_info = cast(Implementation, client_params.clientInfo)
+                    client_str = client_info.title if client_info.title else client_info.name + " " + client_info.version
+                    if client_str != self.get_last_tool_call_client_str():
+                        log.debug(f"Updating client info: {client_info}")
+                        self.set_last_tool_call_client_str(client_str)
+            except Exception as e:
+                log.info(f"Failed to get client info: {e}.")
+
+        def task() -> str:
+            apply_fn = self.get_apply_fn()
+
+            try:
+                if not self.is_active():
+                    return f"Error: Tool '{self.get_name_from_cls()}' is not active. Active tools: {self.agent.get_active_tool_names()}"
+            except Exception as e:
+                return f"RuntimeError while checking if tool {self.get_name_from_cls()} is active: {e}"
+
+            if log_call:
+                self._log_tool_application(inspect.currentframe(), session_id)
+            try:
+                # check whether the tool requires an active project and language server
+                if not isinstance(self, ToolMarkerDoesNotRequireActiveProject):
+                    if self.agent.get_active_project() is None:
+                        return (
+                            "Error: No active project. Ask the user to provide the project path or to select a project from this list of known projects: "
+                            + f"{self.agent.serena_config.project_names}"
+                        )
+
+                # construct apply kwargs, adding session_id if the tool is session-aware
+                apply_kwargs = dict(kwargs)
+                if self._is_session_aware:
+                    apply_kwargs["session_id"] = session_id
+
+                # apply the actual tool
+                try:
+                    result = apply_fn(**apply_kwargs)
+                except SolidLSPException as e:
+                    if e.is_language_server_terminated():
+                        affected_language = e.get_affected_language()
+                        if affected_language is not None:
+                            log.error(
+                                f"Language server terminated while executing tool ({e}). Restarting the language server and retrying ..."
+                            )
+                            self.agent.get_language_server_manager_or_raise().restart_language_server(affected_language)
+                            result = apply_fn(**apply_kwargs)
+                        else:
+                            log.error(
+                                f"Language server terminated while executing tool ({e}), but affected language is unknown. Not retrying."
+                            )
+                            raise
+                    else:
+                        raise
+
+                # record tool usage
+                self.agent.record_tool_usage(apply_kwargs, result, self)
+
+            except Exception as e:
+                if not catch_exceptions:
+                    raise
+                msg = f"Error executing tool: {e.__class__.__name__} - {e}"
+                log.error(f"Error executing tool: {e}", exc_info=e)
+                result = msg
+
+            if log_call:
+                log.info(f"Result: {result}")
+
+            try:
+                ls_manager = self.agent.get_language_server_manager()
+                if ls_manager is not None:
+                    ls_manager.save_all_caches()
+            except Exception as e:
+                log.error(f"Error saving language server cache: {e}")
+
+            return result
+
+        # execute the tool in the agent's task executor, with timeout
+        try:
+            task_exec = self.agent.issue_task(task, name=self.__class__.__name__)
+            return task_exec.result(timeout=self.agent.serena_config.tool_timeout)
+        except Exception as e:  # typically TimeoutError (other exceptions caught in task)
+            msg = f"Error: {e.__class__.__name__} - {e}"
+            log.error(msg)
+            return msg
+
+    @staticmethod
+    def _to_json(x: Any) -> str:
+        return json.dumps(x, ensure_ascii=False)
+
+
+class EditingToolWithDiagnostics(Tool, ToolMarkerCanEdit):
+    """
+    Base class for editing tools that want to capture and report changes in LSP diagnostics before and after the edit.
+    """
+
+    _ENABLE_DIAGNOSTICS: bool = False
+    """
+    Global flag to enable/disable diagnostics for LSP-based editing tools derived from this class.
+    The feature is currently disabled, because per-edit diagnostics are a questionable feature, since individual
+    edits often intentionally introduce diagnostics (e.g. function signature mismatches or even syntax errors) that 
+    are then resolved in subsequent edits.
+    """
+
     @staticmethod
     def _diagnostic_code_repr(code: Any) -> str | None:
         if code is None:
@@ -500,104 +610,6 @@ class Tool(Component):
 
         return f"Edit introduced new warning-or-higher diagnostics: {self._to_json(grouped_result)}"
 
-    def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, mcp_ctx: Context | None = None, **kwargs) -> str:  # type: ignore
-        """
-        Applies the tool with logging and exception handling, using the given keyword arguments
-        """
-        session_id = "global"
-        if mcp_ctx is not None:
-            try:
-                session_id = "%x" % id(mcp_ctx.session)
-                client_params = mcp_ctx.session.client_params
-                if client_params is not None:
-                    client_info = cast(Implementation, client_params.clientInfo)
-                    client_str = client_info.title if client_info.title else client_info.name + " " + client_info.version
-                    if client_str != self.get_last_tool_call_client_str():
-                        log.debug(f"Updating client info: {client_info}")
-                        self.set_last_tool_call_client_str(client_str)
-            except Exception as e:
-                log.info(f"Failed to get client info: {e}.")
-
-        def task() -> str:
-            apply_fn = self.get_apply_fn()
-
-            try:
-                if not self.is_active():
-                    return f"Error: Tool '{self.get_name_from_cls()}' is not active. Active tools: {self.agent.get_active_tool_names()}"
-            except Exception as e:
-                return f"RuntimeError while checking if tool {self.get_name_from_cls()} is active: {e}"
-
-            if log_call:
-                self._log_tool_application(inspect.currentframe(), session_id)
-            try:
-                # check whether the tool requires an active project and language server
-                if not isinstance(self, ToolMarkerDoesNotRequireActiveProject):
-                    if self.agent.get_active_project() is None:
-                        return (
-                            "Error: No active project. Ask the user to provide the project path or to select a project from this list of known projects: "
-                            + f"{self.agent.serena_config.project_names}"
-                        )
-
-                # construct apply kwargs, adding session_id if the tool is session-aware
-                apply_kwargs = dict(kwargs)
-                if self._is_session_aware:
-                    apply_kwargs["session_id"] = session_id
-
-                # apply the actual tool
-                try:
-                    result = apply_fn(**apply_kwargs)
-                except SolidLSPException as e:
-                    if e.is_language_server_terminated():
-                        affected_language = e.get_affected_language()
-                        if affected_language is not None:
-                            log.error(
-                                f"Language server terminated while executing tool ({e}). Restarting the language server and retrying ..."
-                            )
-                            self.agent.get_language_server_manager_or_raise().restart_language_server(affected_language)
-                            result = apply_fn(**apply_kwargs)
-                        else:
-                            log.error(
-                                f"Language server terminated while executing tool ({e}), but affected language is unknown. Not retrying."
-                            )
-                            raise
-                    else:
-                        raise
-
-                # record tool usage
-                self.agent.record_tool_usage(apply_kwargs, result, self)
-
-            except Exception as e:
-                if not catch_exceptions:
-                    raise
-                msg = f"Error executing tool: {e.__class__.__name__} - {e}"
-                log.error(f"Error executing tool: {e}", exc_info=e)
-                result = msg
-
-            if log_call:
-                log.info(f"Result: {result}")
-
-            try:
-                ls_manager = self.agent.get_language_server_manager()
-                if ls_manager is not None:
-                    ls_manager.save_all_caches()
-            except Exception as e:
-                log.error(f"Error saving language server cache: {e}")
-
-            return result
-
-        # execute the tool in the agent's task executor, with timeout
-        try:
-            task_exec = self.agent.issue_task(task, name=self.__class__.__name__)
-            return task_exec.result(timeout=self.agent.serena_config.tool_timeout)
-        except Exception as e:  # typically TimeoutError (other exceptions caught in task)
-            msg = f"Error: {e.__class__.__name__} - {e}"
-            log.error(msg)
-            return msg
-
-    @staticmethod
-    def _to_json(x: Any) -> str:
-        return json.dumps(x, ensure_ascii=False)
-
 
 class EditedFileContext:
     """