tiennm99/serena
1
0
Fork 0
mirror of https://github.com/tiennm99/serena.git synced 2026-09-04 20:18:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove process isolation (SerenaMCPFactoryWithProcessIsolation, ProcessIsolatedSerenaAgent, etc.)

Browse Source
This commit is contained in:
Dominik Jain
2025-07-02 12:20:41 +02:00
committed by Dominik Jain
parent 5edf905833
commit 49515d6235
8 changed files with 7 additions and 1154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.serena/memories/process_isolation_current_status.md
-126
View File
@@ -1,126 +0,0 @@
# Process Isolation: Current Status and Architecture
## Current Status: Disabled by Default
Process isolation is **disabled by default** in the current Serena implementation:
- **Configuration**: `USE_PROCESS_ISOLATION = False` in `src/serena/constants.py:22`
- **Reason**: No longer needed with solid-lsp architecture
- **Default Operation**: Single process mode with MCP server, agent, and language servers in same process
## Why Process Isolation Is No Longer Needed
### Historical Context
Process isolation was **mandatory** when using multilspy because:
1. **Asyncio Contamination**: Multilspy leaked coroutines into MCP server's event loop
2. **Event Loop Conflicts**: Multiple asyncio contexts caused deadlocks
3. **Only Solution**: Complete process separation was the only way to prevent issues
### Current Architecture Benefits
With solid-lsp, process isolation became **unnecessary** because:
1. **Clean Async Boundaries**: No coroutine leakage between MCP server and language server contexts
2. **Single Process Safety**: Solid-lsp designed to work safely within MCP server process
3. **Performance Gains**: Direct method calls instead of expensive IPC
## Architecture Components (Still Present)
The process isolation infrastructure remains **available but unused** by default:
### Core Components (`src/serena/process_isolated_agent.py`)
#### 1. **ProcessIsolatedSerenaAgent** (lines 392-550)
- **Purpose**: Wrapper that manages isolated agent process
- **Communication**: Uses `multiprocessing.Pipe()` for bidirectional communication
- **Process Management**: Creates and manages `SerenaAgentWorker` subprocess
- **Status**: **Available but not used by default**
#### 2. **SerenaAgentWorker** (lines 173-389)
- **Purpose**: Worker process hosting actual SerenaAgent
- **Event Loop**: Polling loop checking for requests every 500ms
- **Request Handling**: INITIALIZE, TOOL_CALL, SHUTDOWN, etc.
- **Status**: **Available but not used by default**
#### 3. **ProcessIsolatedDashboard** (lines 133-170)
- **Purpose**: Runs web dashboard in separate process
- **Integration**: Async web server with port management
- **Status**: **Available but not used by default**
### Global Synchronization (lines 25-28)
```python
_global_log_queue: multiprocessing.Queue = multiprocessing.Queue()
_dashboard_ready_event = multiprocessing.Event()
_dashboard_port_value = multiprocessing.Value("i", 0)
global_shutdown_event = multiprocessing.Event()
```
**Status**: **Available but not actively used**
## Current MCP Factory Selection
In `src/serena/mcp.py:590`:
```python
if not USE_PROCESS_ISOLATION:
mcp_factory = SerenaMCPFactorySingleProcess(context=context, project=project_file)
else:
mcp_factory = SerenaMCPFactoryWithProcessIsolation(context=context, project=project_file)
```
### Default Flow
1. **`USE_PROCESS_ISOLATION = False`** → **`SerenaMCPFactorySingleProcess`**
2. **Single Process**: MCP server, SerenaAgent, and solid-lsp in same process
3. **Direct Communication**: No IPC overhead, direct method calls
### Fallback Option
If `USE_PROCESS_ISOLATION = True`:
1. **`SerenaMCPFactoryWithProcessIsolation`** would be used
2. **Separate Processes**: MCP server and SerenaAgent in different processes
3. **IPC Communication**: Higher latency but complete isolation
## Benefits of Current Single Process Architecture
### 1. **Performance Improvements**
- **Lower Latency**: Direct method calls vs IPC communication
- **Reduced Memory**: No duplicate process memory footprints
- **Faster Startup**: No process creation and initialization overhead
### 2. **Operational Simplicity**
- **Unified Logging**: All components log to same destination
- **Simpler Debugging**: Single process, unified stack traces
- **Resource Management**: Simpler cleanup and shutdown procedures
### 3. **Stability Benefits**
- **No IPC Failures**: Eliminated inter-process communication failure modes
- **Consistent State**: No synchronization issues between processes
- **Reliable Shutdown**: No orphaned processes or cleanup complexity
## When Process Isolation Might Still Be Useful
While not needed by default, process isolation could still be beneficial for:
### 1. **Fault Isolation**
- **Agent Crashes**: Prevent agent failures from affecting MCP server
- **Memory Protection**: Isolate memory leaks to specific processes
- **Recovery**: Restart failed components without full system restart
### 2. **Resource Management**
- **Memory Limits**: Constrain memory usage of specific components
- **CPU Isolation**: Prevent CPU-intensive operations from blocking MCP server
- **Security**: Additional process boundaries for security-sensitive environments
### 3. **Debugging and Development**
- **Component Isolation**: Debug specific components in isolation
- **Performance Analysis**: Measure resource usage per component
- **Development Safety**: Prevent development errors from affecting stable components
## Configuration Management
### Enabling Process Isolation
To re-enable process isolation:
1. **Set**: `USE_PROCESS_ISOLATION = True` in `src/serena/constants.py`
2. **Result**: Automatic fallback to `SerenaMCPFactoryWithProcessIsolation`
3. **Trade-off**: Higher resource usage but complete component isolation
### Current Recommendation
- **Default**: Keep `USE_PROCESS_ISOLATION = False` for optimal performance
- **Special Cases**: Enable only when specific isolation requirements exist
- **Testing**: Both modes should be tested to ensure compatibility
The current architecture successfully eliminated the need for process isolation while maintaining the capability as a fallback option for specialized use cases.
.serena/memories/suggested_commands.md
-1
View File
@@ -17,7 +17,6 @@ The following tasks should generally be executed using `uv run poe <task_name>`.
"typescript: language server running for TypeScript",
"php: language server running for PHP",
"snapshot: snapshot tests for symbolic editing operations",
"isolated_process: test runs with process isolated agent",
]
```
By default, `uv run poe test` uses the markers set in the env var `PYTEST_MARKERS`, or, if it unset, uses `-m "not java and not rust and not isolated process"`.
pyproject.toml
+1 -2
View File
@@ -104,7 +104,7 @@ PYDEVD_DISABLE_FILE_VALIDATION = "1"
# Uses PYTEST_MARKERS env var for default markers
# For custom markers, one can either adjust the env var or just use -m option in the command line,
# as the second -m option will override the first one.
test = "pytest test -vv -m \"${PYTEST_MARKERS:-not java and not rust and not isolated_process}\""
test = "pytest test -vv -m \"${PYTEST_MARKERS:-not java and not rust}\""
_black_check = "black --check src scripts test"
_ruff_check = "ruff check src scripts test"
_black_format = "black src scripts test"
@@ -247,7 +247,6 @@ markers = [
"php: language server running for PHP",
"csharp: language server running for C#",
"snapshot: snapshot tests for symbolic editing operations",
"isolated_process: test runs with process isolated agent",
]
[tool.codespell]
src/serena/constants.py
-2
View File
@@ -19,6 +19,4 @@ DEFAULT_MODES = ("interactive", "editing")
PROJECT_TEMPLATE_FILE = str(_serena_pkg_path / "resources" / "project.template.yml")
SELENA_CONFIG_TEMPLATE_FILE = str(_serena_pkg_path / "resources" / "serena_config.template.yml")
USE_PROCESS_ISOLATION = False
SERENA_LOG_FORMAT = "%(levelname)-5s %(asctime)-15s [%(threadName)s] %(name)s:%(funcName)s:%(lineno)d - %(message)s"
src/serena/mcp.py
+3 -254
View File
@@ -2,13 +2,7 @@
The Serena Model Context Protocol (MCP) Server
"""
import asyncio
import contextlib
import os
import signal
import sys
import threading
import time
from abc import abstractmethod
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import asynccontextmanager
@@ -26,24 +20,14 @@ from pydantic_settings import SettingsConfigDict
from sensai.util import logging
from serena.agent import (
ActivateProjectTool,
Project,
SerenaAgent,
SerenaConfig,
ToolInterface,
ToolRegistry,
create_serena_config,
show_fatal_exception_safe,
)
from serena.config import RegisteredContext, SerenaAgentContext, SerenaAgentMode
from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES, USE_PROCESS_ISOLATION
from serena.process_isolated_agent import (
ProcessIsolatedDashboard,
ProcessIsolatedSerenaAgent,
ProcessIsolatedTool,
global_shutdown_event,
request_global_shutdown,
)
from serena.config import SerenaAgentContext, SerenaAgentMode
from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES
log = logging.getLogger(__name__)
LOG_FORMAT = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s"
@@ -224,235 +208,6 @@ class SerenaMCPFactorySingleProcess(SerenaMCPFactory):
yield
class SerenaMCPFactoryWithProcessIsolation(SerenaMCPFactory):
"""
MCP server factory with process isolation for the SerenaAgent and its language server; they run in a separate process
from the MCP server.
"""
def __init__(self, context: str = DEFAULT_CONTEXT, project: str | None = None):
"""
:param context: The context name or path to context file
:param project: Either an absolute path to the project directory or a name of an already registered project.
If the project passed here hasn't been registered yet, it will be registered automatically and can be activated by its name
afterward.
"""
super().__init__(context=context, project=project)
self.active_tool_names: set[str] | None = None
self.serena_agent_process: ProcessIsolatedSerenaAgent | None = None
self.serena_dashboard_process: ProcessIsolatedDashboard | None = None
@staticmethod
def _determine_active_tool_names(context: SerenaAgentContext, project: Project | None) -> set[str]:
"""
Determine the names of tools that should be included in this session based on the context.
"""
tools_excluded_in_this_session = context.get_excluded_tool_classes()
# if a project has been loaded, it will be activated at startup and in ide-assistant context,
# we assume that no other project will be activated in this session.
# Therefore, we exclude the activate project tool
is_ide_assistant = context.name == RegisteredContext.IDE_ASSISTANT.value
if is_ide_assistant and project is not None:
tools_excluded_in_this_session.extend(project.project_config.get_excluded_tool_classes())
tools_excluded_in_this_session.append(ActivateProjectTool)
tool_names_excluded_in_this_session = {tool.get_name_from_cls() for tool in tools_excluded_in_this_session}
all_tool_names = set(ToolRegistry.get_tool_names())
tool_names_included_in_this_session = all_tool_names - tool_names_excluded_in_this_session
return tool_names_included_in_this_session
@staticmethod
def make_mcp_tool(tool: ToolInterface) -> MCPTool:
func_name = tool.get_name()
func_doc = tool.get_apply_docstring() or ""
func_arg_metadata = tool.get_apply_fn_metadata()
is_async = False
parameters = func_arg_metadata.arg_model.model_json_schema()
docstring = docstring_parser.parse(func_doc)
# Mount the tool description as a combination of the docstring description and
# the return value description, if it exists.
if docstring.description:
func_doc = f"{docstring.description.strip().strip('.')}."
else:
func_doc = ""
if docstring.returns and (docstring_returns_descr := docstring.returns.description):
# Only add a space before "Returns" if func_doc is not empty
prefix = " " if func_doc else ""
func_doc = f"{func_doc}{prefix}Returns {docstring_returns_descr.strip().strip('.')}."
# Parse the parameter descriptions from the docstring and add pass its description
# to the parameter schema.
docstring_params = {param.arg_name: param for param in docstring.params}
parameters_properties: dict[str, dict[str, Any]] = parameters["properties"]
for parameter, properties in parameters_properties.items():
if (param_doc := docstring_params.get(parameter)) and param_doc.description:
param_desc = f"{param_doc.description.strip().strip('.') + '.'}"
properties["description"] = param_desc[0].upper() + param_desc[1:]
def execute_fn(**kwargs) -> str: # type: ignore
return tool.apply_ex(log_call=True, catch_exceptions=True, **kwargs)
return MCPTool(
fn=execute_fn,
name=func_name,
description=func_doc,
parameters=parameters,
fn_metadata=func_arg_metadata,
is_async=is_async,
context_kwarg=None,
annotations=None,
)
def _iter_tools(self) -> Iterator[ToolInterface]:
assert self.active_tool_names is not None
assert self.serena_agent_process is not None
for tool_name in self.active_tool_names:
yield ProcessIsolatedTool(process_agent=self.serena_agent_process, tool_name=tool_name)
# noinspection PyProtectedMember
def _set_mcp_tools(self, mcp: FastMCP) -> None:
"""Update the tools in the MCP server"""
if mcp is not None:
mcp._tool_manager._tools = {}
for tool in self._iter_tools():
mcp_tool = self.make_mcp_tool(tool)
mcp._tool_manager._tools[tool.get_name()] = mcp_tool
def _instantiate_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> None:
if self.project is not None:
self.project_instance = serena_config.get_project(self.project)
self.serena_agent_process = ProcessIsolatedSerenaAgent(
project=self.project, serena_config=serena_config, modes=modes, context=self.context
)
self.active_tool_names = self._determine_active_tool_names(self.context, self.project_instance)
if serena_config.web_dashboard:
assert self.active_tool_names is not None
self.serena_dashboard_process = ProcessIsolatedDashboard(tool_names=sorted(self.active_tool_names))
def create_mcp_server(
self,
host: str = "0.0.0.0",
port: int = 8000,
modes: Sequence[str] = DEFAULT_MODES,
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,
) -> FastMCP:
"""
Create an MCP server with process-isolated SerenaAgent to prevent asyncio contamination.
:param host: The host to bind to
:param port: The port to bind to
:param modes: List of mode names or paths to mode files
:param enable_web_dashboard: Whether to enable the web dashboard. If not specified, will take the value from the serena configuration.
:param enable_gui_log_window: Whether to enable the GUI log window. It currently does not work on macOS, and setting this to True will be ignored then.
If not specified, will take the value from the serena configuration.
:param log_level: Log level. If not specified, will take the value from the serena configuration.
:param trace_lsp_communication: Whether to trace the communication between Serena and the language servers.
This is useful for debugging language server issues.
:param tool_timeout: Timeout in seconds for tool execution. If not specified, will take the value from the serena configuration.
"""
try:
serena_config = create_serena_config(
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,
)
modes_instances = [SerenaAgentMode.load(mode) for mode in modes]
self._instantiate_agent(serena_config, modes_instances)
except Exception as e:
show_fatal_exception_safe(e)
raise
# Override model_config to disable the use of `.env` files for reading settings, because user projects are likely to contain
# `.env` files (e.g. containing LOG_LEVEL) that are not supposed to override the MCP settings;
# retain only FASTMCP_ prefix for already set environment variables.
Settings.model_config = SettingsConfigDict(env_prefix="FASTMCP_")
mcp_settings = Settings(lifespan=self.server_lifespan, host=host, port=port)
mcp = FastMCP(**mcp_settings.model_dump())
return mcp
@asynccontextmanager
async def server_lifespan(self, mcp_server: FastMCP) -> AsyncIterator[None]:
"""Manage server startup and shutdown lifecycle."""
def signal_handler(signum: int, frame: Any) -> None:
log.info(f"Received signal {signum} in main process")
request_global_shutdown()
def force_exit() -> None:
time.sleep(2.0) # Wait 2 seconds for graceful shutdown
log.warning("Forcing exit after timeout")
# noinspection PyProtectedMember
# noinspection PyUnresolvedReferences
os._exit(1)
threading.Thread(target=force_exit, daemon=True).start()
# Install signal handlers
sigint_singal = signal.signal(signal.SIGINT, signal_handler)
sigterm_signal = signal.signal(signal.SIGTERM, signal_handler)
if self.serena_dashboard_process is not None:
log.info("Starting dashboard process")
assert self.serena_dashboard_process is not None
self.serena_dashboard_process.start()
log.info("Starting serena agent process")
assert self.serena_agent_process is not None
self.serena_agent_process.start()
self._set_mcp_tools(mcp_server)
async def monitor_global_shutdown() -> None:
"""Monitor the global shutdown event and trigger local shutdown."""
while not global_shutdown_event.is_set():
# Poll the multiprocessing Event in async context
await asyncio.sleep(0.1)
continue
log.info("Global shutdown event detected, initiating server shutdown")
request_global_shutdown()
# Send SIGTERM to self to trigger graceful shutdown
os.kill(os.getpid(), signal.SIGTERM)
# Start monitoring task
monitor_task = asyncio.create_task(monitor_global_shutdown())
log.info("MCP server lifetime setup complete")
try:
yield
except (KeyboardInterrupt, SystemExit):
log.info("Received shutdown signal")
request_global_shutdown()
except Exception as e:
log.error(f"Error in server lifespan: {e}")
request_global_shutdown()
finally:
# Cancel monitor task
monitor_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await monitor_task
self.serena_agent_process.stop()
if self.serena_dashboard_process is not None:
self.serena_dashboard_process.stop()
request_global_shutdown()
log.info("Shutting down all processes")
signal.signal(signal.SIGINT, sigint_singal)
signal.signal(signal.SIGTERM, sigterm_signal)
class ProjectType(click.ParamType):
name = "[PROJECT_NAME|PROJECT_PATH]"
@@ -588,13 +343,7 @@ def start_mcp_server(
# This is for backward compatibility with the old CLI, should be removed in the future!
project_file = project_file_arg if project_file_arg is not None else project
mcp_factory: SerenaMCPFactory
if not USE_PROCESS_ISOLATION:
mcp_factory = SerenaMCPFactorySingleProcess(context=context, project=project_file)
else:
mcp_factory = SerenaMCPFactoryWithProcessIsolation(context=context, project=project_file)
# Use process isolation by default to prevent asyncio event loop contamination
mcp_factory = SerenaMCPFactorySingleProcess(context=context, project=project_file)
mcp_server = mcp_factory.create_mcp_server(
host=host,
port=port,
src/serena/process_isolated_agent.py
-571
View File
@@ -1,571 +0,0 @@
import asyncio
import contextlib
import logging
import multiprocessing
import os
import threading
import traceback
import webbrowser
from enum import StrEnum
from logging.handlers import QueueHandler
from multiprocessing.connection import Connection
from multiprocessing.sharedctypes import Synchronized
from multiprocessing.synchronize import Event as EventClass
from typing import Any, Literal, Self
from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata
from serena.agent import SerenaAgent, SerenaConfig, SerenaConfigBase, Tool, ToolInterface, ToolRegistry
from serena.config import SerenaAgentContext, SerenaAgentMode
from serena.dashboard import MemoryLogHandler, SerenaDashboardAPI
log = logging.getLogger(__name__)
# Global synchronization primitives
_global_log_queue: multiprocessing.Queue = multiprocessing.Queue()
_dashboard_ready_event = multiprocessing.Event()
_dashboard_port_value = multiprocessing.Value("i", 0)
global_shutdown_event = multiprocessing.Event()
def request_global_shutdown() -> None:
"""Signal the global shutdown event."""
global_shutdown_event.set()
log.info("Global shutdown event set")
def _dashboard_worker(
tool_names: list[str],
log_q: "multiprocessing.Queue[Any]",
dashboard_ready_event: EventClass,
port_value: "Synchronized[int]",
shutdown_evt: EventClass,
) -> None:
"""Entry point for the dashboard process."""
# Route all logging to the shared queue
root = logging.getLogger()
root.handlers.clear()
root.setLevel(logging.DEBUG)
root.addHandler(QueueHandler(log_q))
async def _process_logs(api: SerenaDashboardAPI) -> None:
while not shutdown_evt.is_set():
while not log_q.empty():
record = log_q.get_nowait()
if record is None:
break
api.memory_log_handler.emit(record)
# Small delay to avoid busy waiting
await asyncio.sleep(0.1)
async def _monitor_shutdown() -> None:
# Poll the multiprocessing Event in async context
# Check every 100ms until shutdown is requested
loop = asyncio.get_event_loop()
while True:
# Check in executor to avoid blocking
result = await loop.run_in_executor(None, shutdown_evt.is_set)
if result:
break
await asyncio.sleep(0.1)
async def _async_main() -> None:
api = SerenaDashboardAPI(
memory_log_handler=MemoryLogHandler(),
tool_names=tool_names,
shutdown_callback=shutdown_evt.set,
)
# Pick a free port and signal readiness
port = api._find_first_free_port(0x5EDA)
port_value.value = port
# Start Flask server in a thread
def run_flask_server() -> None:
api._app.run(host="0.0.0.0", port=port, debug=False, use_reloader=False, threaded=True)
server_thread = threading.Thread(target=run_flask_server, daemon=True)
server_thread.start()
shutdown_task = asyncio.create_task(_monitor_shutdown())
logging_loop_task = asyncio.create_task(_process_logs(api))
dashboard_ready_event.set()
done, pending = await asyncio.wait(
[shutdown_task, logging_loop_task],
return_when=asyncio.FIRST_COMPLETED,
)
# Cancel remaining tasks
for task in pending:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
try:
asyncio.run(_async_main())
except BaseException:
logging.exception("Dashboard worker crashed")
finally:
logging.info("Dashboard worker exiting")
def _shutdown_process(proc: multiprocessing.Process, timeout: float = 1.0) -> None:
"""Helper to shutdown a process gracefully."""
if proc.is_alive():
proc.terminate()
proc.join(timeout=timeout)
if proc.is_alive():
log.error("Process did not terminate gracefully, forcing kill")
proc.kill()
proc.join(timeout=timeout)
if proc.is_alive():
log.error("Process did not respond to kill")
class ProcessIsolatedDashboard:
"""Dashboard running in a separate process to avoid asyncio contamination."""
def __init__(self, tool_names: list[str]):
self.tool_names = tool_names
self.process: multiprocessing.Process | None = None
def start(self, timeout: float = 10.0) -> None:
"""Start the dashboard process."""
if self.process is not None:
raise RuntimeError("Dashboard already started")
self.process = multiprocessing.Process(
target=_dashboard_worker,
args=(self.tool_names, _global_log_queue, _dashboard_ready_event, _dashboard_port_value, global_shutdown_event),
daemon=True,
)
self.process.start()
if not _dashboard_ready_event.wait(timeout):
self.process.terminate()
self.process.join(timeout=1.0)
self.process = None
raise RuntimeError("Dashboard failed to start within timeout")
port = _dashboard_port_value.value
log.info(f"Dashboard started on port {port}")
webbrowser.open(f"http://localhost:{port}/dashboard/index.html")
def stop(self, timeout: float = 1.0) -> None:
"""Signal shutdown and wait for the dashboard process to exit."""
if self.process is None:
return
log.info("Stopping dashboard process")
request_global_shutdown()
_shutdown_process(self.process, timeout=timeout)
self.process = None
class SerenaAgentWorker:
"""Worker process that hosts the actual SerenaAgent."""
class RequestMethod(StrEnum):
INITIALIZE = "initialize"
TOOL_CALL = "tool_call"
GET_ACTIVE_TOOL_NAMES = "get_active_tool_names"
IS_LANGUAGE_SERVER_RUNNING = "is_language_server_running"
RESET_LANGUAGE_SERVER = "reset_language_server"
GET_EXPOSED_TOOL_NAMES = "get_exposed_tool_names"
SHUTDOWN = "shutdown"
def __init__(self, conn: Connection):
self.conn = conn
self.agent: SerenaAgent | None = None
def run(self, log_queue: "multiprocessing.Queue[str]") -> None:
"""Main worker loop - runs in separate process."""
qh = QueueHandler(log_queue)
root = logging.getLogger()
root.setLevel(logging.DEBUG)
root.addHandler(qh)
log.info("SerenaAgent worker process started")
try:
while not global_shutdown_event.is_set():
try:
# Use polling to avoid blocking indefinitely
if self.conn.poll(timeout=0.5): # Poll every 500ms
try:
request = self.conn.recv()
if request is None: # Explicit shutdown signal
break
response = self._handle_request(request)
self.conn.send(response)
except EOFError:
# Connection closed - parent process terminated
log.info("Connection closed, worker shutting down")
break
except (BrokenPipeError, ConnectionResetError):
# Connection broken, can't communicate
log.info("Connection broken, worker shutting down")
break
except Exception as e:
log.error(f"Error processing request: {e}")
try:
response = {
"error": f"Worker process error: {e!s}",
"traceback": traceback.format_exc(),
}
self.conn.send(response)
except (EOFError, BrokenPipeError, ConnectionResetError):
# Connection is broken, can't send error response
log.info("Connection broken during error response, shutting down")
break
# Continue polling if no data available
except Exception as e:
log.error(f"Error in worker main loop: {e}")
break
except Exception as e:
log.error(f"Fatal error in worker process: {e}")
finally:
self._cleanup()
log.info("SerenaAgent worker process stopped")
os._exit(0) # Exit without raising any further exceptions
def _handle_request(self, request: dict[str, Any]) -> dict[str, Any]:
"""Handle a single request."""
try:
method = request["method"]
params = request.get("params", {})
match method:
case self.RequestMethod.INITIALIZE:
return self._initialize(params)
case self.RequestMethod.TOOL_CALL:
return self._tool_call(params)
case self.RequestMethod.GET_ACTIVE_TOOL_NAMES:
return self._get_active_tool_names()
case self.RequestMethod.IS_LANGUAGE_SERVER_RUNNING:
return self._is_language_server_running()
case self.RequestMethod.RESET_LANGUAGE_SERVER:
return self._reset_language_server()
case self.RequestMethod.GET_EXPOSED_TOOL_NAMES:
return self._get_exposed_tool_names()
case self.RequestMethod.SHUTDOWN:
return self.shutdown()
case _:
return {"error": f"Unknown method: {method}"}
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()}
def _initialize(self, params: dict[str, Any]) -> dict[str, Any]:
"""Initialize the SerenaAgent."""
if self.agent is not None:
return {"result": "SerenaAgent already initialized"}
try:
# Extract all possible initialization parameters
context_param = params.get("context")
project = params.get("project")
serena_config = SerenaConfig.from_json_dict(params["serena_config"])
context = SerenaAgentContext.from_json_dict(context_param) if context_param is not None else None
modes = [SerenaAgentMode.from_json_dict(m) for m in params["modes"]]
log_level = params.get("log_level")
trace_lsp_communication = params.get("trace_lsp_communication")
tool_timeout = params.get("tool_timeout")
self.agent = SerenaAgent(
project=project,
serena_config=serena_config,
context=context,
modes=modes,
enable_web_dashboard=False,
enable_gui_log_window=False,
log_level=log_level,
trace_lsp_communication=trace_lsp_communication,
tool_timeout=tool_timeout,
)
return {"result": "SerenaAgent initialized successfully"}
except Exception as e:
return {"error": f"Failed to initialize SerenaAgent: {e!s}", "traceback": traceback.format_exc()}
def _tool_call(self, params: dict[str, Any]) -> dict[str, Any]:
"""Execute a tool call."""
if self.agent is None:
return {"error": "SerenaAgent not initialized"}
try:
tool_name = params["tool_name"]
tool_params = params["tool_params"]
# Get the tool by name
tool = None
for tool_instance in self.agent._active_tools.values():
if tool_instance.get_name_from_cls() == tool_name:
tool = tool_instance
break
if tool is None:
return {"error": f"Tool '{tool_name}' not found or not active"}
# Execute the tool
result = tool.apply_ex(**tool_params)
return {"result": result}
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()}
def _get_active_tool_names(self) -> dict[str, Any]:
"""Get list of active tool names."""
if self.agent is None:
return {"error": "SerenaAgent not initialized"}
try:
tool_names = self.agent.get_active_tool_names()
return {"result": tool_names}
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()}
def _is_language_server_running(self) -> dict[str, Any]:
"""Check if language server is running."""
if self.agent is None:
return {"error": "SerenaAgent not initialized"}
try:
is_running = self.agent.is_language_server_running()
return {"result": is_running}
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()}
def _reset_language_server(self) -> dict[str, Any]:
"""Reset the language server."""
if self.agent is None:
return {"error": "SerenaAgent not initialized"}
try:
self.agent.reset_language_server()
return {"result": "Language server reset successfully"}
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()}
def _get_exposed_tool_names(self) -> dict[str, Any]:
"""Get exposed tool names for MCP tool creation."""
if self.agent is None:
return {"error": "SerenaAgent not initialized"}
try:
tool_instances = self.agent.get_exposed_tool_instances()
# Return only tool names - metadata will be reconstructed from ToolRegistry
tool_names = [tool.get_name_from_cls() for tool in tool_instances]
return {"result": tool_names}
except Exception as e:
return {"error": str(e), "traceback": traceback.format_exc()}
def shutdown(self) -> dict[str, Any]:
try:
log.info("Shutting down SerenaAgent worker process on request")
self._cleanup()
request_global_shutdown()
# Return successful response before exiting
return {"result": "Shutdown initiated"}
except Exception as e:
log.error(f"Error during shutdown: {e}")
return {"error": str(e), "traceback": traceback.format_exc()}
def _cleanup(self) -> None:
"""Clean up resources."""
if self.agent is not None:
try:
if self.agent.is_language_server_running() and self.agent.language_server is not None:
self.agent.language_server.stop()
except Exception as e:
log.error(f"Error stopping language server: {e}")
self.agent = None
class ProcessIsolatedSerenaAgent:
"""Process-isolated wrapper for SerenaAgent that prevents asyncio contamination."""
def __init__(
self,
project: str | None = None,
serena_config: SerenaConfigBase | None = None,
context: SerenaAgentContext | None = None,
modes: list[SerenaAgentMode] | None = None,
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None,
trace_lsp_communication: bool | None = None,
tool_timeout: float | None = None,
):
self.project = project
self.serena_config = serena_config or SerenaConfig.from_config_file()
self.context = context
self.modes = modes or []
self.log_level = log_level
self.trace_lsp_communication = trace_lsp_communication
self.tool_timeout = tool_timeout
self.process: multiprocessing.Process | None = None
self.conn: Connection | None = None
def start(self) -> None:
"""Start the worker process."""
if self.process is not None:
raise RuntimeError("ProcessIsolatedSerenaAgent already started")
log.info("Starting process-isolated SerenaAgent")
# Create communication pipe
parent_conn, child_conn = multiprocessing.Pipe()
self.conn = parent_conn # type: ignore
# Create and start worker process, passing along the dashboard's queue if available
worker = SerenaAgentWorker(child_conn) # type: ignore
self.process = multiprocessing.Process(target=worker.run, args=[_global_log_queue])
self.process.start()
# Prepare initialization parameters, converting complex objects to dict if present
init_params = {
"project": self.project,
"serena_config": self.serena_config.to_json_dict(),
"context": self.context.to_json_dict() if self.context is not None else None,
"modes": [m.to_json_dict() for m in self.modes],
"log_level": self.log_level,
"trace_lsp_communication": self.trace_lsp_communication,
"tool_timeout": self.tool_timeout,
}
# Initialize the agent in the worker process
try:
self._make_request_with_result(SerenaAgentWorker.RequestMethod.INITIALIZE, init_params)
except Exception as e:
self.stop()
raise RuntimeError(f"Failed to initialize SerenaAgent: {e}") from e
log.info("Process-isolated SerenaAgent started successfully")
def stop(self) -> None:
"""Stop the worker process."""
if self.process is None:
return
log.info("Stopping SerenaAgent process")
try:
# Close connection to signal worker to shutdown
if self.conn is not None:
self.conn.close()
_shutdown_process(self.process, timeout=2.0)
self.process = None
except KeyboardInterrupt:
log.warning("Keyboard interrupt during shutdown - forcing termination")
if self.process and self.process.is_alive():
self.process.kill()
self.process.join(timeout=1.0)
except Exception as e:
log.error(f"Error stopping worker process: {e}")
finally:
self.process = None
self.conn = None
log.info("SerenaAgent stopped")
def _make_request(self, method: SerenaAgentWorker.RequestMethod, params: dict[str, Any] | None = None) -> dict[str, Any]:
"""Make a request to the worker process."""
if self.process is None or not self.process.is_alive():
raise RuntimeError("Worker process is not running")
if self.conn is None:
raise RuntimeError("Connection is not initialized")
request = {"method": method, "params": params or {}}
# Send request
try:
self.conn.send(request)
except (EOFError, BrokenPipeError) as e:
raise RuntimeError("Failed to send request: worker process may have crashed") from e
# Wait for response with timeout
timeout = self.serena_config.tool_timeout
if self.conn.poll(timeout):
try:
return self.conn.recv()
except (EOFError, BrokenPipeError) as e:
raise RuntimeError("Failed to receive response: worker process may have crashed") from e
else:
raise TimeoutError(f"Request {method} timed out after {timeout} seconds")
def _make_request_with_result(self, method: SerenaAgentWorker.RequestMethod, params: dict[str, Any] | None = None) -> Any:
"""Make a request and return the result, raising an exception if there's an error."""
response = self._make_request(method, params)
if "error" in response:
raise RuntimeError(f"Request {method} failed: {response['error']}")
return response["result"]
def tool_call(self, tool_name: str, **tool_params: Any) -> str:
"""Call a tool in the worker process."""
return self._make_request_with_result(
SerenaAgentWorker.RequestMethod.TOOL_CALL, {"tool_name": tool_name, "tool_params": tool_params}
)
def get_tool(self, tool_cls: type[Tool]) -> "ProcessIsolatedTool":
"""Get a process-isolated tool that delegates to this agent."""
tool_name = tool_cls.get_name_from_cls()
return ProcessIsolatedTool(self, tool_name)
def get_active_tool_names(self) -> list[str]:
"""Get list of active tool names."""
return self._make_request_with_result(SerenaAgentWorker.RequestMethod.GET_ACTIVE_TOOL_NAMES)
def is_language_server_running(self) -> bool:
"""Check if language server is running."""
return self._make_request_with_result(SerenaAgentWorker.RequestMethod.IS_LANGUAGE_SERVER_RUNNING)
def reset_language_server(self) -> None:
"""Reset the language server."""
self._make_request_with_result(SerenaAgentWorker.RequestMethod.RESET_LANGUAGE_SERVER)
def shutdown_from_dashboard(self) -> None:
"""Request shutdown from dashboard."""
self._make_request_with_result(SerenaAgentWorker.RequestMethod.SHUTDOWN)
def get_exposed_tool_names(self) -> list[str]:
"""Get tool names for MCP tool creation."""
return self._make_request_with_result(SerenaAgentWorker.RequestMethod.GET_EXPOSED_TOOL_NAMES)
def __enter__(self) -> Self:
self.start()
return self
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
self.stop()
class ProcessIsolatedTool(ToolInterface):
"""A clean tool wrapper that delegates to ProcessIsolatedSerenaAgent."""
def __init__(self, process_agent: ProcessIsolatedSerenaAgent, tool_name: str):
self.process_agent = process_agent
self._tool_name = tool_name
@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
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.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}"
raise
test/serena/test_make_tool_process_isolation.py
-162
View File
@@ -1,162 +0,0 @@
"""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 SerenaMCPFactory
from serena.process_isolated_agent import ProcessIsolatedSerenaAgent, ProcessIsolatedTool
from test.serena.test_serena_agent import SerenaConfigForTests
make_tool = SerenaMCPFactory.make_mcp_tool
@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(serena_config=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_exposed_tool_names())
# 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}"
test/serena/test_serena_agent.py
+3 -36
View File
@@ -7,7 +7,6 @@ import pytest
import test.solidlsp.clojure as clj
from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, Project, ProjectConfig, SerenaAgent, SerenaConfigBase
from serena.process_isolated_agent import ProcessIsolatedSerenaAgent
from solidlsp.ls_config import Language
from test.conftest import get_repo_path
@@ -70,31 +69,10 @@ def serena_agent(request: pytest.FixtureRequest, serena_config):
language = Language(request.param)
project_name = f"test_repo_{language}"
# Check if this test should use process isolation by looking at the test parameters
isolated_process = False
if hasattr(request, "node") and hasattr(request.node, "callspec"):
# Get the isolated_process parameter value from the test
params = request.node.callspec.params
isolated_process = params.get("isolated_process", False)
if isolated_process:
agent = ProcessIsolatedSerenaAgent(project=project_name, serena_config=serena_config)
agent.start()
# Add cleanup to stop the process
def cleanup():
agent.stop()
request.addfinalizer(cleanup)
return agent
else:
return SerenaAgent(project=project_name, serena_config=serena_config)
return SerenaAgent(project=project_name, serena_config=serena_config)
class TestSerenaAgent:
@pytest.mark.parametrize(
"isolated_process", [pytest.param(False, id="direct"), pytest.param(True, id="isolated", marks=pytest.mark.isolated_process)]
)
@pytest.mark.parametrize(
"serena_agent,symbol_name,expected_kind,expected_file",
[
@@ -115,7 +93,7 @@ class TestSerenaAgent:
],
indirect=["serena_agent"],
)
def test_find_symbol(self, serena_agent, symbol_name: str, expected_kind: str, expected_file: str, isolated_process: bool):
def test_find_symbol(self, serena_agent, symbol_name: str, expected_kind: str, expected_file: str):
agent = serena_agent
find_symbol_tool = agent.get_tool(FindSymbolTool)
result = find_symbol_tool.apply_ex(name_path=symbol_name)
@@ -126,9 +104,6 @@ class TestSerenaAgent:
for s in symbols
), f"Expected to find {symbol_name} ({expected_kind}) in {expected_file}"
@pytest.mark.parametrize(
"isolated_process", [pytest.param(False, id="direct"), pytest.param(True, id="isolated", marks=pytest.mark.isolated_process)]
)
@pytest.mark.parametrize(
"serena_agent,symbol_name,def_file,ref_file",
[
@@ -161,7 +136,7 @@ class TestSerenaAgent:
],
indirect=["serena_agent"],
)
def test_find_symbol_references(self, serena_agent, symbol_name: str, def_file: str, ref_file: str, isolated_process: bool) -> None:
def test_find_symbol_references(self, serena_agent, symbol_name: str, def_file: str, ref_file: str) -> None:
agent = serena_agent
# Find the symbol location first
@@ -182,9 +157,6 @@ class TestSerenaAgent:
ref["relative_path"] == ref_file for ref in refs
), f"Expected to find reference to {symbol_name} in {ref_file}. refs={refs}"
@pytest.mark.parametrize(
"isolated_process", [pytest.param(False, id="direct"), pytest.param(True, id="isolated", marks=pytest.mark.isolated_process)]
)
@pytest.mark.parametrize(
"serena_agent,name_path,substring_matching,expected_symbol_name,expected_kind,expected_file",
[
@@ -259,7 +231,6 @@ class TestSerenaAgent:
expected_symbol_name: str,
expected_kind: str,
expected_file: str,
isolated_process: bool,
):
agent = serena_agent
@@ -282,9 +253,6 @@ class TestSerenaAgent:
for s in symbols
), f"Expected to find {name_path} ({expected_kind}) in {expected_file} for {agent._active_project.language.name}. Symbols: {symbols}"
@pytest.mark.parametrize(
"isolated_process", [pytest.param(False, id="direct"), pytest.param(True, id="isolated", marks=pytest.mark.isolated_process)]
)
@pytest.mark.parametrize(
"serena_agent,name_path",
[
@@ -307,7 +275,6 @@ class TestSerenaAgent:
self,
serena_agent,
name_path: str,
isolated_process: bool,
):
agent = serena_agent