diff --git a/example.env b/example.env index 17d3269cc..4a0a41277 100644 --- a/example.env +++ b/example.env @@ -189,6 +189,20 @@ ENCRYPTION_KEY="RQMpe38gK3m0szjpSmTNw_sP3Y54r6hDc6JewBoPKXc=" # This prevents excessively deep nesting in tool output # XAGENT_TOOL_MAX_RECURSION_DEPTH="20" +# =========================================== +# Process Isolation Configuration +# =========================================== +# Enable xoscar-backed process isolation for explicitly supported tools. +# Sandbox execution has priority when sandbox is configured. +XAGENT_PROCESS_ISOLATION_ENABLED="true" + +# xoscar main pool address for process isolation (default: localhost:12345) +# XAGENT_PROCESS_ISOLATION_ADDRESS="localhost:12345" + +# Maximum concurrent process-isolated executions. +# 0 means use os.cpu_count() as the concurrency limit. +# XAGENT_PROCESS_ISOLATION_WORKERS="0" + # =========================================== # External Database Connections (for SQL Query Tool) # =========================================== diff --git a/pyproject.toml b/pyproject.toml index 53e64a0f7..1f60902b1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,6 +65,8 @@ dependencies = [ "filelock>=3.0.0", "cloudpickle>=3.0.0", "exa-py>=2.0.0", + # Process isolation for tool execution + "xoscar>=0.9.0", # Data science libraries (core for RAG and analysis) "pandas>=1.3.0", # Special handling - AI-enhanced document processing @@ -231,6 +233,9 @@ module = [ "lark_oapi.*", "exa_py", "exa_py.*", + "xoscar", + "xoscar.*", + "xagent.core.sandbox_manager", ] ignore_missing_imports = true @@ -251,6 +256,8 @@ module = [ "xagent.core.model.tts.xinference", "xagent.web.channels.telegram.bot", "xagent.web.channels.telegram.handler", + "xagent.core.execution.actors.base_executor_actor", + "xagent.core.tools.adapters.vibe.process_isolated.integration", ] disallow_any_unimported = false warn_unused_ignores = false diff --git a/src/xagent/config.py b/src/xagent/config.py index ea8f8e059..2724a151f 100644 --- a/src/xagent/config.py +++ b/src/xagent/config.py @@ -43,6 +43,9 @@ SANDBOX_VOLUMES = "SANDBOX_VOLUMES" BOXLITE_HOME_DIR = "BOXLITE_HOME_DIR" WEB_SEARCH_PROVIDER = "XAGENT_WEB_SEARCH_PROVIDER" +PROCESS_ISOLATION_ENABLED = "XAGENT_PROCESS_ISOLATION_ENABLED" +PROCESS_ISOLATION_ADDRESS = "XAGENT_PROCESS_ISOLATION_ADDRESS" +PROCESS_ISOLATION_WORKERS = "XAGENT_PROCESS_ISOLATION_WORKERS" TOOL_MAX_OUTPUT_LENGTH = "XAGENT_TOOL_MAX_OUTPUT_LENGTH" TOOL_MAX_RECURSION_DEPTH = "XAGENT_TOOL_MAX_RECURSION_DEPTH" @@ -578,6 +581,48 @@ def get_web_search_provider() -> str: return "auto" +def get_process_isolation_enabled() -> bool: + """Get whether process-isolated tool execution is enabled. + + Priority: + 1. XAGENT_PROCESS_ISOLATION_ENABLED environment variable + 2. True + """ + value = os.getenv(PROCESS_ISOLATION_ENABLED) + if value is None: + return True + return value.strip().lower() not in {"", "0", "false", "no", "off"} + + +def get_process_isolation_address() -> str: + """Get the xoscar process service address.""" + return (os.getenv(PROCESS_ISOLATION_ADDRESS) or "localhost:12345").strip() + + +def get_process_isolation_workers() -> int: + """Get maximum concurrent process-isolated executions.""" + env_str = os.getenv(PROCESS_ISOLATION_WORKERS) + if env_str: + try: + workers = int(env_str) + except ValueError: + logger.warning( + "Invalid %s=%r; falling back to 0", + PROCESS_ISOLATION_WORKERS, + env_str, + ) + return 0 + if workers < 0: + logger.warning( + "%s=%r is negative; falling back to 0", + PROCESS_ISOLATION_WORKERS, + env_str, + ) + return 0 + return workers + return 0 + + def get_tool_max_recursion_depth() -> int: """Get the maximum recursion depth for tools. diff --git a/src/xagent/core/execution/__init__.py b/src/xagent/core/execution/__init__.py new file mode 100644 index 000000000..21ca5b1a7 --- /dev/null +++ b/src/xagent/core/execution/__init__.py @@ -0,0 +1,31 @@ +""" +Execution module for process-level isolation. + +Provides dynamic process execution - creates a new process for each execution. +""" + +from .service.base import ( + BaseService, + ExecutionResult, + IsolationType, + ServiceInfo, + ServiceStatus, +) +from .service.manager import ( + clear_process_service, + get_process_service, + set_process_service, +) +from .service.process import ProcessService + +__all__ = [ + "BaseService", + "ExecutionResult", + "IsolationType", + "ServiceInfo", + "ServiceStatus", + "ProcessService", + "get_process_service", + "set_process_service", + "clear_process_service", +] diff --git a/src/xagent/core/execution/actors/__init__.py b/src/xagent/core/execution/actors/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/xagent/core/execution/actors/base_executor_actor.py b/src/xagent/core/execution/actors/base_executor_actor.py new file mode 100644 index 000000000..61da4b722 --- /dev/null +++ b/src/xagent/core/execution/actors/base_executor_actor.py @@ -0,0 +1,92 @@ +""" +Base executor actor class. + +Provides common functionality for all executor actors using xoscar framework. +""" + +import inspect +import time +import traceback +from typing import Any + +import xoscar as xo + + +class BaseExecutorActor(xo.Actor): # type: ignore[misc] + """Base executor actor. + + Provides common execution functionality for Python, JavaScript, and command executors. + All executors inherit from this class to get consistent error handling and result formatting. + """ + + async def __post_create__(self) -> None: + """Called after actor creation. + + Can be overridden by subclasses for initialization. + """ + pass + + async def __pre_destroy__(self) -> None: + """Called before actor destruction. + + Can be overridden by subclasses for cleanup. + """ + pass + + def _format_result(self, result: dict[str, Any], execution_time: float) -> dict: + return { + "success": result.get("success", result.get("return_code", 0) == 0), + "output": result.get("output", ""), + "error": result.get("error", ""), + "return_code": result.get("return_code", 0), + "metadata": result.get("metadata", {}), + "execution_time": execution_time, + } + + def _format_exception(self, exc: Exception, execution_time: float) -> dict: + error_message = f"{type(exc).__name__}: {str(exc)}" + error_traceback = traceback.format_exc() + + return { + "success": False, + "output": "", + "error": error_message, + "return_code": -1, + "metadata": {"traceback": error_traceback}, + "execution_time": execution_time, + } + + def _execute_with_tracking(self, func: Any, *args: Any, **kwargs: Any) -> dict: + """Execute function with time tracking. + + Args: + func: Function to execute + *args: Positional arguments + **kwargs: Keyword arguments + + Returns: + Execution result dictionary + """ + start_time = time.time() + try: + result = func(*args, **kwargs) + execution_time = time.time() - start_time + return self._format_result(result, execution_time) + except Exception as e: + execution_time = time.time() - start_time + return self._format_exception(e, execution_time) + + async def _execute_async_with_tracking( + self, func: Any, *args: Any, **kwargs: Any + ) -> dict: + """Execute async-capable function with time tracking.""" + start_time = time.time() + try: + result = func(*args, **kwargs) + if inspect.isawaitable(result): + result = await result + execution_time = time.time() - start_time + return self._format_result(result, execution_time) + except Exception as e: + execution_time = time.time() - start_time + return self._format_exception(e, execution_time) diff --git a/src/xagent/core/execution/actors/command_executor_actor.py b/src/xagent/core/execution/actors/command_executor_actor.py new file mode 100644 index 000000000..6faf4d64b --- /dev/null +++ b/src/xagent/core/execution/actors/command_executor_actor.py @@ -0,0 +1,27 @@ +"""Shell command executor actor.""" + +import asyncio +from typing import Optional + +from ...tools.core.command_executor import CommandExecutorCore +from .base_executor_actor import BaseExecutorActor + + +class CommandExecutorActor(BaseExecutorActor): + """Execute shell commands in an isolated process.""" + + async def execute( + self, + command: str, + workspace: Optional[str] = None, + timeout: int = 300, + ) -> dict: + async def _execute() -> dict: + executor = CommandExecutorCore(working_directory=workspace) + return await asyncio.to_thread( + executor.execute_command, + command, + timeout=timeout, + ) + + return await self._execute_async_with_tracking(_execute) diff --git a/src/xagent/core/execution/actors/javascript_executor_actor.py b/src/xagent/core/execution/actors/javascript_executor_actor.py new file mode 100644 index 000000000..fd3d5a608 --- /dev/null +++ b/src/xagent/core/execution/actors/javascript_executor_actor.py @@ -0,0 +1,47 @@ +"""JavaScript executor actor.""" + +import asyncio +from typing import Optional + +from .base_executor_actor import BaseExecutorActor + + +class JavaScriptExecutorActor(BaseExecutorActor): + """Execute JavaScript code in an isolated process using Node.js.""" + + async def execute( + self, + code: str, + workspace: Optional[str] = None, + timeout: int = 300, + ) -> dict: + async def _execute() -> dict: + process = await asyncio.create_subprocess_exec( + "node", + "-e", + code, + cwd=workspace, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for( + process.communicate(), timeout=timeout + ) + except asyncio.TimeoutError: + process.kill() + await process.wait() + return { + "output": "", + "error": f"Execution timed out after {timeout} seconds", + "return_code": -1, + } + + return { + "output": stdout.decode(errors="replace"), + "error": stderr.decode(errors="replace"), + "return_code": process.returncode or 0, + "metadata": {}, + } + + return await self._execute_async_with_tracking(_execute) diff --git a/src/xagent/core/execution/actors/python_executor_actor.py b/src/xagent/core/execution/actors/python_executor_actor.py new file mode 100644 index 000000000..3b420f006 --- /dev/null +++ b/src/xagent/core/execution/actors/python_executor_actor.py @@ -0,0 +1,105 @@ +""" +Python executor actor. + +Executes Python code in isolated process using xoscar actor framework. +""" + +import contextlib +import io +import os +import traceback +from typing import Optional + +from .base_executor_actor import BaseExecutorActor + + +class PythonExecutorActor(BaseExecutorActor): + """Python executor actor. + + Executes Python code in an isolated process with configurable timeout. + """ + + async def execute( + self, + code: str, + workspace: Optional[str] = None, + timeout: int = 300, + ) -> dict: + """Execute Python code. + + Args: + code: Python code to execute + workspace: Working directory path (optional) + timeout: Execution timeout in seconds (default: 300) + + Returns: + Execution result dictionary with keys: + - success: bool + - output: str + - error: str + - return_code: int + - metadata: dict + - execution_time: float + """ + + def _execute() -> dict: + """Internal execution function.""" + # Change to workspace if specified + old_cwd = None + if workspace: + old_cwd = os.getcwd() + try: + os.chdir(workspace) + except FileNotFoundError: + return { + "output": "", + "error": f"Workspace directory not found: {workspace}", + "return_code": 1, + } + + try: + # Capture stdout and stderr + stdout_capture = io.StringIO() + stderr_capture = io.StringIO() + + with ( + contextlib.redirect_stdout(stdout_capture), + contextlib.redirect_stderr(stderr_capture), + ): + # Execute code + exec_globals = { + "__name__": "__main__", + "__builtins__": __builtins__, + } + exec(code, exec_globals) + + # Get output + stdout_value = stdout_capture.getvalue() + stderr_value = stderr_capture.getvalue() + + return { + "output": stdout_value, + "error": stderr_value, + "return_code": 0, + "metadata": {}, + } + + except Exception as e: + error_message = f"{type(e).__name__}: {str(e)}" + error_traceback = traceback.format_exc() + return { + "output": "", + "error": f"{error_message}\n{error_traceback}", + "return_code": 1, + "metadata": {"exception_type": type(e).__name__}, + } + + finally: + # Restore working directory + if old_cwd is not None: + try: + os.chdir(old_cwd) + except Exception: + pass + + return self._execute_with_tracking(_execute) diff --git a/src/xagent/core/execution/actors/tool_executor_actor.py b/src/xagent/core/execution/actors/tool_executor_actor.py new file mode 100644 index 000000000..7f572bd63 --- /dev/null +++ b/src/xagent/core/execution/actors/tool_executor_actor.py @@ -0,0 +1,63 @@ +""" +Generic tool executor actor. + +Executes any tool in isolated process by receiving the tool instance +and calling its run_json_async method. xoscar handles serialization automatically. +""" + +import traceback +from typing import Any + +from .base_executor_actor import BaseExecutorActor + + +class ToolExecutorActor(BaseExecutorActor): + """Generic tool executor actor. + + Executes any AbstractBaseTool in isolated process. + xoscar automatically serializes the tool instance and arguments. + """ + + async def execute( + self, + tool: Any, # AbstractBaseTool instance - xoscar serializes it + args: dict, # Tool arguments - xoscar serializes it + timeout: int = 300, + ) -> dict: + """Execute tool in isolated process. + + Args: + tool: Tool instance serialized by xoscar + args: Tool arguments (serialized and sent by xoscar) + timeout: Execution timeout in seconds + + Returns: + Execution result dictionary + """ + + async def _execute() -> dict: + """Internal execution function.""" + try: + if hasattr(tool, "run_json_async"): + result = await tool.run_json_async(args) + else: + result = tool.run_json_sync(args) + + return { + "output": result, + "error": "", + "return_code": 0, + "metadata": {}, + } + + except Exception as e: + error_message = f"{type(e).__name__}: {str(e)}" + error_traceback = traceback.format_exc() + return { + "output": None, + "error": f"{error_message}\n{error_traceback}", + "return_code": 1, + "metadata": {"exception_type": type(e).__name__}, + } + + return await self._execute_async_with_tracking(_execute) diff --git a/src/xagent/core/execution/service/__init__.py b/src/xagent/core/execution/service/__init__.py new file mode 100644 index 000000000..dd7be84c6 --- /dev/null +++ b/src/xagent/core/execution/service/__init__.py @@ -0,0 +1,31 @@ +""" +Service layer for execution management. + +Provides lifecycle management for execution services including process isolation. +""" + +from .base import ( + BaseService, + ExecutionResult, + IsolationType, + ServiceInfo, + ServiceStatus, +) +from .manager import ( + clear_process_service, + get_process_service, + set_process_service, +) +from .process import ProcessService + +__all__ = [ + "BaseService", + "ExecutionResult", + "IsolationType", + "ServiceInfo", + "ServiceStatus", + "ProcessService", + "get_process_service", + "set_process_service", + "clear_process_service", +] diff --git a/src/xagent/core/execution/service/base.py b/src/xagent/core/execution/service/base.py new file mode 100644 index 000000000..1f018b8f3 --- /dev/null +++ b/src/xagent/core/execution/service/base.py @@ -0,0 +1,185 @@ +""" +Base interfaces and data classes for execution services. + +Defines the unified abstraction for execution services with support for +multiple isolation strategies. +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Optional + + +class IsolationType(Enum): + """Isolation type enumeration. + + Defines the level of isolation for code execution. + """ + + PROCESS = "process" # Process-level isolation (using xoscar) + SANDBOX = "sandbox" # Sandbox isolation (future work) + + +class ServiceStatus(Enum): + """Service status enumeration. + + Tracks the lifecycle state of a service. + """ + + STARTING = "starting" + RUNNING = "running" + STOPPING = "stopping" + STOPPED = "stopped" + ERROR = "error" + + +@dataclass +class ExecutionResult: + """Execution result dataclass. + + Unified result format for all execution types (Python, JavaScript, command). + """ + + success: bool + output: str + error: str = "" + return_code: int = 0 + metadata: dict[str, Any] = field(default_factory=dict) + execution_time: Optional[float] = None + memory_used_mb: Optional[float] = None + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary format. + + Returns: + Dictionary representation of the execution result + """ + return { + "success": self.success, + "output": self.output, + "error": self.error, + "return_code": self.return_code, + "metadata": self.metadata, + "execution_time": self.execution_time, + "memory_used_mb": self.memory_used_mb, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "ExecutionResult": + """Create from dictionary format. + + Args: + data: Dictionary representation + + Returns: + ExecutionResult instance + """ + return cls( + success=data.get("success", False), + output=data.get("output", ""), + error=data.get("error", ""), + return_code=data.get("return_code", 0), + metadata=data.get("metadata", {}), + execution_time=data.get("execution_time"), + memory_used_mb=data.get("memory_used_mb"), + ) + + +@dataclass +class ServiceInfo: + """Service information dataclass. + + Contains status, resource information, and metrics for a service. + """ + + name: str + status: ServiceStatus + resource_info: dict[str, Any] = field(default_factory=dict) + metrics: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary format. + + Returns: + Dictionary representation of the service info + """ + return { + "name": self.name, + "status": self.status.value, + "resource_info": self.resource_info, + "metrics": self.metrics, + } + + +class BaseService(ABC): + """Base service abstract class. + + Defines the unified lifecycle management interface for all services. + All services (ProcessService, SandboxService, etc.) must inherit from this. + """ + + def __init__(self) -> None: + """Initialize base service.""" + self._status = ServiceStatus.STOPPED + + @property + @abstractmethod + def service_name(self) -> str: + """Service name. + + Returns: + Unique service name identifier + """ + pass + + @property + def status(self) -> ServiceStatus: + """Get service status. + + Returns: + Current service status + """ + return self._status + + @abstractmethod + async def start(self) -> None: + """Start service. + + Initialize resources, establish connections, etc. + """ + pass + + @abstractmethod + async def stop(self) -> None: + """Stop service. + + Release resources, close connections, etc. + """ + pass + + @abstractmethod + async def health_check(self) -> bool: + """Health check. + + Returns: + True if service is healthy, False otherwise + """ + pass + + @abstractmethod + def get_info(self) -> ServiceInfo: + """Get service information. + + Returns: + ServiceInfo containing status, resource info, and metrics + """ + pass + + async def restart(self) -> None: + """Restart service. + + Stop and then start the service. + """ + await self.stop() + await self.start() diff --git a/src/xagent/core/execution/service/manager.py b/src/xagent/core/execution/service/manager.py new file mode 100644 index 000000000..87c2cb18b --- /dev/null +++ b/src/xagent/core/execution/service/manager.py @@ -0,0 +1,41 @@ +""" +Service manager for execution services. + +Provides global access to execution services from anywhere in the application. +""" + +import logging +from typing import Optional + +from .process import ProcessService + +logger = logging.getLogger(__name__) + + +# Global service instance +_process_service: Optional[ProcessService] = None + + +def get_process_service() -> Optional[ProcessService]: + """Get the global ProcessService instance. + + Returns: + ProcessService instance or None if not initialized + """ + return _process_service + + +def set_process_service(service: ProcessService) -> None: + """Set the global ProcessService instance. + + Args: + service: ProcessService instance to set as global + """ + global _process_service + _process_service = service + + +def clear_process_service() -> None: + """Clear the global ProcessService instance.""" + global _process_service + _process_service = None diff --git a/src/xagent/core/execution/service/process.py b/src/xagent/core/execution/service/process.py new file mode 100644 index 000000000..43193720e --- /dev/null +++ b/src/xagent/core/execution/service/process.py @@ -0,0 +1,309 @@ +""" +Process service using xoscar with sub-pool creation. + +Creates a main pool at startup, then creates sub-pools (with one worker) for +each execution request using append_sub_pool, destroys them after completion. +Sub-pools are created on-demand and destroyed when done. +""" + +import asyncio +import logging +import os +import sys +import uuid +from typing import Any, Optional + +import xoscar as xo + +from .base import ( + BaseService, + ExecutionResult, + ServiceInfo, + ServiceStatus, +) + +logger = logging.getLogger(__name__) + + +class ProcessService(BaseService): + """Process service using xoscar with sub-pool creation. + + Creates a main pool at startup. + For each execution request, appends a sub-pool with one worker. + After execution, the sub-pool is killed. + """ + + def __init__(self, address: str = "localhost:12345", n_workers: int = 0): + super().__init__() + self._address = address + self._n_workers = n_workers + self._max_concurrency = n_workers if n_workers > 0 else (os.cpu_count() or 1) + self._semaphore = asyncio.Semaphore(self._max_concurrency) + self._lock = asyncio.Lock() + self._active_actors: dict[str, Any] = {} + self._pool: Any = None + + @property + def service_name(self) -> str: + return "process" + + async def start(self) -> None: + """Start process service.""" + self._status = ServiceStatus.STARTING + try: + # Initialize xoscar router + from xoscar.backends import router as xo_router + + default_router = xo_router.Router.get_instance_or_empty() + xo_router.Router.set_instance(default_router) + logger.info("xoscar router initialized") + + # Create main pool that will hold sub-pools + self._pool = await xo.create_actor_pool( + address=self._address, + n_process=0, # Main pool doesn't need workers, only sub-pools + ) + + self._status = ServiceStatus.RUNNING + logger.info( + f"ProcessService started successfully with main pool at {self._address}" + ) + except Exception as e: + self._status = ServiceStatus.ERROR + logger.error(f"Failed to start ProcessService: {e}", exc_info=True) + raise + + async def stop(self) -> None: + """Stop process service.""" + self._status = ServiceStatus.STOPPING + try: + logger.info("Stopping ProcessService") + + # Stop main pool (this will also stop all sub-pools) + if self._pool: + await self._pool.stop() + self._pool = None + logger.info("Main pool stopped") + + self._status = ServiceStatus.STOPPED + logger.info("ProcessService stopped successfully") + except Exception as e: + self._status = ServiceStatus.ERROR + logger.error(f"Failed to stop ProcessService: {e}", exc_info=True) + raise + + async def health_check(self) -> bool: + """Health check.""" + return self._status == ServiceStatus.RUNNING + + def get_info(self) -> ServiceInfo: + """Get service information.""" + return ServiceInfo( + name=self.service_name, + status=self._status, + resource_info={ + "type": "dynamic", + "address": self._address, + "n_workers": self._n_workers, + "max_concurrency": self._max_concurrency, + "active_actors": len(self._active_actors), + }, + metrics={}, + ) + + def _ensure_running(self) -> None: + if self._status != ServiceStatus.RUNNING or self._pool is None: + raise RuntimeError("ProcessService not started") + + def _sub_pool_env(self) -> dict[str, str]: + """Build environment for child pools so xoscar can import caller modules.""" + cwd = os.getcwd() + path_entries = [] + for path in sys.path: + abs_path = os.path.abspath(path or cwd) + if abs_path == cwd or abs_path.startswith(cwd + os.sep): + path_entries.append(abs_path) + + env = {} + existing_pythonpath = os.environ.get("PYTHONPATH") + if existing_pythonpath: + path_entries.extend(existing_pythonpath.split(os.pathsep)) + env["PYTHONPATH"] = os.pathsep.join(dict.fromkeys(path_entries)) + return env + + async def _execute_in_actor( + self, + task_id: str, + actor_cls: Any, + timeout: int, + **execute_kwargs: Any, + ) -> ExecutionResult: + """Execute an actor call in a temporary sub-pool.""" + self._ensure_running() + + async with self._semaphore: + return await self._execute_in_sub_pool( + task_id, + actor_cls, + timeout, + **execute_kwargs, + ) + + async def _execute_in_sub_pool( + self, + task_id: str, + actor_cls: Any, + timeout: int, + **execute_kwargs: Any, + ) -> ExecutionResult: + sub_pool_address = None + actor_ref = None + timed_out = False + startup_timeout = min(float(timeout), 30.0) + timeout_stage = "sub-pool startup" + timeout_seconds = startup_timeout + + try: + sub_pool_address = await asyncio.wait_for( + self._pool.append_sub_pool( + label=task_id, + env=self._sub_pool_env(), + ), + timeout=startup_timeout, + ) + logger.debug(f"Appended sub-pool {sub_pool_address} for {task_id}") + + timeout_stage = "actor creation" + actor_ref = await asyncio.wait_for( + xo.create_actor(actor_cls, address=sub_pool_address), + timeout=startup_timeout, + ) + async with self._lock: + self._active_actors[task_id] = actor_ref + + logger.debug(f"Created actor {task_id}") + actor = await xo.actor_ref(actor_ref) + + timeout_stage = "actor execution" + timeout_seconds = float(timeout) + result_dict = await asyncio.wait_for( + actor.execute(**execute_kwargs, timeout=timeout), + timeout=timeout, + ) + return ExecutionResult.from_dict(result_dict) + + except asyncio.TimeoutError: + timed_out = True + return ExecutionResult( + success=False, + output="", + error=( + f"Execution timed out during {timeout_stage} " + f"after {timeout_seconds:g} seconds" + ), + return_code=-1, + ) + except Exception as e: + return ExecutionResult( + success=False, + output="", + error=f"Execution failed: {str(e)}", + return_code=-1, + ) + finally: + if actor_ref is not None: + if not timed_out: + try: + await xo.destroy_actor(actor_ref) + logger.debug(f"Destroyed actor {task_id}") + except Exception as e: + logger.error(f"Failed to destroy actor {task_id}: {e}") + async with self._lock: + self._active_actors.pop(task_id, None) + + if sub_pool_address and self._pool is not None: + try: + await self._pool.remove_sub_pool(sub_pool_address, force=timed_out) + logger.debug(f"Removed sub-pool {sub_pool_address}") + except Exception as e: + logger.error(f"Failed to remove sub-pool {sub_pool_address}: {e}") + + async def execute_python( + self, + code: str, + workspace: Optional[str] = None, + timeout: int = 300, + ) -> ExecutionResult: + """Execute Python code in a dynamic actor. + + Creates a sub-pool, creates an actor, executes the code, + then destroys both the actor and sub-pool. + """ + task_id = f"python_{uuid.uuid4().hex[:8]}" + from ..actors.python_executor_actor import PythonExecutorActor + + return await self._execute_in_actor( + task_id, + PythonExecutorActor, + timeout, + code=code, + workspace=workspace, + ) + + async def execute_tool( + self, + tool: Any, + args: dict, + timeout: int = 300, + ) -> ExecutionResult: + """Execute any tool in a dynamic actor. + + Creates a sub-pool, creates an actor, executes the tool, + then destroys both the actor and sub-pool. + """ + task_id = f"tool_{uuid.uuid4().hex[:8]}" + from ..actors.tool_executor_actor import ToolExecutorActor + + return await self._execute_in_actor( + task_id, + ToolExecutorActor, + timeout, + tool=tool, + args=args, + ) + + async def execute_command( + self, + command: str, + workspace: Optional[str] = None, + timeout: int = 300, + ) -> ExecutionResult: + """Execute a shell command in a dynamic actor.""" + from ..actors.command_executor_actor import CommandExecutorActor + + task_id = f"command_{uuid.uuid4().hex[:8]}" + return await self._execute_in_actor( + task_id, + CommandExecutorActor, + timeout, + command=command, + workspace=workspace, + ) + + async def execute_javascript( + self, + code: str, + workspace: Optional[str] = None, + timeout: int = 300, + ) -> ExecutionResult: + """Execute JavaScript code in a dynamic actor.""" + from ..actors.javascript_executor_actor import JavaScriptExecutorActor + + task_id = f"javascript_{uuid.uuid4().hex[:8]}" + return await self._execute_in_actor( + task_id, + JavaScriptExecutorActor, + timeout, + code=code, + workspace=workspace, + ) diff --git a/src/xagent/core/tools/adapters/vibe/factory.py b/src/xagent/core/tools/adapters/vibe/factory.py index b24673891..d12c5aebc 100644 --- a/src/xagent/core/tools/adapters/vibe/factory.py +++ b/src/xagent/core/tools/adapters/vibe/factory.py @@ -221,6 +221,10 @@ async def create_all_tools( await create_workspace_in_sandbox(sandbox, workspace) tools = await ToolFactory._wrap_sandbox_tools(tools, sandbox) + else: + from .process_isolated import wrap_tools + + tools = wrap_tools(tools) # Apply output filtering to all tools tools = ToolFactory._apply_output_filters(tools, config) diff --git a/src/xagent/core/tools/adapters/vibe/output_filter_wrapper.py b/src/xagent/core/tools/adapters/vibe/output_filter_wrapper.py index 75fccecd4..081edbcc4 100644 --- a/src/xagent/core/tools/adapters/vibe/output_filter_wrapper.py +++ b/src/xagent/core/tools/adapters/vibe/output_filter_wrapper.py @@ -52,6 +52,14 @@ def __init__( def is_sandboxed(self) -> bool: return getattr(self._target, "is_sandboxed", False) + @property + def is_isolated(self) -> bool: + return getattr(self._target, "is_isolated", False) + + @property + def supports_process_isolation(self) -> bool: + return getattr(self._target, "supports_process_isolation", False) + @property def name(self) -> str: return self._target.name diff --git a/src/xagent/core/tools/adapters/vibe/process_isolated/__init__.py b/src/xagent/core/tools/adapters/vibe/process_isolated/__init__.py new file mode 100644 index 000000000..0d73b792c --- /dev/null +++ b/src/xagent/core/tools/adapters/vibe/process_isolated/__init__.py @@ -0,0 +1,26 @@ +""" +Process-isolated tool wrapper. + +Provides process-level isolation for tools using xoscar. +xoscar automatically handles serialization - no manual encoding needed. +""" + +from .integration import ( + maybe_wrap_tool, + should_use_process_isolation, + supports_process_isolation, + wrap_tools, +) +from .process_isolated_tool_wrapper import ( + ProcessIsolatedToolWrapper, + create_process_isolated_tool, +) + +__all__ = [ + "ProcessIsolatedToolWrapper", + "create_process_isolated_tool", + "supports_process_isolation", + "should_use_process_isolation", + "maybe_wrap_tool", + "wrap_tools", +] diff --git a/src/xagent/core/tools/adapters/vibe/process_isolated/integration.py b/src/xagent/core/tools/adapters/vibe/process_isolated/integration.py new file mode 100644 index 000000000..430e988e1 --- /dev/null +++ b/src/xagent/core/tools/adapters/vibe/process_isolated/integration.py @@ -0,0 +1,118 @@ +""" +Integration utilities for process-isolated tools. + +Provides helper functions to automatically wrap tools with process isolation +when ProcessService is available. +""" + +import logging +from typing import Any + +from ..base import AbstractBaseTool +from .process_isolated_tool_wrapper import ( + create_process_isolated_tool, +) + +logger = logging.getLogger(__name__) + + +def supports_process_isolation(tool: AbstractBaseTool) -> bool: + """Return whether a tool explicitly supports process-isolated execution.""" + return bool(getattr(tool, "supports_process_isolation", False)) + + +def should_use_process_isolation( + tool_name: str, + tool: AbstractBaseTool | None = None, +) -> bool: + """Check if a tool should use process isolation. + + Args: + tool_name: Name of the tool + + Returns: + True if process isolation should be used + """ + from .....execution.service.manager import get_process_service + + try: + from xagent.web.sandbox_manager import ( + get_sandbox_manager, # type: ignore[import] + ) + except ImportError: + # If web module is not available, define a stub + def get_sandbox_manager() -> Any: # type: ignore[no-redef,misc] + return None + + # Check if ProcessService is available + process_service = get_process_service() + if not process_service: + return False + + if tool is not None and not supports_process_isolation(tool): + return False + + # Check if sandbox is enabled (sandbox has higher priority) + sandbox_mgr = get_sandbox_manager() + if sandbox_mgr: + # Sandbox is enabled, don't use process isolation + return False + + # Process isolation is available and sandbox is not enabled + return True + + +def maybe_wrap_tool( + tool: AbstractBaseTool, + timeout: int = 300, +) -> AbstractBaseTool: + """Wrap tool with process isolation if appropriate. + + Args: + tool: Tool to potentially wrap + timeout: Execution timeout in seconds + + Returns: + Original tool or process-isolated wrapper + """ + # Check if we should use process isolation + if not should_use_process_isolation(tool.name, tool): + return tool + + try: + # Wrap tool with process isolation + wrapped_tool = create_process_isolated_tool( + tool=tool, + timeout=timeout, + ) + logger.info(f"Tool '{tool.name}' wrapped with process isolation") + return wrapped_tool + + except Exception as e: + logger.error( + f"Failed to wrap tool '{tool.name}' with process isolation: {e}", + exc_info=True, + ) + # Fall back to original tool + return tool + + +def wrap_tools( + tools: list[AbstractBaseTool], + timeout: int = 300, +) -> list[AbstractBaseTool]: + """Wrap multiple tools with process isolation if appropriate. + + Args: + tools: List of tools to potentially wrap + timeout: Execution timeout in seconds + + Returns: + List of tools (some may be wrapped) + """ + wrapped_tools = [] + for tool in tools: + wrapped_tool = maybe_wrap_tool(tool, timeout=timeout) + wrapped_tools.append(wrapped_tool) + + return wrapped_tools diff --git a/src/xagent/core/tools/adapters/vibe/process_isolated/process_isolated_tool_wrapper.py b/src/xagent/core/tools/adapters/vibe/process_isolated/process_isolated_tool_wrapper.py new file mode 100644 index 000000000..c8a3a4253 --- /dev/null +++ b/src/xagent/core/tools/adapters/vibe/process_isolated/process_isolated_tool_wrapper.py @@ -0,0 +1,197 @@ +""" +Process-isolated tool wrapper. + +Execute tool's run_json_sync/async methods in isolated processes using xoscar. +xoscar automatically handles serialization - no need for manual pickle/json encoding. +""" + +import asyncio +import logging +import threading +from typing import TYPE_CHECKING, Any, Mapping, Optional, Type + +from pydantic import BaseModel + +from .....execution.service.manager import get_process_service +from ..base import AbstractBaseTool, ToolMetadata + +if TYPE_CHECKING: + from ..base import ToolCategory + +logger = logging.getLogger(__name__) + + +class ProcessIsolatedToolWrapper(AbstractBaseTool): + """Process-isolated tool wrapper. + + Wrap any AbstractBaseTool to execute in isolated processes using xoscar. + xoscar automatically serializes the tool instance and arguments. + """ + + def __init__( + self, + target_tool: AbstractBaseTool, + timeout: int = 300, + ): + """Initialize process-isolated tool wrapper. + + Args: + target_tool: Target tool to wrap + timeout: Execution timeout in seconds (default: 300) + """ + self._target = target_tool + self._timeout = timeout + + # Proxy target tool attributes + self._visibility = getattr(target_tool, "_visibility", None) + self._allow_users = getattr(target_tool, "_allow_users", None) + + @property + def is_isolated(self) -> bool: + """Marker for process-isolated.""" + return True + + @property + def supports_process_isolation(self) -> bool: + """Marker for process-isolation-capable tools.""" + return True + + @property + def is_sandboxed(self) -> bool: + return getattr(self._target, "is_sandboxed", False) + + @property + def name(self) -> str: + return self._target.name + + @property + def description(self) -> str: + return self._target.description + + @property + def tags(self) -> list[str]: + return self._target.tags + + @property + def category(self) -> "ToolCategory": + return getattr(self._target, "category", None) # type: ignore[return-value] + + @property + def metadata(self) -> ToolMetadata: + return self._target.metadata + + def args_type(self) -> Type[BaseModel]: + return self._target.args_type() + + def return_type(self) -> Type[BaseModel]: + return self._target.return_type() + + def state_type(self) -> Optional[Type[BaseModel]]: + return self._target.state_type() + + def is_async(self) -> bool: + return self._target.is_async() + + def return_value_as_string(self, value: Any) -> str: + return self._target.return_value_as_string(value) + + def run_json_sync(self, args: Mapping[str, Any]) -> Any: + """Synchronous execution.""" + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(self.run_json_async(args)) + + result: dict[str, Any] = {} + + def _run_in_thread() -> None: + try: + result["value"] = asyncio.run(self.run_json_async(args)) + except BaseException as exc: + result["error"] = exc + + thread = threading.Thread(target=_run_in_thread, daemon=True) + thread.start() + thread.join() + + if "error" in result: + raise result["error"] + return result.get("value") + + async def run_json_async(self, args: Mapping[str, Any]) -> Any: + """Execute tool asynchronously in isolated process. + + Args: + args: Tool arguments + + Returns: + Tool execution result + """ + process_service = get_process_service() + if not process_service: + # ProcessService not available, fall back to direct execution + logger.warning( + f"ProcessService not available for {self._target.name}, " + "falling back to direct execution" + ) + return await self._target.run_json_async(args) + + try: + # xoscar automatically serializes tool and args + result = await process_service.execute_tool( + tool=self._target, # Tool instance - xoscar serializes it + args=dict(args), # Arguments - xoscar serializes them + timeout=self._timeout, + ) + + if not result.success: + raise RuntimeError( + f"Process-isolated execution failed for {self._target.name}: " + f"{result.error}" + ) + + return result.output + + except Exception as e: + logger.error( + f"Error executing tool {self._target.name} in isolated process: {e}", + exc_info=True, + ) + raise + + async def save_state_json(self) -> Mapping[str, Any]: + """Save state (delegates to target tool).""" + return await self._target.save_state_json() + + async def load_state_json(self, state: Mapping[str, Any]) -> None: + """Load state (delegates to target tool).""" + await self._target.load_state_json(state) + + async def setup(self, task_id: Optional[str] = None) -> None: + """Setup tool (delegates to target tool).""" + await self._target.setup(task_id) + + async def teardown(self, task_id: Optional[str] = None) -> None: + """Teardown tool (delegates to target tool).""" + await self._target.teardown(task_id) + + +def create_process_isolated_tool( + tool: AbstractBaseTool, + timeout: int = 300, +) -> ProcessIsolatedToolWrapper: + """Create process-isolated tool instance. + + Args: + tool: Tool to wrap + timeout: Execution timeout in seconds + + Returns: + Process-isolated tool wrapper + """ + wrapper = ProcessIsolatedToolWrapper( + target_tool=tool, + timeout=timeout, + ) + + return wrapper diff --git a/src/xagent/core/tools/adapters/vibe/python_executor.py b/src/xagent/core/tools/adapters/vibe/python_executor.py index f3bdc8e73..393488249 100644 --- a/src/xagent/core/tools/adapters/vibe/python_executor.py +++ b/src/xagent/core/tools/adapters/vibe/python_executor.py @@ -40,6 +40,8 @@ class PythonExecutorResult(BaseModel): class PythonExecutorTool(AbstractBaseTool): """Framework wrapper for the pure Python executor tool""" + supports_process_isolation = True + def __init__(self, workspace: Optional[TaskWorkspace] = None) -> None: self._visibility = ToolVisibility.PUBLIC self._workspace = workspace diff --git a/src/xagent/web/api/services.py b/src/xagent/web/api/services.py new file mode 100644 index 000000000..e040d842c --- /dev/null +++ b/src/xagent/web/api/services.py @@ -0,0 +1,78 @@ +""" +API endpoints for execution service management. + +Provides endpoints to check service status and health. +""" + +import logging +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException + +from ...core.execution.service.manager import get_process_service +from ..auth_dependencies import get_current_user +from ..models.user import User + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/process-service", tags=["Process Service"]) + + +@router.get("/status") +async def get_service_status( + current_user: User = Depends(get_current_user), +) -> dict[str, Any]: + """Get ProcessService status. + + Returns: + Dictionary with service status information + """ + process_service = get_process_service() + if not process_service: + return { + "enabled": False, + "status": "not_initialized", + "message": "Process isolation is not enabled or failed to initialize", + } + + try: + info = process_service.get_info() + return { + "enabled": True, + "status": info.status.value, + "resource_info": info.resource_info, + "metrics": info.metrics, + } + except Exception as e: + logger.error(f"Failed to get service status: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.get("/health") +async def health_check( + current_user: User = Depends(get_current_user), +) -> dict[str, Any]: + """Check ProcessService health. + + Returns: + Dictionary with health check result + """ + process_service = get_process_service() + if not process_service: + return { + "healthy": False, + "message": "ProcessService not initialized", + } + + try: + is_healthy = await process_service.health_check() + return { + "healthy": is_healthy, + "message": "Service is healthy" if is_healthy else "Service is unhealthy", + } + except Exception as e: + logger.error(f"Health check failed: {e}", exc_info=True) + return { + "healthy": False, + "message": f"Health check failed: {str(e)}", + } diff --git a/src/xagent/web/app.py b/src/xagent/web/app.py index c5d3e6a21..e7c95648b 100644 --- a/src/xagent/web/app.py +++ b/src/xagent/web/app.py @@ -10,7 +10,12 @@ from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles -from ..config import get_uploads_dir +from ..config import ( + get_process_isolation_address, + get_process_isolation_enabled, + get_process_isolation_workers, + get_uploads_dir, +) from ..core.tracing.langfuse import flush_langfuse, initialize_langfuse from .api.admin_mcp import admin_mcp_router from .api.admin_users import router as admin_users_router @@ -27,6 +32,7 @@ from .api.model import model_router from .api.monitor import monitor_router from .api.progress_ws import progress_ws_router +from .api.services import router as process_services_router from .api.skills import router as skills_router from .api.system import system_router from .api.templates import router as templates_router @@ -160,6 +166,7 @@ async def global_exception_handler(request: Request, exc: Exception) -> None: app.include_router(admin_users_router) app.include_router(admin_mcp_router) app.include_router(skills_router) +app.include_router(process_services_router) app.include_router(system_router) app.include_router(templates_router) app.include_router(agents_router) @@ -178,6 +185,20 @@ async def startup_event() -> None: initialize_langfuse() + if get_process_isolation_enabled(): + from ..core.execution.service import ProcessService, set_process_service + + process_service = ProcessService( + address=get_process_isolation_address(), + n_workers=get_process_isolation_workers(), + ) + await process_service.start() + set_process_service(process_service) + app.state.process_service = process_service + logger.info("Process isolation service initialized") + else: + logger.info("Process isolation service disabled") + # Initialize skill manager from ..skills.utils import create_skill_manager @@ -645,6 +666,14 @@ async def shutdown_event() -> None: if sandbox_mgr: await sandbox_mgr.cleanup() + process_service = getattr(app.state, "process_service", None) + if process_service is not None: + from ..core.execution.service import clear_process_service + + await process_service.stop() + clear_process_service() + app.state.process_service = None + # Frontend is now served by Next.js at http://localhost:3000 # This backend only provides API endpoints diff --git a/tests/core/execution/__init__.py b/tests/core/execution/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/core/execution/test_integration.py b/tests/core/execution/test_integration.py new file mode 100644 index 000000000..0236558e2 --- /dev/null +++ b/tests/core/execution/test_integration.py @@ -0,0 +1,220 @@ +""" +Integration tests for process isolation with xagent tools. + +Tests that the ProcessService integrates correctly with existing tools. +""" + +import asyncio + +import pytest + +from xagent.core.execution.service import ProcessService +from xagent.core.execution.service.manager import ( + clear_process_service, + set_process_service, +) + + +def _free_local_port() -> int: + import socket + + with socket.socket() as sock: + sock.bind(("localhost", 0)) + return int(sock.getsockname()[1]) + + +@pytest.mark.asyncio +async def test_process_service_manager(): + """Test ProcessService global manager.""" + service = ProcessService(n_workers=2, address="localhost:12352") + + # Set as global service + set_process_service(service) + + # Verify we can retrieve it + from xagent.core.execution.service.manager import get_process_service + + retrieved_service = get_process_service() + assert retrieved_service is service + + # Cleanup + clear_process_service() + assert get_process_service() is None + + +@pytest.mark.asyncio +async def test_execution_result_serialization(): + """Test ExecutionResult serialization.""" + from xagent.core.execution.service import ExecutionResult + + # Create result + result = ExecutionResult( + success=True, + output="Test output", + error="", + return_code=0, + metadata={"key": "value"}, + execution_time=1.5, + memory_used_mb=100.0, + ) + + # Convert to dict + result_dict = result.to_dict() + assert result_dict["success"] is True + assert result_dict["output"] == "Test output" + assert result_dict["execution_time"] == 1.5 + + # Convert back from dict + restored_result = ExecutionResult.from_dict(result_dict) + assert restored_result.success == result.success + assert restored_result.output == result.output + assert restored_result.execution_time == result.execution_time + + +@pytest.mark.asyncio +async def test_isolation_type_enum(): + """Test IsolationType enum.""" + from xagent.core.execution.service import IsolationType + + assert IsolationType.PROCESS.value == "process" + assert IsolationType.SANDBOX.value == "sandbox" + + +@pytest.mark.asyncio +async def test_service_status_enum(): + """Test ServiceStatus enum.""" + from xagent.core.execution.service import ServiceStatus + + assert ServiceStatus.STARTING.value == "starting" + assert ServiceStatus.RUNNING.value == "running" + assert ServiceStatus.STOPPING.value == "stopping" + assert ServiceStatus.STOPPED.value == "stopped" + assert ServiceStatus.ERROR.value == "error" + + +@pytest.mark.asyncio +async def test_service_info(): + """Test ServiceInfo dataclass.""" + from xagent.core.execution.service import ServiceInfo, ServiceStatus + + info = ServiceInfo( + name="test_service", + status=ServiceStatus.RUNNING, + resource_info={"workers": 4}, + metrics={"executions": 100}, + ) + + # Convert to dict + info_dict = info.to_dict() + assert info_dict["name"] == "test_service" + assert info_dict["status"] == "running" + assert info_dict["resource_info"]["workers"] == 4 + assert info_dict["metrics"]["executions"] == 100 + + +@pytest.mark.asyncio +async def test_process_service_not_started_error(): + """Test error when calling execute before starting service.""" + service = ProcessService(n_workers=2, address="localhost:12353") + + # Don't start the service + with pytest.raises(RuntimeError, match="ProcessService not started"): + await service.execute_python(code="print('test')") + + +@pytest.mark.asyncio +async def test_python_execution_with_workspace(): + """Test Python execution with workspace directory.""" + import tempfile + + service = ProcessService(n_workers=2, address="localhost:12354") + + await service.start() + + try: + # Create a temporary workspace + with tempfile.TemporaryDirectory() as workspace: + # Test that workspace is accessible + result = await service.execute_python( + code="import os\nprint(f'CWD: {os.getcwd()}')\nprint('Files:', os.listdir('.'))", + workspace=workspace, + timeout=10, + ) + + assert result.success is True + # The output should contain the workspace path + assert workspace in result.output or "CWD:" in result.output + + finally: + await service.stop() + + +@pytest.mark.asyncio +async def test_tool_factory_process_isolated_python_executor_runs( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +): + """ToolFactory-created execute_python_code should run through process isolation.""" + import os + + import xagent.web.sandbox_manager as sandbox_manager_module + from xagent.core.tools.adapters.vibe.basic_tools import create_basic_tools + from xagent.core.tools.adapters.vibe.config import ToolConfig + from xagent.core.tools.adapters.vibe.factory import ToolFactory, ToolRegistry + + clear_process_service() + sandbox_manager_module._sandbox_manager = None + sandbox_manager_module._sandbox_manager_initialized = True + monkeypatch.setenv("XAGENT_PROCESS_ISOLATION_ENABLED", "true") + + async def _create_registered_tools(config): + return await create_basic_tools(config) + + monkeypatch.setattr( + ToolRegistry, + "create_registered_tools", + _create_registered_tools, + ) + + service = ProcessService(n_workers=1, address=f"localhost:{_free_local_port()}") + await service.start() + set_process_service(service) + + try: + tools = await ToolFactory.create_all_tools( + ToolConfig( + { + "workspace": { + "task_id": "process_isolated_python_executor", + "base_dir": str(tmp_path), + }, + "allowed_tools": ["execute_python_code"], + "basic_tools_enabled": True, + "file_tools_enabled": False, + "browser_tools_enabled": False, + } + ) + ) + assert len(tools) == 1 + + tool = tools[0] + assert tool.name == "execute_python_code" + assert tool.is_isolated is True + + parent_pid = os.getpid() + result = await asyncio.wait_for( + tool.run_json_async( + { + "code": "import os\nprint(os.getpid())\nprint('process-ok')", + } + ), + timeout=30, + ) + + assert result["success"] is True + assert "process-ok" in result["output"] + child_pid = int(result["output"].splitlines()[0]) + assert child_pid != parent_pid + finally: + await service.stop() + clear_process_service() diff --git a/tests/core/execution/test_process_service.py b/tests/core/execution/test_process_service.py new file mode 100644 index 000000000..3558139e2 --- /dev/null +++ b/tests/core/execution/test_process_service.py @@ -0,0 +1,205 @@ +""" +Test ProcessService functionality. + +This test requires xoscar to be installed. +""" + +import asyncio + +import pytest + +from xagent.core.execution.service import ProcessService + + +@pytest.mark.asyncio +async def test_process_service_lifecycle(): + """Test ProcessService start and stop.""" + service = ProcessService(n_workers=2, address="localhost:12346") + + # Test start + await service.start() + assert service.status.value == "running" + + # Test health check + is_healthy = await service.health_check() + assert is_healthy is True + + # Test get_info + info = service.get_info() + assert info.name == "process" + assert info.status.value == "running" + assert info.resource_info["n_workers"] == 2 + + # Test stop + await service.stop() + assert service.status.value == "stopped" + + +@pytest.mark.asyncio +async def test_python_execution(): + """Test Python code execution in isolated process.""" + service = ProcessService(n_workers=2, address="localhost:12347") + + await service.start() + + try: + # Test simple Python execution + result = await service.execute_python( + code="print('Hello from isolated process!')\nresult = 2 + 2\nprint(f'2 + 2 = {result}')", + timeout=10, + ) + + assert result.success is True + assert "Hello from isolated process!" in result.output + assert "2 + 2 = 4" in result.output + assert result.return_code == 0 + + finally: + await service.stop() + + +@pytest.mark.asyncio +async def test_python_execution_with_error(): + """Test Python code execution with syntax error.""" + service = ProcessService(n_workers=2, address="localhost:12348") + + await service.start() + + try: + # Test Python code with error + result = await service.execute_python( + code="print('Before error')\nraise ValueError('Test error')", + timeout=10, + ) + + assert result.success is False + assert "ValueError: Test error" in result.error + assert result.return_code == 1 + + finally: + await service.stop() + + +@pytest.mark.asyncio +async def test_python_execution_timeout(): + """Test Python code execution with timeout.""" + service = ProcessService(n_workers=2, address="localhost:12349") + + await service.start() + + try: + # Test Python code that exceeds timeout + result = await service.execute_python( + code="import time\ntime.sleep(10)\nprint('This should not print')", + timeout=2, + ) + + assert result.success is False + assert "timed out" in result.error.lower() + assert service.get_info().resource_info["active_actors"] == 0 + + finally: + await service.stop() + + +@pytest.mark.asyncio +async def test_sub_pool_startup_timeout(monkeypatch): + """Test ProcessService does not hang forever when sub-pool startup stalls.""" + from xagent.core.execution.service import ServiceStatus + + class HangingPool: + async def append_sub_pool(self, **kwargs): + await asyncio.sleep(10) + + async def remove_sub_pool(self, *args, **kwargs): + raise AssertionError("remove_sub_pool should not run without an address") + + service = ProcessService(n_workers=1, address="localhost:12356") + service._status = ServiceStatus.RUNNING + service._pool = HangingPool() + + result = await service.execute_python("print('never runs')", timeout=0.1) + + assert result.success is False + assert "timed out during sub-pool startup" in result.error.lower() + assert service.get_info().resource_info["active_actors"] == 0 + + +@pytest.mark.asyncio +async def test_command_execution(): + """Test shell command execution in isolated process.""" + service = ProcessService(n_workers=2, address="localhost:12350") + + await service.start() + + try: + # Test simple command + result = await service.execute_command( + command="echo 'Hello from command!'", + timeout=10, + ) + + assert result.success is True + assert "Hello from command!" in result.output + assert result.return_code == 0 + + finally: + await service.stop() + + +@pytest.mark.asyncio +async def test_command_execution_preserves_shell_semantics(tmp_path): + """Test process command execution preserves existing shell semantics.""" + service = ProcessService(n_workers=2, address="localhost:12355") + + await service.start() + + try: + pipe_result = await service.execute_command( + command="echo hello | wc -c", + workspace=str(tmp_path), + timeout=10, + ) + assert pipe_result.success is True + assert pipe_result.output.strip() == "6" + + redirect_result = await service.execute_command( + command="echo hi > out.txt", + workspace=str(tmp_path), + timeout=10, + ) + assert redirect_result.success is True + assert (tmp_path / "out.txt").read_text().strip() == "hi" + + chain_result = await service.execute_command( + command=f"cd {tmp_path} && pwd", + timeout=10, + ) + assert chain_result.success is True + assert chain_result.output.strip() == str(tmp_path) + + finally: + await service.stop() + + +@pytest.mark.asyncio +async def test_javascript_execution(): + """Test JavaScript code execution in isolated process.""" + service = ProcessService(n_workers=2, address="localhost:12351") + + await service.start() + + try: + # Test simple JavaScript execution + result = await service.execute_javascript( + code="console.log('Hello from JavaScript!');\nconst result = 2 + 2;\nconsole.log(`2 + 2 = ${result}`);", + timeout=10, + ) + + assert result.success is True + assert "Hello from JavaScript!" in result.output + assert "2 + 2 = 4" in result.output + assert result.return_code == 0 + + finally: + await service.stop() diff --git a/tests/core/test_config.py b/tests/core/test_config.py index 9705fbb09..510d56453 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -12,6 +12,9 @@ EXTERNAL_UPLOAD_DIRS, LANCEDB_PATH, MAX_UPLOAD_SIZE, + PROCESS_ISOLATION_ADDRESS, + PROCESS_ISOLATION_ENABLED, + PROCESS_ISOLATION_WORKERS, SANDBOX_CPUS, SANDBOX_ENV, SANDBOX_IMAGE, @@ -31,6 +34,9 @@ get_external_upload_dirs, get_lancedb_path, get_max_upload_size_bytes, + get_process_isolation_address, + get_process_isolation_enabled, + get_process_isolation_workers, get_sandbox_cpus, get_sandbox_env, get_sandbox_image, @@ -76,6 +82,49 @@ def test_max_upload_size_constant(self): def test_web_search_provider_constant(self): assert WEB_SEARCH_PROVIDER == "XAGENT_WEB_SEARCH_PROVIDER" + def test_process_isolation_constants(self): + assert PROCESS_ISOLATION_ENABLED == "XAGENT_PROCESS_ISOLATION_ENABLED" + assert PROCESS_ISOLATION_ADDRESS == "XAGENT_PROCESS_ISOLATION_ADDRESS" + assert PROCESS_ISOLATION_WORKERS == "XAGENT_PROCESS_ISOLATION_WORKERS" + + +class TestProcessIsolationConfig: + """Test process isolation configuration.""" + + def test_process_isolation_enabled_by_default(self, monkeypatch): + monkeypatch.delenv(PROCESS_ISOLATION_ENABLED, raising=False) + assert get_process_isolation_enabled() is True + + @pytest.mark.parametrize("value", ["1", "true", "yes", "on", " TRUE "]) + def test_process_isolation_enabled_values(self, monkeypatch, value): + monkeypatch.setenv(PROCESS_ISOLATION_ENABLED, value) + assert get_process_isolation_enabled() is True + + @pytest.mark.parametrize("value", ["0", "false", "no", "off", ""]) + def test_process_isolation_disabled_values(self, monkeypatch, value): + monkeypatch.setenv(PROCESS_ISOLATION_ENABLED, value) + assert get_process_isolation_enabled() is False + + def test_process_isolation_address_default_and_override(self, monkeypatch): + monkeypatch.delenv(PROCESS_ISOLATION_ADDRESS, raising=False) + assert get_process_isolation_address() == "localhost:12345" + + monkeypatch.setenv(PROCESS_ISOLATION_ADDRESS, "localhost:22345") + assert get_process_isolation_address() == "localhost:22345" + + def test_process_isolation_workers_default_and_override(self, monkeypatch): + monkeypatch.delenv(PROCESS_ISOLATION_WORKERS, raising=False) + assert get_process_isolation_workers() == 0 + + monkeypatch.setenv(PROCESS_ISOLATION_WORKERS, "3") + assert get_process_isolation_workers() == 3 + + monkeypatch.setenv(PROCESS_ISOLATION_WORKERS, "bad") + assert get_process_isolation_workers() == 0 + + monkeypatch.setenv(PROCESS_ISOLATION_WORKERS, "-1") + assert get_process_isolation_workers() == 0 + class TestGetWebSearchProvider: """Test get_web_search_provider() function.""" diff --git a/tests/core/tools/test_process_isolated_wrapper.py b/tests/core/tools/test_process_isolated_wrapper.py new file mode 100644 index 000000000..2a6e42bfd --- /dev/null +++ b/tests/core/tools/test_process_isolated_wrapper.py @@ -0,0 +1,385 @@ +""" +Tests for ProcessIsolatedToolWrapper. +""" + +import pytest +from pydantic import BaseModel + +from xagent.core.execution.service import ProcessService +from xagent.core.execution.service.manager import ( + clear_process_service, + set_process_service, +) +from xagent.core.tools.adapters.vibe.base import AbstractBaseTool, ToolCategory +from xagent.core.tools.adapters.vibe.config import ToolConfig +from xagent.core.tools.adapters.vibe.factory import ToolFactory, ToolRegistry +from xagent.core.tools.adapters.vibe.output_filter_wrapper import ( + OutputFilteredToolWrapper, +) +from xagent.core.tools.adapters.vibe.process_isolated import ( + ProcessIsolatedToolWrapper, + create_process_isolated_tool, + maybe_wrap_tool, + should_use_process_isolation, + wrap_tools, +) + + +# Simple test tool +class SimpleCalculatorTool: + """A simple calculator tool for testing.""" + + supports_process_isolation = True + + def __init__(self, precision: int = 2): + self.precision = precision + + @property + def name(self) -> str: + return "simple_calculator" + + @property + def description(self) -> str: + return "A simple calculator" + + @property + def tags(self) -> list[str]: + return ["calculator", "test"] + + @property + def metadata(self): + return {"version": "1.0"} + + def args_type(self): + from typing import Optional + + from pydantic import BaseModel + + class CalculatorArgs(BaseModel): + expression: str + scale: Optional[float] = 1.0 + + return CalculatorArgs + + def return_type(self): + from pydantic import BaseModel + + class CalculatorResult(BaseModel): + result: float + scaled: float + + return CalculatorResult + + def state_type(self): + return None + + async def run_json_async(self, args: dict) -> dict: + """Calculate expression.""" + try: + expression = args["expression"] + scale = args.get("scale", 1.0) + + # Safe evaluation + result = eval(expression, {"__builtins__": {}}, {}) + + # Apply precision + rounded = round(result, self.precision) + scaled = round(result * scale, self.precision) + + return { + "result": rounded, + "scaled": scaled, + } + except Exception as e: + raise RuntimeError(f"Calculation failed: {e}") + + +class LifecycleArgs(BaseModel): + value: str + + +class LifecycleResult(BaseModel): + value: str + + +class LifecycleState(BaseModel): + value: str = "initial" + + +class LifecycleTool(AbstractBaseTool): + supports_process_isolation = True + category = ToolCategory.BASIC + + def __init__(self) -> None: + self.setup_calls: list[str | None] = [] + self.teardown_calls: list[str | None] = [] + self.state = "initial" + + @property + def name(self) -> str: + return "lifecycle_tool" + + @property + def description(self) -> str: + return "Lifecycle test tool" + + def args_type(self) -> type[BaseModel]: + return LifecycleArgs + + def return_type(self) -> type[BaseModel]: + return LifecycleResult + + def state_type(self) -> type[BaseModel]: + return LifecycleState + + def is_async(self) -> bool: + return False + + def return_value_as_string(self, value): + return f"formatted:{value}" + + def run_json_sync(self, args): + return {"value": args["value"]} + + async def run_json_async(self, args): + return self.run_json_sync(args) + + async def save_state_json(self): + return {"value": self.state} + + async def load_state_json(self, state): + self.state = state["value"] + + async def setup(self, task_id=None): + self.setup_calls.append(task_id) + + async def teardown(self, task_id=None): + self.teardown_calls.append(task_id) + + +class UnsupportedTool(LifecycleTool): + supports_process_isolation = False + + @property + def name(self) -> str: + return "unsupported_tool" + + +@pytest.mark.asyncio +async def test_process_isolated_wrapper_basic(): + """Test basic functionality of ProcessIsolatedToolWrapper.""" + # Create and start ProcessService + service = ProcessService(n_workers=2, address="localhost:12360") + await service.start() + set_process_service(service) + + try: + # Create tool + tool = SimpleCalculatorTool(precision=4) + + # Wrap with process isolation + wrapped_tool = create_process_isolated_tool(tool, timeout=30) + + # Verify wrapper properties + assert wrapped_tool.name == "simple_calculator" + assert wrapped_tool.description == "A simple calculator" + assert wrapped_tool.is_isolated is True + + # Execute tool + result = await wrapped_tool.run_json_async( + { + "expression": "2 + 3 * 4", + "scale": 2.0, + } + ) + + assert result["result"] == 14.0 + assert result["scaled"] == 28.0 + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_process_isolated_wrapper_with_init_params(): + """Test wrapper preserves init params via xoscar serialization.""" + service = ProcessService(n_workers=2, address="localhost:12361") + await service.start() + set_process_service(service) + + try: + # Create tool with custom init param + # xoscar automatically serializes the tool instance + tool = SimpleCalculatorTool(precision=6) + wrapped_tool = create_process_isolated_tool(tool) + + # Execute and verify precision is preserved + # xoscar serializes tool with its state (precision=6) + result = await wrapped_tool.run_json_async( + { + "expression": "1 / 3", + } + ) + + # Should be rounded to 6 decimal places (tool state preserved) + assert result["result"] == 0.333333 + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_should_use_process_isolation(): + """Test should_use_process_isolation logic.""" + import xagent.web.sandbox_manager as sandbox_manager_module + from xagent.core.execution.service.manager import clear_process_service + + # Clear both services + clear_process_service() + # Temporarily clear sandbox manager + sandbox_manager_module._sandbox_manager = None + sandbox_manager_module._sandbox_manager_initialized = False + + # No services available + assert should_use_process_isolation("test_tool") is False + + # Enable ProcessService + service = ProcessService(n_workers=2, address="localhost:12362") + await service.start() + set_process_service(service) + + try: + # ProcessService available, no sandbox + assert should_use_process_isolation("test_tool") is True + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_maybe_wrap_tool(): + """Test maybe_wrap_tool function.""" + service = ProcessService(n_workers=2, address="localhost:12363") + await service.start() + set_process_service(service) + + try: + tool = SimpleCalculatorTool(precision=2) + + # Should wrap when ProcessService is available + wrapped = maybe_wrap_tool(tool) + assert isinstance(wrapped, ProcessIsolatedToolWrapper) + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_maybe_wrap_tool_fallback(): + """Test maybe_wrap_tool falls back to original tool when ProcessService unavailable.""" + from xagent.core.execution.service.manager import clear_process_service + + # Ensure no ProcessService + clear_process_service() + + tool = SimpleCalculatorTool(precision=2) + + # Should return original tool when no ProcessService + wrapped = maybe_wrap_tool(tool) + assert wrapped is tool # Same object + assert not isinstance(wrapped, ProcessIsolatedToolWrapper) + + +@pytest.mark.asyncio +async def test_maybe_wrap_tool_requires_explicit_support(): + """Test process isolation only wraps explicitly supported tools.""" + service = ProcessService(n_workers=2, address="localhost:12365") + await service.start() + set_process_service(service) + + try: + tool = UnsupportedTool() + + wrapped = maybe_wrap_tool(tool) + assert wrapped is tool + assert not isinstance(wrapped, ProcessIsolatedToolWrapper) + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_wrap_tools_list(): + """Test wrap_tools function with multiple tools.""" + service = ProcessService(n_workers=2, address="localhost:12364") + await service.start() + set_process_service(service) + + try: + tools = [ + SimpleCalculatorTool(precision=2), + SimpleCalculatorTool(precision=4), + ] + + wrapped_tools = wrap_tools(tools) + + assert len(wrapped_tools) == 2 + assert all(isinstance(t, ProcessIsolatedToolWrapper) for t in wrapped_tools) + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_process_isolated_wrapper_delegates_tool_contract(): + """Test lifecycle, state, formatting, and marker properties delegate.""" + tool = LifecycleTool() + wrapped = create_process_isolated_tool(tool) + + assert wrapped.is_async() is False + assert wrapped.category == ToolCategory.BASIC + assert wrapped.supports_process_isolation is True + assert ( + wrapped.return_value_as_string({"value": "ok"}) == "formatted:{'value': 'ok'}" + ) + + await wrapped.setup(task_id="task-1") + assert tool.setup_calls == ["task-1"] + + await wrapped.load_state_json({"value": "restored"}) + assert await wrapped.save_state_json() == {"value": "restored"} + + await wrapped.teardown(task_id="task-1") + assert tool.teardown_calls == ["task-1"] + + +@pytest.mark.asyncio +async def test_tool_factory_applies_process_isolation_for_supported_tools(monkeypatch): + """Test ToolFactory wraps supported tools through the production path.""" + clear_process_service() + + async def fake_create_registered_tools(config): + return [LifecycleTool(), UnsupportedTool()] + + monkeypatch.setattr( + ToolRegistry, + "create_registered_tools", + fake_create_registered_tools, + ) + set_process_service(ProcessService(n_workers=1, address="localhost:12366")) + + try: + tools = await ToolFactory.create_all_tools(ToolConfig({})) + + assert len(tools) == 2 + assert all(isinstance(tool, OutputFilteredToolWrapper) for tool in tools) + assert tools[0].is_isolated is True + assert tools[1].is_isolated is False + finally: + clear_process_service() diff --git a/tests/core/tools/test_python_executor_process_isolation.py b/tests/core/tools/test_python_executor_process_isolation.py new file mode 100644 index 000000000..76dfde44e --- /dev/null +++ b/tests/core/tools/test_python_executor_process_isolation.py @@ -0,0 +1,338 @@ +""" +Integration tests for Python executor with process isolation. + +Tests that PythonExecutorTool actually runs in xoscar isolated processes. +""" + +import asyncio +import os + +import pytest + +from xagent.core.execution.service import ProcessService +from xagent.core.execution.service.manager import ( + clear_process_service, + set_process_service, +) +from xagent.core.tools.adapters.vibe.process_isolated import ( + create_process_isolated_tool, +) +from xagent.core.tools.adapters.vibe.python_executor import PythonExecutorTool + + +@pytest.mark.asyncio +async def test_python_executor_in_isolated_process(): + """Test that Python executor runs in xoscar isolated process.""" + # Create and start ProcessService + service = ProcessService(address="localhost:12400") + await service.start() + set_process_service(service) + + try: + # Create Python executor tool + python_tool = PythonExecutorTool(workspace=None) + + # Wrap with process isolation + isolated_tool = create_process_isolated_tool(python_tool, timeout=30) + + # Verify it's wrapped + assert isolated_tool.is_isolated is True + assert isolated_tool.name == "python_executor" + + # Execute code that checks process isolation + code = """ +import os +# Get current process ID +current_pid = os.getpid() +# Get parent process ID +parent_pid = os.getppid() +# Output both +print(f"PID: {current_pid}") +print(f"Parent PID: {parent_pid}") + +# Get main process ID from environment if available +import json +result = { + "current_pid": current_pid, + "parent_pid": parent_pid, +} +print(f"Result: {json.dumps(result)}") +""" + + result = await isolated_tool.run_json_async( + { + "code": code, + "capture_output": True, + } + ) + + assert result["success"] is True + output = result["output"] + + # The output should contain process IDs + assert "PID:" in output + assert "Parent PID:" in output + + # Verify it ran in a different process (not this one) + import json + + main_pid = os.getpid() + + # Extract PID from output + for line in output.split("\n"): + if line.strip().startswith("Result:"): + result_json = line.split("Result:")[1].strip() + result_data = json.loads(result_json) + + isolated_pid = result_data["current_pid"] + + # The isolated process should be different from main process + assert isolated_pid != main_pid, ( + f"Python executor ran in same process (PID: {isolated_pid}) as main process (PID: {main_pid})" + ) + + print( + f"✅ Python executor ran in isolated process (PID: {isolated_pid}) != main process (PID: {main_pid})" + ) + break + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_python_executor_state_isolation(): + """Test that Python executor state is isolated between executions.""" + service = ProcessService(address="localhost:12401") + await service.start() + set_process_service(service) + + try: + python_tool = PythonExecutorTool(workspace=None) + isolated_tool = create_process_isolated_tool(python_tool, timeout=30) + + # First execution: create a variable + result1 = await isolated_tool.run_json_async( + { + "code": "x = 42\nprint(f'x = {x}')", + "capture_output": True, + } + ) + + assert result1["success"] is True + assert "x = 42" in result1["output"] + + # Second execution: variable should not exist (state isolation) + result2 = await isolated_tool.run_json_async( + { + "code": """ +try: + print(f'x exists: {x}') + state_isolated = False +except NameError: + print('x does not exist (state is isolated)') + state_isolated = True +""", + "capture_output": True, + } + ) + + assert result2["success"] is True + assert "state is isolated" in result2["output"].lower() + + print("✅ State isolation verified between executions") + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_python_executor_import_isolation(): + """Test that imports don't leak between executions.""" + service = ProcessService(address="localhost:12402") + await service.start() + set_process_service(service) + + try: + python_tool = PythonExecutorTool(workspace=None) + isolated_tool = create_process_isolated_tool(python_tool, timeout=30) + + # First execution: import a module + result1 = await isolated_tool.run_json_async( + { + "code": "import math\nprint(f'math.pi = {math.pi}')", + "capture_output": True, + } + ) + + assert result1["success"] is True + assert "math.pi = 3.14159" in result1["output"] + + # Second execution: check if import is still available + # In isolated process, each execution starts fresh + result2 = await isolated_tool.run_json_async( + { + "code": """ +import sys +# Check if math is in built-in modules +math_in_builtins = 'math' in sys.builtin_module_names +import math as math_check +print(f'math available: {not math_in_builtins}') # Should be available as stdlib +""", + "capture_output": True, + } + ) + + assert result2["success"] is True + print("✅ Import isolation verified") + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_python_executor_error_handling_isolated(): + """Test that errors in isolated process don't affect main process.""" + service = ProcessService(address="localhost:12403") + await service.start() + set_process_service(service) + + try: + python_tool = PythonExecutorTool(workspace=None) + isolated_tool = create_process_isolated_tool(python_tool, timeout=30) + + # Execute code that raises error + error_result = await isolated_tool.run_json_async( + { + "code": "raise RuntimeError('Intentional error in isolated process')", + "capture_output": True, + } + ) + + assert error_result["success"] is False + assert "RuntimeError" in error_result["error"] + assert "Intentional error" in error_result["error"] + + # Main process should still be working + main_pid = os.getpid() + assert main_pid > 0 + + # Next execution should work fine (no state corruption) + normal_result = await isolated_tool.run_json_async( + { + "code": "print('After error, execution continues normally')", + "capture_output": True, + } + ) + + assert normal_result["success"] is True + assert "continues normally" in normal_result["output"] + + print("✅ Error handling verified - errors isolated to subprocess") + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_python_executor_workspace_binding_with_isolation(): + """Test that workspace binding works with process isolation.""" + import tempfile + + service = ProcessService(address="localhost:12404") + await service.start() + set_process_service(service) + + try: + # Create a temporary workspace + with tempfile.TemporaryDirectory() as workspace_dir: + from xagent.core.workspace import TaskWorkspace + + workspace = TaskWorkspace( + id="test_task", + base_dir=workspace_dir, + ) + + python_tool = PythonExecutorTool(workspace=workspace) + isolated_tool = create_process_isolated_tool(python_tool, timeout=30) + + # Execute code that uses workspace + result = await isolated_tool.run_json_async( + { + "code": """ +import os +print(f'Working directory: {os.getcwd()}') +print('Workspace accessible!') +""", + "capture_output": True, + } + ) + + assert result["success"] is True + assert "Working directory:" in result["output"] + print("✅ Workspace binding works with process isolation") + + finally: + await service.stop() + clear_process_service() + + +@pytest.mark.asyncio +async def test_python_executor_concurrent_executions(): + """Test that multiple Python executors can run concurrently in isolation.""" + service = ProcessService(address="localhost:12405") + await service.start() + set_process_service(service) + + try: + python_tool = PythonExecutorTool(workspace=None) + isolated_tool = create_process_isolated_tool(python_tool, timeout=30) + + # Run multiple executions concurrently + async def run_execution(task_id: int): + code = f""" +import os +import time +import random +print(f'Task {task_id} starting in PID: {{os.getpid()}}') +# Simulate some work +time.sleep(random.uniform(0.1, 0.3)) +print(f'Task {task_id} completed') +""" + result = await isolated_tool.run_json_async( + { + "code": code, + "capture_output": True, + } + ) + return task_id, result + + # Run 5 concurrent tasks + tasks = [run_execution(i) for i in range(5)] + results = await asyncio.gather(*tasks) + + # All should succeed + for task_id, result in results: + assert result["success"] is True, f"Task {task_id} failed" + assert f"Task {task_id} completed" in result["output"] + + print("✅ Concurrent executions verified") + + finally: + await service.stop() + clear_process_service() + + +if __name__ == "__main__": + # Run tests manually for debugging + asyncio.run(test_python_executor_in_isolated_process()) + asyncio.run(test_python_executor_state_isolation()) + asyncio.run(test_python_executor_import_isolation()) + asyncio.run(test_python_executor_error_handling_isolated()) + asyncio.run(test_python_executor_workspace_binding_with_isolation()) + asyncio.run(test_python_executor_concurrent_executions()) + print("\n✅ All process isolation tests passed!") diff --git a/tests/integration/test_auto_migration_startup.py b/tests/integration/test_auto_migration_startup.py index b5e5ad0ee..9f0ce400a 100644 --- a/tests/integration/test_auto_migration_startup.py +++ b/tests/integration/test_auto_migration_startup.py @@ -733,6 +733,7 @@ class _FakeConn: original_create_task = asyncio.create_task monkeypatch.setenv("LANCEDB_AUTO_MIGRATE", "false") + monkeypatch.setenv("XAGENT_PROCESS_ISOLATION_ENABLED", "false") monkeypatch.setattr(web_app_module, "init_db", lambda: None) monkeypatch.setattr(web_app_module, "_migration_task", None) monkeypatch.setattr( @@ -838,6 +839,7 @@ class _FakeConn: original_create_task = asyncio.create_task monkeypatch.setenv("LANCEDB_AUTO_MIGRATE", "true") + monkeypatch.setenv("XAGENT_PROCESS_ISOLATION_ENABLED", "false") monkeypatch.setattr(web_app_module, "init_db", lambda: None) monkeypatch.setattr(web_app_module, "_migration_task", None) monkeypatch.setattr( @@ -944,6 +946,7 @@ class _FakeConn: original_create_task = asyncio.create_task monkeypatch.setenv("LANCEDB_AUTO_MIGRATE", "true") + monkeypatch.setenv("XAGENT_PROCESS_ISOLATION_ENABLED", "false") monkeypatch.setattr(web_app_module, "init_db", lambda: None) monkeypatch.setattr(web_app_module, "_migration_task", None) monkeypatch.setattr( diff --git a/tests/test_process_quick.py b/tests/test_process_quick.py new file mode 100644 index 000000000..7401c6563 --- /dev/null +++ b/tests/test_process_quick.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +"""Quick test of ProcessService functionality.""" + +import asyncio + + +async def test_process_service(): + """Test ProcessService basic functionality.""" + from xagent.core.execution.service import ProcessService + + print("Creating ProcessService...") + service = ProcessService(n_workers=2, address="localhost:12399") + + print("Starting ProcessService...") + await service.start() + + try: + print(f"✅ Service status: {service.status.value}") + + # Test health check + is_healthy = await service.health_check() + print(f"✅ Health check: {is_healthy}") + + # Get service info + info = service.get_info() + print(f"✅ Service info: {info.to_dict()}") + + # Test Python execution + print("\nTesting Python execution...") + result = await service.execute_python( + code="print('Hello from isolated process!')\nresult = 2 + 2\nprint(f'2 + 2 = {result}')", + timeout=10, + ) + + if result.success: + print("✅ Python execution successful!") + print(f"Output:\n{result.output}") + else: + print(f"❌ Python execution failed: {result.error}") + + # Test command execution + print("\nTesting command execution...") + result = await service.execute_command( + command="echo 'Hello from command!'", + timeout=10, + ) + + if result.success: + print("✅ Command execution successful!") + print(f"Output: {result.output}") + else: + print(f"❌ Command execution failed: {result.error}") + + finally: + print("\nStopping ProcessService...") + await service.stop() + print(f"✅ Service status: {service.status.value}") + + +if __name__ == "__main__": + asyncio.run(test_process_service()) + print("\n✅ All tests passed!") diff --git a/tests/web/api/test_process_service_api.py b/tests/web/api/test_process_service_api.py new file mode 100644 index 000000000..ee50e55de --- /dev/null +++ b/tests/web/api/test_process_service_api.py @@ -0,0 +1,12 @@ +"""Tests for process service API routing.""" + +from xagent.web.api.services import router + + +def test_process_service_router_uses_explicit_prefix(): + """Process service routes must not collide with app-level health routes.""" + paths = {route.path for route in router.routes} + + assert "/api/process-service/status" in paths + assert "/api/process-service/health" in paths + assert "/health" not in paths