diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 4e5de1c6f..c7e07216b 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -583,9 +583,10 @@ copying a raw checkpoint because delta state is not self-contained in one tuple. - `BoxliteProvider` (`packages/harness/deerflow/community/boxlite/`) - BoxLite micro-VM isolation. The `boxlite` runtime is optional (`deerflow-harness[boxlite]`) and lazy-imported only when this provider is selected. The provider owns one private asyncio event loop on a daemon thread because BoxLite handles are loop-affine; sync `Sandbox` calls marshal onto that loop with `run_coroutine_threadsafe`. Boxes are named deterministically from `user_id:thread_id`, released into an in-process warm pool after each agent turn, and reclaimed only by the same user/thread. Warm-pool health checks use a short explicit timeout and forward that timeout through both BoxLite `exec(timeout=...)` and the private-loop `.result(timeout)` bridge so a hung VM cannot pin the per-thread acquire lock indefinitely. `sandbox.replicas` caps active + warm VMs per gateway process; if capacity is exhausted, only warm-pool VMs are evicted. `sandbox.idle_timeout` stops idle warm VMs after the configured seconds. `reset()` is intentionally a lightweight registry clear for `reset_sandbox_provider()` and does not close boxes, stop the idle reaper, or close the private loop; full teardown remains `shutdown()`. +- `TenkiSandboxProvider` (`packages/harness/deerflow/community/tenki/`) - Tenki cloud microVM isolation. The `tenki-sandbox` SDK is optional (`deerflow-harness[tenki]`) and lazy-imported (`_import_client`) only when this provider is selected. Unlike Boxlite, the SDK is synchronous, so the adapter calls it directly with no event-loop bridge. File transport uses Tenki's native `sandbox.fs` API (`read_text`/`read_stream`/`write_stream`/`mkdir`/`stat`) — binary-safe and streaming, no base64/shell hop; only directory/content *search* (`list_dir`/`glob`/`grep`) shells out to busybox-portable `find`/`grep`, parsed with the shared `deerflow.sandbox.search` helpers like `community/e2b_sandbox`. Sandboxes run as the unprivileged `tenki` user, so DeerFlow's `/mnt/user-data` prefix is remapped under a writable HOME (`_resolve_path`) and best-effort `sudo`-symlinked at bootstrap. Boxes are named deterministically from `sha256(user_id:thread_id)[:16]` (64-bit, matching E2B; the warm pool is keyed by this id alone with no full-seed fallback), released into an in-process warm pool, and reclaimed only by the same user/thread after a liveness check. A terminal session error (named SDK errors plus builtin `ConnectionError`/`BrokenPipeError`/`EOFError`) routes through `_invalidate_sandbox` to evict the dead microVM. Cross-process orphan reconciliation is a follow-up (single-process warm pool today). -**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles. AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. +**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper. **Virtual Path System**: - Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills` @@ -642,7 +643,7 @@ Scheduled-task runtime note: - `browser_automation/` - Agentic browser control (stateful `navigate → observe → click/type` loop) via Playwright, distinct from the read-only `web_fetch`/`web_capture` tools. Tools: `browser_navigate`, `browser_snapshot`, `browser_click`, `browser_type`, `browser_get_text`, `browser_back`, `browser_screenshot`, `browser_close` (config `group: browser`). A process-local `BrowserSessionManager` owns one private, loop-affine Playwright event-loop thread (same pattern as the BoxLite provider) so a per-thread browser session survives across turns regardless of the caller's loop (Gateway / TUI / test). Each action returns a fresh page snapshot whose interactive elements are addressed by a stable numeric `[ref]` index (stamped as `data-df-ref` during snapshot), so the model acts on what it just observed instead of holding stale handles or guessing selectors. URLs are SSRF-screened via the shared `validate_public_http_url` (opt-out `allow_private_addresses` only for intentional internal targets). CDP attachment cannot install the request guard on an existing Chrome context, so `cdp_url` fails closed unless the operator explicitly sets `allow_unguarded_cdp: true` for a trusted local browser. Browser REST/Live access also requires an exact non-NULL thread owner, rather than the general legacy shared-thread policy, because retained pages may contain authenticated state. Session admission is a hard `max_sessions` cap: pinned Live/operation sessions are never evicted, and a new thread is rejected when no unpinned session can be closed; one Live viewer owns a session at a time. Optional dependency: `cd backend && uv sync --extra browser && uv run playwright install chromium`; `scripts/detect_uv_extras.py` preserves the extra when `config.yaml` enables `browser_navigate`, and Gateway startup fails fast if configured browser control cannot import Playwright. Tests: `tests/test_browser_automation.py` (mocked tools + a real-Chromium integration test guarded by `importorskip`); `tests/manual_browser_live_check.py` is a manual DeepSeek-driven end-to-end check (not collected by pytest). Live UI input dispatch is kept independent from JPEG capture: non-move actions start a rate-limited background refresh loop, so pointer, wheel, or keyboard input stays responsive while continuous gestures still produce frames throughout the interaction. -Additional providers also live here (`boxlite`, `brave`, `browserless`, `crawl4ai`, `ddg_search`, `e2b_sandbox`, `exa`, `fastcrw`, `groundroute`, `infoquest`, `searxng`, `serper`); see each subpackage for specifics. E2B bootstrap is required. If it fails, the provider kills and closes the unusable remote sandbox. New sandbox creation raises an error. Warm-pool reclaim and remote discovery discard the sandbox and continue acquisition. E2B mounts remain optional. +Additional providers also live here (`boxlite`, `brave`, `browserless`, `crawl4ai`, `ddg_search`, `e2b_sandbox`, `exa`, `fastcrw`, `groundroute`, `infoquest`, `searxng`, `serper`, `tenki`); see each subpackage for specifics. E2B bootstrap is required. If it fails, the provider kills and closes the unusable remote sandbox. New sandbox creation raises an error. Warm-pool reclaim and remote discovery discard the sandbox and continue acquisition. E2B mounts remain optional. E2B output sync records remote file versions and actual host file metadata in a thread-local manifest. The manifest binds to the remote sandbox ID. A complete output listing removes entries for deleted files. This avoids repeat downloads when the host filesystem rounds modification times. A single release-time sync pass is bounded by aggregate ceilings (`_MAX_SYNC_TOTAL_BYTES`, `_MAX_SYNC_FILES`, `_SYNC_DEADLINE_SECONDS`) on top of the per-file `_MAX_DOWNLOAD_SIZE` cap, so a pathological outputs tree cannot make release download unboundedly; a truncated pass logs what it dropped and leaves the manifest un-pruned (only entries observed in that pass are reconciled), so files it never reached are retried on the next release rather than being forgotten. diff --git a/backend/packages/harness/deerflow/community/tenki/README.md b/backend/packages/harness/deerflow/community/tenki/README.md new file mode 100644 index 000000000..2f9ebbf34 --- /dev/null +++ b/backend/packages/harness/deerflow/community/tenki/README.md @@ -0,0 +1,97 @@ +# Tenki backend + +Runs each DeerFlow sandbox as a [Tenki](https://tenki.cloud) cloud sandbox — an +isolated microVM created from a stock base image, with no daemon or local +virtualization to manage. A cloud-hosted alternative to the container-based AIO +sandbox and the local-virtualization BoxLite backend. + +## Configuration + +```yaml +sandbox: + use: deerflow.community.tenki:TenkiSandboxProvider + api_key: $TENKI_API_KEY # falls back to TENKI_API_KEY / TENKI_AUTH_TOKEN env var + base_url: https://tenki.cloud # optional; SDK default when omitted + image: my-base-image # optional; Tenki account default base image when omitted + project_id: proj_... # optional; auto-selected if the account has exactly one + workspace_id: ws_... # optional; auto-selected if the account has exactly one + cpu_cores: 2 # optional per-sandbox vCPUs + memory_mb: 2048 # optional per-sandbox memory + replicas: 3 # active + warm microVM cap per gateway process (default: 3) + idle_timeout: 600 # warm microVM idle seconds before terminate; 0 disables + max_duration: 14400 # Tenki sandbox lifetime in seconds (default: 4h); 0 uses the account default + sticky: false # pin the microVM to its host (only matters with pause/resume) + home_dir: /home/tenki # writable dir backing /mnt/user-data (default: /home/tenki) + environment: # injected into every command (and as create-time env) + PYTHONUNBUFFERED: "1" +``` + +Install the optional SDK before selecting this provider: + +```bash +pip install "deerflow-harness[tenki]" +``` + +The `tenki-sandbox` package is an optional DeerFlow harness extra, not part of +the default install. Get an API key from . + +## Design + +Tenki's Python SDK is synchronous, so — unlike BoxLite — the adapter calls the +SDK directly with no event-loop bridge. Sandboxes are named deterministically +from `user_id:thread_id`, released into an in-process warm pool after each agent +turn, and reclaimed (after a liveness health check) by the same thread on the +next acquire. A terminal session error evicts the sandbox and the next acquire +rebuilds it; other transport errors surface to the caller (`exec` is never +auto-retried — it is not idempotent, so re-running could double a command's side +effects). Sandboxes are created with `wait=False` and awaited via `wait_ready()` +so a readiness failure still leaves this provider holding the handle to +terminate, and with an explicit `max_duration` so a long-lived thread does not +lose its sandbox to Tenki's default lifetime mid-conversation. + +| File | Role | +| --- | --- | +| `provider.py` | `SandboxProvider` lifecycle, warm pool, scope resolution | +| `sandbox.py` | `Sandbox` adapter; `execute_command` + file ops | + +## Contract coverage + +The full `Sandbox` surface is implemented. File transport uses Tenki's native `sandbox.fs` +API; directory and content search shell out and reuse `deerflow.sandbox.search`, +mirroring `e2b_sandbox`: + +- `execute_command` — `sh -lc`, with per-call env and timeout. +- `read_file` / `write_file` / `update_file` — native `fs.read_text` / `fs.mkdir` / `fs.write_stream` (binary-safe, streamed). +- `download_file` — native `fs.read_stream`, restricted to the `/mnt/user-data` prefix; the 100 MB cap is enforced on bytes actually received, so a file growing mid-transfer cannot slip past it. +- `list_dir` / `glob` / `grep` — `find` / `grep` with busybox-portable flags (the fs API is single-level and has no content search); results filtered/capped in Python and reported back under `/mnt/user-data`. + +Tenki sandboxes run as the unprivileged `tenki` user with `/mnt` root-owned, so +DeerFlow's `/mnt/user-data` virtual prefix is remapped under the writable +`home_dir` (like `e2b_sandbox`). The provider also best-effort `sudo`-symlinks +`/mnt/user-data` → `home_dir` at create time so agent shell commands using the +literal `/mnt/...` path still work; if `sudo` is unavailable the file APIs keep +working via the remap. + +Warm-pool capacity is governed by `sandbox.replicas` across active + warm +sandboxes. `sandbox.idle_timeout` controls how long released warm sandboxes stay +running; `0` disables idle reaping. Active sandboxes are never evicted to satisfy +the cap. + +## Scope: stable features only + +Only the stable Tenki surface is used — sandbox create/terminate plus +exec/shell/filesystem over shell commands. Volumes, snapshots, and template +builds are intentionally **not** used, so no prebaked image or unstable Tenki +feature is required and any stock base image works. + +Not yet implemented (follow-ups): cross-process orphan reconciliation (adopting +sandboxes left by a previous gateway process) and a preview-URL surface. + +## Status + +Verified end-to-end against live Tenki sandboxes: provider resolution → +`execute_command` → full file-op surface (`read`/`write`/`update`/`download`, +`list_dir`/`glob`/`grep`) → warm-pool reclaim → terminate, plus the +`/mnt/user-data` sudo symlink. Unit tests run in CI without `tenki-sandbox` +installed; `test_integration_real_sandbox` exercises a real microVM when +`TENKI_API_KEY` is set. diff --git a/backend/packages/harness/deerflow/community/tenki/__init__.py b/backend/packages/harness/deerflow/community/tenki/__init__.py new file mode 100644 index 000000000..247a83137 --- /dev/null +++ b/backend/packages/harness/deerflow/community/tenki/__init__.py @@ -0,0 +1,46 @@ +"""Tenki cloud sandbox provider for DeerFlow. + +Integrates `Tenki `_ cloud sandboxes behind DeerFlow's +:class:`Sandbox` / :class:`SandboxProvider` contract. Each sandbox is an +isolated cloud microVM created from a stock base image; the full contract is +implemented — ``execute_command`` plus ``read_file`` / ``write_file`` / +``update_file`` / ``download_file`` / ``list_dir`` / ``glob`` / ``grep`` (file +transport uses Tenki's native ``sandbox.fs`` API; search shells out to +``find`` / ``grep``). + +Configuration example (``config.yaml``):: + + sandbox: + use: deerflow.community.tenki:TenkiSandboxProvider + # Tenki-specific options (read via SandboxConfig's ``extra="allow"``): + api_key: $TENKI_API_KEY # falls back to TENKI_API_KEY / TENKI_AUTH_TOKEN env var + base_url: https://tenki.cloud # optional; SDK default when omitted + image: my-base-image # optional; Tenki account default base image when omitted + project_id: proj_... # optional; auto-selected if the account has exactly one + workspace_id: ws_... # optional; auto-selected if the account has exactly one + cpu_cores: 2 # optional per-sandbox vCPUs + memory_mb: 2048 # optional per-sandbox memory + replicas: 3 # active + warm microVM cap per gateway process + idle_timeout: 600 # warm microVM idle seconds before terminate; 0 disables + max_duration: 14400 # Tenki sandbox lifetime in seconds; 0 uses the account default + sticky: false # pin the microVM to its host (only matters with pause/resume) + home_dir: /home/tenki # writable dir backing /mnt/user-data + environment: # injected into every command (and as create-time env) + PYTHONUNBUFFERED: "1" + +Install the optional SDK before selecting this provider:: + + pip install "deerflow-harness[tenki]" + +Only the stable Tenki surface is used — sandbox create/terminate plus +exec/shell/filesystem. Volumes, snapshots, and template builds are intentionally +not used, so no prebaked image or unstable Tenki feature is required. +""" + +from .provider import TenkiSandboxProvider +from .sandbox import TenkiSandbox + +__all__ = [ + "TenkiSandbox", + "TenkiSandboxProvider", +] diff --git a/backend/packages/harness/deerflow/community/tenki/provider.py b/backend/packages/harness/deerflow/community/tenki/provider.py new file mode 100644 index 000000000..355f5b8a2 --- /dev/null +++ b/backend/packages/harness/deerflow/community/tenki/provider.py @@ -0,0 +1,450 @@ +"""``TenkiSandboxProvider`` — DeerFlow :class:`SandboxProvider` backed by Tenki. + +Integrates `Tenki `_ cloud sandboxes as a DeerFlow sandbox +backend. Each sandbox is an isolated cloud microVM created from a stock base +image; the provider creates one per ``(user, thread)`` and reuses it within the +process, parking released sandboxes in a warm pool (shared +:class:`WarmPoolLifecycleMixin` machinery) for fast reclaim. + +Config is read off :class:`SandboxConfig` (``extra="allow"``), so Tenki keys may +appear under ``sandbox:`` in ``config.yaml`` even though they are not declared on +the model — see this package's ``__init__`` docstring for the full set. + +The Tenki SDK is imported lazily (``_import_client``) so the harness — and every +other provider — installs without ``tenki-sandbox``; the dependency is only +needed once this provider is selected. +""" + +from __future__ import annotations + +import atexit +import hashlib +import logging +import shlex +import threading +import time +import uuid +from typing import TYPE_CHECKING, Any + +from deerflow.config import get_app_config +from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env +from deerflow.sandbox.sandbox_provider import SandboxProvider + +from ..warm_pool_lifecycle import WarmPoolLifecycleMixin +from .sandbox import DEFAULT_TENKI_HOME_DIR, TenkiSandbox + +if TYPE_CHECKING: + from tenki_sandbox import Client + +logger = logging.getLogger(__name__) + +_SANDBOX_NAME_PREFIX = "deer-flow-tenki-" +# Tenki terminates a sandbox at its max lifetime (~30 min by default), which +# would silently drop a long-running thread's state mid-conversation. DeerFlow +# owns the lifecycle here — the warm pool's idle_timeout reaps unused sandboxes +# — so ask for a lifetime that comfortably outlives a research run. Override +# with sandbox.max_duration (seconds); 0 falls back to the Tenki account default. +DEFAULT_MAX_DURATION = 4 * 60 * 60 +# Bootstrap is best-effort and holds the per-scope acquire lock, so bound its +# exec: a hang here (e.g. a stuck sudo) would otherwise stall acquire for that +# scope indefinitely and queue later acquires behind it. A timeout drops it to +# the existing warning path instead. +_BOOTSTRAP_TIMEOUT = 30.0 + + +def _bootstrap_script(home_dir: str) -> str: + """Materialise DeerFlow's virtual path layout inside the sandbox. + + Tenki sandboxes run as the unprivileged ``tenki`` user with ``/mnt`` + root-owned, so ``mkdir -p /mnt/user-data/...`` fails with Permission denied. + Mirroring ``community/e2b_sandbox``, we create the real backing directories + under the writable HOME and best-effort ``sudo``-symlink ``/mnt/user-data`` + to it so commands using the documented ``/mnt/...`` paths still work. If + sudo is unavailable the symlink step is skipped; the file APIs keep working + via :meth:`TenkiSandbox._resolve_path`'s home remap. + + ``sudo -n`` (non-interactive) is deliberate: if the tenki user's sudoers + entry needs a password, a bare ``sudo`` on a tty-allocating exec would block + on the password prompt and — with this call's exec timeout — stall the whole + acquire. ``-n`` fails fast instead, and the existing ``|| true`` swallows it. + """ + home = shlex.quote(home_dir) + return ( + f"mkdir -p {home}/workspace {home}/uploads {home}/outputs; " + f"if command -v sudo >/dev/null 2>&1; then " + f" if [ ! -e /mnt/user-data ] || [ -L /mnt/user-data ]; then " + f" sudo -n ln -sfn {home} /mnt/user-data 2>/dev/null || true; " + f" fi; " + f"fi; " + f"echo BOOTSTRAP_OK" + ) + + +def _import_client() -> type[Client]: + """Import Tenki's ``Client`` lazily. + + Kept out of module import so the harness (and every other provider) installs + without Tenki; the dependency is only needed once this provider is selected. + """ + try: + from tenki_sandbox import Client + except ImportError as e: # pragma: no cover - depends on the optional dependency + raise ImportError("TenkiSandboxProvider requires the optional 'tenki-sandbox' dependency. Install it with: pip install 'deerflow-harness[tenki]' or pip install tenki-sandbox.") from e + return Client + + +class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider): + """Run each DeerFlow sandbox as a Tenki cloud microVM.""" + + uses_thread_data_mounts = False + needs_upload_permission_adjustment = True + _idle_checker_thread_name = "tenki-idle-reaper" + + @staticmethod + def _sandbox_id(thread_id: str, user_id: str) -> str: + """Deterministic sandbox ID from user/thread scope. + + Includes user_id so a sandbox created for one user's bucket cannot be + reclaimed by another user's thread with the same thread_id. The warm + pool is keyed by this id alone (``_reclaim_warm_pool`` looks it up + directly, with no full-seed fallback), so on a hosted multi-tenant + gateway a hash collision would let one user reclaim another's parked + sandbox. 64 bits, like community/e2b_sandbox, keeps that negligible. + """ + return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16] + + # ── Provider lifecycle ─────────────────────────────────────────────── + + def __init__(self) -> None: + self._lock = threading.Lock() + self._sandboxes: dict[str, TenkiSandbox] = {} + self._thread_sandboxes: dict[tuple[str, str], str] = {} + self._warm_pool: dict[str, tuple[TenkiSandbox, float]] = {} + self._acquire_locks: dict[str, threading.Lock] = {} + self._idle_checker_stop = threading.Event() + self._idle_checker_thread: threading.Thread | None = None + self._shutdown_called = False + self._client: Client | None = None + self._config = self._load_config() + atexit.register(self.shutdown) + self._start_idle_checker() + + def _load_config(self) -> dict[str, Any]: + 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") + replicas = _opt("replicas") + idle_timeout = _opt("idle_timeout") + max_duration = _opt("max_duration") + environment = dict(_opt("environment") or {}) + # Fail fast on a misconfigured key (e.g. "bad-key"): the per-call env goes + # through the same POSIX-name check in execute_command, but this static + # config env is merged into every command and would otherwise only surface + # as a confusing SDK error at create/exec time. + _validate_extra_env(environment) + return { + "max_duration": float(max_duration if max_duration is not None else DEFAULT_MAX_DURATION), + # Off by default (the SDK default). Warm-pool sandboxes stay running + # between turns, so host pinning only matters to deployments that + # also pause/resume; expose it rather than decide for them. + "sticky": bool(_opt("sticky", False)), + "api_key": api_key, # None → SDK falls back to TENKI_API_KEY / TENKI_AUTH_TOKEN + "base_url": _opt("base_url"), + "image": _opt("image"), # None → Tenki account default base image + "home_dir": _opt("home_dir") or DEFAULT_TENKI_HOME_DIR, + "project_id": _opt("project_id"), + "workspace_id": _opt("workspace_id"), + "cpu_cores": _opt("cpu_cores"), + "memory_mb": _opt("memory_mb"), + "environment": environment, + "replicas": replicas if replicas is not None else self.DEFAULT_REPLICAS, + "idle_timeout": idle_timeout if idle_timeout is not None else self.DEFAULT_IDLE_TIMEOUT, + } + + def _get_client(self) -> Client: + with self._lock: + if self._client is not None: + return self._client + client_cls = _import_client() + client = client_cls(auth_token=self._config["api_key"], base_url=self._config["base_url"]) + with self._lock: + if self._client is None: + self._client = client + return self._client + + def _resolve_scope(self) -> tuple[str | None, str | None]: + """Return (project_id, workspace_id), auto-selecting when unambiguous. + + Tenki's ``create`` needs a project scope. When the caller didn't set one + in config, pick it if the account has exactly one workspace and project; + otherwise raise with the choices so the operator can set ``project_id``. + """ + project_id = self._config["project_id"] + workspace_id = self._config["workspace_id"] + if project_id is not None: + return project_id, workspace_id + + identity = self._get_client().who_am_i() + workspaces = list(identity.workspaces or []) + if workspace_id is not None: + workspaces = [w for w in workspaces if w.id == workspace_id] + workspace = self._require_single(workspaces, "workspace", "workspace_id") + project = self._require_single(list(workspace.projects or []), "project", "project_id") + return project.id, workspace.id + + @staticmethod + def _require_single(items: list[Any], kind: str, param: str) -> Any: + if len(items) != 1: + raise ValueError(f"Could not auto-select a Tenki {kind}; set sandbox.{param} in config.yaml. Found: {[(getattr(i, 'id', None), getattr(i, 'name', None)) for i in items]}") + return items[0] + + @staticmethod + def _thread_key(thread_id: str, user_id: str | None) -> tuple[str, str]: + return (user_id or "", thread_id) + + @classmethod + def _sandbox_name(cls, sandbox_id: str) -> str: + return f"{_SANDBOX_NAME_PREFIX}{sandbox_id}" + + def _lock_for_sandbox(self, sandbox_id: str) -> threading.Lock: + with self._lock: + lock = self._acquire_locks.get(sandbox_id) + if lock is None: + lock = threading.Lock() + self._acquire_locks[sandbox_id] = lock + return lock + + def _start_idle_checker(self) -> None: + """Start idle cleanup when enabled; idle_timeout=0 keeps it disabled.""" + if self._config["idle_timeout"] <= 0: + return + super()._start_idle_checker() + + def _active_count_locked(self) -> int: + return len(self._sandboxes) + + def _destroy_warm_entry(self, sandbox_id: str, entry: TenkiSandbox, *, reason: str) -> None: + self._close_quietly(entry, context=f"warm pool, reason={reason}") + + def _invalidate_sandbox(self, sandbox_id: str, reason: str) -> None: + """Destroy and deregister a sandbox after a terminal command-path failure.""" + to_close: TenkiSandbox | None = None + with self._lock: + active = self._sandboxes.pop(sandbox_id, None) + warm_entry = self._warm_pool.pop(sandbox_id, None) + for key in [k for k, sid in self._thread_sandboxes.items() if sid == sandbox_id]: + self._thread_sandboxes.pop(key, None) + to_close = active or (warm_entry[0] if warm_entry is not None else None) + + if to_close is None: + logger.warning("Tenki sandbox %s failed terminally but was not tracked: %s", sandbox_id, reason) + return + logger.warning("Invalidating Tenki sandbox %s after terminal failure: %s", sandbox_id, reason) + self._close_quietly(to_close, context="terminal failure") + + # ── Acquire / release ──────────────────────────────────────────────── + + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + if thread_id is None: + sandbox_id = str(uuid.uuid4())[:8] + sandbox = self._create_sandbox(sandbox_id) + with self._lock: + self._sandboxes[sandbox.id] = sandbox + return sandbox.id + + key = self._thread_key(thread_id, user_id) + sandbox_id = self._sandbox_id(thread_id, user_id or "") + acquire_lock = self._lock_for_sandbox(sandbox_id) + with acquire_lock: + with self._lock: + existing = self._thread_sandboxes.get(key) + if existing is not None and existing in self._sandboxes: + return existing + + reclaimed = self._reclaim_warm_pool(sandbox_id) + if reclaimed is not None: + with self._lock: + self._thread_sandboxes[key] = reclaimed + return reclaimed + + sandbox = self._create_sandbox(sandbox_id) + with self._lock: + self._sandboxes[sandbox.id] = sandbox + self._thread_sandboxes[key] = sandbox.id + return sandbox.id + + def _create_sandbox(self, sandbox_id: str) -> TenkiSandbox: + # Enforce replica soft cap: evict the oldest warm sandbox if active + warm + # are at capacity. + replicas, total = self._replica_count() + if total >= replicas: + evicted = self._evict_oldest_warm() + self._log_replicas_soft_cap(replicas, sandbox_id, evicted) + + client = self._get_client() + project_id, workspace_id = self._resolve_scope() + create_kwargs: dict[str, Any] = { + "name": self._sandbox_name(sandbox_id), + "project_id": project_id, + "workspace_id": workspace_id, + "sticky": self._config["sticky"], + # Wait for readiness ourselves (below) instead of inside create(): + # create(wait=True) raises with the session handle still local to the + # SDK, so a readiness failure would leak a running, billed microVM + # that this provider never sees and can never terminate. + "wait": False, + } + if self._config["max_duration"] > 0: + create_kwargs["max_duration"] = self._config["max_duration"] + for key in ("image", "cpu_cores", "memory_mb"): + if self._config[key] is not None: + create_kwargs[key] = self._config[key] + if self._config["environment"]: + create_kwargs["env"] = self._config["environment"] + + remote = client.create(**create_kwargs) + try: + remote.wait_ready() + except Exception: + self._terminate_orphan(sandbox_id, remote) + raise + + # Materialise DeerFlow's virtual path layout under the writable HOME. + # Best-effort: on failure the file APIs still work via the home remap. + try: + result = remote.exec("sh", "-lc", _bootstrap_script(self._config["home_dir"]), timeout=_BOOTSTRAP_TIMEOUT) + if result.exit_code not in (0, None) or "BOOTSTRAP_OK" not in (result.stdout_text or ""): + logger.warning( + "Tenki bootstrap for %s exited code=%s stderr=%s", + sandbox_id, + result.exit_code, + (result.stderr_text or "").strip(), + ) + except Exception as e: + logger.warning("Tenki bootstrap for %s raised: %s", sandbox_id, e) + logger.info("Created Tenki sandbox %s (name=%s)", sandbox_id, self._sandbox_name(sandbox_id)) + return TenkiSandbox( + sandbox_id, + remote, + default_env=self._config["environment"], + home_dir=self._config["home_dir"], + on_terminal_failure=self._invalidate_sandbox, + ) + + @staticmethod + def _terminate_orphan(sandbox_id: str, remote: Any) -> None: + """Terminate a microVM that was created but never handed to the adapter.""" + try: + remote.close() + logger.warning("Terminated Tenki sandbox %s after it failed to become ready", sandbox_id) + except Exception as e: + logger.error("Leaked Tenki sandbox %s (id=%s): could not terminate after readiness failure: %s", sandbox_id, getattr(remote, "id", "?"), e) + + @staticmethod + def _close_quietly(sandbox: TenkiSandbox, *, context: str) -> None: + """Close a sandbox where the caller has no way to act on a failure.""" + try: + sandbox.close() + except Exception as e: + logger.warning("Error closing Tenki sandbox %s (%s): %s", sandbox.id, context, e) + + def get(self, sandbox_id: str) -> Sandbox | None: + with self._lock: + return self._sandboxes.get(sandbox_id) + + def release(self, sandbox_id: str) -> None: + """Release a sandbox into the warm pool — the microVM stays running. + + The sandbox moves from ``_sandboxes`` to ``_warm_pool`` and its + ``_thread_sandboxes`` entries are cleared. It is NOT terminated unless + shutdown has already begun. + """ + close_sandbox: TenkiSandbox | None = None + with self._lock: + sandbox = self._sandboxes.pop(sandbox_id, None) + for key in [k for k, sid in self._thread_sandboxes.items() if sid == sandbox_id]: + self._thread_sandboxes.pop(key, None) + if sandbox is None: + return + if self._shutdown_called: + close_sandbox = sandbox + else: + self._warm_pool[sandbox_id] = (sandbox, time.time()) + + if close_sandbox is not None: + self._close_quietly(close_sandbox, context="released during shutdown") + logger.info("Closed released Tenki sandbox %s because shutdown is in progress", sandbox_id) + else: + logger.info("Released Tenki sandbox %s to warm pool (microVM still running)", sandbox_id) + + def _reclaim_warm_pool(self, sandbox_id: str) -> str | None: + """Reclaim a warm-pool sandbox by id after a liveness health check. + + Returns sandbox_id on success, None if not present or the health check + fails (the dead entry is destroyed). + """ + with self._lock: + if sandbox_id not in self._warm_pool: + return None + sandbox, _ = self._warm_pool[sandbox_id] + + try: + result = sandbox.execute_command("echo ok", timeout=10) + healthy = "ok" in result + except Exception as e: + logger.warning("Tenki warm-pool sandbox %s health check error: %s", sandbox_id, e) + healthy = False + + if not healthy: + with self._lock: + warm_entry = self._warm_pool.pop(sandbox_id, None) + if warm_entry is not None: + self._destroy_warm_entry(sandbox_id, warm_entry[0], reason="health_check_failed") + return None + + with self._lock: + warm_entry = self._warm_pool.pop(sandbox_id, None) + if warm_entry is None: + return None # Raced with another thread + self._sandboxes[sandbox_id] = warm_entry[0] + + logger.info("Reclaimed warm-pool Tenki sandbox %s", sandbox_id) + return sandbox_id + + def reset(self) -> None: + """Park tracked sandboxes into this instance's warm pool for later cleanup. + + ``reset_sandbox_provider()`` drops the singleton and calls this so config + changes take effect on the next construction. Teardown belongs to + ``shutdown()``; reset leaves running microVMs alive but visible to this + instance's idle reaper and atexit shutdown instead of orphaning them. + """ + with self._lock: + now = time.time() + for sandbox_id, sandbox in self._sandboxes.items(): + self._warm_pool.setdefault(sandbox_id, (sandbox, now)) + self._sandboxes.clear() + self._thread_sandboxes.clear() + self._acquire_locks.clear() + + def shutdown(self) -> None: + with self._lock: + if self._shutdown_called: + return + self._shutdown_called = True + + self._stop_idle_checker() + + with self._lock: + active = list(self._sandboxes.values()) + warm = [sandbox for sandbox, _ in self._warm_pool.values()] + self._sandboxes.clear() + self._warm_pool.clear() + self._thread_sandboxes.clear() + self._acquire_locks.clear() + + for sandbox in active + warm: + self._close_quietly(sandbox, context="shutdown") diff --git a/backend/packages/harness/deerflow/community/tenki/sandbox.py b/backend/packages/harness/deerflow/community/tenki/sandbox.py new file mode 100644 index 000000000..995a7fa62 --- /dev/null +++ b/backend/packages/harness/deerflow/community/tenki/sandbox.py @@ -0,0 +1,464 @@ +"""``TenkiSandbox`` — DeerFlow :class:`Sandbox` backed by a Tenki cloud sandbox. + +Tenki's Python SDK (``tenki-sandbox``) is synchronous, so — unlike +``community/boxlite`` — this adapter calls the SDK directly with no event-loop +bridge. File transport uses Tenki's native ``sandbox.fs`` API (``read_text`` / +``read_stream`` / ``write_stream`` / ``mkdir`` / ``stat``), which is binary-safe +and streams, so no base64/shell encoding is involved. Directory and content +*search* (``list_dir`` / ``glob`` / ``grep``) still shells out to ``find`` / +``grep`` — the fs API is single-level and has no content search — and is parsed +with the shared ``deerflow.sandbox.search`` helpers, the same approach as +``community/e2b_sandbox``. Those commands use only busybox-portable flags so any +Tenki base image works. + +The Tenki SDK is not imported at module load (only its exception *class names* +are matched, as strings), so importing this package never requires +``tenki-sandbox`` to be installed — it is needed only once the provider is +selected and a sandbox is actually created. +""" + +from __future__ import annotations + +import errno +import logging +import posixpath +import re +import shlex +import threading +from typing import TYPE_CHECKING, Any, TypeVar + +from deerflow.config.paths import VIRTUAL_PATH_PREFIX +from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env +from deerflow.sandbox.search import GrepMatch, path_matches, should_ignore_path, truncate_line + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from tenki_sandbox import Sandbox as TenkiClientSandbox + from tenki_sandbox.fs import SandboxFS + +T = TypeVar("T") + +logger = logging.getLogger(__name__) + +_MAX_DOWNLOAD_SIZE = 100 * 1024 * 1024 # 100 MB +# Tenki sandboxes run as the unprivileged ``tenki`` user (HOME=/home/tenki) and +# ``/mnt`` is root-owned, so DeerFlow's ``/mnt/user-data`` virtual prefix is not +# writable directly. Like ``community/e2b_sandbox``, file ops are remapped under +# this home dir (the provider also best-effort symlinks /mnt/user-data → here so +# agent shell commands using the literal path still work). +DEFAULT_TENKI_HOME_DIR = "/home/tenki" +# Frame size for fs.write_stream uploads. +_STREAM_CHUNK = 1024 * 1024 + +# Tenki SDK exception *class names* that mean the remote session is gone for +# good — matched as strings so this module imports without ``tenki-sandbox``. +# A terminated/not-found/closed session is unrecoverable; the provider drops it +# and rebuilds on the next call. This is only the named-error half of the rule: +# _is_terminal_failure ALSO treats the builtin ConnectionError / BrokenPipeError +# / EOFError as terminal via isinstance, so a transport reset evicts the sandbox +# and cold-starts the next acquire too. That is a deliberate fail-safe (a reset +# often means the microVM is gone); the cost is churning a warm sandbox on a +# one-off flaky-network blip. +_TERMINAL_ERROR_NAMES = frozenset( + { + "SessionTerminatedError", + "SessionNotFoundError", + "InvalidStateError", + "StreamClosedError", + } +) + + +class TenkiSandbox(Sandbox): + """DeerFlow Sandbox adapter that delegates to a live Tenki cloud sandbox. + + Args: + id: DeerFlow-side sandbox id (the provider's cache key). + sandbox: A live, started ``tenki_sandbox.Sandbox``. The provider owns + its lifecycle; this adapter terminates it on :meth:`close`. + default_env: Static environment merged into every command, overridden + per-call by the ``env`` passed to :meth:`execute_command` + (request-scoped secrets). + home_dir: Writable directory that backs the ``VIRTUAL_PATH_PREFIX`` + (``/mnt/user-data``) prefix inside the sandbox. Defaults to + :data:`DEFAULT_TENKI_HOME_DIR`. + on_terminal_failure: Optional callback ``(sandbox_id, reason)`` invoked + when an operation fails with a terminal Tenki error, so the provider + can evict the dead sandbox. + """ + + def __init__( + self, + id: str, + sandbox: TenkiClientSandbox, + *, + default_env: dict[str, str] | None = None, + home_dir: str = DEFAULT_TENKI_HOME_DIR, + on_terminal_failure: Callable[[str, str], None] | None = None, + ) -> None: + super().__init__(id) + self._sandbox = sandbox + self._default_env = dict(default_env or {}) + self._home_dir = home_dir.rstrip("/") or "/" + self._on_terminal_failure = on_terminal_failure + self._lock = threading.Lock() + # Serialises the append read-modify-write across its three fs ops. A + # lock distinct from _lock, so it can wrap the whole sequence without the + # per-op eviction callback (which reaches back into the provider) ever + # running under it. + self._write_lock = threading.Lock() + self._closed = False + + @property + def is_closed(self) -> bool: + with self._lock: + return self._closed + + @staticmethod + def _is_terminal_failure(error: Exception) -> bool: + if isinstance(error, (BrokenPipeError, ConnectionError, EOFError)): + return True + return type(error).__name__ in _TERMINAL_ERROR_NAMES + + def close(self) -> None: + """Terminate the underlying Tenki session (idempotent). + + The microVM is terminated *first*; the adapter is only marked closed once + the session is actually gone, so a failed termination stays retryable + instead of silently leaking a running (billed) sandbox. A terminal + session error means it is already gone, which counts as closed; anything + else is raised so the caller can retry or alert. + """ + with self._lock: + if self._closed: + return + sandbox = self._sandbox + try: + sandbox.close() + except Exception as e: + if not self._is_terminal_failure(e): + logger.error("Error terminating Tenki sandbox %s: %s", self.id, e) + raise + logger.info("Tenki sandbox %s was already gone at close: %s", self.id, e) + with self._lock: + self._closed = True + + # ── bridge helpers ────────────────────────────────────────────────── + + def _note_failure(self, error: Exception) -> None: + """Evict this sandbox when an operation failed with a terminal error.""" + if self._on_terminal_failure is None or not self._is_terminal_failure(error): + return + try: + self._on_terminal_failure(self.id, str(error)) + except Exception: + logger.exception("Terminal Tenki failure callback errored for %s", self.id) + + def _fs_op(self, op: Callable[[SandboxFS], T]) -> T: + """Run a native ``sandbox.fs`` call, evicting the sandbox on terminal errors. + + The lock is held across ``op`` (not just the fs lookup) so concurrent + calls on the same sandbox serialise: the Tenki SDK shares one connection + per instance, like community/e2b_sandbox. ``_note_failure`` runs *after* + the lock is released — it reaches back into the provider, which locks in + the opposite order (provider then sandbox), so holding both at once could + deadlock. + """ + with self._lock: + if self._closed: + raise RuntimeError("sandbox has been closed") + fs = self._sandbox.fs + try: + return op(fs) + except Exception as e: + failure = e + self._note_failure(failure) + raise failure + + def _exec(self, *argv: str, env: dict[str, str] | None = None, timeout: float | None = None) -> Any: + # No forced cwd: commands run in the sandbox default working directory + # (like community/e2b_sandbox and community/boxlite); file ops address + # absolute, home-remapped paths, so cwd is irrelevant to them. + # + # No auto-retry: exec is not idempotent (the command may have run + # server-side before a transport ack dropped), so re-running it risks + # double side effects. Like boxlite, a transient error is surfaced to the + # caller (returned as text by execute_command); a terminal session error + # additionally evicts the sandbox so the next acquire rebuilds it. + with self._lock: + if self._closed: + raise RuntimeError("sandbox has been closed") + sandbox = self._sandbox + try: + return sandbox.exec(*argv, env=env, timeout=timeout) + except Exception as e: + self._note_failure(e) + raise + + def _sh(self, script: str, env: dict[str, str] | None = None, timeout: float | None = None) -> Any: + return self._exec("sh", "-lc", script, env=env, timeout=timeout) + + # ── path safety (mirrors community/e2b_sandbox) ────────────────────── + + @staticmethod + def _guard_traversal(path: str) -> str: + if not path: + raise ValueError("path must be a non-empty string") + normalized = path.replace("\\", "/") + for segment in normalized.split("/"): + if segment == "..": + raise PermissionError(f"Access denied: path traversal detected in '{path}'") + return normalized + + def _resolve_path(self, path: str) -> str: + """Map DeerFlow virtual paths into the writable sandbox home dir. + + ``VIRTUAL_PATH_PREFIX`` (``/mnt/user-data``) is rewritten under + :attr:`_home_dir`; other absolute paths pass through so the sandbox can + reach system directories when needed. Traversal is always rejected. + """ + normalized = self._guard_traversal(path) + if normalized == VIRTUAL_PATH_PREFIX or normalized.startswith(f"{VIRTUAL_PATH_PREFIX}/"): + tail = normalized[len(VIRTUAL_PATH_PREFIX) :].lstrip("/") + return f"{self._home_dir}/{tail}".rstrip("/") if tail else self._home_dir + return normalized + + def _virtual_path(self, resolved: str) -> str: + """Inverse of :meth:`_resolve_path` — the form callers gave us. + + Everything that *returns* paths (``list_dir``/``glob``/``grep``) reports + them under ``VIRTUAL_PATH_PREFIX``, not the sandbox-internal home dir, so + results can be fed straight back into the other file APIs. + """ + if resolved == self._home_dir: + return VIRTUAL_PATH_PREFIX + if resolved.startswith(f"{self._home_dir}/"): + return f"{VIRTUAL_PATH_PREFIX}/{resolved[len(self._home_dir) :].lstrip('/')}" + return resolved + + # ── command execution ─────────────────────────────────────────────── + + def execute_command( + self, + command: str, + env: dict[str, str] | None = None, + timeout: float | None = None, + ) -> str: + """Run ``command`` through a shell in the Tenki sandbox and return output. + + DeerFlow passes a bash command *string*; it runs through ``sh -lc``. + Per-call ``env`` is layered over the static config environment and + scoped to this command only (request-scoped secrets, issue #3861). + """ + _validate_extra_env(env) # POSIX env-var key rule; raises ValueError on a bad key + if self.is_closed: + return "Error: sandbox has been closed" + merged_env = {**self._default_env, **(env or {})} or None + try: + result = self._sh(command, env=merged_env, timeout=timeout) + except Exception as e: + logger.error("Failed to execute command in Tenki sandbox %s: %s", self.id, e) + return f"Error: {e}" + + stdout = result.stdout_text or "" + stderr = result.stderr_text or "" + if stdout and stderr: + output = f"{stdout}\n{stderr}" + else: + output = stdout or stderr + if result.exit_code not in (0, None) and not output: + output = f"Command exited with code {result.exit_code}" + return output if output else "(no output)" + + # ── file operations ───────────────────────────────────────────────── + + def read_file(self, path: str) -> str: + resolved = self._resolve_path(path) + try: + return self._fs_op(lambda fs: fs.read_text(resolved)) + except Exception as e: + logger.error("read_file %s failed: %s", resolved, e) + return f"Error: {e}" + + def write_file(self, path: str, content: str, append: bool = False) -> None: + self._write_bytes(self._resolve_path(path), content.encode("utf-8"), append=append) + + def update_file(self, path: str, content: bytes) -> None: + self._write_bytes(self._resolve_path(path), content, append=False) + + def _write_bytes(self, resolved: str, data: bytes, *, append: bool) -> None: + parent = posixpath.dirname(resolved) + if not append: + if parent: + self._fs_op(lambda fs: fs.mkdir(parent)) + self._fs_op(lambda fs: fs.write_stream(resolved, _frames(data))) + return + + # Tenki's write stream has no append mode (it starts at offset 0), so we + # read-modify-write like community/e2b_sandbox. The read and the write are + # separate fs ops, so two concurrent appends could both read the same + # pre-image and the second would clobber the first; _write_lock makes the + # whole sequence atomic. + with self._write_lock: + if parent: + self._fs_op(lambda fs: fs.mkdir(parent)) + try: + data = self._fs_op(lambda fs: fs.read_bytes(resolved)) + data + except Exception as e: + if type(e).__name__ != "FileNotFoundError": + raise + self._fs_op(lambda fs: fs.write_stream(resolved, _frames(data))) + + def download_file(self, path: str) -> bytes: + normalized = self._guard_traversal(path) + stripped = normalized.lstrip("/") + allowed = VIRTUAL_PATH_PREFIX.lstrip("/") + if stripped != allowed and not stripped.startswith(f"{allowed}/"): + raise PermissionError(f"Access denied: path must be under '{VIRTUAL_PATH_PREFIX}': '{path}'") + resolved = self._resolve_path(path) + + with self._lock: + if self._closed: + raise RuntimeError("sandbox has been closed") + fs = self._sandbox.fs + + # Deliberate: the lock is dropped before streaming, unlike _fs_op which + # holds it across its op. _fs_op's serialization guards short, bounded + # calls; a download can be up to _MAX_DOWNLOAD_SIZE (100 MB), and holding + # the instance lock across it would block every other tool on this + # sandbox for the whole transfer. The Tenki read stream is safe to run + # alongside other ops (the SDK multiplexes over its connection), so we + # accept the interleave here for latency and still evict on a terminal + # transport error via _note_failure below. + # + # The cap is enforced on bytes actually received, so a file that grows + # mid-transfer still can't exceed it (a stat-then-read check could). + chunks: list[bytes] = [] + total = 0 + try: + for chunk in fs.read_stream(resolved): + total += len(chunk) + if total > _MAX_DOWNLOAD_SIZE: + raise OSError(errno.EFBIG, f"File exceeds maximum download size of {_MAX_DOWNLOAD_SIZE} bytes", path) + chunks.append(chunk) + except OSError as e: + # Our own EFBIG size-cap is not a session death — let it pass through + # without evicting. Every other OSError is a real transport failure: + # ConnectionError / BrokenPipeError / EOFError are OSError subclasses + # that _is_terminal_failure treats as terminal, so they must route + # through _note_failure like _fs_op/_exec do. Without this, a session + # that dies mid-download is never evicted and the agent keeps hitting + # OSErrors until some other op happens to reap it. + if e.errno == errno.EFBIG: + raise + self._note_failure(e) + raise + except Exception as e: + self._note_failure(e) + raise OSError(f"cannot read '{path}' from sandbox: {e}") from e + return b"".join(chunks) + + def list_dir(self, path: str, max_depth: int = 2) -> list[str]: + resolved = self._resolve_path(path) + r = self._sh(f"find {shlex.quote(resolved)} -maxdepth {int(max_depth)} \\( -type f -o -type d \\) 2>/dev/null | head -500") + return [self._virtual_path(line.strip()) for line in (r.stdout_text or "").splitlines() if line.strip()] + + 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",) + type_expr = " -o ".join(f"-type {t}" for t in types) + hard_limit = max(max_results * 4, max_results + 50) + r = self._sh(f"find {shlex.quote(resolved)} \\( {type_expr} \\) -print 2>/dev/null | head -{hard_limit}") + + matches: list[str] = [] + root = resolved.rstrip("/") or "/" + root_prefix = root if root == "/" else f"{root}/" + for entry in (r.stdout_text or "").splitlines(): + entry = entry.strip() + if not entry or (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(self._virtual_path(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]: + # Validate a regex pattern at the boundary (grep uses POSIX ERE, but this + # catches gross errors); a literal needs none. grep receives the RAW + # pattern: -F matches it literally, -E as a regex. + if not literal: + re.compile(pattern, 0 if case_sensitive else re.IGNORECASE) + + resolved = self._resolve_path(path) + # busybox+GNU-portable flags: -r recursive, -H always print the filename + # (without it, grep -r on a path that resolves to a single file prints + # "line:text" and the file:line:text unpack below drops every match), -n + # line numbers, -I skip binary, -E/-F regex vs fixed. --include and -m are + # omitted for busybox portability; glob-scoping and the result cap are + # applied in Python below. + flags = ["-r", "-H", "-n", "-I"] + if not case_sensitive: + flags.append("-i") + flags.append("-F" if literal else "-E") + total_cap = max(max_results * 4, max_results + 50) + cmd = "grep " + " ".join(flags) + f" -e {shlex.quote(pattern)} {shlex.quote(resolved)} 2>/dev/null | head -{total_cap}" + r = self._sh(cmd) + + root = resolved.rstrip("/") or "/" + root_prefix = root if root == "/" else f"{root}/" + matches: list[GrepMatch] = [] + truncated = False + for raw in (r.stdout_text or "").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 + if glob is not None: + # Match the caller's real directory scope: a pattern like + # "src/*.js" must not broaden to every *.js in the tree. Same + # helper, same relative-to-root semantics as glob() above. + if file_path != root and not file_path.startswith(root_prefix): + continue + rel_path = file_path[len(root) :].lstrip("/") + if not rel_path or not path_matches(glob, rel_path): + continue + matches.append(GrepMatch(path=self._virtual_path(file_path), line_number=line_number, line=truncate_line(line_text))) + if len(matches) >= max_results: + truncated = True + break + return matches, truncated + + +def _frames(data: bytes) -> Iterator[bytes]: + """Slice ``data`` into upload frames for ``fs.write_stream``.""" + for i in range(0, len(data), _STREAM_CHUNK): + yield data[i : i + _STREAM_CHUNK] diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index ee29c65fc..03f9652f2 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -67,6 +67,9 @@ postgres = [ redis = ["redis>=5.0.0"] pymupdf = ["pymupdf4llm>=0.0.17"] boxlite = ["boxlite>=0.9.7"] +# Tenki cloud sandbox provider (deerflow.community.tenki). Optional so a default +# install stays free of the Tenki SDK; only pulled in when the provider is used. +tenki = ["tenki-sandbox>=0.4.0"] # Agent observability (Monocle). Optional so a default install stays free of the # OpenTelemetry stack; only pulled in when MONOCLE_TRACING is used. monocle = ["monocle_apptrace>=0.8.8"] diff --git a/backend/tests/test_tenki_provider.py b/backend/tests/test_tenki_provider.py new file mode 100644 index 000000000..2fe8ba93b --- /dev/null +++ b/backend/tests/test_tenki_provider.py @@ -0,0 +1,933 @@ +"""Unit tests for the Tenki community sandbox provider. + +These run in CI without ``tenki-sandbox`` installed: they cover the lazy-import +error path, provider lifecycle, path-safety guards, the native ``fs`` file +round-trip, warm-pool mechanics, and scope resolution — none of which need a live +sandbox. A single opt-in integration test (``test_integration_real_sandbox``) +exercises a real Tenki microVM end to end when ``TENKI_API_KEY`` is set. +""" + +from __future__ import annotations + +import errno +import os +import shlex +import sys +import threading +import time +import types + +import pytest + +from deerflow.community.tenki.provider import _BOOTSTRAP_TIMEOUT, TenkiSandboxProvider, _import_client +from deerflow.community.tenki.sandbox import TenkiSandbox + +# ── Fake Tenki SDK ──────────────────────────────────────────────────── + + +class _FakeResult: + def __init__(self, exit_code: int = 0, stdout: bytes = b"", stderr: bytes = b"") -> None: + self.exit_code = exit_code + self.stdout = stdout + self.stderr = stderr + + @property + def stdout_text(self) -> str: + return self.stdout.decode(errors="replace") + + @property + def stderr_text(self) -> str: + return self.stderr.decode(errors="replace") + + +class _FakeSessionTerminatedError(RuntimeError): + """Stand-in whose class name matches a Tenki terminal error.""" + + +# Rename so ``type(e).__name__`` matches the adapter's terminal-name set. +_FakeSessionTerminatedError.__name__ = "SessionTerminatedError" + + +class _FakeMissingFileError(Exception): + """Stand-in for ``tenki_sandbox.errors.FileNotFoundError`` (matched by name).""" + + +_FakeMissingFileError.__name__ = "FileNotFoundError" + + +class _FakeFileInfo: + def __init__(self, path: str, size: int) -> None: + self.path = path + self.size = size + self.is_dir = False + + +class _FakeFS: + """In-memory stand-in for Tenki's native ``sandbox.fs`` API.""" + + _READ_FRAME = 64 # small, so multi-frame reads are exercised + + def __init__(self, owner: _FakeSandbox) -> None: + self._owner = owner + + def _guard(self) -> None: + if self._owner.fs_error is not None: + raise self._owner.fs_error + + def mkdir(self, path: str, *, recursive: bool = True, mode: int = 0o755) -> None: + self._guard() + self._owner.dirs.add(path) + + def read_bytes(self, path: str) -> bytes: + self._guard() + if path not in self._owner.files: + raise _FakeMissingFileError(f"no such file or directory: {path}") + return self._owner.files[path] + + def read_text(self, path: str, *, encoding: str = "utf-8") -> str: + return self.read_bytes(path).decode(encoding, errors="replace") + + def read_stream(self, path: str, *, offset: int = 0, length: int = 0, chunk_bytes: int = 0): + # A generator, like the real SDK: errors surface while iterating. + data = self.read_bytes(path) + for i in range(0, len(data), self._READ_FRAME): + yield data[i : i + self._READ_FRAME] + + def write_stream(self, path: str, chunks, *, mode: int = 0o644, truncate: bool = True, sync: bool = False) -> None: + self._guard() + frames = list(chunks) + self._owner.write_frame_counts.append(len(frames)) + self._owner.files[path] = b"".join(frames) + + def stat(self, path: str) -> _FakeFileInfo: + self._guard() + if path not in self._owner.files: + raise _FakeMissingFileError(f"no such file or directory: {path}") + return _FakeFileInfo(path, len(self._owner.files[path])) + + +class _FakeSandbox: + """A fake tenki ``Sandbox``: a native ``fs`` API plus the handful of shell + commands the adapter still emits for search, backed by one in-memory + filesystem — so file and search round-trips are exercised for real. + """ + + def __init__( + self, + *, + exec_error: Exception | None = None, + close_error: Exception | None = None, + fs_error: Exception | None = None, + wait_ready_error: Exception | None = None, + ) -> None: + self.id = "remote-session-id" + self.files: dict[str, bytes] = {} + self.dirs: set[str] = set() + self.write_frame_counts: list[int] = [] + self.exec_calls: list[dict] = [] + self.exec_error = exec_error + self.close_error = close_error + self.fs_error = fs_error + self.wait_ready_error = wait_ready_error + self.wait_ready_calls = 0 + self.closed = False + self.fs = _FakeFS(self) + + def wait_ready(self, timeout: float = 180) -> None: + self.wait_ready_calls += 1 + if self.wait_ready_error is not None: + raise self.wait_ready_error + + def exec(self, *argv: str, cwd=None, env=None, timeout=None): + self.exec_calls.append({"argv": argv, "cwd": cwd, "env": env, "timeout": timeout}) + if self.exec_error is not None: + raise self.exec_error + if argv[:2] == ("sh", "-lc"): + return self._run_script(argv[2]) + return _FakeResult() + + def _run_script(self, script: str) -> _FakeResult: + if script == "echo ok": + return _FakeResult(stdout=b"ok\n") + if "BOOTSTRAP_OK" in script: # provider create-time bootstrap script + return _FakeResult(stdout=b"BOOTSTRAP_OK\n") + if script.startswith("find "): + root = shlex.split(script)[1] + hits = [p for p in self.files if p == root or p.startswith(f"{root.rstrip('/')}/")] + return _FakeResult(stdout=("\n".join(hits) + "\n").encode() if hits else b"") + if script.startswith("grep "): + # grep -e 2>/dev/null | head -N + tokens = shlex.split(script) + needle, root = tokens[tokens.index("-e") + 1], tokens[tokens.index("-e") + 2] + lines = [] + for path, blob in sorted(self.files.items()): + if not path.startswith(root): + continue + for n, text in enumerate(blob.decode(errors="replace").splitlines(), start=1): + if needle in text: + lines.append(f"{path}:{n}:{text}") + return _FakeResult(stdout=("\n".join(lines) + "\n").encode() if lines else b"") + return _FakeResult() + + def close(self): + self.closed = True + if self.close_error is not None: + raise self.close_error + + +class _FakeProject: + def __init__(self, id: str, name: str) -> None: + self.id = id + self.name = name + + +class _FakeWorkspace: + def __init__(self, id: str, name: str, projects: list[_FakeProject]) -> None: + self.id = id + self.name = name + self.projects = projects + + +class _FakeIdentity: + def __init__(self, workspaces: list[_FakeWorkspace]) -> None: + self.workspaces = workspaces + + +class _FakeClient: + def __init__(self, *, workspaces=None, sandbox_factory=None, **kwargs) -> None: + self.create_count = 0 + self.create_kwargs: list[dict] = [] + self._sandbox_factory = sandbox_factory or (lambda: _FakeSandbox()) + self.last_sandbox: _FakeSandbox | None = None + self._by_id: dict[str, _FakeSandbox] = {} + self._workspaces = workspaces if workspaces is not None else [_FakeWorkspace("ws1", "Workspace", [_FakeProject("proj1", "Project")])] + + def who_am_i(self): + return _FakeIdentity(self._workspaces) + + def create(self, **kwargs): + self.create_count += 1 + self.create_kwargs.append(kwargs) + sandbox = self._sandbox_factory() + sandbox.id = f"sb{self.create_count}" + self._by_id[sandbox.id] = sandbox + self.last_sandbox = sandbox + return sandbox + + def get(self, sandbox_id): + return self._by_id[sandbox_id] + + +# ── Config stub ─────────────────────────────────────────────────────── + + +def _stub_config(sandbox_attrs=None): + attrs = sandbox_attrs or {} + return types.SimpleNamespace(sandbox=types.SimpleNamespace(**attrs)) + + +def _install(monkeypatch, *, client=None, config_attrs=None): + """Construct a provider with get_app_config + _import_client stubbed.""" + monkeypatch.setattr( + "deerflow.community.tenki.provider.get_app_config", + lambda: _stub_config(config_attrs), + ) + if client is not None: + monkeypatch.setattr( + "deerflow.community.tenki.provider._import_client", + lambda: lambda **kw: client, + ) + provider = TenkiSandboxProvider() + return provider + + +def _no_tenki(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setitem(sys.modules, "tenki_sandbox", None) + + +# ── Lazy import ──────────────────────────────────────────────────────── + + +def test_import_client_missing_raises_actionable(monkeypatch: pytest.MonkeyPatch) -> None: + _no_tenki(monkeypatch) + with pytest.raises(ImportError, match=r"deerflow-harness\[tenki\]"): + _import_client() + + +def test_acquire_without_tenki_raises_and_shuts_down_cleanly(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("deerflow.community.tenki.provider.get_app_config", lambda: _stub_config()) + _no_tenki(monkeypatch) + provider = TenkiSandboxProvider() + try: + with pytest.raises(ImportError, match=r"deerflow-harness\[tenki\]"): + provider.acquire("thread-1", user_id="u") + finally: + provider.shutdown() + provider.shutdown() # idempotent + + +# ── Adapter: guards, env, output, terminal detection ────────────────── + + +def test_guard_traversal() -> None: + assert TenkiSandbox._guard_traversal("/mnt/user-data/workspace/a.txt") == "/mnt/user-data/workspace/a.txt" + assert TenkiSandbox._guard_traversal("relative/ok.txt") == "relative/ok.txt" + with pytest.raises(PermissionError): + TenkiSandbox._guard_traversal("/mnt/user-data/../etc/passwd") + with pytest.raises(ValueError): + TenkiSandbox._guard_traversal("") + + +def test_resolve_path_remaps_virtual_prefix_to_home() -> None: + box = TenkiSandbox("sb", _FakeSandbox(), home_dir="/home/tenki") + assert box._resolve_path("/mnt/user-data") == "/home/tenki" + assert box._resolve_path("/mnt/user-data/workspace/a.txt") == "/home/tenki/workspace/a.txt" + # Non-virtual absolute paths pass through unchanged. + assert box._resolve_path("/etc/hostname") == "/etc/hostname" + with pytest.raises(PermissionError): + box._resolve_path("/mnt/user-data/../etc/passwd") + + +def test_download_file_guards_reject_before_touching_sandbox() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + with pytest.raises(PermissionError): + box.download_file("/etc/passwd") # outside the virtual prefix + with pytest.raises(PermissionError): + box.download_file("/mnt/user-data/../etc/passwd") # traversal + assert fake.exec_calls == [] # guards raised before any exec + + +def test_execute_command_rejects_invalid_env_key() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + with pytest.raises(ValueError, match=r"POSIX"): + box.execute_command("echo hi", env={"BAD KEY": "x"}) + + +def test_execute_command_formats_stdout_and_forwards_env_timeout() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake, default_env={"BASE": "1"}) + out = box.execute_command("echo ok", env={"EXTRA": "2"}, timeout=5) + assert out == "ok\n" # combined output is returned verbatim (matches boxlite/e2b) + call = fake.exec_calls[-1] + assert call["argv"] == ("sh", "-lc", "echo ok") + assert call["env"] == {"BASE": "1", "EXTRA": "2"} + assert call["timeout"] == 5 + assert call["cwd"] is None # no forced cwd; runs in sandbox default dir + + +def test_execute_command_returns_error_as_text() -> None: + box = TenkiSandbox("sb", _FakeSandbox(exec_error=RuntimeError("boom"))) + assert box.execute_command("echo hi") == "Error: boom" + + +def test_execute_command_closed_returns_error() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.close() + assert box.execute_command("echo hi") == "Error: sandbox has been closed" + + +def test_terminal_failure_triggers_callback() -> None: + invalidated: list[tuple[str, str]] = [] + box = TenkiSandbox( + "sb", + _FakeSandbox(exec_error=_FakeSessionTerminatedError("session gone")), + on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason)), + ) + out = box.execute_command("echo hi") + assert out == "Error: session gone" + assert invalidated == [("sb", "session gone")] + + +def test_regular_error_does_not_trigger_terminal_callback() -> None: + invalidated: list[tuple[str, str]] = [] + box = TenkiSandbox( + "sb", + _FakeSandbox(exec_error=RuntimeError("user command failed")), + on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason)), + ) + box.execute_command("echo hi") + assert invalidated == [] + + +def test_transient_transport_error_is_not_retried_and_not_evicted() -> None: + # exec is not idempotent, so a transient transport error must NOT be retried + # (re-running could double a command's side effects or duplicate a base64 + # write chunk). It surfaces as text, and — not being a terminal session + # error — does not evict the sandbox. + invalidated: list[tuple[str, str]] = [] + fake = _FakeSandbox(exec_error=RuntimeError("UNAVAILABLE: Socket closed")) + box = TenkiSandbox("sb", fake, on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason))) + out = box.execute_command("echo ok") + assert out.startswith("Error:") + assert len(fake.exec_calls) == 1 # executed exactly once — no retry + assert invalidated == [] # transient, not terminal → sandbox not evicted + + +def test_connection_error_is_terminal() -> None: + invalidated: list[tuple[str, str]] = [] + box = TenkiSandbox( + "sb", + _FakeSandbox(exec_error=ConnectionError("reset")), + on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason)), + ) + box.execute_command("echo hi") + assert invalidated == [("sb", "reset")] + + +def test_close_is_idempotent() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.close() + box.close() # idempotent + assert box.is_closed is True + assert fake.closed is True + + +def test_close_failure_leaves_sandbox_retryable() -> None: + # Marking the adapter closed before a failed termination would strand a + # running (billed) microVM with no handle left to retry it. + fake = _FakeSandbox(close_error=RuntimeError("terminate rejected")) + box = TenkiSandbox("sb", fake) + with pytest.raises(RuntimeError, match="terminate rejected"): + box.close() + assert box.is_closed is False # still ours to terminate + + fake.close_error = None + box.close() + assert box.is_closed is True + + +def test_close_treats_terminal_error_as_already_gone() -> None: + fake = _FakeSandbox(close_error=_FakeSessionTerminatedError("session gone")) + box = TenkiSandbox("sb", fake) + box.close() # nothing left to terminate → not an error + assert box.is_closed is True + + +# ── Adapter: native fs file round-trip ──────────────────────────────── + + +def test_file_round_trip_text() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.write_file("/mnt/user-data/workspace/note.txt", "hello world") + assert box.read_file("/mnt/user-data/workspace/note.txt") == "hello world" + + +def test_write_creates_parent_directory() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.write_file("/mnt/user-data/workspace/nested/note.txt", "hi") + assert "/home/tenki/workspace/nested" in fake.dirs + + +def test_file_round_trip_large_binary_streams_in_frames() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + data = bytes(range(256)) * 8192 # 2 MB → more than one 1 MiB upload frame + box.update_file("/mnt/user-data/outputs/blob.bin", data) + assert fake.write_frame_counts[-1] > 1 + assert box.download_file("/mnt/user-data/outputs/blob.bin") == data + + +def test_append_accumulates() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.write_file("/mnt/user-data/workspace/log.txt", "a") + box.write_file("/mnt/user-data/workspace/log.txt", "b", append=True) + assert box.read_file("/mnt/user-data/workspace/log.txt") == "ab" + + +def test_append_to_missing_file_creates_it() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.write_file("/mnt/user-data/workspace/new.txt", "first", append=True) + assert box.read_file("/mnt/user-data/workspace/new.txt") == "first" + + +def test_read_missing_file_returns_error() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + assert box.read_file("/mnt/user-data/workspace/nope.txt").startswith("Error:") + + +def test_download_missing_file_raises_oserror() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + with pytest.raises(OSError): + box.download_file("/mnt/user-data/outputs/nope.bin") + + +def test_download_cap_counts_bytes_actually_received(monkeypatch) -> None: + # The cap must not rely on a separate size probe: a file that grows between + # the probe and the read would slip past it. + monkeypatch.setattr("deerflow.community.tenki.sandbox._MAX_DOWNLOAD_SIZE", 100) + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + fake.files["/home/tenki/outputs/big.bin"] = b"x" * 500 + with pytest.raises(OSError) as excinfo: + box.download_file("/mnt/user-data/outputs/big.bin") + assert excinfo.value.errno == errno.EFBIG + + +def test_download_size_cap_does_not_evict_sandbox(monkeypatch) -> None: + # Hitting the size cap is a client-side limit, not a dead session — the + # sandbox must stay live. + monkeypatch.setattr("deerflow.community.tenki.sandbox._MAX_DOWNLOAD_SIZE", 100) + invalidated: list[tuple[str, str]] = [] + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake, on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason))) + fake.files["/home/tenki/outputs/big.bin"] = b"x" * 500 + with pytest.raises(OSError): + box.download_file("/mnt/user-data/outputs/big.bin") + assert invalidated == [] + + +def test_download_terminal_transport_error_evicts_sandbox() -> None: + # A ConnectionError mid-stream is an OSError subclass and terminal; the old + # `except OSError: raise` re-raised it before _note_failure, so a session + # that died mid-download never got evicted. + invalidated: list[tuple[str, str]] = [] + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake, on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason))) + box.write_file("/mnt/user-data/outputs/f.bin", "payload") + fake.fs_error = ConnectionError("connection reset mid-stream") + with pytest.raises(ConnectionError): + box.download_file("/mnt/user-data/outputs/f.bin") + assert invalidated == [("sb", "connection reset mid-stream")] + + +def test_fs_terminal_failure_evicts_sandbox() -> None: + invalidated: list[tuple[str, str]] = [] + fake = _FakeSandbox(fs_error=_FakeSessionTerminatedError("session gone")) + box = TenkiSandbox("sb", fake, on_terminal_failure=lambda sid, reason: invalidated.append((sid, reason))) + with pytest.raises(Exception, match="session gone"): + box.write_file("/mnt/user-data/workspace/a.txt", "x") + assert invalidated == [("sb", "session gone")] + + +# ── Adapter: returned paths stay caller-facing ──────────────────────── + + +def test_search_results_use_virtual_paths() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.write_file("/mnt/user-data/workspace/pkg/mod.py", "def foo():\n return 42\n") + + assert box.list_dir("/mnt/user-data/workspace") == ["/mnt/user-data/workspace/pkg/mod.py"] + + found, truncated = box.glob("/mnt/user-data/workspace", "**/*.py") + assert found == ["/mnt/user-data/workspace/pkg/mod.py"] and not truncated + + matches, _ = box.grep("/mnt/user-data/workspace", "return") + assert [m.path for m in matches] == ["/mnt/user-data/workspace/pkg/mod.py"] + assert matches[0].line_number == 2 + + # Results feed straight back into the file APIs. + assert box.read_file(found[0]).startswith("def foo()") + + +def test_grep_glob_keeps_its_directory_prefix() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.write_file("/mnt/user-data/workspace/src/a.js", "const x = 1;\n") + box.write_file("/mnt/user-data/workspace/vendor/b.js", "const x = 2;\n") + + matches, _ = box.grep("/mnt/user-data/workspace", "const x", glob="src/*.js") + assert [m.path for m in matches] == ["/mnt/user-data/workspace/src/a.js"] + + +def test_grep_passes_capital_h_so_single_file_matches_parse() -> None: + # Without -H, `grep -r` on a path that resolves to a single file prints + # "line:text" and the file:line:text unpack drops every match. The flag must + # always be present regardless of the other options. + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.write_file("/mnt/user-data/workspace/a.txt", "needle here\n") + box.grep("/mnt/user-data/workspace", "needle") + grep_scripts = [c["argv"][2] for c in fake.exec_calls if c["argv"][:2] == ("sh", "-lc") and c["argv"][2].startswith("grep ")] + assert grep_scripts and "-H" in shlex.split(grep_scripts[0]) + + +def _grep_script(fake: _FakeSandbox) -> list[str]: + scripts = [c["argv"][2] for c in fake.exec_calls if c["argv"][:2] == ("sh", "-lc") and c["argv"][2].startswith("grep ")] + assert scripts, "no grep command was issued" + return shlex.split(scripts[0]) + + +def test_grep_literal_uses_fixed_string_flag() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.write_file("/mnt/user-data/workspace/a.txt", "a.b\n") + box.grep("/mnt/user-data/workspace", "a.b", literal=True) + tokens = _grep_script(fake) + assert "-F" in tokens and "-E" not in tokens # -F matches the pattern literally + assert "-i" in tokens # case-insensitive is still the default + + +def test_grep_case_sensitive_omits_ignore_case_flag() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.write_file("/mnt/user-data/workspace/a.txt", "Needle\n") + box.grep("/mnt/user-data/workspace", "Needle", case_sensitive=True) + tokens = _grep_script(fake) + assert "-i" not in tokens # case-sensitive → no -i + assert "-E" in tokens + + +def test_glob_include_dirs_adds_directory_type_to_find() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.glob("/mnt/user-data/workspace", "*", include_dirs=True) + find_scripts = [c["argv"][2] for c in fake.exec_calls if c["argv"][:2] == ("sh", "-lc") and c["argv"][2].startswith("find ")] + assert find_scripts and "-type d" in find_scripts[-1] # dirs requested, not just files + + +def test_list_dir_forwards_max_depth() -> None: + fake = _FakeSandbox() + box = TenkiSandbox("sb", fake) + box.list_dir("/mnt/user-data/workspace", max_depth=4) + find_scripts = [c["argv"][2] for c in fake.exec_calls if c["argv"][:2] == ("sh", "-lc") and c["argv"][2].startswith("find ")] + assert find_scripts and "-maxdepth 4" in find_scripts[-1] + + +def test_paths_outside_the_virtual_prefix_are_reported_as_is() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + assert box._virtual_path("/etc/hostname") == "/etc/hostname" + assert box._virtual_path("/home/tenki") == "/mnt/user-data" + + +# ── Provider: id derivation ─────────────────────────────────────────── + + +def test_sandbox_id_deterministic(monkeypatch): + provider = _install(monkeypatch) + assert provider._sandbox_id("t1", "u1") == provider._sandbox_id("t1", "u1") + assert len(provider._sandbox_id("t1", "u1")) == 16 # 64-bit, like community/e2b_sandbox + + +def test_sandbox_id_distinct_users_and_threads(monkeypatch): + provider = _install(monkeypatch) + assert provider._sandbox_id("t1", "u1") != provider._sandbox_id("t1", "u2") + assert provider._sandbox_id("t1", "u1") != provider._sandbox_id("t2", "u1") + + +def test_idle_timeout_zero_disables_reaper(monkeypatch): + provider = _install(monkeypatch, config_attrs={"idle_timeout": 0}) + assert provider._config["idle_timeout"] == 0 + assert provider._idle_checker_thread is None + provider.shutdown() + + +def test_load_config_rejects_invalid_environment_key(monkeypatch): + # The config `environment` is merged into every command; a bad key must + # fail fast at load time, like the per-call env in execute_command, not + # surface as a confusing SDK error at create/exec time. + monkeypatch.setattr("deerflow.community.tenki.provider.get_app_config", lambda: _stub_config({"environment": {"bad-key": "1"}})) + with pytest.raises(ValueError, match=r"POSIX"): + TenkiSandboxProvider() + + +def test_load_config_accepts_valid_environment_key(monkeypatch): + provider = _install(monkeypatch, config_attrs={"environment": {"PYTHONUNBUFFERED": "1"}}) + assert provider._config["environment"] == {"PYTHONUNBUFFERED": "1"} + provider.shutdown() + + +# ── Provider: acquire / create / warm pool ──────────────────────────── + + +def test_create_passes_prefixed_name_and_scope(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid = provider.acquire("thread-1", user_id="u1") + assert sid in provider._sandboxes + kwargs = client.create_kwargs[0] + assert kwargs["name"].startswith("deer-flow-tenki-") + assert kwargs["project_id"] == "proj1" + assert kwargs["workspace_id"] == "ws1" + provider.shutdown() + + +def test_create_waits_client_side_and_configures_lifetime(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client, config_attrs={"max_duration": 7200}) + provider.acquire("thread-1", user_id="u1") + kwargs = client.create_kwargs[0] + # wait=False keeps the handle on our side of create() so a readiness failure + # is still terminable (see the next test). + assert kwargs["wait"] is False + assert client.last_sandbox.wait_ready_calls == 1 + # Without an explicit lifetime, Tenki reaps the sandbox out from under a + # long-lived thread at its default (~30 min). + assert kwargs["max_duration"] == 7200 + assert kwargs["sticky"] is False + provider.shutdown() + + +def test_create_default_lifetime_is_explicit(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + provider.acquire("thread-1", user_id="u1") + assert client.create_kwargs[0]["max_duration"] == pytest.approx(4 * 60 * 60) + provider.shutdown() + + +def test_create_terminates_the_microvm_when_readiness_fails(monkeypatch): + client = _FakeClient(sandbox_factory=lambda: _FakeSandbox(wait_ready_error=RuntimeError("never became ready"))) + provider = _install(monkeypatch, client=client) + with pytest.raises(RuntimeError, match="never became ready"): + provider.acquire("thread-1", user_id="u1") + # The session exists remotely even though create() failed — it must not leak. + assert client.last_sandbox.closed is True + assert provider._sandboxes == {} + provider.shutdown() + + +def test_create_survives_bootstrap_failure(monkeypatch, caplog): + # Bootstrap is best-effort: the file APIs still work via the home remap, so a + # non-zero bootstrap must warn, not fail the acquire. + class _BootstrapFailSandbox(_FakeSandbox): + def _run_script(self, script: str) -> _FakeResult: + if "BOOTSTRAP_OK" in script: + return _FakeResult(exit_code=1, stderr=b"permission denied") + return super()._run_script(script) + + client = _FakeClient(sandbox_factory=lambda: _BootstrapFailSandbox()) + provider = _install(monkeypatch, client=client) + with caplog.at_level("WARNING"): + sid = provider.acquire("thread-1", user_id="u1") + box = provider.get(sid) + assert box is not None and not box.is_closed # usable despite bootstrap failure + assert any("bootstrap" in r.message.lower() for r in caplog.records) + provider.shutdown() + + +def test_bootstrap_is_non_interactive_and_time_bounded(monkeypatch): + # The bootstrap runs under the per-scope acquire lock, so it must not hang: + # `sudo -n` fails fast instead of blocking on a password prompt, and the exec + # carries a timeout so any other stall drops to the warning path. + client = _FakeClient() + provider = _install(monkeypatch, client=client) + provider.acquire("thread-1", user_id="u1") + bootstrap = next(c for c in client.last_sandbox.exec_calls if "BOOTSTRAP_OK" in (c["argv"][2] if len(c["argv"]) > 2 else "")) + assert "sudo -n ln -sfn" in bootstrap["argv"][2] + # Pin the actual bounded value, not merely "some timeout": a regression to + # timeout=0 (no timeout in some SDKs) would slip past an `is not None` check. + assert bootstrap["timeout"] == _BOOTSTRAP_TIMEOUT + provider.shutdown() + + +def test_release_parks_in_warm_pool(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid = provider.acquire("thread-1", user_id="u1") + provider.release(sid) + assert sid not in provider._sandboxes + assert sid in provider._warm_pool + sandbox, _ = provider._warm_pool[sid] + assert not sandbox.is_closed # microVM not terminated + provider.shutdown() + + +def test_acquire_reclaims_from_warm_pool(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid1 = provider.acquire("thread-1", user_id="u1") + provider.release(sid1) + sid2 = provider.acquire("thread-1", user_id="u1") + assert sid1 == sid2 + assert client.create_count == 1 # reused, not recreated + assert sid2 in provider._sandboxes + provider.shutdown() + + +def test_different_threads_dont_reclaim_each_other(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid_a = provider.acquire("thread-a", user_id="u1") + provider.release(sid_a) + sid_b = provider.acquire("thread-b", user_id="u1") + assert sid_b != sid_a + assert sid_a in provider._warm_pool + assert sid_b in provider._sandboxes + provider.shutdown() + + +def test_warm_pool_reclaim_failed_health_check_creates_new(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid1 = provider.acquire("thread-1", user_id="u1") + provider.release(sid1) + # Kill the warm sandbox so the health check fails. + sandbox, _ = provider._warm_pool[sid1] + sandbox.close() + sid2 = provider.acquire("thread-1", user_id="u1") + assert sid2 == sid1 # same deterministic id + assert client.create_count == 2 # a fresh sandbox was created + replacement = provider.get(sid2) + assert replacement is not None and not replacement.is_closed + provider.shutdown() + + +def test_terminal_failure_evicts_active_sandbox(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid = provider.acquire("thread-1", user_id="u1") + box = provider.get(sid) + assert box is not None + # Only fail command-time, not the create-time mkdir bootstrap. + client.last_sandbox.exec_error = _FakeSessionTerminatedError("gone") + box.execute_command("echo hi") # terminal failure → invalidate + assert provider.get(sid) is None + provider.shutdown() + + +def test_concurrent_same_thread_acquire_creates_one_sandbox(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + original_create = provider._create_sandbox + create_started = threading.Event() + + def slow_create(sandbox_id: str): + create_started.set() + time.sleep(0.1) + return original_create(sandbox_id) + + provider._create_sandbox = slow_create # type: ignore[method-assign] + results: list[str] = [] + + def worker(): + results.append(provider.acquire("thread-1", user_id="u1")) + + first = threading.Thread(target=worker) + second = threading.Thread(target=worker) + first.start() + assert create_started.wait(timeout=2) + second.start() + first.join(timeout=2) + second.join(timeout=2) + + assert len(results) == 2 + assert results[0] == results[1] + assert client.create_count == 1 + provider.shutdown() + + +def test_release_during_shutdown_closes_instead_of_reparking(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid = provider.acquire("thread-1", user_id="u1") + box = provider._sandboxes[sid] + with provider._lock: + provider._shutdown_called = True + provider.release(sid) + assert sid not in provider._sandboxes + assert sid not in provider._warm_pool + assert box.is_closed + + +def test_reset_parks_running_for_later_cleanup(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid_active = provider.acquire("thread-active", user_id="u1") + sid_warm = provider.acquire("thread-warm", user_id="u1") + provider.release(sid_warm) + active_box = provider._sandboxes[sid_active] + provider.reset() + assert provider._sandboxes == {} + assert provider._warm_pool[sid_active][0] is active_box + assert provider._thread_sandboxes == {} + assert not active_box.is_closed + provider.shutdown() + assert active_box.is_closed + + +def test_idle_reaper_destroys_expired_warm(monkeypatch): + client = _FakeClient() + monkeypatch.setattr(TenkiSandboxProvider, "IDLE_CHECK_INTERVAL", 0.1) + provider = _install(monkeypatch, client=client) + sid = provider.acquire("thread-1", user_id="u1") + provider.release(sid) + warm_box = provider._warm_pool[sid][0] + provider._warm_pool[sid] = (warm_box, time.time() - 9999) + time.sleep(0.3) + assert sid not in provider._warm_pool + assert warm_box.is_closed + provider.shutdown() + + +def test_replica_enforcement_evicts_oldest_warm(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client, config_attrs={"replicas": 2}) + sid_a = provider.acquire("thread-a", user_id="u1") + provider.release(sid_a) + sid_b = provider.acquire("thread-b", user_id="u1") + provider.release(sid_b) + box_a = provider._warm_pool[sid_a][0] + provider._warm_pool[sid_a] = (box_a, time.time() - 100) + provider._warm_pool[sid_b] = (provider._warm_pool[sid_b][0], time.time()) + provider.acquire("thread-c", user_id="u1") + assert sid_a not in provider._warm_pool + assert box_a.is_closed + assert sid_b in provider._warm_pool + provider.shutdown() + + +def test_shutdown_destroys_all_and_stops_reaper(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid_active = provider.acquire("thread-1", user_id="u1") + sid_warm = provider.acquire("thread-2", user_id="u1") + provider.release(sid_warm) + box_active = provider._sandboxes[sid_active] + box_warm = provider._warm_pool[sid_warm][0] + checker = provider._idle_checker_thread + provider.shutdown() + assert provider._idle_checker_stop.is_set() + assert checker is not None and not checker.is_alive() + assert box_active.is_closed and box_warm.is_closed + assert provider._sandboxes == {} and provider._warm_pool == {} + + +# ── Provider: scope resolution ───────────────────────────────────────── + + +def test_scope_auto_resolves_single(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + provider.acquire("thread-1", user_id="u1") + assert client.create_kwargs[0]["project_id"] == "proj1" + assert client.create_kwargs[0]["workspace_id"] == "ws1" + provider.shutdown() + + +def test_explicit_project_id_skips_lookup(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client, config_attrs={"project_id": "explicit"}) + provider.acquire("thread-1", user_id="u1") + assert client.create_kwargs[0]["project_id"] == "explicit" + provider.shutdown() + + +def test_ambiguous_project_raises(monkeypatch): + client = _FakeClient(workspaces=[_FakeWorkspace("ws1", "W", [_FakeProject("p1", "A"), _FakeProject("p2", "B")])]) + provider = _install(monkeypatch, client=client) + with pytest.raises(ValueError, match="project_id"): + provider.acquire("thread-1", user_id="u1") + provider.shutdown() + + +# ── Live integration (opt-in) ────────────────────────────────────────── + + +@pytest.mark.skipif( + not os.getenv("TENKI_API_KEY") and not os.getenv("TENKI_AUTH_TOKEN"), + reason="requires a real Tenki API key (TENKI_API_KEY); network integration test", +) +def test_integration_real_sandbox(monkeypatch): + monkeypatch.setattr("deerflow.community.tenki.provider.get_app_config", lambda: _stub_config()) + provider = TenkiSandboxProvider() + try: + sid = provider.acquire("it-thread", user_id="it-user") + box = provider.get(sid) + assert box is not None + assert "42" in box.execute_command("python3 -c 'print(6 * 7)'") + box.write_file("/mnt/user-data/workspace/it.txt", "tenki-e2e") + assert box.read_file("/mnt/user-data/workspace/it.txt") == "tenki-e2e" + finally: + provider.shutdown() diff --git a/backend/uv.lock b/backend/uv.lock index 086224d29..075f36ebf 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -935,6 +935,9 @@ pymupdf = [ redis = [ { name = "redis" }, ] +tenki = [ + { name = "tenki-sandbox" }, +] tui = [ { name = "textual" }, ] @@ -985,10 +988,11 @@ requires-dist = [ { name = "redis", marker = "extra == 'redis'", specifier = ">=5.0.0" }, { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0,<3.0" }, { name = "tavily-python", specifier = ">=0.7.17" }, + { name = "tenki-sandbox", marker = "extra == 'tenki'", specifier = ">=0.4.0" }, { name = "textual", marker = "extra == 'tui'", specifier = ">=0.80" }, { name = "tiktoken", specifier = ">=0.8.0" }, ] -provides-extras = ["tui", "groundroute", "ollama", "postgres", "redis", "pymupdf", "boxlite", "monocle", "browser"] +provides-extras = ["tui", "groundroute", "ollama", "postgres", "redis", "pymupdf", "boxlite", "tenki", "monocle", "browser"] [[package]] name = "defusedxml" @@ -4511,6 +4515,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d7/c1/eb8f9debc45d3b7918a32ab756658a0904732f75e555402972246b0b8e71/tenacity-9.1.4-py3-none-any.whl", hash = "sha256:6095a360c919085f28c6527de529e76a06ad89b23659fa881ae0649b867a9d55", size = 28926, upload-time = "2026-02-07T10:45:32.24Z" }, ] +[[package]] +name = "tenki-sandbox" +version = "0.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "grpcio" }, + { name = "protobuf" }, + { name = "websocket-client" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/8d/58/f6527c63b2e4c94fd00a48ba4bd93ef52de22dbc72e247110949a17511f6/tenki_sandbox-0.4.0.tar.gz", hash = "sha256:2836e6e7100dfc81715b4d93ecfe80e3f055867498cb9f34be921a02969bb15f", size = 179392, upload-time = "2026-07-17T22:18:09.869Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3b/f7/3c02da98793dc0e56230c520064d7d233c6e0edab13b410c031e1cfe7fd9/tenki_sandbox-0.4.0-py3-none-any.whl", hash = "sha256:76d5ece2e1607b43705587d86277e983ef43601a86993077e379be8033a40b3e", size = 155246, upload-time = "2026-07-17T22:18:08.383Z" }, +] + [[package]] name = "textual" version = "8.2.7" diff --git a/config.example.yaml b/config.example.yaml index b75641790..dafd8dd2b 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1354,6 +1354,27 @@ sandbox: # # Note: provisioner-created Pods use the provisioner's SANDBOX_IMAGE # # environment variable, not sandbox.image from this config file. +# Option 5: Tenki cloud microVM Sandbox +# Runs each sandbox as an isolated Tenki cloud microVM. Released sandboxes stay +# in an in-process warm pool and are reclaimed by the same user/thread without a +# cold start. Requires the optional SDK: pip install "deerflow-harness[tenki]". +# sandbox: +# use: deerflow.community.tenki:TenkiSandboxProvider +# # api_key: $TENKI_API_KEY # falls back to TENKI_API_KEY / TENKI_AUTH_TOKEN env var +# # base_url: https://tenki.cloud # optional; SDK default when omitted +# # image: my-base-image # optional; Tenki account default base image when omitted +# # project_id: proj_... # optional; auto-selected if the account has exactly one +# # workspace_id: ws_... # optional; auto-selected if the account has exactly one +# # cpu_cores: 2 # optional per-sandbox vCPUs +# # memory_mb: 2048 # optional per-sandbox memory +# # replicas: 3 # active + warm microVM cap per gateway process +# # idle_timeout: 600 # warm microVM idle seconds before terminate; 0 disables +# # max_duration: 14400 # Tenki sandbox lifetime in seconds; 0 uses the account default +# # sticky: false # pin the microVM to its host (only matters with pause/resume) +# # home_dir: /home/tenki # writable dir backing /mnt/user-data +# # environment: # injected into every command (and as create-time env) +# # PYTHONUNBUFFERED: "1" + # ============================================================================ # Subagents Configuration # ============================================================================