diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 644a0a4f0..314d3deff 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -519,6 +519,7 @@ sandbox: home_dir: /home/user # /mnt/user-data is remapped under this directory idle_timeout: 600 # forwarded to e2b's server-side set_timeout() replicas: 3 # max concurrent sandboxes per gateway process + mount_upload_deadline_seconds: 120 # per-sandbox time budget for mount uploads (seconds) ownership: # use Redis when more than one gateway shares E2B type: redis redis_url: $REDIS_URL @@ -564,6 +565,11 @@ Notes specific to `E2BSandboxProvider`: `/mnt/user-data/outputs/` (which is mapped to `home_dir/outputs/` inside the sandbox and surfaced through the standard artifact pipeline) to ship files back to the gateway. +- `mount_upload_deadline_seconds` sets the per-sandbox time budget for mount + uploads. The provider checks it before each mount, during directory preflight, + and before each SDK write. The deadline does not interrupt active filesystem or + E2B SDK calls. Omitting the key preserves the 120-second default. Values below + 1 are clamped to 1; non-numeric or null values fall back to the default. **OpenSandbox Remote Sandbox** (runs code through an OpenSandbox deployment): diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index b098aec02..79e49ea7d 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -90,7 +90,8 @@ Each mount has these fixed limits: The full sandbox creation pass also allows 512 MiB and 2,000 files. Skill projections and configured mounts share this budget. -The pass has a cooperative 120-second deadline. The provider checks it before +The pass has a cooperative deadline controlled by +``mount_upload_deadline_seconds`` (default: 120 seconds). The provider checks it before each mount, during directory preflight, and before each SDK write. The deadline does not interrupt active filesystem or E2B SDK calls. diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py b/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py index dd9e36ef7..91bd4eeff 100644 --- a/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/__init__.py @@ -22,6 +22,7 @@ Configuration example (``config.yaml``):: reconciliation_max_pages: 10 reconciliation_max_items: 200 reconciliation_max_seconds: 15 + mount_upload_deadline_seconds: 120 # mount upload pass deadline; default: 120 mounts: # one-shot upload of host files into the sandbox - host_path: /path/on/host container_path: /path/in/sandbox diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py index 42e378cde..7dbcdb805 100644 --- a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py @@ -21,6 +21,7 @@ provider fields during startup. reconciliation_max_pages: 10 reconciliation_max_items: 200 reconciliation_max_seconds: 15 + mount_upload_deadline_seconds: 120 # mount upload pass deadline; default: 120 ownership: type: redis # shares ownership and capacity across Gateways redis_url: redis://redis:6379/0 @@ -45,6 +46,7 @@ import threading import time import uuid from collections import OrderedDict +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from decimal import Decimal, InvalidOperation @@ -111,8 +113,8 @@ _MAX_MOUNT_PASS_FILES = 2000 _MOUNT_PASS_DEADLINE_SECONDS = 120 -def _mount_deadline_reason() -> str: - return f"time budget {_MOUNT_PASS_DEADLINE_SECONDS}s" +def _mount_deadline_reason(deadline_seconds: int) -> str: + return f"time budget {deadline_seconds}s" class _MountPassLimitExceeded(Exception): @@ -122,6 +124,7 @@ class _MountPassLimitExceeded(Exception): @dataclass class _MountUploadBudget: deadline: float + deadline_seconds: int attempted_bytes: int = 0 attempted_files: int = 0 completed_bytes: int = 0 @@ -133,7 +136,7 @@ class _MountUploadBudget: def check_deadline(self) -> None: if self.expired: - raise _MountPassLimitExceeded(_mount_deadline_reason()) + raise _MountPassLimitExceeded(_mount_deadline_reason(self.deadline_seconds)) # Metadata keys we attach to every sandbox so we can discover ours via @@ -151,6 +154,7 @@ E2B_EXTRA_CONFIG_KEYS = frozenset( "api_key", "domain", "home_dir", + "mount_upload_deadline_seconds", "reconciliation_grace_seconds", "reconciliation_interval_seconds", "reconciliation_max_items", @@ -329,6 +333,7 @@ class E2BSandboxProvider(SandboxProvider): 0.1, float(_opt("reconciliation_max_seconds", DEFAULT_RECONCILIATION_MAX_SECONDS)), ), + "mount_upload_deadline_seconds": self._resolve_mount_upload_deadline(_opt), } @staticmethod @@ -341,6 +346,28 @@ class E2BSandboxProvider(SandboxProvider): resolved[key] = "" if value is None else str(value) return resolved + @staticmethod + def _resolve_mount_upload_deadline(_opt: Callable[[str, Any], Any]) -> int: + raw = _opt("mount_upload_deadline_seconds", _MOUNT_PASS_DEADLINE_SECONDS) + if raw is None: + return _MOUNT_PASS_DEADLINE_SECONDS + try: + value = int(raw) + except (TypeError, ValueError, OverflowError): + logger.warning( + "E2BSandboxProvider: non-numeric mount_upload_deadline_seconds=%r; falling back to %ds", + raw, + _MOUNT_PASS_DEADLINE_SECONDS, + ) + return _MOUNT_PASS_DEADLINE_SECONDS + if value < 1: + logger.warning( + "E2BSandboxProvider: invalid mount_upload_deadline_seconds=%d; clamping to 1", + value, + ) + return 1 + return value + def _get_sandbox_cls(self) -> type[E2BClientSandbox]: """Return the e2b SDK Sandbox class.""" return E2BClientSandbox @@ -1798,7 +1825,11 @@ class E2BSandboxProvider(SandboxProvider): def _apply_mounts(self, client: E2BClientSandbox, *, user_id: str | None = None) -> None: started_at = time.monotonic() - budget = _MountUploadBudget(deadline=started_at + _MOUNT_PASS_DEADLINE_SECONDS) + deadline_seconds = self._config.get("mount_upload_deadline_seconds", _MOUNT_PASS_DEADLINE_SECONDS) + budget = _MountUploadBudget( + deadline=started_at + deadline_seconds, + deadline_seconds=deadline_seconds, + ) def warn_pass_stopped(reason: str) -> None: elapsed_ms = int((time.monotonic() - started_at) * 1000) @@ -1835,7 +1866,7 @@ class E2BSandboxProvider(SandboxProvider): for host_path, container_path, read_only in mounts: if budget.expired: - warn_pass_stopped(_mount_deadline_reason()) + warn_pass_stopped(_mount_deadline_reason(deadline_seconds)) break if not host_path.exists(): logger.warning("Skipping e2b mount: host_path %s does not exist", host_path) diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py index b80671704..7b205e57a 100644 --- a/backend/tests/test_e2b_sandbox_provider.py +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -746,6 +746,209 @@ def test_apply_mounts_deadline_stops_before_next_mount_preflight(monkeypatch, tm assert "time budget 1s" in caplog.text +def test_apply_mounts_deadline_defaults_to_120_when_not_configured(monkeypatch, tmp_path, caplog): + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + monkeypatch.setattr( + mod, + "get_app_config", + lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")), + ) + clock = [0.0] + monkeypatch.setattr(mod.time, "monotonic", lambda: clock[0]) + + class DeadlineFilesAPI(FakeFilesAPI): + def write(self, path: str, content: Any) -> None: + super().write(path, content) + clock[0] = 121.0 + + source = tmp_path / "mount" + source.mkdir() + (source / "first.txt").write_text("first", encoding="utf-8") + (source / "second.txt").write_text("second", encoding="utf-8") + provider = _make_provider() + assert "mount_upload_deadline_seconds" not in provider._config + monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: []) + provider._config["mounts"] = [ + SimpleNamespace(host_path=str(source), container_path="/mnt/data", read_only=False), + ] + client = FakeClient(files=DeadlineFilesAPI()) + + with caplog.at_level("WARNING"): + provider._apply_mounts(client, user_id="user-1") + + assert len(client.files.write_calls) == 1 + assert "time budget 120s" in caplog.text + + +def test_apply_mounts_deadline_uses_configured_value(monkeypatch, tmp_path, caplog): + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + monkeypatch.setattr( + mod, + "get_app_config", + lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")), + ) + clock = [0.0] + monkeypatch.setattr(mod.time, "monotonic", lambda: clock[0]) + + class DeadlineFilesAPI(FakeFilesAPI): + def write(self, path: str, content: Any) -> None: + super().write(path, content) + clock[0] = 61.0 + + source = tmp_path / "mount" + source.mkdir() + (source / "first.txt").write_text("first", encoding="utf-8") + (source / "second.txt").write_text("second", encoding="utf-8") + provider = _make_provider() + provider._config["mount_upload_deadline_seconds"] = 60 + monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: []) + provider._config["mounts"] = [ + SimpleNamespace(host_path=str(source), container_path="/mnt/data", read_only=False), + ] + client = FakeClient(files=DeadlineFilesAPI()) + + with caplog.at_level("WARNING"): + provider._apply_mounts(client, user_id="user-1") + + assert len(client.files.write_calls) == 1 + assert "time budget 60s" in caplog.text + + +@pytest.mark.parametrize( + "raw,expected", + [ + (0, 1), + (-5, 1), + (-100, 1), + (None, 120), + ("120s", 120), + ("abc", 120), + (float("inf"), 120), + ], + ids=["zero", "negative", "large_negative", "none", "suffix", "alpha", "infinity"], +) +def test_load_config_clamps_invalid_mount_upload_deadline(monkeypatch, caplog, raw, expected): + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + + class FakeConfig: + sandbox = SimpleNamespace( + model_extra={"mount_upload_deadline_seconds": raw}, + api_key="test-key", + template=None, + image=None, + domain=None, + home_dir=None, + idle_timeout=None, + replicas=None, + overflow_policy=None, + acquire_timeout=None, + burst_limit=None, + mounts=[], + environment=None, + ownership=None, + mount_upload_deadline_seconds=raw, + ) + + monkeypatch.setattr(mod, "get_app_config", lambda: FakeConfig()) + provider = mod.E2BSandboxProvider.__new__(mod.E2BSandboxProvider) + with caplog.at_level("WARNING"): + config = provider._load_config() + assert config["mount_upload_deadline_seconds"] == expected + if raw is None: + assert "clamping" not in caplog.text + else: + assert "mount_upload_deadline_seconds" in caplog.text + + +def test_load_config_custom_mount_upload_deadline_flows_to_apply_mounts(monkeypatch, tmp_path, caplog): + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + monkeypatch.setattr( + mod, + "get_app_config", + lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")), + ) + clock = [0.0] + monkeypatch.setattr(mod.time, "monotonic", lambda: clock[0]) + + class DeadlineFilesAPI(FakeFilesAPI): + def write(self, path: str, content: Any) -> None: + super().write(path, content) + clock[0] = 61.0 + + source = tmp_path / "mount" + source.mkdir() + (source / "first.txt").write_text("first", encoding="utf-8") + (source / "second.txt").write_text("second", encoding="utf-8") + + class FakeConfig: + sandbox = SimpleNamespace( + model_extra={"mount_upload_deadline_seconds": 60}, + api_key="test-key", + template=None, + image=None, + domain=None, + home_dir=None, + idle_timeout=None, + replicas=None, + overflow_policy=None, + acquire_timeout=None, + burst_limit=None, + mounts=[SimpleNamespace(host_path=str(source), container_path="/mnt/data", read_only=False)], + environment=None, + ownership=None, + mount_upload_deadline_seconds=60, + ) + skills = SimpleNamespace(container_path="/mnt/skills") + + monkeypatch.setattr(mod, "get_app_config", lambda: FakeConfig()) + provider = mod.E2BSandboxProvider.__new__(mod.E2BSandboxProvider) + provider._config = provider._load_config() + monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: []) + client = FakeClient(files=DeadlineFilesAPI()) + + with caplog.at_level("WARNING"): + provider._apply_mounts(client, user_id="user-1") + + assert provider._config["mount_upload_deadline_seconds"] == 60 + assert len(client.files.write_calls) == 1 + assert "time budget 60s" in caplog.text + + +def test_apply_mounts_deadline_reason_shows_configured_value(monkeypatch, tmp_path, caplog): + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + monkeypatch.setattr( + mod, + "get_app_config", + lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")), + ) + clock = [0.0] + monkeypatch.setattr(mod.time, "monotonic", lambda: clock[0]) + + class DeadlineFilesAPI(FakeFilesAPI): + def write(self, path: str, content: Any) -> None: + super().write(path, content) + clock[0] = 200.0 + + source = tmp_path / "mount" + source.mkdir() + (source / "first.txt").write_text("first", encoding="utf-8") + (source / "second.txt").write_text("second", encoding="utf-8") + provider = _make_provider() + provider._config["mount_upload_deadline_seconds"] = 180 + monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: []) + provider._config["mounts"] = [ + SimpleNamespace(host_path=str(source), container_path="/mnt/data", read_only=False), + ] + client = FakeClient(files=DeadlineFilesAPI()) + + with caplog.at_level("WARNING"): + provider._apply_mounts(client, user_id="user-1") + + assert len(client.files.write_calls) == 1 + assert "time budget 180s" in caplog.text + assert "attempted_files=1" in caplog.text + + def test_skill_projection_and_configured_mount_share_upload_budget(monkeypatch, tmp_path): mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") monkeypatch.setattr(mod, "_MAX_MOUNT_PASS_FILES", 1)