feat(sandbox): add E2B-compatible remote backend

ms_enclave/sandbox/boxes/__init__.py

@@ -3,6 +3,7 @@
 from .base import Sandbox, SandboxFactory, register_sandbox
 from .docker_notebook import DockerNotebookSandbox
 from .docker_sandbox import DockerSandbox
+from .e2b_sandbox import E2BSandbox
 from .stateless_sandbox import StatelessSandbox
 from .volcengine import VolcengineSandbox
 
@@ -14,6 +15,7 @@
     'StatelessSandbox',
     # Implementations
     'DockerSandbox',
+    'E2BSandbox',
     'DockerNotebookSandbox',
     'VolcengineSandbox',
 ]

ms_enclave/sandbox/boxes/base.py

@@ -15,6 +15,7 @@
     CommandResult,
     DockerNotebookConfig,
     DockerSandboxConfig,
+    E2BSandboxConfig,
     SandboxConfig,
     SandboxInfo,
     SandboxStatus,
@@ -265,6 +266,8 @@ def create_sandbox(
                 config = DockerNotebookConfig()
             elif sandbox_type == SandboxType.VOLCENGINE:
                 config = VolcengineSandboxConfig()
+            elif sandbox_type == SandboxType.E2B:
+                config = E2BSandboxConfig()
             else:
                 config = SandboxConfig()
         elif isinstance(config, dict):
@@ -274,6 +277,8 @@ def create_sandbox(
                 config = DockerNotebookConfig(**config)
             elif sandbox_type == SandboxType.VOLCENGINE:
                 config = VolcengineSandboxConfig(**config)
+            elif sandbox_type == SandboxType.E2B:
+                config = E2BSandboxConfig(**config)
             else:
                 config = SandboxConfig(**config)
 

ms_enclave/sandbox/boxes/docker_sandbox.py

@@ -536,6 +536,8 @@ async def _create_container(self) -> None:
 
             # Privileged mode
             container_config['privileged'] = self.config.privileged
+            if self.config.cap_add:
+                container_config['cap_add'] = list(self.config.cap_add)
 
             # Platform (e.g. linux/amd64 for amd64 images on Apple Silicon)
             if self.config.platform:

ms_enclave/sandbox/boxes/e2b_connection.py

@@ -0,0 +1,96 @@
+"""Optional SDK boundary and connection settings for E2B services."""
+
+import asyncio
+from dataclasses import dataclass
+from typing import Any, Awaitable, Callable, Dict, Optional, TypeVar
+
+import httpx
+from pydantic import SecretStr
+
+from ms_enclave.utils import get_logger
+
+from ..model import E2BSandboxManagerConfig
+from ..utils.e2b_helpers import BuildMode, E2BProvider, redact
+
+
+class _MissingE2BError(Exception):
+    """Placeholder exception when the optional SDK is unavailable."""
+
+
+try:
+    from e2b import AsyncSandbox, AsyncTemplate, CommandExitException, RateLimitException
+    from e2b.api.client_async import get_api_client
+    from e2b.connection_config import ConnectionConfig
+
+    _E2B_IMPORT_ERROR: Optional[ImportError] = None
+except ImportError as exc:
+    AsyncSandbox = AsyncTemplate = get_api_client = ConnectionConfig = None
+    CommandExitException = RateLimitException = _MissingE2BError
+    _E2B_IMPORT_ERROR = exc
+
+logger = get_logger()
+E2B_REQUEST_TIMEOUT_S = 60.0
+_DISPATCH_ATTEMPTS = 3
+T = TypeVar('T')
+
+
+def require_e2b() -> None:
+    """Fail with an installation hint when E2B was not installed."""
+    if _E2B_IMPORT_ERROR is not None:
+        raise ImportError(
+            "The E2B sandbox requires the e2b SDK (Python >= 3.11): pip install 'ms-enclave[e2b]'"
+        ) from _E2B_IMPORT_ERROR
+
+
+@dataclass(frozen=True)
+class E2BConnection:
+    """Resolved settings shared by the manager, sandboxes and template resolver."""
+
+    api_key: SecretStr
+    api_url: Optional[str]
+    domain: Optional[str]
+    provider: E2BProvider
+    region: Optional[str]
+    build_mode: BuildMode
+
+    @classmethod
+    def from_config(cls, config: E2BSandboxManagerConfig) -> 'E2BConnection':
+        """Resolve a validated manager configuration."""
+        return cls(
+            config.api_key,
+            config.api_url,
+            config.domain,
+            config.provider,
+            config.target_region,
+            config.resolved_build_mode,
+        )
+
+    def api_params(self, **extra: Any) -> Dict[str, Any]:
+        """Build SDK options, omitting unset endpoints to preserve SDK defaults."""
+        params = {'api_key': self.api_key.get_secret_value(), 'request_timeout': E2B_REQUEST_TIMEOUT_S}
+        if self.api_url:
+            params['api_url'] = self.api_url
+        if self.domain:
+            params['domain'] = self.domain
+        return {**params, **extra}
+
+    def redact(self, text: str) -> str:
+        """Remove the connection key from diagnostic text."""
+        return redact(text, self.api_key.get_secret_value())
+
+
+async def retry_dispatch(
+    op_name: str, call: Callable[[], Awaitable[T]], *, run_id: str, redactor: Callable[[str], str]
+) -> T:
+    """Retry only connection establishment failures and explicit rate limits."""
+    retryable = (ConnectionError, httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout, RateLimitException)
+    for attempt in range(1, _DISPATCH_ATTEMPTS + 1):
+        try:
+            return await call()
+        except Exception as exc:
+            reason = redactor(str(exc))
+            rate_limited = isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code == 429
+            if (not isinstance(exc, retryable) and not rate_limited) or attempt == _DISPATCH_ATTEMPTS:
+                raise RuntimeError(f'E2B {op_name} failed (run_id={run_id}): {reason}') from None
+            logger.warning(f'E2B {op_name} attempt={attempt} run_id={run_id}: {reason}; retrying')
+            await asyncio.sleep(2 ** (attempt - 1))

ms_enclave/sandbox/boxes/e2b_sandbox.py

@@ -0,0 +1,201 @@
+"""Stateful remote sandbox backed by the optional E2B SDK."""
+
+import asyncio
+import posixpath
+import shlex
+import time
+import uuid
+from typing import Any, Dict, List, Optional, Union
+
+from ms_enclave.utils import get_logger
+
+from ..model import (
+    CommandResult,
+    E2BSandboxConfig,
+    E2BSandboxManagerConfig,
+    ExecutionStatus,
+    SandboxStatus,
+    SandboxType,
+)
+from .base import Sandbox, register_sandbox
+from .e2b_connection import AsyncSandbox, CommandExitException, E2BConnection, require_e2b, retry_dispatch
+from .e2b_template import E2BTemplateResolver
+
+logger = get_logger()
+
+
+@register_sandbox(SandboxType.E2B)
+class E2BSandbox(Sandbox):
+    """Execute commands and transfer archives without a local Docker daemon."""
+
+    def __init__(
+        self,
+        config: E2BSandboxConfig,
+        sandbox_id: Optional[str] = None,
+        *,
+        connection: Optional[E2BConnection] = None,
+        templates: Optional[E2BTemplateResolver] = None,
+        run_id: Optional[str] = None,
+    ) -> None:
+        require_e2b()
+        super().__init__(config, sandbox_id)
+        self.config: E2BSandboxConfig = config
+        self._conn = connection
+        self._templates = templates
+        self._run_id = run_id or uuid.uuid4().hex[:12]
+        self._sbx: Any = None
+        self._started = 0.0
+
+    @property
+    def sandbox_type(self) -> SandboxType:
+        """Return the registered E2B type."""
+        return SandboxType.E2B
+
+    async def start(self) -> None:
+        """Create the remote sandbox with a fixed lifetime and run metadata."""
+        if self._conn is None:
+            cfg = E2BSandboxManagerConfig()
+            self._conn = E2BConnection.from_config(cfg)
+        if self._templates is None:
+            self._templates = E2BTemplateResolver(self._conn, {}, 4)
+        self.update_status(SandboxStatus.INITIALIZING)
+        self._started = time.monotonic()
+        template = await self._templates.resolve(self.config, run_id=self._run_id)
+        try:
+            self._sbx = await retry_dispatch(
+                'create sandbox',
+                lambda: AsyncSandbox.create(
+                    template=template,
+                    timeout=self.config.sandbox_timeout,
+                    metadata={'created_by': 'ms-enclave', 'run_id': self._run_id},
+                    allow_internet_access=self.config.allow_internet_access,
+                    **self._conn.api_params(retries=0),
+                ),
+                run_id=self._run_id,
+                redactor=self._conn.redact,
+            )
+        except Exception as exc:
+            self.update_status(SandboxStatus.ERROR)
+            message = (
+                f'Failed to create E2B sandbox (template={template}, run_id={self._run_id}, '
+                f'sandbox_timeout={self.config.sandbox_timeout}): {exc}'
+            )
+            if 'cannot be greater than' in str(exc):
+                message += (
+                    '; sandbox_timeout may exceed the plan limit (e2b official Hobby: 3600s); '
+                    'lower sandbox.default_config.sandbox_timeout or upgrade the plan.'
+                )
+            raise RuntimeError(self._conn.redact(message)) from None
+        self.metadata.update(remote_id=self._sbx.sandbox_id, template=template, run_id=self._run_id)
+        await self.initialize_tools()
+        self.update_status(SandboxStatus.RUNNING)
+        logger.info(
+            f'E2B sandbox {self.id} remote={self._sbx.sandbox_id} template={template} '
+            f'run_id={self._run_id} elapsed={time.monotonic() - self._started:.1f}s'
+        )
+
+    async def stop(self) -> None:
+        """Kill once, removing the local reference even if the service rejects kill."""
+        remote, self._sbx = self._sbx, None
+        if remote is not None:
+            try:
+                await remote.kill(**self._conn.api_params(retries=0))
+            except Exception as exc:
+                logger.warning(
+                    self._conn.redact(f'E2B sandbox {self.id} remote={remote.sandbox_id}: kill failed: {exc}')
+                )
+            logger.info(
+                f'E2B sandbox {self.id} stopped run_id={self._run_id} lifetime={time.monotonic() - self._started:.1f}s'
+            )
+        self.update_status(SandboxStatus.STOPPED)
+
+    async def cleanup(self) -> None:
+        """Destroy the sandbox."""
+        await self.stop()
+
+    async def get_execution_context(self) -> Any:
+        """Return the underlying SDK sandbox."""
+        return self._sbx
+
+    def _result(
+        self,
+        command: Union[str, List[str]],
+        status: ExecutionStatus,
+        code: int,
+        stdout: str,
+        stderr: str,
+        started: float,
+    ) -> CommandResult:
+        return CommandResult(
+            command=command,
+            status=status,
+            exit_code=code,
+            stdout=stdout,
+            stderr=stderr,
+            execution_time=time.monotonic() - started,
+        )
+
+    async def execute_command(
+        self,
+        command: Union[str, List[str]],
+        timeout: Optional[int] = None,
+        stream: bool = True,
+        *,
+        user: Optional[str] = None,
+        cwd: Optional[str] = None,
+        env: Optional[Dict[str, str]] = None,
+    ) -> CommandResult:
+        """Run once, preserve command output and kill the process on local timeout."""
+        if self._sbx is None:
+            raise RuntimeError('E2B sandbox is not running')
+        cmd = command if isinstance(command, str) else shlex.join(command)
+        started = time.monotonic()
+        try:
+            handle = await self._sbx.commands.run(
+                cmd,
+                background=True,
+                user=user if user is not None else self.config.user,
+                cwd=cwd if cwd is not None else self.config.working_dir,
+                envs={**self.config.env_vars, **(env or {})} or None,
+                timeout=0,
+            )
+        except Exception as exc:
+            return self._result(command, ExecutionStatus.ERROR, -1, '', self._conn.redact(str(exc)), started)
+        try:
+            result = await asyncio.wait_for(handle.wait(), timeout=timeout if timeout and timeout > 0 else None)
+            return self._result(command, ExecutionStatus.SUCCESS, 0, result.stdout, result.stderr, started)
+        except CommandExitException as exc:
+            return self._result(command, ExecutionStatus.ERROR, exc.exit_code, exc.stdout, exc.stderr, started)
+        except asyncio.TimeoutError:
+            try:
+                await handle.kill()
+            except Exception as exc:
+                logger.warning(self._conn.redact(f'E2B sandbox {self.id}: process kill failed: {exc}'))
+            stderr = handle.stderr or f'Command timed out after {timeout} seconds'
+            return self._result(command, ExecutionStatus.TIMEOUT, -1, handle.stdout, stderr, started)
+        except Exception as exc:
+            return self._result(command, ExecutionStatus.ERROR, -1, '', self._conn.redact(str(exc)), started)
+
+    async def put_archive(self, target_dir: str, data: bytes) -> bool:
+        """Upload a tar archive, extract as the configured user, and remove it."""
+        if not posixpath.isabs(target_dir):
+            raise ValueError(f'target_dir must be an absolute path: {target_dir}')
+        if self._sbx is None:
+            raise RuntimeError('E2B sandbox is not running')
+        tmp = f'/tmp/ms-enclave-upload-{uuid.uuid4().hex}.tar'
+        try:
+            await self._sbx.files.write(tmp, data, user=self.config.user)
+        except Exception as exc:
+            logger.error(self._conn.redact(f'E2B sandbox {self.id}: upload to {target_dir} failed: {exc}'))
+            return False
+        q = shlex.quote
+        result = await self.execute_command(
+            f'mkdir -p {q(target_dir)} && tar -xf {q(tmp)} -C {q(target_dir)}; rc=$?; rm -f {q(tmp)}; exit $rc',
+            timeout=300,
+        )
+        if result.exit_code != 0:
+            logger.error(
+                self._conn.redact(f'E2B sandbox {self.id}: extracting into {target_dir} failed: {result.stderr}')
+            )
+            return False
+        return True