From 37d100446057d8c9a3a568c87bfa880c0af9fd74 Mon Sep 17 00:00:00 2001 From: Michael Panchenko Date: Sat, 14 Jun 2025 01:59:29 +0200 Subject: [PATCH] Fixes in SerenaAgent isolation in process creation make_tool previously depended on tool instances in the main process, but the tool instances are no longer available there To solve this, the tool interface was adjusted to provide static information which is used in the ProcessIsolatedTool wrapper. I also tried another approach before, where ProcessIsolatedTool would accept tool metadata at init, but this lead to serialization issues that couldn't be easily solved (mcp's FuncMetadata is not serializable, despite being a BaseModel...). Tests for checking that the wrapped tool instance has the same metadata as the wrapper have been added. --- src/serena/agent.py | 189 ++++++++++++++---- src/serena/agno.py | 4 +- src/serena/mcp.py | 68 +++---- src/serena/process_isolated_agent.py | 60 +++--- .../test_make_tool_process_isolation.py | 160 +++++++++++++++ test/serena/test_mcp.py | 4 +- test/serena/test_serena_agent.py | 37 +++- 7 files changed, 414 insertions(+), 108 deletions(-) create mode 100644 test/serena/test_make_tool_process_isolation.py diff --git a/src/serena/agent.py b/src/serena/agent.py index 75d0e77c..ebb8a36a 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -21,9 +21,10 @@ from functools import cached_property from logging import Logger from pathlib import Path from types import TracebackType -from typing import TYPE_CHECKING, Any, Literal, Protocol, Self, TypeVar, Union, cast +from typing import TYPE_CHECKING, Any, Literal, Self, TypeVar, Union, cast import yaml +from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata from overrides import override from pathspec import PathSpec from ruamel.yaml.comments import CommentedMap @@ -548,6 +549,77 @@ class MemoriesManagerMDFilesInProject(MemoriesManager): return f"Memory {name} deleted." +def create_serena_config( + project: str | None = None, + serena_config: SerenaConfigBase | None = None, + context: SerenaAgentContext | None = None, + modes: list[SerenaAgentMode] | None = None, + enable_web_dashboard: bool | None = None, + enable_gui_log_window: bool | None = None, + log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None, + trace_lsp_communication: bool | None = None, + tool_timeout: float | None = None, +) -> SerenaConfig: + """ + Create a SerenaConfig instance without instantiating a full SerenaAgent. + + This function extracts the configuration creation logic from SerenaAgent.__init__ + to allow creating configurations independently for process isolation and other use cases. + + :param project: the project to register/activate or None + :param serena_config: the base Serena configuration or None to read from default location + :param context: the context in which the agent will operate + :param modes: list of modes in which the agent will operate + :param enable_web_dashboard: Whether to enable the web dashboard + :param enable_gui_log_window: Whether to enable the GUI log window + :param log_level: Log level + :param trace_lsp_communication: Whether to trace LSP communication + :param tool_timeout: Timeout in seconds for tool execution + :return: A fully configured SerenaConfig instance + """ + # obtain serena configuration + if serena_config is not None: + # If a complete SerenaConfig is provided, use it directly + if isinstance(serena_config, SerenaConfig): + config = serena_config + else: + # For SerenaConfigBase instances (like test configs), create an in-memory SerenaConfig + # that preserves the base config attributes without loading from file + from ruamel.yaml.comments import CommentedMap + + config = SerenaConfig.__new__(SerenaConfig) # Create without calling __init__ + # Initialize basic attributes from base config + config.projects = getattr(serena_config, "projects", []) + config.gui_log_window_enabled = serena_config.gui_log_window_enabled + config.log_level = serena_config.log_level + config.trace_lsp_communication = serena_config.trace_lsp_communication + config.web_dashboard = serena_config.web_dashboard + config.tool_timeout = serena_config.tool_timeout + # Set empty yaml for in-memory config + config.loaded_commented_yaml = CommentedMap() + else: + config = SerenaConfig.from_config_file() + + # Apply parameter overrides + if enable_web_dashboard is not None: + config.web_dashboard = enable_web_dashboard + if enable_gui_log_window is not None: + config.gui_log_window_enabled = enable_gui_log_window + if log_level is not None: + log_level = cast(Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], log_level.upper()) + # transform to int + config.log_level = logging.getLevelNamesMapping()[log_level] + if trace_lsp_communication is not None: + config.trace_lsp_communication = trace_lsp_communication + if tool_timeout is not None: + config.tool_timeout = tool_timeout + + # Note: Project registration/activation is handled separately by the caller + # since it involves complex logic that may require the full agent context + + return config + + class SerenaAgent: def __init__( self, @@ -577,20 +649,18 @@ class SerenaAgent: :param log_level: Log level for the GUI log window. If not specified, will take the value from the serena configuration. :param tool_timeout: Timeout in seconds for tool execution. If not specified, will take the value from the serena configuration. """ - # obtain serena configuration - self.serena_config = serena_config or SerenaConfig.from_config_file() - if enable_web_dashboard is not None: - self.serena_config.web_dashboard = enable_web_dashboard - if enable_gui_log_window is not None: - self.serena_config.gui_log_window_enabled = enable_gui_log_window - if log_level is not None: - log_level = cast(Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], log_level.upper()) - # transform to int - self.serena_config.log_level = logging.getLevelNamesMapping()[log_level] - if trace_lsp_communication is not None: - self.serena_config.trace_lsp_communication = trace_lsp_communication - if tool_timeout is not None: - self.serena_config.tool_timeout = tool_timeout + # obtain serena configuration using the decoupled factory function + self.serena_config = create_serena_config( + project=None, # Don't pass project here, handle activation separately + serena_config=serena_config, + context=context, + modes=modes, + enable_web_dashboard=enable_web_dashboard, + enable_gui_log_window=enable_gui_log_window, + log_level=log_level, + trace_lsp_communication=trace_lsp_communication, + tool_timeout=tool_timeout, + ) # adjust log level serena_log_level = self.serena_config.log_level @@ -617,7 +687,7 @@ class SerenaAgent: # instantiate all tool classes self._all_tools: dict[type[Tool], Tool] = {tool_class: tool_class(self) for tool_class in ToolRegistry.get_all_tool_classes()} - tool_names = [tool.get_name() for tool in self._all_tools.values()] + tool_names = [tool.get_name_from_cls() for tool in self._all_tools.values()] # If GUI log window is enabled, set the tool names for highlighting if self._gui_log_handler is not None: @@ -654,7 +724,7 @@ class SerenaAgent: modes = SerenaAgentMode.load_default_modes() self._context = context self._modes = modes - log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name() for tool in self._all_tools.values()])}") + log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name_from_cls() for tool in self._all_tools.values()])}") self._active_tools: dict[type[Tool], Tool] = {} self._update_active_tools() @@ -839,7 +909,7 @@ class SerenaAgent: """ :return: the list of names of the active tools for the current project """ - return sorted([tool.get_name() for tool in self.get_active_tool_classes()]) + return sorted([tool.get_name_from_cls() for tool in self.get_active_tool_classes()]) def tool_is_active(self, tool_class: type["Tool"] | str) -> bool: """ @@ -885,7 +955,7 @@ class SerenaAgent: result_str += " " + ", ".join(chunk) + "\n" # Available but not active tools - all_tool_names = sorted([tool.get_name() for tool in self._all_tools.values()]) + all_tool_names = sorted([tool.get_name_from_cls() for tool in self._all_tools.values()]) inactive_tool_names = [tool for tool in all_tool_names if tool not in active_tool_names] if inactive_tool_names: result_str += "Available but not active tools:\n" @@ -1026,23 +1096,31 @@ class ToolMarkerDoesNotRequireActiveProject: pass -class ToolProtocol(Protocol): +class ToolInterface(ABC): """Protocol defining the complete interface that make_tool() expects from a tool.""" + @abstractmethod def get_name(self) -> str: """Get the tool name.""" ... - def apply(self, **kwargs: Any) -> str: - """The application logic of the tool, docstring will be set here and used for MCP tool description.""" + @abstractmethod + def get_apply_docstring(self) -> str: + """Get the docstring for the tool application, used by the MCP server.""" ... + @abstractmethod + def get_apply_fn_metadata(self) -> FuncMetadata: + """Get the metadata for the tool application function, used by the MCP server.""" + ... + + @abstractmethod def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs: Any) -> str: """Apply the tool with logging and exception handling.""" ... -class Tool(Component): +class Tool(Component, ToolInterface): # NOTE: each tool should implement the apply method, which is then used in # the central method of the Tool class `apply_ex`. # Failure to do so will result in a RuntimeError at tool execution time. @@ -1054,7 +1132,7 @@ class Tool(Component): # and to validate the tool call arguments. @classmethod - def get_name(cls) -> str: + def get_name_from_cls(cls) -> str: name = cls.__name__ if name.endswith("Tool"): name = name[:-4] @@ -1062,6 +1140,9 @@ class Tool(Component): name = "".join(["_" + c.lower() if c.isupper() else c for c in name]).lstrip("_") return name + def get_name(self) -> str: + return self.get_name_from_cls() + def get_apply_fn(self) -> Callable: apply_fn = getattr(self, "apply") if apply_fn is None: @@ -1084,12 +1165,48 @@ class Tool(Component): return "" return docstring.strip() - def get_function_description(self) -> str: - apply_fn = self.get_apply_fn() + @classmethod + def get_apply_docstring_from_cls(cls) -> str: + """Get the docstring for the apply method from the class (static metadata). + Needed for creating MCP tools in a separate process without running into serialization issues. + """ + # First try to get from __dict__ to handle dynamic docstring changes + if "apply" in cls.__dict__: + apply_fn = cls.__dict__["apply"] + else: + # Fall back to getattr for inherited methods + apply_fn = getattr(cls, "apply", None) + if apply_fn is None: + raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?") + docstring = apply_fn.__doc__ - if docstring is None: - raise Exception(f"Missing docstring for {self}") - return docstring + if not docstring: + raise AttributeError(f"apply method has no (or empty) docstring in {cls}. Did you forget to implement it?") + return docstring.strip() + + def get_apply_docstring(self) -> str: + """Get the docstring for the apply method (instance method implementing ToolProtocol).""" + return self.get_apply_docstring_from_cls() + + def get_apply_fn_metadata(self) -> FuncMetadata: + """Get the metadata for the apply method (instance method implementing ToolProtocol).""" + return self.get_apply_fn_metadata_from_cls() + + @classmethod + def get_apply_fn_metadata_from_cls(cls) -> FuncMetadata: + """Get the metadata for the apply method from the class (static metadata). + Needed for creating MCP tools in a separate process without running into serialization issues. + """ + # First try to get from __dict__ to handle dynamic docstring changes + if "apply" in cls.__dict__: + apply_fn = cls.__dict__["apply"] + else: + # Fall back to getattr for inherited methods + apply_fn = getattr(cls, "apply", None) + if apply_fn is None: + raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?") + + return func_metadata(apply_fn, skip_names=["self", "cls"]) def _log_tool_application(self, frame: Any) -> None: params = {} @@ -1101,7 +1218,7 @@ class Tool(Component): params.update(value) else: params[param] = value - log.info(f"{self.get_name()}: {dict_string(params)}") + log.info(f"{self.get_name_from_cls()}: {dict_string(params)}") @staticmethod def _limit_length(result: str, max_answer_chars: int) -> str: @@ -1123,9 +1240,9 @@ class Tool(Component): try: if not self.is_active(): - return f"Error: Tool '{self.get_name()}' is not active. Active tools: {self.agent.get_active_tool_names()}" + 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()} is active: {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()) @@ -1143,7 +1260,7 @@ class Tool(Component): # apply the actual tool with a timeout execution_fn = lambda: apply_fn(**kwargs) - execution_result = execute_with_timeout(execution_fn, self.agent.serena_config.tool_timeout, self.get_name()) + execution_result = execute_with_timeout(execution_fn, self.agent.serena_config.tool_timeout, self.get_name_from_cls()) if execution_result.status == ExecutionResult.Status.SUCCESS: result = cast(str, execution_result.result_value) else: @@ -1696,7 +1813,7 @@ class DeleteLinesTool(Tool, ToolMarkerCanEdit): """ if not self.lines_read.were_lines_read(relative_path, (start_line, end_line)): read_lines_tool = self.agent.get_tool(ReadFileTool) - return f"Error: Must call `{read_lines_tool.get_name()}` first to read exactly the affected lines." + return f"Error: Must call `{read_lines_tool.get_name_from_cls()}` first to read exactly the affected lines." self.symbol_manager.delete_lines(relative_path, start_line, end_line) return SUCCESS_RESULT @@ -2158,7 +2275,7 @@ def _iter_tool_classes(same_module_only: bool = True) -> Generator[type[Tool], N yield tool_class -_TOOL_REGISTRY_DICT: dict[str, type[Tool]] = {tool_class.get_name(): tool_class for tool_class in _iter_tool_classes()} +_TOOL_REGISTRY_DICT: dict[str, type[Tool]] = {tool_class.get_name_from_cls(): tool_class for tool_class in _iter_tool_classes()} """maps tool name to the corresponding tool class""" @@ -2194,7 +2311,7 @@ class ToolRegistry: tool_dict: dict[str, type[Tool] | Tool] = {} for tool_class in tools: - tool_dict[tool_class.get_name()] = tool_class + tool_dict[tool_class.get_name_from_cls()] = tool_class for tool_name in sorted(tool_dict.keys()): tool_class = tool_dict[tool_name] print(f" * `{tool_name}`: {tool_class.get_tool_description().strip()}") diff --git a/src/serena/agno.py b/src/serena/agno.py index c9150693..0bb0e70f 100644 --- a/src/serena/agno.py +++ b/src/serena/agno.py @@ -25,7 +25,7 @@ class SerenaAgnoToolkit(Toolkit): def __init__(self, serena_agent: SerenaAgent): super().__init__("Serena") for tool in serena_agent.get_exposed_tool_instances(): - self.functions[tool.get_name()] = self._create_agno_function(tool) + self.functions[tool.get_name_from_cls()] = self._create_agno_function(tool) log.info("Agno agent functions: %s", list(self.functions.keys())) @staticmethod @@ -39,7 +39,7 @@ class SerenaAgnoToolkit(Toolkit): return tool.apply_ex(log_call=True, catch_exceptions=True, **kwargs) function = Function.from_callable(tool.get_apply_fn()) - function.name = tool.get_name() + function.name = tool.get_name_from_cls() function.entrypoint = entrypoint function.skip_entrypoint_processing = True return function diff --git a/src/serena/mcp.py b/src/serena/mcp.py index f499b3d5..db188a2f 100644 --- a/src/serena/mcp.py +++ b/src/serena/mcp.py @@ -8,21 +8,23 @@ from contextlib import asynccontextmanager from dataclasses import dataclass from logging import Formatter, Logger, StreamHandler from pathlib import Path -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal import click # Add click import import docstring_parser from mcp.server.fastmcp import server from mcp.server.fastmcp.server import FastMCP, Settings from mcp.server.fastmcp.tools.base import Tool as MCPTool -from mcp.server.fastmcp.utilities.func_metadata import func_metadata from sensai.util import logging from sensai.util.helper import mark_used -from serena.agent import SerenaAgent, SerenaConfig, ToolProtocol, show_fatal_exception_safe +from serena.agent import SerenaAgent, ToolInterface, create_serena_config, show_fatal_exception_safe from serena.config import SerenaAgentContext, SerenaAgentMode from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES -from serena.process_isolated_agent import ProcessIsolatedSerenaAgent, ProcessIsolatedTool + +# Import for type annotations +if TYPE_CHECKING: + from serena.process_isolated_agent import ProcessIsolatedSerenaAgent log = logging.getLogger(__name__) LOG_FORMAT = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s" @@ -47,18 +49,12 @@ class SerenaMCPRequestContext: def make_tool( - tool: ToolProtocol, + tool: ToolInterface, ) -> MCPTool: func_name = tool.get_name() - - apply_fn = getattr(tool, "apply") - if apply_fn is None: - raise ValueError(f"Tool does not have an apply method: {tool}") - - func_doc = apply_fn.__doc__ or "" + func_doc = tool.get_apply_docstring() or "" + func_arg_metadata = tool.get_apply_fn_metadata() is_async = False - - func_arg_metadata = func_metadata(apply_fn) parameters = func_arg_metadata.arg_model.model_json_schema() docstring = docstring_parser.parse(func_doc) @@ -94,6 +90,7 @@ def make_tool( fn_metadata=func_arg_metadata, is_async=is_async, context_kwarg=None, + annotations=None, ) @@ -108,7 +105,7 @@ def create_mcp_server_and_agent( log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None, trace_lsp_communication: bool | None = None, tool_timeout: float | None = None, -) -> tuple[FastMCP, SerenaAgent]: +) -> tuple[FastMCP, "ProcessIsolatedSerenaAgent"]: """ Create an MCP server (legacy - now uses process isolation). @@ -141,7 +138,7 @@ def create_mcp_server_and_agent( tool_timeout=tool_timeout, ) # Return a dummy agent for compatibility - return mcp, process_agent # type: ignore + return mcp, process_agent def create_mcp_server_and_process_isolated_agent( @@ -155,7 +152,7 @@ def create_mcp_server_and_process_isolated_agent( log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None, trace_lsp_communication: bool | None = None, tool_timeout: float | None = None, -) -> tuple[FastMCP, ProcessIsolatedSerenaAgent]: +) -> tuple[FastMCP, "ProcessIsolatedSerenaAgent"]: """ Create an MCP server with process-isolated SerenaAgent to prevent asyncio contamination. @@ -179,8 +176,8 @@ def create_mcp_server_and_process_isolated_agent( modes_instances = [SerenaAgentMode.load(mode) for mode in modes] try: - # First create a regular agent to get the config - temp_agent = SerenaAgent( + # Create configuration without instantiating a full SerenaAgent + serena_config = create_serena_config( project=project, context=context_instance, modes=modes_instances, @@ -191,15 +188,10 @@ def create_mcp_server_and_process_isolated_agent( tool_timeout=tool_timeout, ) - # Create process-isolated agent with the same config - if not isinstance(temp_agent.serena_config, SerenaConfig): - raise TypeError("Expected SerenaConfig instance") - process_agent = ProcessIsolatedSerenaAgent(temp_agent.serena_config) + # Create process-isolated agent with the configuration + from serena.process_isolated_agent import ProcessIsolatedSerenaAgent - # Clean up temp agent - if temp_agent.language_server is not None: - temp_agent.language_server.stop() - del temp_agent + process_agent = ProcessIsolatedSerenaAgent(serena_config) except Exception as e: show_fatal_exception_safe(e) @@ -209,23 +201,29 @@ def create_mcp_server_and_process_isolated_agent( """Update the tools in the MCP server - adapted for process isolation.""" nonlocal mcp, process_agent - # Get tool instances data from process-isolated agent + # Check if process agent is running + if process_agent.process is None or not process_agent.process.is_alive(): + log.debug("Process agent not running yet, skipping tool update") + return + + # Get tool names from process-isolated agent try: - tool_instances_data = process_agent.get_tool_instances() + tool_names = process_agent.get_tool_instances() except Exception as e: - log.error(f"Failed to get tool instances from process agent: {e}") + log.error(f"Failed to get tool names from process agent: {e}") return if mcp is not None: + from serena.process_isolated_agent import ProcessIsolatedTool + mcp._tool_manager._tools = {} - for tool_data in tool_instances_data: + for tool_name in tool_names: # Create ProcessIsolatedTool instance that can be used with make_tool - tool_instance = ProcessIsolatedTool( - process_agent=process_agent, tool_name=tool_data["name"], tool_doc=tool_data["apply_method_doc"] - ) + # Metadata is extracted from ToolRegistry using classmethods + tool_instance = ProcessIsolatedTool(process_agent=process_agent, tool_name=tool_name) # Use the original make_tool function to get full docstring parsing mcp_tool = make_tool(tool_instance) - mcp._tool_manager._tools[tool_data["name"]] = mcp_tool + mcp._tool_manager._tools[tool_name] = mcp_tool @asynccontextmanager async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[None]: @@ -235,6 +233,8 @@ def create_mcp_server_and_process_isolated_agent( try: process_agent.start() + # Update tools now that the process agent is running + update_tools() yield finally: process_agent.stop() diff --git a/src/serena/process_isolated_agent.py b/src/serena/process_isolated_agent.py index de562a11..543b5ec2 100644 --- a/src/serena/process_isolated_agent.py +++ b/src/serena/process_isolated_agent.py @@ -15,7 +15,9 @@ import time import traceback from typing import Any, Self -from serena.agent import SerenaAgent, SerenaConfig, ToolProtocol +from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata + +from serena.agent import SerenaAgent, SerenaConfig, SerenaConfigBase, Tool, ToolInterface, ToolRegistry log = logging.getLogger(__name__) @@ -110,7 +112,7 @@ class SerenaAgentWorker: # Get the tool by name tool = None for tool_instance in self.agent._active_tools.values(): - if tool_instance.get_name() == tool_name: + if tool_instance.get_name_from_cls() == tool_name: tool = tool_instance break @@ -159,23 +161,15 @@ class SerenaAgentWorker: return {"id": request_id, "error": str(e), "traceback": traceback.format_exc()} def _get_tool_instances(self, request_id: str) -> dict[str, Any]: - """Get exposed tool instances for MCP tool creation.""" + """Get exposed tool names for MCP tool creation.""" if self.agent is None: return {"id": request_id, "error": "SerenaAgent not initialized"} try: tool_instances = self.agent.get_exposed_tool_instances() - # Serialize tool information that can be used to recreate tools on the MCP side - tool_data = [] - for tool in tool_instances: - tool_data.append( - { - "name": tool.get_name(), - "class_name": tool.__class__.__name__, - "apply_method_doc": getattr(tool, "apply").__doc__, - } - ) - return {"id": request_id, "result": tool_data} + # Return only tool names - metadata will be reconstructed from ToolRegistry + tool_names = [tool.get_name_from_cls() for tool in tool_instances] + return {"id": request_id, "result": tool_names} except Exception as e: return {"id": request_id, "error": str(e), "traceback": traceback.format_exc()} @@ -193,7 +187,7 @@ class SerenaAgentWorker: class ProcessIsolatedSerenaAgent: """Process-isolated wrapper for SerenaAgent that prevents asyncio contamination.""" - def __init__(self, serena_config: SerenaConfig): + def __init__(self, serena_config: SerenaConfigBase): self.serena_config = serena_config self.process: multiprocessing.Process | None = None self.request_queue: multiprocessing.Queue | None = None @@ -320,8 +314,8 @@ class ProcessIsolatedSerenaAgent: """Reset the language server.""" self._make_request_with_result("reset_language_server") - def get_tool_instances(self) -> list[dict[str, Any]]: - """Get tool instances data for MCP tool creation.""" + def get_tool_instances(self) -> list[str]: + """Get tool names for MCP tool creation.""" return self._make_request_with_result("get_tool_instances") def __enter__(self) -> Self: @@ -332,30 +326,36 @@ class ProcessIsolatedSerenaAgent: self.stop() -class ProcessIsolatedTool(ToolProtocol): +class ProcessIsolatedTool(ToolInterface): """A clean tool wrapper that delegates to ProcessIsolatedSerenaAgent.""" - def __init__(self, process_agent: ProcessIsolatedSerenaAgent, tool_name: str, tool_doc: str | None = None): + def __init__(self, process_agent: ProcessIsolatedSerenaAgent, tool_name: str): self.process_agent = process_agent - self.tool_name = tool_name - self._tool_doc = tool_doc - # Set the docstring for the apply method if provided - if tool_doc: - self.apply.__doc__ = tool_doc + self._tool_name = tool_name - def apply(self, **kwargs: Any) -> str: - """Apply the tool through the process-isolated agent.""" - return self.process_agent.tool_call(self.tool_name, **kwargs) + @property + def _tool_class(self) -> type[Tool]: + return ToolRegistry.get_tool_class_by_name(self._tool_name) def get_name(self) -> str: """Get the tool name for this process-isolated tool.""" - return self.tool_name + return self._tool_name + + def get_apply_docstring(self) -> str: + """Get the docstring for the apply method.""" + # in the actual tool, this is a classmethod + return self._tool_class.get_apply_docstring_from_cls() + + def get_apply_fn_metadata(self) -> FuncMetadata: + """Get the metadata for the apply method.""" + # in the actual tool, this is a classmethod + return self._tool_class.get_apply_fn_metadata_from_cls() def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs: Any) -> str: """Apply the tool with logging and exception handling.""" try: - return self.apply(**kwargs) + return self.process_agent.tool_call(self._tool_name, **kwargs) except Exception as e: if catch_exceptions: - return f"Error executing tool {self.tool_name}: {e!s}" + return f"Error executing tool {self._tool_name}: {e!s}" raise diff --git a/test/serena/test_make_tool_process_isolation.py b/test/serena/test_make_tool_process_isolation.py new file mode 100644 index 00000000..f0fdda26 --- /dev/null +++ b/test/serena/test_make_tool_process_isolation.py @@ -0,0 +1,160 @@ +"""Tests for make_tool consistency between regular tools and ProcessIsolatedTool.""" + +import pytest +from mcp.server.fastmcp.tools.base import Tool as MCPTool + +from serena.agent import SerenaAgent, ToolRegistry +from serena.mcp import make_tool +from serena.process_isolated_agent import ProcessIsolatedSerenaAgent, ProcessIsolatedTool +from test.serena.test_serena_agent import SerenaConfigForTests + + +@pytest.fixture +def in_memory_config(): + """Create an in-memory configuration for tests.""" + return SerenaConfigForTests() + + +@pytest.fixture +def regular_agent(in_memory_config): + """Create a regular SerenaAgent for comparison.""" + return SerenaAgent(serena_config=in_memory_config) + + +@pytest.fixture +def process_isolated_agent(in_memory_config): + """Create a ProcessIsolatedSerenaAgent for comparison.""" + agent = ProcessIsolatedSerenaAgent(in_memory_config) + agent.start() + yield agent + agent.stop() + + +class TestMakeToolProcessIsolation: + """Test that make_tool produces identical metadata for regular and process-isolated tools.""" + + @pytest.mark.parametrize("tool_name", ToolRegistry.get_tool_names()) + def test_make_tool_metadata_consistency( + self, tool_name: str, regular_agent: SerenaAgent, process_isolated_agent: ProcessIsolatedSerenaAgent + ): + """Test that make_tool produces identical metadata for regular and process-isolated tools.""" + # Get regular tool instance + tool_class = ToolRegistry.get_tool_class_by_name(tool_name) + regular_tool = regular_agent.get_tool(tool_class) + + # Get ProcessIsolatedTool instance + isolated_tool = ProcessIsolatedTool(process_isolated_agent, tool_name) + + # Create MCP tools from both + regular_mcp_tool = make_tool(regular_tool) + isolated_mcp_tool = make_tool(isolated_tool) + + # Verify both are MCPTool instances + assert isinstance(regular_mcp_tool, MCPTool) + assert isinstance(isolated_mcp_tool, MCPTool) + + # Test name consistency + assert regular_mcp_tool.name == isolated_mcp_tool.name + assert regular_mcp_tool.name == tool_name + + # Test description consistency + assert regular_mcp_tool.description == isolated_mcp_tool.description, ( + f"Tool {tool_name}: descriptions differ\n" + f"Regular: {regular_mcp_tool.description}\n" + f"Isolated: {isolated_mcp_tool.description}" + ) + + # Test parameters schema consistency + assert regular_mcp_tool.parameters == isolated_mcp_tool.parameters, ( + f"Tool {tool_name}: parameter schemas differ\n" + f"Regular: {regular_mcp_tool.parameters}\n" + f"Isolated: {isolated_mcp_tool.parameters}" + ) + + # Test function metadata consistency (compare schemas, not class objects) + regular_schema = regular_mcp_tool.fn_metadata.arg_model.model_json_schema() + isolated_schema = isolated_mcp_tool.fn_metadata.arg_model.model_json_schema() + assert ( + regular_schema == isolated_schema + ), f"Tool {tool_name}: function metadata schemas differ\nRegular: {regular_schema}\nIsolated: {isolated_schema}" + + # Test async flag consistency + assert regular_mcp_tool.is_async == isolated_mcp_tool.is_async + + # Test context kwarg consistency + assert regular_mcp_tool.context_kwarg == isolated_mcp_tool.context_kwarg + + @pytest.mark.parametrize("tool_name", ToolRegistry.get_tool_names()[:5]) # Test first 5 tools for faster execution + def test_tool_protocol_methods_consistency( + self, tool_name: str, regular_agent: SerenaAgent, process_isolated_agent: ProcessIsolatedSerenaAgent + ): + """Test that Tool methods return identical results for regular and process-isolated tools.""" + # Get regular tool instance + tool_class = ToolRegistry.get_tool_class_by_name(tool_name) + regular_tool = regular_agent.get_tool(tool_class) + + # Get ProcessIsolatedTool instance + isolated_tool = ProcessIsolatedTool(process_isolated_agent, tool_name) + + # Test get_name() + assert regular_tool.get_name_from_cls() == isolated_tool.get_name() + assert regular_tool.get_name_from_cls() == tool_name + + # Test get_apply_docstring() + regular_docstring = regular_tool.get_apply_docstring() + isolated_docstring = isolated_tool.get_apply_docstring() + assert ( + regular_docstring == isolated_docstring + ), f"Tool {tool_name}: docstrings differ\nRegular: {regular_docstring}\nIsolated: {isolated_docstring}" + + # Test get_apply_fn_metadata() + regular_metadata = regular_tool.get_apply_fn_metadata() + isolated_metadata = isolated_tool.get_apply_fn_metadata() + + # Compare metadata properties (compare schemas, not class objects) + regular_schema = regular_metadata.arg_model.model_json_schema() + isolated_schema = isolated_metadata.arg_model.model_json_schema() + assert ( + regular_schema == isolated_schema + ), f"Tool {tool_name}: metadata schemas differ\nRegular: {regular_schema}\nIsolated: {isolated_schema}" + + def test_process_isolated_tool_uses_tool_registry(self, process_isolated_agent: ProcessIsolatedSerenaAgent): + """Test that ProcessIsolatedTool correctly uses ToolRegistry for metadata.""" + tool_name = ToolRegistry.get_tool_names()[0] # Use first available tool + isolated_tool = ProcessIsolatedTool(process_isolated_agent, tool_name) + + # Verify that the tool uses ToolRegistry + assert isolated_tool._tool_class == ToolRegistry.get_tool_class_by_name(tool_name) + + # Verify that metadata comes from the tool class + expected_docstring = isolated_tool._tool_class.get_apply_docstring_from_cls() + expected_metadata = isolated_tool._tool_class.get_apply_fn_metadata_from_cls() + + assert isolated_tool.get_apply_docstring() == expected_docstring + # Compare schemas, not class objects + isolated_schema = isolated_tool.get_apply_fn_metadata().arg_model.model_json_schema() + expected_schema = expected_metadata.arg_model.model_json_schema() + assert isolated_schema == expected_schema + + def test_tool_registry_completeness(self, regular_agent: SerenaAgent, process_isolated_agent: ProcessIsolatedSerenaAgent): + """Test that all tools are available in both agents and the registry.""" + # Get tool names from both agents + regular_active_tools = set(regular_agent.get_active_tool_names()) + regular_all_tools = set(tool.get_name_from_cls() for tool in regular_agent.get_exposed_tool_instances()) + isolated_tool_names = set(process_isolated_agent.get_tool_instances()) + + # The process isolated agent should have all tools (exposed, not just active) + assert regular_all_tools == isolated_tool_names, ( + f"Tool sets differ:\n" + f"Regular exposed only: {regular_all_tools - isolated_tool_names}\n" + f"Isolated only: {isolated_tool_names - regular_all_tools}" + ) + + # Active tools should be a subset of all tools + assert regular_active_tools.issubset( + regular_all_tools + ), f"Some active tools not in exposed tools: {regular_active_tools - regular_all_tools}" + + # All tools should be in the registry + registry_tool_names = set(ToolRegistry.get_tool_names()) + assert regular_all_tools.issubset(registry_tool_names), f"Some tools not in registry: {regular_all_tools - registry_tool_names}" diff --git a/test/serena/test_mcp.py b/test/serena/test_mcp.py index f179b857..eed27a7f 100644 --- a/test/serena/test_mcp.py +++ b/test/serena/test_mcp.py @@ -224,7 +224,7 @@ def test_make_tool_missing_apply() -> None: """, "", ), - ("", ""), + ("Description without params.", "Description without params."), ], ) def test_make_tool_descriptions(docstring, expected_description) -> None: @@ -287,7 +287,7 @@ def test_make_tool_all_tools(tool_class) -> None: # Basic validation assert isinstance(mcp_tool, MCPTool) - assert mcp_tool.name == tool_class.get_name() + assert mcp_tool.name == tool_class.get_name_from_cls() # The description should be a string (either from docstring or default) assert isinstance(mcp_tool.description, str) diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index f1208dd3..49a47d88 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -6,7 +6,7 @@ from dataclasses import dataclass import pytest from multilspy.multilspy_config import Language -from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, SerenaAgent, SerenaConfigBase +from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, Project, ProjectConfig, SerenaAgent, SerenaConfigBase from test.conftest import LanguageParamRequest, get_repo_path @@ -19,17 +19,46 @@ class SerenaConfigForTests(SerenaConfigBase): gui_log_window_enabled: bool = False web_dashboard: bool = False + def __post_init__(self): + # Initialize with empty projects list if not already set + if not hasattr(self, "projects") or self.projects is None: + self.projects = [] + @pytest.fixture def serena_config(): - return SerenaConfigForTests() + """Create an in-memory configuration for tests with test repositories pre-registered.""" + # Create test projects for all supported languages + test_projects = [] + for language in [Language.PYTHON, Language.GO, Language.JAVA, Language.RUST, Language.TYPESCRIPT, Language.PHP]: + repo_path = get_repo_path(language) + if repo_path.exists(): + project_name = f"test_repo_{language}" + project = Project( + project_root=str(repo_path), + project_config=ProjectConfig( + project_name=project_name, + language=language, + ignored_paths=[], + excluded_tools=set(), + read_only=False, + ignore_all_files_in_gitignore=True, + initial_prompt="", + encoding="utf-8", + ), + ) + test_projects.append(project) + + config = SerenaConfigForTests() + config.projects = test_projects + return config @pytest.fixture def serena_agent(request: LanguageParamRequest, serena_config) -> SerenaAgent: language = Language(request.param) - repo = get_repo_path(language) - return SerenaAgent(project=str(repo), serena_config=serena_config) + project_name = f"test_repo_{language}" + return SerenaAgent(project=project_name, serena_config=serena_config) class TestSerenaAgent: