feat: add process isolation for tool execution

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)
 # ===========================================

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

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.
 

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",
+]

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)

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)

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)