diff --git a/.env.example b/.env.example index 971cf8b2b..c71296a8b 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,10 @@ INFOQUEST_API_KEY=your-infoquest-api-key # MINIMAX_API_KEY=your-minimax-api-key # OpenAI-compatible, see https://platform.minimax.io # STEPFUN_API_KEY=your-stepfun-api-key # OpenAI-compatible, see https://platform.stepfun.com # VLLM_API_KEY=your-vllm-api-key # OpenAI-compatible + +# E2B cloud sandbox API key — required only when using E2BSandboxProvider. +# Sign up at https://e2b.dev/dashboard +# E2B_API_KEY=your-e2b-api-key # FEISHU_APP_ID=your-feishu-app-id # FEISHU_APP_SECRET=your-feishu-app-secret diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 2a9d908e6..dc4162df6 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -274,6 +274,45 @@ When using Docker development (`make docker-start`), DeerFlow starts the `provis See [Provisioner Setup Guide](../../docker/provisioner/README.md) for detailed configuration, prerequisites, and troubleshooting. +**E2B Cloud Sandbox** (runs sandbox code in [E2B](https://e2b.dev) cloud micro-VMs): + +```yaml +sandbox: + use: deerflow.community.e2b_sandbox:E2BSandboxProvider + api_key: $E2B_API_KEY # required; or set the E2B_API_KEY env var + template: code-interpreter-v1 # e2b sandbox template id + # domain: e2b.dev # optional; for self-hosted e2b deployments + home_dir: /home/user # /mnt/user-data is remapped under this directory + idle_timeout: 600 # forwarded to e2b's server-side set_timeout() + replicas: 3 # max concurrent sandboxes per gateway process + mounts: # one-shot upload of host files at sandbox start + - host_path: /path/on/host + container_path: /home/user/shared + read_only: false + environment: # forwarded to the sandbox at create time + OPENAI_API_KEY: $OPENAI_API_KEY +``` + +`e2b-code-interpreter` is bundled as a core dependency of `deerflow-harness`, +so no extra install step is needed; just supply your API key and switch the +provider in `config.yaml`. + +Notes specific to `E2BSandboxProvider`: + +- Each DeerFlow thread is bound to its e2b sandbox via metadata + (`deer_flow_user`, `deer_flow_thread`), so the same thread reuses the same + sandbox across gateway restarts and across processes — no cross-process + file lock is needed because the e2b control plane is the source of truth. +- Idle expiry is enforced server-side by e2b's `set_timeout()`. The provider + refreshes the timeout on every release so warm sandboxes stay alive long + enough for the next acquire. +- `mounts` are uploaded once when the sandbox starts; e2b cannot host bind-mount + the gateway filesystem, so changes inside the sandbox are not reflected back + on disk automatically. Use the `download_file` tool or write outputs under + `/mnt/user-data/outputs/` (which is mapped to `home_dir/outputs/` inside the + sandbox and surfaced through the standard artifact pipeline) to ship files + back to the gateway. + Choose between local execution or Docker-based isolation: **Option 1: Local Sandbox** (default, simpler setup): diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py b/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py new file mode 100644 index 000000000..1a92f8eb6 --- /dev/null +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py @@ -0,0 +1,30 @@ +"""E2B cloud sandbox provider for DeerFlow. + +This package implements DeerFlow's :class:`Sandbox` / :class:`SandboxProvider` +contract on top of the `e2b` / `e2b_code_interpreter` cloud sandbox SDK. + +Configuration example (``config.yaml``):: + + sandbox: + use: deerflow.community.e2b_sandbox:E2BSandboxProvider + # E2B specific options (read via SandboxConfig's ``extra="allow"``): + api_key: $E2B_API_KEY # falls back to E2B_API_KEY env var + template: code-interpreter-v1 # e2b template id; defaults to e2b code-interpreter + domain: e2b.dev # optional e2b domain (e.g. self-hosted) + idle_timeout: 600 # forwarded to e2b ``set_timeout`` (seconds) + replicas: 3 # max concurrent sandboxes (LRU eviction beyond) + mounts: # one-shot upload of host files into the sandbox + - host_path: /path/on/host + container_path: /path/in/sandbox + read_only: false + environment: # forwarded as e2b ``envs`` on create + OPENAI_API_KEY: $OPENAI_API_KEY +""" + +from .e2b_sandbox import E2BSandbox +from .e2b_sandbox_provider import E2BSandboxProvider + +__all__ = [ + "E2BSandbox", + "E2BSandboxProvider", +] diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py new file mode 100644 index 000000000..21619dc4d --- /dev/null +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py @@ -0,0 +1,453 @@ +from __future__ import annotations + +import errno +import logging +import re +import shlex +import threading + +from e2b_code_interpreter import Sandbox as E2BClientSandbox + +from deerflow.config.paths import VIRTUAL_PATH_PREFIX +from deerflow.sandbox.sandbox import Sandbox +from deerflow.sandbox.search import GrepMatch, path_matches, should_ignore_path, truncate_line + +logger = logging.getLogger(__name__) + +_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024 # 100 MB + +# Where DeerFlow's ``/mnt/user-data`` virtual prefix is materialised inside +# the e2b sandbox. e2b code-interpreter templates default to ``/home/user`` +# as the working directory. +DEFAULT_E2B_HOME_DIR = "/home/user" + +_E2B_NOT_FOUND_SIGNATURES = ( + "sandbox was not found", + "sandbox not found", + "paused sandbox", +) + + +def _is_sandbox_gone_error(exc: BaseException) -> bool: + msg = str(exc).lower() + return any(sig in msg for sig in _E2B_NOT_FOUND_SIGNATURES) + + +class E2BSandbox(Sandbox): + """DeerFlow Sandbox adapter that delegates to an e2b cloud sandbox. + + Args: + id: DeerFlow-side sandbox id (used as cache key in the provider). + client: A live ``e2b_code_interpreter.Sandbox`` (sync) instance. + The caller owns the connection and is responsible for ``kill()``; + this wrapper only calls ``close()`` on its host-side HTTP client + during release. + home_dir: Directory inside the sandbox that backs the + ``VIRTUAL_PATH_PREFIX`` (``/mnt/user-data``) prefix. Defaults to + :data:`DEFAULT_E2B_HOME_DIR`. + """ + + def __init__( + self, + id: str, + client: E2BClientSandbox, + *, + home_dir: str = DEFAULT_E2B_HOME_DIR, + ) -> None: + super().__init__(id) + self._client = client + self._home_dir = home_dir.rstrip("/") or "/" + self._lock = threading.Lock() + self._closed = False + self._dead = False + + # ── Properties / lifecycle ─────────────────────────────────────────── + + @property + def client(self) -> E2BClientSandbox: + return self._client + + @property + def home_dir(self) -> str: + return self._home_dir + + @property + def sandbox_id(self) -> str: + """e2b-side sandbox id (different from DeerFlow's ``self.id`` cache key).""" + return getattr(self._client, "sandbox_id", self.id) + + def close(self) -> None: + with self._lock: + if self._closed: + return + self._closed = True + client = self._client + self._client = None + + if client is None: + return + + for closer in ( + getattr(client, "close", None), + getattr(getattr(client, "_transport", None), "close", None), + ): + if callable(closer): + try: + closer() + except Exception as e: + logger.warning("Error closing E2BSandbox %s: %s", self.id, e) + return + + def _resolve_path(self, path: str) -> str: + """Map DeerFlow virtual paths into the e2b sandbox filesystem. + + ``VIRTUAL_PATH_PREFIX`` (``/mnt/user-data``) is rewritten under + :attr:`home_dir`, mirroring how ``LocalContainerBackend`` bind-mounts + the host workspace into the AIO container at ``/mnt/user-data``. + Other absolute paths are returned verbatim so the sandbox can reach + system directories (``/tmp``, ``/etc``, …) when needed. + """ + if not path: + raise ValueError("path must be a non-empty string") + normalised = path.replace("\\", "/") + for segment in normalised.split("/"): + if segment == "..": + raise PermissionError(f"Access denied: path traversal detected in '{path}'") + if normalised == VIRTUAL_PATH_PREFIX or normalised.startswith(f"{VIRTUAL_PATH_PREFIX}/"): + tail = normalised[len(VIRTUAL_PATH_PREFIX) :].lstrip("/") + return f"{self._home_dir}/{tail}".rstrip("/") if tail else self._home_dir + return normalised + + def execute_command(self, command: str) -> str: + """Execute a shell command via ``sandbox.commands.run``. + + Returns the combined stdout/stderr. + The lock serialises concurrent calls on the same instance + because the e2b SDK shares a single HTTP/2 connection per sandbox. + """ + with self._lock: + client = self._client + if client is None: + return "Error: sandbox client has been closed" + if self._dead: + return "Error: e2b sandbox has been reaped by the control plane (idle timeout or explicit pause). The provider will rebuild a fresh sandbox on the next tool call." + try: + result = client.commands.run(command) + stdout = getattr(result, "stdout", "") or "" + stderr = getattr(result, "stderr", "") or "" + exit_code = getattr(result, "exit_code", 0) + if stdout and stderr: + output = f"{stdout}\n{stderr}" + else: + output = stdout or stderr + if exit_code not in (0, None) and not output: + output = f"Command exited with code {exit_code}" + return output if output else "(no output)" + except Exception as e: + if _is_sandbox_gone_error(e): + self._dead = True + logger.error("Failed to execute command in e2b sandbox: %s", e) + return f"Error: {e}" + + @property + def is_dead(self) -> bool: + """Whether the underlying e2b VM is known to be reaped. + + Updated lazily by ``execute_command`` and the provider's ``ping`` / + bootstrap calls — there is no proactive heartbeat. Reading the value + does *not* round-trip to the API. + """ + with self._lock: + return self._dead + + def ping(self) -> bool: + """Cheap health check: returns False if the e2b VM has been reaped. + + Run as ``commands.run("true")`` so successful execution implies the + full HTTP path (auth + control plane + envd) is alive. Sets + ``_dead = True`` on the same "sandbox not found" signature + :func:`_is_sandbox_gone_error` recognises so subsequent calls + short-circuit. + """ + with self._lock: + if self._dead or self._client is None: + return False + client = self._client + try: + client.commands.run("true") + return True + except Exception as e: + if _is_sandbox_gone_error(e): + with self._lock: + self._dead = True + return False + logger.warning("e2b sandbox ping raised non-fatal error: %s", e) + return True + + def read_file(self, path: str) -> str: + resolved = self._resolve_path(path) + try: + content = self._client.files.read(resolved) + if isinstance(content, bytes): + return content.decode("utf-8", errors="replace") + return content if content is not None else "" + except Exception as e: + logger.error("Failed to read file %s in e2b sandbox: %s", resolved, e) + return f"Error: {e}" + + def download_file(self, path: str) -> bytes: + normalised = path.replace("\\", "/") + for segment in normalised.split("/"): + if segment == "..": + logger.error("Refused download due to path traversal: %s", path) + raise PermissionError(f"Access denied: path traversal detected in '{path}'") + + stripped_path = normalised.lstrip("/") + allowed_prefix = VIRTUAL_PATH_PREFIX.lstrip("/") + if stripped_path != allowed_prefix and not stripped_path.startswith(f"{allowed_prefix}/"): + logger.error( + "Refused download outside allowed directory: path=%s, allowed_prefix=%s", + path, + VIRTUAL_PATH_PREFIX, + ) + raise PermissionError(f"Access denied: path must be under '{VIRTUAL_PATH_PREFIX}': '{path}'") + + resolved = self._resolve_path(path) + # Prefer the streaming API so the 100 MB cap is enforced *before* the + # whole payload is buffered in the gateway process. ``format="bytes"`` + # is implemented by the e2b SDK as ``bytearray(r.content)`` — i.e. the + # entire file is materialised in memory before returning — which would + # let a multi-GB artifact OOM the shared gateway on hosted deployments. + # ``format="stream"`` returns a ``FileStreamReader`` (an + # ``Iterator[bytes]``) that owns its HTTP response and releases the + # pooled connection on exhaustion / close / error. + with self._lock: + client = self._client + if client is None: + raise RuntimeError("sandbox client has been closed") + try: + data = client.files.read(resolved, format="stream") + except TypeError: + try: + data = client.files.read(resolved, format="bytes") + except Exception as e: + logger.error("Failed to download file %s from e2b sandbox: %s", resolved, e) + raise OSError(f"Failed to download file '{path}' from sandbox: {e}") from e + except Exception as e: + logger.error("Failed to download file %s from e2b sandbox: %s", resolved, e) + raise OSError(f"Failed to download file '{path}' from sandbox: {e}") from e + + if data is None: + return b"" + + # Buffered fallbacks (bytes/bytearray/str): apply the cap up front so + # we still refuse oversize payloads even on this path. + if isinstance(data, (bytes, bytearray)): + if len(data) > _MAX_DOWNLOAD_SIZE: + raise OSError( + errno.EFBIG, + f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", + path, + ) + return bytes(data) + if isinstance(data, str): + encoded = data.encode("utf-8") + if len(encoded) > _MAX_DOWNLOAD_SIZE: + raise OSError( + errno.EFBIG, + f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", + path, + ) + return encoded + + chunks: list[bytes] = [] + total = 0 + close = getattr(data, "close", None) + try: + try: + for chunk in data: + if not chunk: + continue + chunk_bytes = chunk if isinstance(chunk, bytes) else bytes(chunk) + total += len(chunk_bytes) + if total > _MAX_DOWNLOAD_SIZE: + raise OSError( + errno.EFBIG, + f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", + path, + ) + chunks.append(chunk_bytes) + except OSError: + raise + except Exception as e: + logger.error("Failed to stream file %s from e2b sandbox: %s", resolved, e) + raise OSError(f"Failed to download file '{path}' from sandbox: {e}") from e + finally: + if callable(close): + try: + close() + except Exception: + pass + return b"".join(chunks) + + def list_dir(self, path: str, max_depth: int = 2) -> list[str]: + resolved = self._resolve_path(path) + with self._lock: + client = self._client + if client is None: + return [] + try: + result = client.commands.run(f"find {shlex.quote(resolved)} -maxdepth {int(max_depth)} \\( -type f -o -type d \\) 2>/dev/null | head -500") + output = getattr(result, "stdout", "") or "" + return [line.strip() for line in output.splitlines() if line.strip()] + except Exception as e: + logger.error("Failed to list_dir %s in e2b sandbox: %s", resolved, e) + return [] + + def write_file(self, path: str, content: str, append: bool = False) -> None: + resolved = self._resolve_path(path) + with self._lock: + client = self._client + if client is None: + raise RuntimeError("sandbox client has been closed") + try: + if append: + existing = "" + try: + existing = client.files.read(resolved) or "" + if isinstance(existing, bytes): + existing = existing.decode("utf-8", errors="replace") + except Exception: + existing = "" + content = (existing or "") + content + client.files.write(resolved, content) + except Exception as e: + logger.error("Failed to write file %s in e2b sandbox: %s", resolved, e) + raise + + def update_file(self, path: str, content: bytes) -> None: + resolved = self._resolve_path(path) + with self._lock: + client = self._client + if client is None: + raise RuntimeError("sandbox client has been closed") + try: + # e2b's ``files.write`` accepts either ``str`` or ``bytes`` — + # passing bytes preserves binary content losslessly. + client.files.write(resolved, content) + except Exception as e: + logger.error("Failed to update file %s in e2b sandbox: %s", resolved, e) + raise + + def glob( + self, + path: str, + pattern: str, + *, + include_dirs: bool = False, + max_results: int = 200, + ) -> tuple[list[str], bool]: + resolved = self._resolve_path(path) + types = "f,d" if include_dirs else "f" + with self._lock: + client = self._client + if client is None: + return [], False + try: + hard_limit = max(max_results * 4, max_results + 50) + cmd = f"find {shlex.quote(resolved)} \\( " + " -o ".join(f"-type {t}" for t in types.split(",")) + f" \\) -print 2>/dev/null | head -{hard_limit}" + result = client.commands.run(cmd) + output = getattr(result, "stdout", "") or "" + except Exception as e: + logger.error("Failed to glob in e2b sandbox: %s", e) + return [], False + + matches: list[str] = [] + root = resolved.rstrip("/") or "/" + root_prefix = root if root == "/" else f"{root}/" + for entry in output.splitlines(): + entry = entry.strip() + if not entry: + continue + if entry != root and not entry.startswith(root_prefix): + continue + if should_ignore_path(entry): + continue + rel_path = entry[len(root) :].lstrip("/") + if not rel_path: + continue + if path_matches(pattern, rel_path): + matches.append(entry) + if len(matches) >= max_results: + return matches, True + return matches, False + + def grep( + self, + path: str, + pattern: str, + *, + glob: str | None = None, + literal: bool = False, + case_sensitive: bool = False, + max_results: int = 100, + ) -> tuple[list[GrepMatch], bool]: + regex_source = re.escape(pattern) if literal else pattern + re.compile(regex_source, 0 if case_sensitive else re.IGNORECASE) + + resolved = self._resolve_path(path) + # Build a portable ``grep`` invocation: + # -r recursive, -n line numbers, -H always print filename, -I skip + # binary files, -E extended regex (or -F for literal/fixed strings). + flags = ["-r", "-n", "-H", "-I"] + if not case_sensitive: + flags.append("-i") + if literal: + flags.append("-F") + else: + flags.append("-E") + if glob is not None: + include_pattern = glob.split("/")[-1] or glob + flags.append(f"--include={include_pattern}") + + per_file_cap = max(max_results, 50) + total_cap = max(max_results * 4, max_results + 50) + flags.append(f"-m{per_file_cap}") + + cmd = "grep " + " ".join(flags) + f" -- {shlex.quote(regex_source)} {shlex.quote(resolved)} 2>/dev/null" + f" | head -{total_cap}" + + with self._lock: + client = self._client + if client is None: + return [], False + try: + result = client.commands.run(cmd) + output = getattr(result, "stdout", "") or "" + except Exception as e: + logger.error("Failed to grep in e2b sandbox: %s", e) + return [], False + + matches: list[GrepMatch] = [] + truncated = False + for raw in output.splitlines(): + try: + file_path, line_no_str, line_text = raw.split(":", 2) + except ValueError: + continue + try: + line_number = int(line_no_str) + except ValueError: + continue + if should_ignore_path(file_path): + continue + matches.append( + GrepMatch( + path=file_path, + line_number=line_number, + line=truncate_line(line_text), + ) + ) + if len(matches) >= max_results: + truncated = True + break + return matches, truncated diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py new file mode 100644 index 000000000..55b7f7873 --- /dev/null +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py @@ -0,0 +1,1084 @@ +"""``E2BSandboxProvider`` — DeerFlow :class:`SandboxProvider` for e2b cloud. + +Configuration is read from :class:`SandboxConfig` (which has +``extra="allow"``), so any keys below can appear under ``sandbox:`` in +``config.yaml`` even though they are not declared on the model: + +.. code-block:: yaml + + sandbox: + use: deerflow.community.e2b_sandbox:E2BSandboxProvider + api_key: $E2B_API_KEY # required (or via E2B_API_KEY env var) + template: code-interpreter-v1 # default: e2b code-interpreter template + domain: e2b.dev # optional; for self-hosted e2b + idle_timeout: 600 # forwarded to ``set_timeout`` + replicas: 3 # max concurrent sandboxes + mounts: # one-shot uploads on sandbox start + - host_path: /data/skills + container_path: /home/user/skills + read_only: true + environment: # forwarded as e2b ``envs`` on create + OPENAI_API_KEY: $OPENAI_API_KEY +""" + +from __future__ import annotations + +import asyncio +import atexit +import hashlib +import logging +import os +import shlex +import signal +import threading +import time +import uuid +from collections import OrderedDict +from pathlib import Path +from typing import Any + +from e2b_code_interpreter import Sandbox as E2BClientSandbox + +from deerflow.config import get_app_config +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.sandbox.sandbox import Sandbox +from deerflow.sandbox.sandbox_provider import SandboxProvider + +from .e2b_sandbox import DEFAULT_E2B_HOME_DIR, E2BSandbox, _is_sandbox_gone_error + +logger = logging.getLogger(__name__) + + +# ── Defaults ───────────────────────────────────────────────────────────── +DEFAULT_TEMPLATE = "code-interpreter-v1" # the public e2b code-interpreter template +DEFAULT_IDLE_TIMEOUT = 1800 # 30 minutes; passed to ``Sandbox.set_timeout``. +DEFAULT_REPLICAS = 3 +# Hard upper bound for ``set_timeout`` (e2b currently caps at 24h on the +# free plan; passing an excessive value is rejected by the control-plane). +MAX_E2B_TIMEOUT = 24 * 60 * 60 + +# Metadata keys we attach to every sandbox so we can discover ours via +# ``Sandbox.list(query={...})`` from any gateway process. +META_KEY_USER = "deer_flow_user" +META_KEY_THREAD = "deer_flow_thread" +META_KEY_PROVIDER = "deer_flow_provider" +META_VAL_PROVIDER = "e2b_sandbox_provider" + + +class E2BSandboxProvider(SandboxProvider): + """Sandbox provider backed by the e2b code-interpreter cloud SDK.""" + + # e2b sandboxes are remote: there is no shared host filesystem with the + # gateway, so the framework must explicitly sync uploaded files (the + # remote backend in AioSandboxProvider sets the same flag). + uses_thread_data_mounts = False + needs_upload_permission_adjustment = True + + # ── Construction & config ──────────────────────────────────────────── + + def __init__(self) -> None: + self._lock = threading.Lock() + # Active sandboxes, keyed by DeerFlow-side sandbox id (== e2b id). + self._sandboxes: dict[str, E2BSandbox] = {} + # (user_id, thread_id) -> sandbox id for fast in-process lookup. + self._thread_sandboxes: dict[tuple[str, str], str] = {} + # Per-(user,thread) lock to serialise acquire() against itself. + self._thread_locks: dict[tuple[str, str], threading.Lock] = {} + # Warm pool: released sandboxes whose remote micro-VM is still alive. + # ``OrderedDict`` maintains insertion / move_to_end order for LRU. + self._warm_pool: OrderedDict[str, tuple[str, float]] = OrderedDict() + self._shutdown_called = False + + self._config = self._load_config() + + atexit.register(self.shutdown) + self._register_signal_handlers() + + def _load_config(self) -> dict[str, Any]: + """Read e2b options off ``SandboxConfig`` (``extra="allow"``).""" + sandbox_config = get_app_config().sandbox + + def _opt(name: str, default: Any = None) -> Any: + return getattr(sandbox_config, name, default) + + api_key = _opt("api_key") or os.environ.get("E2B_API_KEY") + if not api_key: + logger.warning("E2BSandboxProvider: no api_key configured (set sandbox.api_key in config.yaml or the E2B_API_KEY environment variable). The SDK will fail on the first acquire() until this is provided.") + + idle_timeout = _opt("idle_timeout") + if idle_timeout is None: + idle_timeout = DEFAULT_IDLE_TIMEOUT + idle_timeout = max(0, min(int(idle_timeout), MAX_E2B_TIMEOUT)) + + replicas = _opt("replicas") + replicas = DEFAULT_REPLICAS if replicas is None else max(1, int(replicas)) + + return { + "api_key": api_key, + "template": _opt("template") or _opt("image") or DEFAULT_TEMPLATE, + "domain": _opt("domain"), + "home_dir": _opt("home_dir") or DEFAULT_E2B_HOME_DIR, + "idle_timeout": idle_timeout, + "replicas": replicas, + "mounts": _opt("mounts") or [], + "environment": self._resolve_env_vars(_opt("environment") or {}), + } + + @staticmethod + def _resolve_env_vars(env_config: dict[str, str]) -> dict[str, str]: + resolved: dict[str, str] = {} + for key, value in env_config.items(): + if isinstance(value, str) and value.startswith("$"): + resolved[key] = os.environ.get(value[1:], "") + else: + resolved[key] = "" if value is None else str(value) + return resolved + + def _get_sandbox_cls(self) -> type[E2BClientSandbox]: + """Return the e2b SDK Sandbox class.""" + return E2BClientSandbox + + # ── Identity helpers ──────────────────────────────────────────────── + + @staticmethod + def _effective_acquire_user_id(user_id: str | None) -> str: + return user_id or get_effective_user_id() + + @staticmethod + def _thread_key(thread_id: str, user_id: str) -> tuple[str, str]: + return (user_id, thread_id) + + @staticmethod + def _stable_seed(thread_id: str, user_id: str) -> str: + return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16] + + # ── Signal / shutdown handling ─────────────────────────────────────── + + def _register_signal_handlers(self) -> None: + try: + self._original_sigterm = signal.getsignal(signal.SIGTERM) + self._original_sigint = signal.getsignal(signal.SIGINT) + self._original_sighup = signal.getsignal(signal.SIGHUP) if hasattr(signal, "SIGHUP") else None + except (ValueError, OSError): + return + + def _handler(signum, frame): + self.shutdown() + if signum == signal.SIGTERM: + original = self._original_sigterm + elif hasattr(signal, "SIGHUP") and signum == signal.SIGHUP: + original = self._original_sighup + else: + original = self._original_sigint + if callable(original): + original(signum, frame) + elif original == signal.SIG_DFL: + signal.signal(signum, signal.SIG_DFL) + signal.raise_signal(signum) + + for sig_name in ("SIGTERM", "SIGINT", "SIGHUP"): + sig = getattr(signal, sig_name, None) + if sig is None: + continue + try: + signal.signal(sig, _handler) + except (ValueError, OSError): + logger.debug( + "Could not register %s handler (likely not running on main thread)", + sig_name, + ) + + def _get_thread_lock(self, thread_id: str, user_id: str) -> threading.Lock: + key = self._thread_key(thread_id, user_id) + with self._lock: + lock = self._thread_locks.get(key) + if lock is None: + lock = threading.Lock() + self._thread_locks[key] = lock + return lock + + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + effective_user_id = self._effective_acquire_user_id(user_id) + if thread_id: + with self._get_thread_lock(thread_id, effective_user_id): + return self._acquire_internal(thread_id, user_id=effective_user_id) + return self._acquire_internal(thread_id, user_id=effective_user_id) + + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + effective_user_id = self._effective_acquire_user_id(user_id) + return await asyncio.to_thread(self.acquire, thread_id, user_id=effective_user_id) + + def _acquire_internal(self, thread_id: str | None, *, user_id: str) -> str: + if thread_id: + cached = self._reuse_in_process_sandbox(thread_id, user_id=user_id) + if cached is not None: + return cached + + if thread_id: + reclaimed = self._reclaim_warm_pool_sandbox(thread_id, user_id=user_id) + if reclaimed is not None: + return reclaimed + + if thread_id: + discovered = self._discover_remote_sandbox(thread_id, user_id=user_id) + if discovered is not None: + return discovered + return self._create_sandbox(thread_id, user_id=user_id) + + def _reuse_in_process_sandbox(self, thread_id: str, *, user_id: str) -> str | None: + key = self._thread_key(thread_id, user_id) + with self._lock: + sid = self._thread_sandboxes.get(key) + if sid is None: + return None + sandbox = self._sandboxes.get(sid) + if sandbox is None: + # The mapping pointed at a dead entry — clean it up. + self._thread_sandboxes.pop(key, None) + return None + + # Drop the cached entry if the e2b VM has been reaped (control-plane + # idle-timeout, manual pause, etc.). We learn about this either via + # ``execute_command`` flipping ``is_dead`` from a previous tool call, + # or by an explicit ping below — without this check the agent loops + # for ever on "sandbox not found" errors before the next acquire + # finally rebuilds the sandbox. + if sandbox.is_dead or not sandbox.ping(): + logger.warning( + "In-process e2b sandbox %s is dead (reaped by e2b control plane); evicting cache so acquire() can rebuild a fresh sandbox", + sid, + ) + with self._lock: + self._sandboxes.pop(sid, None) + self._thread_sandboxes.pop(key, None) + try: + sandbox.close() + except Exception: + pass + return None + + try: + self._refresh_remote_timeout(sandbox.client) + except Exception as e: # pragma: no cover - defensive + logger.debug("Failed to refresh timeout on reuse: %s", e) + + logger.info( + "Reusing in-process e2b sandbox %s for user/thread %s/%s", + sid, + user_id, + thread_id, + ) + return sid + + def _reclaim_warm_pool_sandbox(self, thread_id: str, *, user_id: str) -> str | None: + key = self._thread_key(thread_id, user_id) + seed = self._stable_seed(thread_id, user_id) + with self._lock: + target_id = next( + (sid for sid, (s, _) in self._warm_pool.items() if s == seed), + None, + ) + if target_id is None: + return None + self._warm_pool.pop(target_id) + + sandbox_cls = self._get_sandbox_cls() + try: + client = self._reconnect_client(sandbox_cls, target_id) + except Exception as e: + logger.warning( + "Warm-pool e2b sandbox %s failed to reconnect, dropping: %s", + target_id, + e, + ) + return None + + # Verify the reconnected client actually corresponds to a live VM. + # ``Sandbox.connect`` succeeds for paused/expired sandboxes too on + # some SDK versions, but the very next command then fails with + # "sandbox not found" mid-tool-call. Pinging here moves that failure + # into the acquire path, where we cleanly fall back to creating a + # fresh sandbox. + if not self._client_alive(client): + logger.warning( + "Warm-pool e2b sandbox %s is no longer alive (reaped by control plane); dropping and falling back to create", + target_id, + ) + self._safe_close_client(client) + return None + + self._refresh_remote_timeout(client) + try: + self._bootstrap_sandbox_paths(client) + except Exception as e: + logger.debug("bootstrap on warm-pool reclaim failed: %s", e) + sandbox = E2BSandbox(id=target_id, client=client, home_dir=self._config["home_dir"]) + with self._lock: + self._sandboxes[target_id] = sandbox + self._thread_sandboxes[key] = target_id + logger.info( + "Reclaimed warm-pool e2b sandbox %s for user/thread %s/%s", + target_id, + user_id, + thread_id, + ) + return target_id + + def _discover_remote_sandbox(self, thread_id: str, *, user_id: str) -> str | None: + """Look for a running e2b sandbox tagged with this (user, thread). + + Other gateway processes (or this process before a restart) may have + created the sandbox already. e2b sandboxes survive across reconnects + as long as the server-side timeout has not fired. + """ + sandbox_cls = self._get_sandbox_cls() + seed = self._stable_seed(thread_id, user_id) + list_kwargs = self._common_kwargs() + try: + running = sandbox_cls.list( # type: ignore[attr-defined] + query={ + "metadata": { + META_KEY_PROVIDER: META_VAL_PROVIDER, + META_KEY_USER: user_id, + META_KEY_THREAD: thread_id, + } + }, + **list_kwargs, + ) + except TypeError: + try: + running = sandbox_cls.list( + metadata={ + META_KEY_PROVIDER: META_VAL_PROVIDER, + META_KEY_USER: user_id, + META_KEY_THREAD: thread_id, + }, + **list_kwargs, + ) + except Exception as e: + logger.debug("e2b Sandbox.list() unavailable, skipping discovery: %s", e) + return None + except Exception as e: + logger.debug( + "e2b Sandbox.list() raised while discovering thread %s: %s", + thread_id, + e, + ) + return None + + # Pick the first matching candidate; tolerate either ``SandboxInfo`` + # objects with ``sandbox_id`` or plain dicts. + # Normalise the return value of ``Sandbox.list()``: + # * Older SDKs (<= 1.x) returned a plain ``list[SandboxInfo]`` — directly iterable. + # * e2b-code-interpreter >= 2.x returns a ``SandboxPaginator`` exposing + # ``has_next: bool`` and ``next_items() -> list[SandboxInfo]`` instead + # of being iterable. Walking pages keeps discovery correct when the + # org has more sandboxes than fit in a single page. + def _iter_running(obj): + if obj is None: + return + if hasattr(obj, "next_items") and hasattr(obj, "has_next"): + for _ in range(50): + try: + page = obj.next_items() + except Exception as exc: + logger.debug("SandboxPaginator.next_items() failed: %s", exc) + return + if not page: + return + yield from page + if not getattr(obj, "has_next", False): + return + return + try: + yield from obj + except TypeError: + logger.debug("Sandbox.list() returned non-iterable %s; ignoring", type(obj).__name__) + + target_id: str | None = None + for entry in _iter_running(running): + sid = getattr(entry, "sandbox_id", None) or (entry.get("sandbox_id") if isinstance(entry, dict) else None) + metadata = getattr(entry, "metadata", None) or (entry.get("metadata") if isinstance(entry, dict) else {}) or {} + if metadata.get(META_KEY_USER) != user_id: + continue + if metadata.get(META_KEY_THREAD) != thread_id: + continue + target_id = sid + break + + if not target_id: + return None + + try: + client = self._reconnect_client(sandbox_cls, target_id) + except Exception as e: + logger.warning( + "Discovered e2b sandbox %s could not be reconnected: %s", + target_id, + e, + ) + return None + + if not self._client_alive(client): + logger.warning( + "Discovered e2b sandbox %s is no longer alive; falling back to create", + target_id, + ) + self._safe_close_client(client) + return None + + self._refresh_remote_timeout(client) + try: + self._bootstrap_sandbox_paths(client) + except Exception as e: + logger.debug("bootstrap on remote discovery failed: %s", e) + sandbox = E2BSandbox(id=target_id, client=client, home_dir=self._config["home_dir"]) + with self._lock: + self._sandboxes[target_id] = sandbox + self._thread_sandboxes[self._thread_key(thread_id, user_id)] = target_id + logger.info( + "Discovered remote e2b sandbox %s for user/thread %s/%s (seed=%s)", + target_id, + user_id, + thread_id, + seed, + ) + return target_id + + def _create_sandbox(self, thread_id: str | None, *, user_id: str) -> str: + """Allocate a fresh e2b sandbox and hydrate it with configured mounts.""" + replicas = int(self._config["replicas"]) + with self._lock: + in_use = len(self._sandboxes) + len(self._warm_pool) + if in_use >= replicas: + evicted = self._evict_oldest_warm() + if evicted is None: + logger.warning( + "All %d e2b replica slots are in active use; creating a new sandbox beyond the soft limit (active=%d, warm=%d)", + replicas, + len(self._sandboxes), + len(self._warm_pool), + ) + + sandbox_cls = self._get_sandbox_cls() + metadata: dict[str, str] = { + META_KEY_PROVIDER: META_VAL_PROVIDER, + } + if thread_id: + metadata[META_KEY_USER] = user_id + metadata[META_KEY_THREAD] = thread_id + + create_kwargs: dict[str, Any] = { + "template": self._config["template"], + "metadata": metadata, + **self._common_kwargs(), + } + if self._config["idle_timeout"] > 0: + create_kwargs["timeout"] = self._config["idle_timeout"] + if self._config["environment"]: + create_kwargs["envs"] = self._config["environment"] + + try: + client = sandbox_cls.create(**create_kwargs) # type: ignore[attr-defined] + except Exception as e: + logger.error("Failed to create e2b sandbox: %s", e) + raise + + sandbox_id: str = getattr(client, "sandbox_id", None) or str(uuid.uuid4())[:8] + + # Materialise DeerFlow's virtual path layout (/mnt/user-data/...) inside + # the e2b VM. Without this step shell commands the agent emits — which + # use the same /mnt/user-data prefix as LocalSandbox / AioSandbox — fail + # with PermissionError because /mnt is owned by root in the e2b + # template. See the path-mapping note in :class:`E2BSandbox`. + try: + self._bootstrap_sandbox_paths(client) + except Exception as e: + logger.warning( + "Failed to bootstrap virtual paths in e2b sandbox %s: %s", + sandbox_id, + e, + ) + + # One-shot mount uploads. e2b has no host bind-mount, so we copy + # files from ``host_path`` into ``container_path`` at sandbox start. + try: + self._apply_mounts(client) + except Exception as e: + logger.warning("Failed to apply some mounts to e2b sandbox %s: %s", sandbox_id, e) + + sandbox = E2BSandbox(id=sandbox_id, client=client, home_dir=self._config["home_dir"]) + with self._lock: + self._sandboxes[sandbox_id] = sandbox + if thread_id: + self._thread_sandboxes[self._thread_key(thread_id, user_id)] = sandbox_id + + logger.info( + "Created e2b sandbox %s for user/thread %s/%s (template=%s, replicas=%d)", + sandbox_id, + user_id, + thread_id, + self._config["template"], + replicas, + ) + return sandbox_id + + def _common_kwargs(self) -> dict[str, Any]: + """Kwargs shared by ``Sandbox.create``, ``Sandbox.connect`` and ``Sandbox.list``.""" + kwargs: dict[str, Any] = {} + if self._config["api_key"]: + kwargs["api_key"] = self._config["api_key"] + if self._config["domain"]: + kwargs["domain"] = self._config["domain"] + return kwargs + + def _reconnect_client(self, sandbox_cls: type[E2BClientSandbox], sandbox_id: str) -> E2BClientSandbox: + """Connect to an existing e2b sandbox by id, with consistent kwargs.""" + return sandbox_cls.connect(sandbox_id, **self._common_kwargs()) # type: ignore[attr-defined] + + def _refresh_remote_timeout(self, client: E2BClientSandbox) -> None: + """Push the configured idle timeout to the e2b control plane.""" + idle_timeout = int(self._config["idle_timeout"]) + if idle_timeout <= 0: + return + set_timeout = getattr(client, "set_timeout", None) + if not callable(set_timeout): + return + try: + set_timeout(idle_timeout) + except Exception as e: # pragma: no cover - defensive + logger.debug("Failed to set timeout on e2b sandbox: %s", e) + + @staticmethod + def _client_alive(client: E2BClientSandbox) -> bool: + """Best-effort liveness probe for a freshly reconnected e2b client. + + ``Sandbox.connect`` may succeed against a paused/expired sandbox on + some SDK versions — the failure only surfaces on the first command. + We send a trivial ``true`` shell command here so the failure happens + in the acquire path (where we can transparently fall through to + creating a fresh sandbox) instead of mid-tool-call (where the agent + would see a confusing "sandbox not found" stack trace). + + Returns ``True`` if the command succeeds, ``False`` if it raises a + "sandbox not found / paused" error. Other transient errors are + treated as alive so a single network blip does not nuke the cache. + """ + try: + client.commands.run("true") + return True + except Exception as e: + if _is_sandbox_gone_error(e): + return False + logger.debug("e2b client liveness probe non-fatal error: %s", e) + return True + + @staticmethod + def _safe_close_client(client: E2BClientSandbox | None) -> None: + """Close the host-side HTTP client of *client* without ever raising. + + Used in cleanup paths where we already know the e2b VM is unreachable + (paused/expired) and we just want to release sockets in the gateway + process. Any exception is logged at debug level and swallowed. + """ + if client is None: + return + for attr in ("close", "_transport"): + target = getattr(client, attr, None) + if target is None: + continue + close = target if callable(target) else getattr(target, "close", None) + if not callable(close): + continue + try: + close() + return + except Exception as e: # pragma: no cover - defensive + logger.debug("e2b client close raised: %s", e) + return + + def _bootstrap_sandbox_paths(self, client: E2BClientSandbox) -> None: + """Materialise DeerFlow's virtual path layout inside the e2b VM. + + The local / docker sandboxes expose ``/mnt/user-data/{workspace,uploads, + outputs}`` and ``/mnt/acp-workspace`` as writable directories, and the + agent prompts (and the lead-agent system prompt in particular) instruct + the model to write outputs there. e2b's default ``code-interpreter`` + template runs as the unprivileged ``user`` (uid 1000) with ``/mnt`` + owned by ``root``, so any ``mkdir -p /mnt/user-data/...`` issued by + the agent fails with ``Permission denied``. + + We fix that once at sandbox start by: + + 1. Creating ``/home/user/{workspace,uploads,outputs}`` as the real, + writable backing directories (they live under the agent's HOME so + there is no permission issue). + 2. Symlinking ``/mnt/user-data`` to ``/home/user`` and + ``/mnt/acp-workspace`` to ``/home/user/acp-workspace`` via ``sudo``, + so commands using the documented ``/mnt/...`` paths "just work" and + land in the same physical location :class:`E2BSandbox._resolve_path` + already remaps to. + 3. Chowning the symlinks (and ``/mnt`` itself if needed) so subsequent + writes through the symlink target succeed. + + The e2b code-interpreter template puts ``user`` in the ``sudo`` group + with passwordless sudo, so the ``sudo`` calls below succeed without + interactive prompts. If the customer template removes that, the + commands fail loudly here and we fall back to silently relying on the + path remap inside ``E2BSandbox`` — agent shell commands will still + fail, but the read/write/list APIs continue to work. + """ + # Use the configured ``home_dir`` so a custom template can move HOME. + home_dir = self._config["home_dir"].rstrip("/") or "/home/user" + bootstrap_script = ( + f"set -e; " + f"mkdir -p {shlex.quote(home_dir)}/workspace " + f"{shlex.quote(home_dir)}/uploads " + f"{shlex.quote(home_dir)}/outputs " + f"{shlex.quote(home_dir)}/acp-workspace; " + # /mnt/user-data -> $home_dir + f"if [ ! -e /mnt/user-data ] || [ -L /mnt/user-data ]; then " + f" sudo ln -sfn {shlex.quote(home_dir)} /mnt/user-data; " + f"fi; " + # /mnt/acp-workspace -> $home_dir/acp-workspace + f"if [ ! -e /mnt/acp-workspace ] || [ -L /mnt/acp-workspace ]; then " + f" sudo ln -sfn {shlex.quote(home_dir)}/acp-workspace /mnt/acp-workspace; " + f"fi; " + # /mnt/skills is left alone here; the optional ``mounts`` config + # uploads its content via _apply_mounts and creates the directory + # on demand. We only ensure that /mnt itself is traversable. + f"sudo chmod a+rx /mnt 2>/dev/null || true; " + f"echo BOOTSTRAP_OK" + ) + + try: + result = client.commands.run(bootstrap_script) + except Exception as e: + logger.warning( + "e2b bootstrap script raised: %s (agent shell commands using /mnt/user-data may fail until the VM is recycled)", + e, + ) + return + + stdout = getattr(result, "stdout", "") or "" + stderr = getattr(result, "stderr", "") or "" + exit_code = getattr(result, "exit_code", 0) + if exit_code not in (0, None) or "BOOTSTRAP_OK" not in stdout: + logger.warning( + "e2b bootstrap script exited with code=%s; stderr=%s", + exit_code, + stderr.strip(), + ) + + def _apply_mounts(self, client: E2BClientSandbox) -> None: + mounts = self._config.get("mounts") or [] + if not mounts: + return + for mount in mounts: + try: + host_path = Path(getattr(mount, "host_path", "") or "") + container_path = (getattr(mount, "container_path", "") or "").rstrip("/") + read_only = bool(getattr(mount, "read_only", False)) + except AttributeError: + host_path = Path(mount.get("host_path", "")) + container_path = (mount.get("container_path", "") or "").rstrip("/") + read_only = bool(mount.get("read_only", False)) + + if not host_path.exists(): + logger.warning("Skipping e2b mount: host_path %s does not exist", host_path) + continue + if not container_path.startswith("/"): + logger.warning( + "Skipping e2b mount: container_path %s must be absolute", + container_path, + ) + continue + + try: + make_dir = getattr(client.files, "make_dir", None) + if callable(make_dir): + make_dir(container_path) + except Exception as e: + logger.debug("make_dir(%s) failed (continuing): %s", container_path, e) + + try: + self._upload_tree(client, host_path, container_path, read_only) + except Exception as e: + logger.warning("Failed to upload mount %s -> %s: %s", host_path, container_path, e) + + # ── Output mirroring ──────────────────────────────────────────────── + _SYNC_BACK_SUBDIRS = ("outputs", "workspace") + + def _sync_outputs_to_host( + self, + sandbox: E2BSandbox, + *, + thread_id: str, + user_id: str, + ) -> None: + """Mirror agent artifacts from the e2b VM back to host thread dirs. + + DeerFlow's ``/api/threads/{tid}/artifacts/...`` endpoint resolves + files against the host-side per-thread ``user-data/`` tree (see + :meth:`Paths.sandbox_outputs_dir`). LocalSandbox writes there + directly via path mappings, so the endpoint just works for the + local provider. The e2b VM has no shared host filesystem, so we + explicitly pull artifacts back at release time. + + We only mirror files whose host-side counterpart is missing or has a + different size — this gives an effective per-file dedup with a single + round-trip per release for unchanged trees, and avoids re-downloading + large generated files (e.g. PDFs, datasets) on every tool turn that + triggers a release. + + Failures are logged at WARNING level but never raised: artifact + download is non-critical for sandbox lifecycle, and we already log + the underlying e2b SDK errors elsewhere. + """ + from deerflow.config.paths import get_paths # lazy import to avoid cycles + + client = sandbox.client + if client is None: + logger.debug("Skip output sync: e2b client already closed for sandbox %s", sandbox.id) + return + + home_dir = sandbox.home_dir.rstrip("/") or "/home/user" + paths = get_paths() + + thread_root = paths.thread_dir(thread_id, user_id=user_id) / "user-data" + host_targets: dict[str, Path] = {sub: thread_root / sub for sub in self._SYNC_BACK_SUBDIRS} + + # Build a single shell command that lists all files in the sync dirs + # with size + path, NUL-separated for safe parsing of weird filenames. + # find -printf '%s\t%p\0' keeps us to one round-trip regardless of + # how many subdirs we mirror. + # + # We list using the *physical* /home/user paths (the bootstrap symlink + # /mnt/user-data -> /home/user follows transparently), then translate + # each hit back to the /mnt/user-data prefix before calling + # ``E2BSandbox.download_file``: that method enforces a security check + # that the path is under ``VIRTUAL_PATH_PREFIX`` (/mnt/user-data) and + # internally re-resolves it to /home/user via ``_resolve_path``. + find_targets = " ".join(shlex.quote(f"{home_dir}/{sub}") for sub in self._SYNC_BACK_SUBDIRS) + list_cmd = f'for d in {find_targets}; do [ -d "$d" ] && find "$d" -type f -printf \'%s\\t%p\\0\' 2>/dev/null; done' + + try: + result = client.commands.run(list_cmd) + except Exception as e: + logger.warning("e2b sync: list command failed: %s", e) + if _is_sandbox_gone_error(e): + with sandbox._lock: + sandbox._dead = True + return + + stdout = getattr(result, "stdout", "") or "" + if not stdout: + return + + synced = 0 + skipped = 0 + from .e2b_sandbox import _MAX_DOWNLOAD_SIZE + + for entry in stdout.split("\0"): + entry = entry.strip() + if not entry: + continue + try: + size_str, remote_path = entry.split("\t", 1) + remote_size = int(size_str) + except ValueError: + logger.debug("e2b sync: unparseable entry %r", entry) + continue + + if remote_size > _MAX_DOWNLOAD_SIZE: + logger.warning( + "e2b sync: skipping oversize artefact %s (%d bytes > %d cap)", + remote_path, + remote_size, + _MAX_DOWNLOAD_SIZE, + ) + skipped += 1 + continue + + # Determine which subdir this file belongs to so we can compute + # the relative path on the host side. remote_path is absolute, + # e.g. /home/user/outputs/foo/bar.pdf + sub_match: tuple[str, Path, str] | None = None + for sub, host_root in host_targets.items(): + prefix = f"{home_dir}/{sub}/" + if remote_path == f"{home_dir}/{sub}": + continue + if remote_path.startswith(prefix): + rel = remote_path[len(prefix) :] + virtual_path = f"/mnt/user-data/{sub}/{rel}" + sub_match = (sub, host_root / rel, virtual_path) + break + if sub_match is None: + continue + _sub, host_path, virtual_path = sub_match + + try: + if host_path.exists() and host_path.stat().st_size == remote_size: + skipped += 1 + continue + except OSError: + pass + + try: + data = sandbox.download_file(virtual_path) + except Exception as e: + logger.warning( + "e2b sync: failed to download %s from sandbox %s: %s", + virtual_path, + sandbox.id, + e, + ) + continue + + try: + host_path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = host_path.with_name(host_path.name + ".e2bsync.tmp") + tmp_path.write_bytes(data) + tmp_path.replace(host_path) + synced += 1 + except OSError as e: + logger.warning("e2b sync: failed to write %s on host: %s", host_path, e) + + if synced or skipped: + logger.info( + "e2b sync: sandbox=%s thread=%s synced=%d skipped=%d", + sandbox.id, + thread_id, + synced, + skipped, + ) + + @staticmethod + def _upload_tree( + client: E2BClientSandbox, + src: Path, + dest_dir: str, + read_only: bool, + ) -> None: + """Recursively upload ``src`` into ``dest_dir`` inside the sandbox.""" + if src.is_file(): + target = f"{dest_dir}/{src.name}" + with src.open("rb") as fh: + client.files.write(target, fh.read()) + if read_only: + try: + client.commands.run(f"chmod a-w {shlex.quote(target)}") + except Exception: + pass + return + + for path in src.rglob("*"): + if not path.is_file(): + continue + rel = path.relative_to(src).as_posix() + target = f"{dest_dir}/{rel}" + try: + make_dir = getattr(client.files, "make_dir", None) + if callable(make_dir): + parent = target.rsplit("/", 1)[0] + if parent and parent != dest_dir: + make_dir(parent) + except Exception: + pass + with path.open("rb") as fh: + client.files.write(target, fh.read()) + if read_only: + try: + client.commands.run(f"chmod -R a-w {shlex.quote(dest_dir)}") + except Exception: + pass + + def _evict_oldest_warm(self) -> str | None: + with self._lock: + if not self._warm_pool: + return None + evict_id, (_, _) = self._warm_pool.popitem(last=False) + + try: + client = self._reconnect_client(self._get_sandbox_cls(), evict_id) + except Exception as e: + logger.warning( + "Evicted warm-pool e2b sandbox %s could not be reconnected for kill: %s", + evict_id, + e, + ) + return evict_id + + try: + kill = getattr(client, "kill", None) + if callable(kill): + kill() + except Exception as e: + logger.warning("Failed to kill evicted e2b sandbox %s: %s", evict_id, e) + finally: + close = getattr(client, "close", None) + if callable(close): + try: + close() + except Exception: + pass + logger.info("Evicted warm-pool e2b sandbox %s", evict_id) + return evict_id + + def get(self, sandbox_id: str) -> Sandbox | None: + with self._lock: + return self._sandboxes.get(sandbox_id) + + def release(self, sandbox_id: str) -> None: + """Park a sandbox in the warm pool while keeping the cloud VM alive. + + e2b sandboxes have a server-enforced timeout — we refresh it here so + the warm-pool entry stays valid for at least one ``idle_timeout`` + window after release. + """ + sandbox: E2BSandbox | None = None + seed: str | None = None + + with self._lock: + sandbox = self._sandboxes.pop(sandbox_id, None) + # Find the (user, thread) the sandbox was bound to. + removed_keys = [key for key, sid in self._thread_sandboxes.items() if sid == sandbox_id] + for key in removed_keys: + self._thread_sandboxes.pop(key, None) + if removed_keys: + user_id, thread_id = removed_keys[0] + seed = self._stable_seed(thread_id, user_id) + + if sandbox is None: + return + + if sandbox.is_dead: + logger.info( + "Releasing dead e2b sandbox %s; skipping output sync and warm pool, killing remote VM", + sandbox_id, + ) + self._kill_and_close(sandbox) + return + + sync_failed_due_to_dead_vm = False + if seed is not None and removed_keys: + user_id_sync, thread_id_sync = removed_keys[0] + try: + self._sync_outputs_to_host(sandbox, thread_id=thread_id_sync, user_id=user_id_sync) + except Exception as e: # pragma: no cover - defensive + logger.warning( + "Failed to mirror e2b sandbox %s outputs to host: %s", + sandbox_id, + e, + ) + if sandbox.is_dead: + sync_failed_due_to_dead_vm = True + + if sync_failed_due_to_dead_vm: + logger.info( + "Sandbox %s was reaped during release; not parking in warm pool", + sandbox_id, + ) + self._kill_and_close(sandbox) + return + + try: + self._refresh_remote_timeout(sandbox.client) + except Exception as e: + logger.debug("Failed to refresh timeout during release: %s", e) + + try: + sandbox.close() + except Exception as e: + logger.warning("Error closing e2b sandbox %s during release: %s", sandbox_id, e) + + with self._lock: + self._warm_pool[sandbox_id] = (seed or "", time.time()) + self._warm_pool.move_to_end(sandbox_id) + logger.info("Released e2b sandbox %s to warm pool", sandbox_id) + + def _kill_and_close(self, sandbox: E2BSandbox) -> None: + client = getattr(sandbox, "_client", None) + if client is not None: + kill = getattr(client, "kill", None) + if callable(kill): + try: + kill() + except Exception as e: + logger.debug( + "kill() on e2b sandbox %s raised (probably already gone): %s", + sandbox.id, + e, + ) + try: + sandbox.close() + except Exception: + pass + + def reset(self) -> None: + with self._lock: + self._sandboxes.clear() + self._thread_sandboxes.clear() + self._thread_locks.clear() + self._warm_pool.clear() + + def shutdown(self) -> None: + with self._lock: + if self._shutdown_called: + return + self._shutdown_called = True + active = list(self._sandboxes.items()) + warm_ids = list(self._warm_pool.keys()) + self._sandboxes.clear() + self._warm_pool.clear() + self._thread_sandboxes.clear() + + logger.info( + "Shutting down E2BSandboxProvider: %d active + %d warm sandboxes", + len(active), + len(warm_ids), + ) + + for sandbox_id, sandbox in active: + try: + kill = getattr(sandbox.client, "kill", None) + if callable(kill): + kill() + except Exception as e: + logger.warning( + "Failed to kill active e2b sandbox %s during shutdown: %s", + sandbox_id, + e, + ) + try: + sandbox.close() + except Exception: + pass + + sandbox_cls = self._get_sandbox_cls() + for sandbox_id in warm_ids: + try: + client = self._reconnect_client(sandbox_cls, sandbox_id) + except Exception as e: + logger.warning( + "Failed to reconnect warm-pool e2b sandbox %s for shutdown: %s", + sandbox_id, + e, + ) + continue + try: + kill = getattr(client, "kill", None) + if callable(kill): + kill() + except Exception as e: + logger.warning( + "Failed to kill warm-pool e2b sandbox %s during shutdown: %s", + sandbox_id, + e, + ) + close = getattr(client, "close", None) + if callable(close): + try: + close() + except Exception: + pass diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 20e3a9f42..15bbc4a71 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -37,6 +37,7 @@ dependencies = [ "aiosqlite>=0.19", "alembic>=1.13", "cryptography>=48.0.1", + "e2b-code-interpreter>=2.8.0", ] [project.scripts] diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 0b4ca357a..6c086d8ee 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -21,6 +21,7 @@ dependencies = [ "bcrypt>=4.0.0", "pyjwt>=2.13.0", "email-validator>=2.0.0", + "e2b-code-interpreter>=2.8.1", ] [project.optional-dependencies] diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py new file mode 100644 index 000000000..1c5af376a --- /dev/null +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -0,0 +1,779 @@ +"""Unit tests for ``E2BSandboxProvider`` and its companion ``E2BSandbox``.""" + +from __future__ import annotations + +import importlib +import threading +from collections import OrderedDict +from types import SimpleNamespace +from typing import Any + +from deerflow.config.paths import Paths + +# ────────────────────────────────────────────────────────────────────────────── +# Fakes for the e2b SDK +# ────────────────────────────────────────────────────────────────────────────── + + +class FakeCommandsAPI: + """Stand-in for ``client.commands``.""" + + GONE = "__GONE__" + NOT_FOUND_MSG = "The sandbox was not found: This error is likely due to sandbox timeout." + + def __init__(self, responses: list[Any] | None = None) -> None: + self.calls: list[str] = [] + self._responses = list(responses or []) + + def _next(self) -> Any: + if not self._responses: + return SimpleNamespace(stdout="BOOTSTRAP_OK", stderr="", exit_code=0) + head = self._responses.pop(0) + return head + + def run(self, cmd: str) -> SimpleNamespace: + self.calls.append(cmd) + head = self._next() + if head == self.GONE: + raise RuntimeError(self.NOT_FOUND_MSG) + if callable(head): + return head(cmd) + if isinstance(head, SimpleNamespace): + return head + return SimpleNamespace(stdout=str(head), stderr="", exit_code=0) + + +class _FakeFileStream: + """Minimal stand-in for ``e2b.FileStreamReader``. + + Yields fixed-size chunks and tracks whether ``close()`` was invoked so + tests can assert we release the connection on both success and abort. + """ + + def __init__(self, data: bytes, *, chunk_size: int = 4096) -> None: + self._data = bytes(data) + self._chunk_size = max(1, int(chunk_size)) + self._offset = 0 + self.closed = False + + def __iter__(self): + return self + + def __next__(self) -> bytes: + if self.closed or self._offset >= len(self._data): + raise StopIteration + end = min(self._offset + self._chunk_size, len(self._data)) + chunk = self._data[self._offset : end] + self._offset = end + return chunk + + def close(self) -> None: + self.closed = True + + def __enter__(self) -> _FakeFileStream: + return self + + def __exit__(self, *exc_info) -> None: + self.close() + + +class FakeFilesAPI: + def __init__( + self, + store: dict[str, bytes] | None = None, + *, + stream_chunk_size: int = 4096, + ) -> None: + self.store = dict(store or {}) + self.read_calls: list[tuple[str, str | None]] = [] + self.write_calls: list[tuple[str, bytes]] = [] + self.streams: list[_FakeFileStream] = [] + self._stream_chunk_size = stream_chunk_size + + def read(self, path: str, *, format: str | None = None): + self.read_calls.append((path, format)) + if path not in self.store: + raise FileNotFoundError(path) + data = self.store[path] + if format == "bytes": + return data + if format == "stream": + stream = _FakeFileStream(data, chunk_size=self._stream_chunk_size) + self.streams.append(stream) + return stream + try: + return data.decode("utf-8") + except UnicodeDecodeError: + return data + + def write(self, path: str, content: bytes) -> None: + self.write_calls.append((path, content)) + self.store[path] = content + + +class FakeClient: + """Lightweight ``e2b.Sandbox`` substitute used by the provider tests.""" + + def __init__( + self, + sandbox_id: str = "fake-sb-1", + *, + commands: FakeCommandsAPI | None = None, + files: FakeFilesAPI | None = None, + ) -> None: + self.sandbox_id = sandbox_id + self.commands = commands or FakeCommandsAPI() + self.files = files or FakeFilesAPI() + self.timeouts_set: list[int] = [] + self.killed = False + self.closed = False + + def set_timeout(self, seconds: int) -> None: + self.timeouts_set.append(int(seconds)) + + def kill(self) -> None: + self.killed = True + + def close(self) -> None: + self.closed = True + + +class FakeSandboxClass: + """Stand-in for ``e2b_code_interpreter.Sandbox`` (the class itself).""" + + def __init__(self) -> None: + self.create_calls: list[dict[str, Any]] = [] + self.connect_calls: list[tuple[str, dict[str, Any]]] = [] + self.list_calls: list[dict[str, Any]] = [] + self.create_factory = lambda **kw: FakeClient(sandbox_id=f"created-{len(self.create_calls)}") + self.connect_factory = lambda sid, **kw: FakeClient(sandbox_id=sid) + self.list_return: Any = [] + + def create(self, **kwargs: Any) -> FakeClient: + self.create_calls.append(kwargs) + return self.create_factory(**kwargs) + + def connect(self, sandbox_id: str, **kwargs: Any) -> FakeClient: + self.connect_calls.append((sandbox_id, kwargs)) + return self.connect_factory(sandbox_id, **kwargs) + + def list(self, **kwargs: Any) -> Any: + self.list_calls.append(kwargs) + return self.list_return + + +def _make_provider(*, replicas: int = 3, idle_timeout: int = 1800) -> Any: + """Build a ``E2BSandboxProvider`` instance bypassing ``__init__``.""" + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + provider = mod.E2BSandboxProvider.__new__(mod.E2BSandboxProvider) + provider._lock = threading.Lock() + provider._sandboxes = {} + provider._thread_sandboxes = {} + provider._thread_locks = {} + provider._warm_pool = OrderedDict() + provider._shutdown_called = False + provider._config = { + "api_key": "test-key", + "template": "code-interpreter-v1", + "domain": None, + "home_dir": "/home/user", + "idle_timeout": idle_timeout, + "replicas": replicas, + "mounts": [], + "environment": {}, + } + return provider + + +def _install_fake_sdk(monkeypatch, provider) -> FakeSandboxClass: + fake_cls = FakeSandboxClass() + monkeypatch.setattr(provider, "_get_sandbox_cls", lambda: fake_cls) + return fake_cls + + +def _make_sandbox(client: FakeClient, *, sandbox_id: str | None = None) -> Any: + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox") + return mod.E2BSandbox( + id=sandbox_id or client.sandbox_id, + client=client, + home_dir="/home/user", + ) + + +def test_thread_key_returns_user_thread_tuple(): + p = _make_provider() + assert p._thread_key("t1", "u1") == ("u1", "t1") + + +def test_stable_seed_is_deterministic_and_user_scoped(): + p = _make_provider() + s_a = p._stable_seed("t1", "u1") + s_b = p._stable_seed("t1", "u1") + s_other_user = p._stable_seed("t1", "u2") + s_other_thread = p._stable_seed("t2", "u1") + assert s_a == s_b + assert s_a != s_other_user + assert s_a != s_other_thread + + +def test_is_sandbox_gone_error_matches_known_signatures(): + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox") + f = mod._is_sandbox_gone_error + assert f(RuntimeError("Paused sandbox abcdef not found")) + assert f(Exception("The sandbox was not found: due to timeout")) + assert f(Exception("sandbox not found")) + # Unrelated errors must not flip the dead flag. + assert not f(Exception("Connection reset by peer")) + assert not f(ValueError("invalid path")) + + +def test_execute_command_marks_dead_on_sandbox_gone_error(): + client = FakeClient(commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + sb = _make_sandbox(client) + out = sb.execute_command("echo hi") + assert "Error: " in out and "sandbox was not found" in out + assert sb.is_dead is True + out2 = sb.execute_command("echo again") + assert "reaped" in out2.lower() + assert client.commands.calls == ["echo hi"] + + +def test_execute_command_returns_stdout_on_success(): + client = FakeClient(commands=FakeCommandsAPI([SimpleNamespace(stdout="hello\n", stderr="", exit_code=0)])) + sb = _make_sandbox(client) + assert sb.execute_command("printf hello").rstrip() == "hello" + assert sb.is_dead is False + + +def test_execute_command_does_not_mark_dead_on_unrelated_error(): + + def boom(_cmd: str) -> Any: + raise RuntimeError("Connection reset by peer") + + client = FakeClient(commands=FakeCommandsAPI([boom])) + sb = _make_sandbox(client) + out = sb.execute_command("echo hi") + assert "Error" in out + assert sb.is_dead is False + + +def test_ping_returns_false_when_sandbox_gone(): + client = FakeClient(commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + sb = _make_sandbox(client) + assert sb.ping() is False + assert sb.is_dead is True + + +def test_ping_returns_true_on_unknown_error(): + def boom(_cmd: str) -> Any: + raise RuntimeError("upstream timeout") + + client = FakeClient(commands=FakeCommandsAPI([boom])) + sb = _make_sandbox(client) + assert sb.ping() is True + assert sb.is_dead is False + + +def test_client_alive_true_for_healthy_client(): + p = _make_provider() + client = FakeClient() + assert p._client_alive(client) is True + assert client.commands.calls == ["true"] + + +def test_client_alive_false_when_sandbox_gone(): + p = _make_provider() + client = FakeClient(commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + assert p._client_alive(client) is False + + +def test_client_alive_treats_unknown_errors_as_alive(): + p = _make_provider() + + def boom(_cmd: str) -> Any: + raise RuntimeError("flaky network") + + client = FakeClient(commands=FakeCommandsAPI([boom])) + assert p._client_alive(client) is True + + +def test_safe_close_client_swallows_close_failures(): + p = _make_provider() + + class BadCloseClient: + def close(self) -> None: + raise RuntimeError("boom") + + p._safe_close_client(BadCloseClient()) + p._safe_close_client(None) + + +def test_kill_and_close_invokes_kill_and_close_in_order(): + p = _make_provider() + client = FakeClient() + sb = _make_sandbox(client) + p._kill_and_close(sb) + assert client.killed is True + + +def test_kill_and_close_swallows_kill_exceptions(): + p = _make_provider() + client = FakeClient() + sb = _make_sandbox(client) + + def explode(): + raise RuntimeError("already gone") + + client.kill = explode + p._kill_and_close(sb) + + +def test_reuse_in_process_sandbox_returns_cached_id_on_healthy_reuse(): + p = _make_provider() + client = FakeClient() + sb = _make_sandbox(client, sandbox_id="sb-1") + p._sandboxes["sb-1"] = sb + p._thread_sandboxes[("u1", "t1")] = "sb-1" + + sid = p._reuse_in_process_sandbox("t1", user_id="u1") + assert sid == "sb-1" + assert client.timeouts_set, "expected set_timeout to be called on reuse" + + +def test_reuse_in_process_sandbox_evicts_dead_sandbox(): + p = _make_provider() + client = FakeClient(commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + sb = _make_sandbox(client, sandbox_id="sb-dead") + sb._dead = True + p._sandboxes["sb-dead"] = sb + p._thread_sandboxes[("u1", "t1")] = "sb-dead" + + sid = p._reuse_in_process_sandbox("t1", user_id="u1") + assert sid is None + assert "sb-dead" not in p._sandboxes + assert ("u1", "t1") not in p._thread_sandboxes + + +def test_reuse_in_process_sandbox_evicts_when_ping_fails(): + p = _make_provider() + client = FakeClient(commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + sb = _make_sandbox(client, sandbox_id="sb-stale") + p._sandboxes["sb-stale"] = sb + p._thread_sandboxes[("u1", "t1")] = "sb-stale" + + sid = p._reuse_in_process_sandbox("t1", user_id="u1") + assert sid is None + assert sb.is_dead is True + assert "sb-stale" not in p._sandboxes + + +def test_reuse_in_process_sandbox_cleans_dangling_mapping(): + p = _make_provider() + p._thread_sandboxes[("u1", "t1")] = "ghost" + sid = p._reuse_in_process_sandbox("t1", user_id="u1") + assert sid is None + assert ("u1", "t1") not in p._thread_sandboxes + + +def test_reuse_in_process_sandbox_returns_none_when_no_mapping(): + p = _make_provider() + assert p._reuse_in_process_sandbox("t-x", user_id="u-x") is None + + +def test_reclaim_warm_pool_sandbox_happy_path(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + seed = p._stable_seed("t1", "u1") + p._warm_pool["sb-warm"] = (seed, 12345.0) + + sid = p._reclaim_warm_pool_sandbox("t1", user_id="u1") + assert sid == "sb-warm" + assert "sb-warm" in p._sandboxes + assert p._thread_sandboxes[("u1", "t1")] == "sb-warm" + assert "sb-warm" not in p._warm_pool + assert [c[0] for c in fake_cls.connect_calls] == ["sb-warm"] + + +def test_reclaim_warm_pool_sandbox_drops_dead_entry(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + fake_cls.connect_factory = lambda sid, **kw: FakeClient(sandbox_id=sid, commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + seed = p._stable_seed("t1", "u1") + p._warm_pool["sb-zombie"] = (seed, 12345.0) + + sid = p._reclaim_warm_pool_sandbox("t1", user_id="u1") + assert sid is None + assert "sb-zombie" not in p._sandboxes + assert "sb-zombie" not in p._warm_pool + + +def test_reclaim_warm_pool_sandbox_handles_reconnect_exception(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + + def boom(sid, **kw): + raise RuntimeError("404 Not Found") + + fake_cls.connect_factory = boom + seed = p._stable_seed("t1", "u1") + p._warm_pool["sb-broken"] = (seed, 12345.0) + + sid = p._reclaim_warm_pool_sandbox("t1", user_id="u1") + assert sid is None + assert "sb-broken" not in p._warm_pool + + +def test_reclaim_warm_pool_sandbox_returns_none_on_seed_mismatch(monkeypatch): + p = _make_provider() + _install_fake_sdk(monkeypatch, p) + p._warm_pool["sb-other"] = ("some-other-seed", 12345.0) + assert p._reclaim_warm_pool_sandbox("t1", user_id="u1") is None + # The unrelated entry must remain untouched. + assert "sb-other" in p._warm_pool + + +class _FakePaginator: + """Mirror of e2b SDK's ``SandboxPaginator``: items via ``next_items``.""" + + def __init__(self, pages: list[list[Any]]) -> None: + self._pages = list(pages) + self.has_next = bool(self._pages) + self.calls = 0 + + def next_items(self) -> list[Any]: + self.calls += 1 + if not self._pages: + self.has_next = False + return [] + page = self._pages.pop(0) + self.has_next = bool(self._pages) + return page + + +def _info(sandbox_id: str, user_id: str, thread_id: str): + return SimpleNamespace( + sandbox_id=sandbox_id, + metadata={ + "deer_flow_provider": "e2b_sandbox_provider", + "deer_flow_user": user_id, + "deer_flow_thread": thread_id, + }, + ) + + +def test_discover_remote_sandbox_walks_paginator(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + fake_cls.list_return = _FakePaginator( + [ + [_info("sb-other", "u-x", "t-x")], + [_info("sb-match", "u1", "t1")], + ] + ) + + sid = p._discover_remote_sandbox("t1", user_id="u1") + assert sid == "sb-match" + assert p._thread_sandboxes[("u1", "t1")] == "sb-match" + + +def test_discover_remote_sandbox_accepts_legacy_list(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + fake_cls.list_return = [_info("sb-legacy", "u1", "t1")] + + sid = p._discover_remote_sandbox("t1", user_id="u1") + assert sid == "sb-legacy" + + +def test_discover_remote_sandbox_skips_dead_candidate(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + fake_cls.list_return = [_info("sb-dead", "u1", "t1")] + fake_cls.connect_factory = lambda sid, **kw: FakeClient(sandbox_id=sid, commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + + assert p._discover_remote_sandbox("t1", user_id="u1") is None + assert ("u1", "t1") not in p._thread_sandboxes + + +def test_discover_remote_sandbox_returns_none_when_list_raises(monkeypatch): + p = _make_provider() + fake_cls = _install_fake_sdk(monkeypatch, p) + + def boom(**kw): + raise RuntimeError("API unreachable") + + fake_cls.list = boom + assert p._discover_remote_sandbox("t1", user_id="u1") is None + + +def test_bootstrap_sandbox_paths_emits_expected_script(): + p = _make_provider() + client = FakeClient() + p._bootstrap_sandbox_paths(client) + assert len(client.commands.calls) == 1 + script = client.commands.calls[0] + assert "ln -sfn" in script + assert "/mnt/user-data" in script + assert "/mnt/acp-workspace" in script + assert "BOOTSTRAP_OK" in script + for sub in ("workspace", "uploads", "outputs", "acp-workspace"): + assert f"/home/user/{sub}" in script + + +def test_bootstrap_sandbox_paths_swallows_command_failure(): + p = _make_provider() + + def boom(_cmd: str) -> Any: + raise RuntimeError("sudo not allowed") + + client = FakeClient(commands=FakeCommandsAPI([boom])) + p._bootstrap_sandbox_paths(client) + + +def test_release_unknown_sandbox_id_is_noop(): + p = _make_provider() + p.release("nonexistent") + assert p._warm_pool == OrderedDict() + + +def test_release_dead_sandbox_skips_warm_pool(monkeypatch): + p = _make_provider() + client = FakeClient() + sb = _make_sandbox(client, sandbox_id="sb-dead") + sb._dead = True + p._sandboxes["sb-dead"] = sb + p._thread_sandboxes[("u1", "t1")] = "sb-dead" + + p.release("sb-dead") + + assert "sb-dead" not in p._warm_pool, "dead sandbox must not be parked" + assert "sb-dead" not in p._sandboxes + assert ("u1", "t1") not in p._thread_sandboxes + assert client.killed is True, "release of dead sandbox must kill the remote VM" + + +def test_release_healthy_sandbox_parks_in_warm_pool(monkeypatch, tmp_path): + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + cmds = FakeCommandsAPI([SimpleNamespace(stdout="", stderr="", exit_code=0)]) + client = FakeClient(commands=cmds) + sb = _make_sandbox(client, sandbox_id="sb-warm-1") + p._sandboxes["sb-warm-1"] = sb + p._thread_sandboxes[("u1", "t1")] = "sb-warm-1" + + p.release("sb-warm-1") + + assert "sb-warm-1" in p._warm_pool + seed_in_pool, _ts = p._warm_pool["sb-warm-1"] + assert seed_in_pool == p._stable_seed("t1", "u1") + assert client.killed is False + assert client.timeouts_set + + +def test_release_skips_warm_pool_when_sync_reveals_dead_vm(monkeypatch, tmp_path): + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + client = FakeClient(commands=FakeCommandsAPI([FakeCommandsAPI.GONE])) + sb = _make_sandbox(client, sandbox_id="sb-died-during-sync") + p._sandboxes["sb-died-during-sync"] = sb + p._thread_sandboxes[("u1", "t1")] = "sb-died-during-sync" + + p.release("sb-died-during-sync") + + assert sb.is_dead is True + assert "sb-died-during-sync" not in p._warm_pool + assert client.killed is True + + +def _setup_paths(monkeypatch, tmp_path): + paths_mod = importlib.import_module("deerflow.config.paths") + monkeypatch.setattr(paths_mod, "get_paths", lambda: Paths(base_dir=tmp_path), raising=False) + + +def test_sync_outputs_to_host_writes_new_files(monkeypatch, tmp_path): + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + listing = "13\t/home/user/outputs/random.pdf\x00" + files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + client = FakeClient(commands=cmds, files=files) + sb = _make_sandbox(client, sandbox_id="sb-sync-1") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + expected = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs" / "random.pdf" + assert expected.exists() + assert expected.read_bytes() == b"%PDF-1.4hello" + + +def test_sync_outputs_to_host_skips_unchanged_files(monkeypatch, tmp_path): + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + out_dir = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs" + out_dir.mkdir(parents=True, exist_ok=True) + target = out_dir / "random.pdf" + target.write_bytes(b"%PDF-1.4hello") + + listing = "13\t/home/user/outputs/random.pdf\x00" + files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"DIFFERENT-SAME-LEN"}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + client = FakeClient(commands=cmds, files=files) + sb = _make_sandbox(client, sandbox_id="sb-sync-2") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + assert files.read_calls == [], "size match should skip the download round-trip" + assert target.read_bytes() == b"%PDF-1.4hello" + + +def test_sync_outputs_to_host_marks_dead_on_sandbox_gone(monkeypatch, tmp_path): + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + cmds = FakeCommandsAPI([FakeCommandsAPI.GONE]) + client = FakeClient(commands=cmds) + sb = _make_sandbox(client, sandbox_id="sb-sync-dead") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + assert sb.is_dead is True + + +def test_sync_outputs_to_host_uses_virtual_path_for_download(monkeypatch, tmp_path): + """`download_file` requires paths under ``/mnt/user-data``; the sync + helper must translate the physical /home/user/... back to the virtual + prefix before calling it.""" + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + + listing = "5\t/home/user/outputs/sub/x.txt\x00" + files = FakeFilesAPI(store={"/home/user/outputs/sub/x.txt": b"hello"}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + client = FakeClient(commands=cmds, files=files) + sb = _make_sandbox(client, sandbox_id="sb-sync-3") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + read_paths = [r[0] for r in files.read_calls] + assert "/home/user/outputs/sub/x.txt" in read_paths + + +def test_sync_outputs_to_host_is_noop_when_client_closed(): + p = _make_provider() + sb = _make_sandbox(FakeClient(), sandbox_id="sb-x") + sb.close() + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + +def test_download_file_uses_streaming_read_and_returns_full_bytes(): + payload = b"A" * (128 * 1024) # 128 KiB — well below the cap. + files = FakeFilesAPI(store={"/home/user/outputs/small.bin": payload}) + client = FakeClient(files=files) + sb = _make_sandbox(client, sandbox_id="sb-stream-1") + + data = sb.download_file("/mnt/user-data/outputs/small.bin") + + assert data == payload + formats_used = [fmt for _p, fmt in files.read_calls] + assert "stream" in formats_used, f"expected download_file to invoke read(format='stream'), got {formats_used!r}" + assert files.streams, "download_file must actually consume a stream" + assert files.streams[-1].closed, "stream must be closed after successful read" + + +def test_download_file_streaming_raises_efbig_before_full_buffering(): + import errno as _errno + + from deerflow.community.e2b_sandbox import e2b_sandbox as e2b_sb_mod + + cap = e2b_sb_mod._MAX_DOWNLOAD_SIZE + + class _OversizeStream: + def __init__(self) -> None: + self.bytes_yielded = 0 + self.closed = False + self._chunk = b"X" * (1024 * 1024) # 1 MiB per chunk + + def __iter__(self): + return self + + def __next__(self) -> bytes: + if self.closed: + raise StopIteration + # Yield up to ``cap + a bit`` — the caller must abort before + # actually buffering all of that in memory. + if self.bytes_yielded > cap + 4 * len(self._chunk): + raise StopIteration + self.bytes_yielded += len(self._chunk) + return self._chunk + + def close(self) -> None: + self.closed = True + + stream = _OversizeStream() + + class _StubFilesAPI: + def __init__(self) -> None: + self.calls: list[tuple[str, str | None]] = [] + + def read(self, path: str, *, format: str | None = None): + self.calls.append((path, format)) + assert format == "stream", "provider must request a streamed download" + return stream + + files = _StubFilesAPI() + client = FakeClient(files=files) # type: ignore[arg-type] + sb = _make_sandbox(client, sandbox_id="sb-stream-oversize") + + try: + sb.download_file("/mnt/user-data/outputs/huge.bin") + except OSError as exc: + assert exc.errno == _errno.EFBIG, f"expected EFBIG, got errno={exc.errno!r} ({exc})" + else: # pragma: no cover - defensive + raise AssertionError("download_file must raise OSError(EFBIG) on oversize stream") + + assert stream.closed is True, "stream must be closed on abort so the pooled connection is released" + assert stream.bytes_yielded <= cap + 2 * 1024 * 1024, f"aborted too late: yielded={stream.bytes_yielded} vs cap={cap}" + + +def test_download_file_falls_back_to_buffered_read_for_legacy_sdk(): + + class _LegacyFilesAPI: + def __init__(self, data: bytes) -> None: + self._data = data + self.calls: list[tuple[str, str | None]] = [] + + def read(self, path: str, *, format: str | None = None): + self.calls.append((path, format)) + if format == "stream": + raise TypeError("format='stream' unsupported") + if format == "bytes": + return self._data + return self._data.decode("utf-8", errors="replace") + + files = _LegacyFilesAPI(b"legacy-payload") + client = FakeClient(files=files) # type: ignore[arg-type] + sb = _make_sandbox(client, sandbox_id="sb-legacy") + + data = sb.download_file("/mnt/user-data/outputs/legacy.bin") + assert data == b"legacy-payload" + formats_used = [fmt for _p, fmt in files.calls] + assert formats_used == ["stream", "bytes"], f"expected stream then bytes fallback, got {formats_used!r}" + + +def test_sync_outputs_to_host_skips_oversize_files(monkeypatch, tmp_path): + from deerflow.community.e2b_sandbox import e2b_sandbox as e2b_sb_mod + + p = _make_provider() + _setup_paths(monkeypatch, tmp_path) + + oversize = e2b_sb_mod._MAX_DOWNLOAD_SIZE + 1 + listing = f"{oversize}\t/home/user/outputs/huge.bin\x00" + files = FakeFilesAPI() # no store entry: any read attempt would raise + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + client = FakeClient(commands=cmds, files=files) + sb = _make_sandbox(client, sandbox_id="sb-sync-oversize") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + assert files.read_calls == [], "oversize files must be skipped without invoking download_file" + host_target = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs" / "huge.bin" + assert not host_target.exists(), "no oversize artefact must be written to host" diff --git a/backend/uv.lock b/backend/uv.lock index 8c2ecc08b..de7f2da52 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -485,6 +485,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/95/c1/84fc6811122f54b20de2e5afb312ee07a3a47a328755587d1e505475239b/blockbuster-1.5.26-py3-none-any.whl", hash = "sha256:f8e53fb2dd4b6c6ec2f04907ddbd063ca7cd1ef587d24448ef4e50e81e3a79bb", size = 13226, upload-time = "2025-12-05T10:43:48.778Z" }, ] +[[package]] +name = "bracex" +version = "2.7" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/37/7c/a2a8a52db0ee751007507ddad3a1ddf1b0f763de546c588e7a828579bdad/bracex-2.7.tar.gz", hash = "sha256:4cb5d415a707f6beeb2779099486090bf98cbd8b7edbdfcb7cbea2f5fe6bdb48", size = 42150, upload-time = "2026-06-28T18:48:39.276Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ba/24/67865d7a710d86de496c7984e06023aa3656b5fae16ee229a530b57c0491/bracex-2.7-py3-none-any.whl", hash = "sha256:025043774188f8a05db36de9e3d4f7d82a8509a41a115cc134c44a60c36375eb", size = 11508, upload-time = "2026-06-28T18:48:38.138Z" }, +] + [[package]] name = "certifi" version = "2026.4.22" @@ -762,6 +771,7 @@ dependencies = [ { name = "bcrypt" }, { name = "deerflow-harness" }, { name = "dingtalk-stream" }, + { name = "e2b-code-interpreter" }, { name = "email-validator" }, { name = "fastapi" }, { name = "httpx" }, @@ -802,6 +812,7 @@ requires-dist = [ { name = "deerflow-harness", extras = ["postgres"], marker = "extra == 'postgres'", editable = "packages/harness" }, { name = "dingtalk-stream", specifier = ">=0.24.3" }, { name = "discord-py", marker = "extra == 'discord'", specifier = ">=2.7.0" }, + { name = "e2b-code-interpreter", specifier = ">=2.8.1" }, { name = "email-validator", specifier = ">=2.0.0" }, { name = "fastapi", specifier = ">=0.115.0" }, { name = "httpx", specifier = ">=0.28.0" }, @@ -841,6 +852,7 @@ dependencies = [ { name = "ddgs" }, { name = "dotenv" }, { name = "duckdb" }, + { name = "e2b-code-interpreter" }, { name = "exa-py" }, { name = "firecrawl-py" }, { name = "httpx" }, @@ -896,6 +908,7 @@ requires-dist = [ { name = "ddgs", specifier = ">=9.10.0" }, { name = "dotenv", specifier = ">=0.9.9" }, { name = "duckdb", specifier = ">=1.4.4" }, + { name = "e2b-code-interpreter", specifier = ">=2.8.0" }, { name = "exa-py", specifier = ">=1.0.0" }, { name = "firecrawl-py", specifier = ">=1.15.0" }, { name = "httpx", specifier = ">=0.28.0" }, @@ -983,6 +996,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ba/5a/18ad964b0086c6e62e2e7500f7edc89e3faa45033c71c1893d34eed2b2de/dnspython-2.8.0-py3-none-any.whl", hash = "sha256:01d9bbc4a2d76bf0db7c1f729812ded6d912bd318d3b1cf81d30c0f845dbf3af", size = 331094, upload-time = "2025-09-07T18:57:58.071Z" }, ] +[[package]] +name = "dockerfile-parse" +version = "2.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/92/df/929ee0b5d2c8bd8d713c45e71b94ab57c7e11e322130724d54f469b2cd48/dockerfile-parse-2.0.1.tar.gz", hash = "sha256:3184ccdc513221983e503ac00e1aa504a2aa8f84e5de673c46b0b6eee99ec7bc", size = 24556, upload-time = "2023-07-18T13:36:07.897Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7a/6c/79cd5bc1b880d8c1a9a5550aa8dacd57353fa3bb2457227e1fb47383eb49/dockerfile_parse-2.0.1-py2.py3-none-any.whl", hash = "sha256:bdffd126d2eb26acf1066acb54cb2e336682e1d72b974a40894fac76a4df17f6", size = 14845, upload-time = "2023-07-18T13:36:06.052Z" }, +] + [[package]] name = "docstring-parser" version = "0.18.0" @@ -1041,6 +1063,42 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b0/0d/9feae160378a3553fa9a339b0e9c1a048e147a4127210e286ef18b730f03/durationpy-0.10-py3-none-any.whl", hash = "sha256:3b41e1b601234296b4fb368338fdcd3e13e0b4fb5b67345948f4f2bf9868b286", size = 3922, upload-time = "2025-05-17T13:52:36.463Z" }, ] +[[package]] +name = "e2b" +version = "2.30.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "attrs" }, + { name = "dockerfile-parse" }, + { name = "h2" }, + { name = "httpcore" }, + { name = "httpx" }, + { name = "packaging" }, + { name = "protobuf" }, + { name = "python-dateutil" }, + { name = "rich" }, + { name = "typing-extensions" }, + { name = "wcmatch" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/fd/3f/5cfb744bae4d57f253ee95cc3291a60ab944ea0fdb34d2d4b9ef05eaf0a7/e2b-2.30.0.tar.gz", hash = "sha256:96515e7512caf26bc975b516372063015fa6b48b386482952b2ab9aeabaee879", size = 179811, upload-time = "2026-06-25T18:03:57.52Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b1/ee/bfce7382df70d2fa252969b083e2bbf5e85d22b73e36cc6d4c663df81670/e2b-2.30.0-py3-none-any.whl", hash = "sha256:37e122fee9e5b36f31980221f3a41db9b33847c2d0866a9f07bc083f9aa4ce96", size = 330349, upload-time = "2026-06-25T18:03:55.975Z" }, +] + +[[package]] +name = "e2b-code-interpreter" +version = "2.8.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "attrs" }, + { name = "e2b" }, + { name = "httpx" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d9/9b/e24eeb48fba1342f6a81e98bf6ff51622a7d5c3353a89d452623f158bcb8/e2b_code_interpreter-2.8.1.tar.gz", hash = "sha256:900bc847a4a6f90a571e0de4112531c79c550fbdb1ec94a1d2b2c9f0d058a674", size = 11467, upload-time = "2026-06-17T09:58:51.217Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/47/c9acfeb0a63925d5b2e782b496be1287665bb649eee48d947fc46b9bdeee/e2b_code_interpreter-2.8.1-py3-none-any.whl", hash = "sha256:8449ff1abb507e4c28134c9cbc4e59f76c09fadceab879c8c4031399b31a126a", size = 15210, upload-time = "2026-06-17T09:58:50.043Z" }, +] + [[package]] name = "email-validator" version = "2.3.0" @@ -1431,6 +1489,28 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, ] +[[package]] +name = "h2" +version = "4.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "hpack" }, + { name = "hyperframe" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1d/17/afa56379f94ad0fe8defd37d6eb3f89a25404ffc71d4d848893d270325fc/h2-4.3.0.tar.gz", hash = "sha256:6c59efe4323fa18b47a632221a1888bd7fde6249819beda254aeca909f221bf1", size = 2152026, upload-time = "2025-08-23T18:12:19.778Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/69/b2/119f6e6dcbd96f9069ce9a2665e0146588dc9f88f29549711853645e736a/h2-4.3.0-py3-none-any.whl", hash = "sha256:c438f029a25f7945c69e0ccf0fb951dc3f73a5f6412981daee861431b70e2bdd", size = 61779, upload-time = "2025-08-23T18:12:17.779Z" }, +] + +[[package]] +name = "hpack" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/26/5b/fcabf6028144a8723726318b07a32c2f3314acdff6265743cf08a344b18e/hpack-4.2.0.tar.gz", hash = "sha256:0895cfa3b5531fc65fe439c05eb65144f123bf7a394fcaa56aa423548d8e45c0", size = 51300, upload-time = "2026-06-23T18:34:46.667Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/b4/4a9fcfb2aef6ba44d9073ecd301443aa00b3dac95de5619f2a7de7ec8a91/hpack-4.2.0-py3-none-any.whl", hash = "sha256:858ac0b02280fa582b5080d68db0899c62a80375e0e5413a74970c5e518b6986", size = 34246, upload-time = "2026-06-23T18:34:45.472Z" }, +] + [[package]] name = "html5lib" version = "1.1" @@ -1527,6 +1607,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f0/0f/310fb31e39e2d734ccaa2c0fb981ee41f7bd5056ce9bc29b2248bd569169/humanfriendly-10.0-py2.py3-none-any.whl", hash = "sha256:1697e1a8a8f550fd43c2865cd84542fc175a61dcb779b6fee18cf6b6ccba1477", size = 86794, upload-time = "2021-09-17T21:40:39.897Z" }, ] +[[package]] +name = "hyperframe" +version = "6.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/02/e7/94f8232d4a74cc99514c13a9f995811485a6903d48e5d952771ef6322e30/hyperframe-6.1.0.tar.gz", hash = "sha256:f630908a00854a7adeabd6382b43923a4c4cd4b821fcb527e6ab9e15382a3b08", size = 26566, upload-time = "2025-01-22T21:41:49.302Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/48/30/47d0bf6072f7252e6521f3447ccfa40b421b6824517f82854703d0f5a98b/hyperframe-6.1.0-py3-none-any.whl", hash = "sha256:b03380493a519fce58ea5af42e4a42317bf9bd425596f7a0835ffce80f1a42e5", size = 13007, upload-time = "2025-01-22T21:41:47.295Z" }, +] + [[package]] name = "idna" version = "3.15" @@ -4537,6 +4626,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e3/bd/fa9bb053192491b3867ba07d2343d9f2252e00811567d30ae8d0f78136fe/watchfiles-1.1.1-cp314-cp314t-musllinux_1_1_x86_64.whl", hash = "sha256:a916a2932da8f8ab582f242c065f5c81bed3462849ca79ee357dd9551b0e9b01", size = 622112, upload-time = "2025-10-14T15:05:50.941Z" }, ] +[[package]] +name = "wcmatch" +version = "10.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "bracex" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/79/3e/c0bdc27cf06f4e47680bd5803a07cb3dfd17de84cde92dd217dcb9e05253/wcmatch-10.1.tar.gz", hash = "sha256:f11f94208c8c8484a16f4f48638a85d771d9513f4ab3f37595978801cb9465af", size = 117421, upload-time = "2025-06-22T19:14:02.49Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/eb/d8/0d1d2e9d3fabcf5d6840362adcf05f8cf3cd06a73358140c3a97189238ae/wcmatch-10.1-py3-none-any.whl", hash = "sha256:5848ace7dbb0476e5e55ab63c6bbd529745089343427caa5537f230cc01beb8a", size = 39854, upload-time = "2025-06-22T19:14:00.978Z" }, +] + [[package]] name = "wcwidth" version = "0.6.0" diff --git a/frontend/src/content/en/application/deployment-guide.mdx b/frontend/src/content/en/application/deployment-guide.mdx index c30f02257..969884a15 100644 --- a/frontend/src/content/en/application/deployment-guide.mdx +++ b/frontend/src/content/en/application/deployment-guide.mdx @@ -114,6 +114,7 @@ For production, use a named volume or a Persistent Volume Claim (PVC) instead of | -------------------------------------- | ------------------------------------------ | | `LocalSandboxProvider` | Single-user, trusted local workflows | | `AioSandboxProvider` (Docker) | Multi-user, moderate isolation requirement | +| `E2BSandboxProvider` (e2b cloud) | Hosted/serverless, no Docker/K8s ops, instant Jupyter | | `AioSandboxProvider` + K8s Provisioner | Production, strong isolation, multi-user | For any deployment with more than one concurrent user, use a container-based sandbox to prevent users from interfering with each other's execution environments. diff --git a/frontend/src/content/en/harness/sandbox.mdx b/frontend/src/content/en/harness/sandbox.mdx index 71cda67c6..6668bf5f4 100644 --- a/frontend/src/content/en/harness/sandbox.mdx +++ b/frontend/src/content/en/harness/sandbox.mdx @@ -17,7 +17,7 @@ The sandbox gives the Lead Agent a controlled environment where it can read file ## Sandbox modes -DeerFlow supports three sandbox modes. Choose the one that fits your deployment: +DeerFlow supports four sandbox modes. Choose the one that fits your deployment: ### LocalSandbox (default) @@ -67,6 +67,61 @@ sandbox: Install: `cd backend && uv add 'deerflow-harness[aio-sandbox]'` +### E2B Cloud Sandbox + +Commands run in an [E2B](https://e2b.dev) cloud micro-VM. Each thread is bound to its own sandbox via metadata, so the same DeerFlow thread keeps reaching the same e2b sandbox even across gateway restarts and across processes. The cloud VM is reaped automatically by e2b once the configured idle timeout elapses. + +- **Best for**: hosted/serverless deployments, teams that do not want to operate Docker or Kubernetes themselves, workloads that need a real Jupyter kernel out of the box. +- **Trade-off**: requires network access to the e2b control plane and an API key. Mounts are **one-shot uploads** at sandbox start (host bind-mounts are not possible because the sandbox lives in the cloud). + +```yaml +sandbox: + use: deerflow.community.e2b_sandbox:E2BSandboxProvider + + # Required: e2b API key. Falls back to the E2B_API_KEY environment variable. + api_key: $E2B_API_KEY + + # Optional: e2b sandbox template id (default: code-interpreter-v1) + template: code-interpreter-v1 + + # Optional: self-hosted e2b deployment domain + # domain: e2b.dev + + # Optional: directory inside the sandbox that backs /mnt/user-data + # (default: /home/user) + home_dir: /home/user + + # Optional: server-enforced idle timeout in seconds (default: 600, max 86400). + # The provider refreshes this on every release so warm sandboxes stay alive. + idle_timeout: 600 + + # Optional: max concurrent sandboxes per gateway process (default: 3). + # When exceeded, the least-recently-used warm sandbox is killed. + replicas: 3 + + # Optional: one-shot upload of host files into the sandbox at start. + # Read-only mounts are chmod'd to read-only after upload. + mounts: + - host_path: /path/on/host + container_path: /home/user/shared + read_only: false + + # Optional: environment variables forwarded to the sandbox on create. + # Values starting with "$" are resolved from the host environment. + environment: + OPENAI_API_KEY: $OPENAI_API_KEY +``` + +Install: `e2b-code-interpreter` is already a core dependency of `deerflow-harness`, so no extra install step is needed — just provide your API key and switch the provider in `config.yaml`. + +Get an API key at [e2b.dev/dashboard](https://e2b.dev/dashboard) and either set `sandbox.api_key` directly in `config.yaml` or export `E2B_API_KEY` in your `.env`. + + + Because e2b is a remote sandbox, DeerFlow's `/mnt/user-data` virtual prefix is + remapped onto `home_dir` (default `/home/user`) inside the cloud VM. Tools + that read or write under `/mnt/user-data/...` continue to work transparently. + + ### Provisioner-managed Sandbox (Kubernetes) Each sandbox gets a dedicated Pod in a Kubernetes cluster, managed by the provisioner service. This provides the strongest isolation and is recommended for production environments with multiple concurrent users. diff --git a/frontend/src/content/zh/application/deployment-guide.mdx b/frontend/src/content/zh/application/deployment-guide.mdx index 9e3f07e03..d5220168a 100644 --- a/frontend/src/content/zh/application/deployment-guide.mdx +++ b/frontend/src/content/zh/application/deployment-guide.mdx @@ -111,6 +111,7 @@ BETTER_AUTH_SECRET=your-secret-here-min-32-chars | -------------------------------------- | -------------------------- | | `LocalSandboxProvider` | 单用户、受信任的本地工作流 | | `AioSandboxProvider`（Docker） | 多用户、中等隔离需求 | +| `E2BSandboxProvider`（e2b 云） | 托管/Serverless、无需运维 Docker/K8s、开箱即用 Jupyter | | `AioSandboxProvider` + K8s Provisioner | 生产环境、强隔离、多用户 | 对于有多个并发用户的任何部署，使用基于容器的沙箱，防止用户之间的执行环境相互干扰。 diff --git a/frontend/src/content/zh/harness/sandbox.mdx b/frontend/src/content/zh/harness/sandbox.mdx index f4f3c88d7..a3e2a5685 100644 --- a/frontend/src/content/zh/harness/sandbox.mdx +++ b/frontend/src/content/zh/harness/sandbox.mdx @@ -16,7 +16,7 @@ import { Callout, Tabs } from "nextra/components"; ## 沙箱模式 -DeerFlow 支持三种沙箱模式，选择适合你部署的一种： +DeerFlow 支持四种沙箱模式，选择适合你部署的一种： ### LocalSandbox（默认） @@ -59,6 +59,60 @@ sandbox: 安装：`cd backend && uv add 'deerflow-harness[aio-sandbox]'` +### E2B 云端沙箱 + +命令在 [E2B](https://e2b.dev) 云端的 micro-VM 中执行。每个线程通过 metadata 绑定到一个独立的 e2b 沙箱，因此即使 Gateway 重启或在多进程之间，同一个 DeerFlow 线程也始终复用同一个 e2b 沙箱。当达到配置的空闲超时时，e2b 会自动回收云端 VM。 + +- **适合**：托管/Serverless 部署、不想自行运维 Docker 或 Kubernetes 的团队、开箱即用需要真实 Jupyter 内核的工作流。 +- **取舍**：需要访问 e2b 控制面网络以及一个 API Key；挂载只能在沙箱启动时**一次性上传**（云端沙箱无法做主机 bind-mount）。 + +```yaml +sandbox: + use: deerflow.community.e2b_sandbox:E2BSandboxProvider + + # 必填：e2b API Key。若不写则回退读取环境变量 E2B_API_KEY。 + api_key: $E2B_API_KEY + + # 可选：e2b 沙箱模板 id（默认：code-interpreter-v1） + template: code-interpreter-v1 + + # 可选：自部署 e2b 实例的 domain + # domain: e2b.dev + + # 可选：沙箱内部 /mnt/user-data 实际指向的目录（默认：/home/user） + home_dir: /home/user + + # 可选：服务端强制的空闲超时（秒，默认 600，最大 86400）。 + # 每次 release 时 provider 会自动续期，让 warm 沙箱保持存活。 + idle_timeout: 600 + + # 可选：单个 Gateway 进程内的最大并发沙箱数（默认：3）。 + # 超出时按 LRU 回收 warm 沙箱。 + replicas: 3 + + # 可选：启动时一次性把主机文件上传到沙箱。 + # read_only 的目录在上传后会被 chmod 为只读。 + mounts: + - host_path: /path/on/host + container_path: /home/user/shared + read_only: false + + # 可选：创建沙箱时传入的环境变量。 + # 以 "$" 开头的值会从主机环境变量解析。 + environment: + OPENAI_API_KEY: $OPENAI_API_KEY +``` + +安装：`e2b-code-interpreter` 已经是 `deerflow-harness` 的核心依赖，无需额外安装 —— 只需配置 API Key，并在 `config.yaml` 中切换 provider 即可。 + +在 [e2b.dev/dashboard](https://e2b.dev/dashboard) 申请 API Key，然后在 `config.yaml` 中直接配置 `sandbox.api_key`，或在 `.env` 中导出 `E2B_API_KEY`。 + + + 因为 e2b 是远程沙箱，DeerFlow 的虚拟前缀 `/mnt/user-data` + 会被自动重映射到云端 VM 的 `home_dir`（默认 `/home/user`）。 + 原本读写 `/mnt/user-data/...` 的工具无需任何改动即可正常工作。 + + ### Provisioner 管理的沙箱（Kubernetes） 每个沙箱在 Kubernetes 集群中获得一个专用 Pod，由 Provisioner 服务管理。这提供最强的隔离性，适合有多个并发用户的生产环境。