fix(sandbox): bound E2B mount upload resource use (#4812)

* fix(sandbox): bound E2B mount uploads

* fix(sandbox): revalidate E2B mount files

backend/packages/harness/deerflow/AGENTS.md

@@ -76,3 +76,19 @@ files. It is marked `live`, excluded from `make test`, and skipped in default
 CI.
 
 **Gateway Conformance Tests** (`TestGatewayConformance`): Validate that every dict-returning client method conforms to the corresponding Gateway Pydantic response model. Each test parses the client output through the Gateway model — if Gateway adds a required field that the client doesn't provide, Pydantic raises `ValidationError` and CI catches the drift. Covers: `ModelsListResponse`, `ModelResponse`, `SkillsListResponse`, `SkillResponse`, `SkillInstallResponse`, `McpConfigResponse`, `UploadResponse`, `MemoryConfigResponse`, `MemoryStatusResponse`.
+
+### E2B Mount Uploads
+
+The E2B provider uploads host mounts during sandbox creation. It passes binary file objects to the E2B SDK.
+
+Each mount has these fixed limits:
+
+- 100 MiB for one file.
+- 512 MiB for all files.
+- 2,000 files.
+
+The provider checks mount limits before upload. It rechecks each opened file descriptor against its preflight size before SDK upload.
+
+An invalid mount does not block later mounts.
+
+Each successful upload logs its source, destination, file count, byte count, and elapsed time.

backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py

@@ -99,6 +99,10 @@ MIN_CAPACITY_RESERVATION_SECONDS = 120.0
 # Hard upper bound for ``set_timeout`` (e2b currently caps at 24h on the
 # free plan; passing an excessive value is rejected by the control-plane).
 MAX_E2B_TIMEOUT = 24 * 60 * 60
+# These limits bound Gateway work during each E2B mount upload.
+_MAX_MOUNT_FILE_SIZE = 100 * 1024 * 1024
+_MAX_MOUNT_TOTAL_SIZE = 512 * 1024 * 1024
+_MAX_MOUNT_FILES = 2000
 
 # Metadata keys we attach to every sandbox so we can discover ours via
 # ``Sandbox.list(query={...})`` from any gateway process.
@@ -2095,22 +2099,32 @@ class E2BSandboxProvider(SandboxProvider):
         read_only: bool,
     ) -> None:
         """Recursively upload ``src`` into ``dest_dir`` inside the sandbox."""
-        if src.is_file():
-            target = f"{dest_dir}/{src.name}"
-            with src.open("rb") as fh:
-                client.files.write(target, fh.read())
-            if read_only:
-                try:
-                    client.commands.run(f"chmod a-w {shlex.quote(target)}")
-                except Exception:
-                    pass
-            return
+        started_at = time.monotonic()
+        source_is_file = src.is_file()
+        files: list[tuple[Path, str, int]] = []
+        total_size = 0
 
-        for path in src.rglob("*"):
-            if not path.is_file():
-                continue
-            rel = path.relative_to(src).as_posix()
-            target = f"{dest_dir}/{rel}"
+        def add_file(path: Path, target: str) -> None:
+            nonlocal total_size
+            file_size = path.stat().st_size
+            if file_size > _MAX_MOUNT_FILE_SIZE:
+                raise ValueError(f"Mount file {path} is {file_size} bytes and exceeds the {_MAX_MOUNT_FILE_SIZE}-byte file limit")
+            if len(files) >= _MAX_MOUNT_FILES:
+                raise ValueError(f"Mount {src} contains more than {_MAX_MOUNT_FILES} files and exceeds the {_MAX_MOUNT_FILES}-file limit")
+            total_size += file_size
+            if total_size > _MAX_MOUNT_TOTAL_SIZE:
+                raise ValueError(f"Mount {src} is at least {total_size} bytes and exceeds the {_MAX_MOUNT_TOTAL_SIZE}-byte total limit")
+            files.append((path, target, file_size))
+
+        if source_is_file:
+            add_file(src, f"{dest_dir}/{src.name}")
+        else:
+            for path in src.rglob("*"):
+                if path.is_file():
+                    rel = path.relative_to(src).as_posix()
+                    add_file(path, f"{dest_dir}/{rel}")
+
+        for path, target, expected_size in files:
             try:
                 make_dir = getattr(client.files, "make_dir", None)
                 if callable(make_dir):
@@ -2120,12 +2134,26 @@ class E2BSandboxProvider(SandboxProvider):
             except Exception:
                 pass
             with path.open("rb") as fh:
-                client.files.write(target, fh.read())
+                actual_size = os.fstat(fh.fileno()).st_size
+                if actual_size != expected_size:
+                    raise ValueError(f"Mount file {path} changed during upload preflight")
+                client.files.write(target, fh)
         if read_only:
             try:
-                client.commands.run(f"chmod -R a-w {shlex.quote(dest_dir)}")
+                chmod_target = files[0][1] if source_is_file else dest_dir
+                chmod_flag = "" if source_is_file else "-R "
+                client.commands.run(f"chmod {chmod_flag}a-w {shlex.quote(chmod_target)}")
             except Exception:
                 pass
+        elapsed_ms = int((time.monotonic() - started_at) * 1000)
+        logger.info(
+            "e2b mount upload: source=%s destination=%s files=%d bytes=%d elapsed_ms=%d",
+            src,
+            dest_dir,
+            len(files),
+            total_size,
+            elapsed_ms,
+        )
 
     def _evict_oldest_warm(self) -> str | None:
         """Evict the oldest warm entry, holding a transitioning slot.

backend/tests/test_e2b_sandbox_provider.py

@@ -105,6 +105,7 @@ class FakeFilesAPI:
         self.store = dict(store or {})
         self.read_calls: list[tuple[str, str | None]] = []
         self.write_calls: list[tuple[str, bytes]] = []
+        self.write_streamed: list[bool] = []
         self.streams: list[_FakeFileStream] = []
         self._stream_chunk_size = stream_chunk_size
 
@@ -124,9 +125,12 @@ class FakeFilesAPI:
         except UnicodeDecodeError:
             return data
 
-    def write(self, path: str, content: bytes) -> None:
-        self.write_calls.append((path, content))
-        self.store[path] = content
+    def write(self, path: str, content: Any) -> None:
+        is_stream = hasattr(content, "read")
+        data = content.read() if is_stream else content
+        self.write_streamed.append(is_stream)
+        self.write_calls.append((path, data))
+        self.store[path] = data
 
 
 class FakeClient:
@@ -361,6 +365,130 @@ def test_apply_mounts_uploads_only_enabled_skill_projection(monkeypatch, tmp_pat
     assert "/mnt/skills/integrations/lark-cli/disabled-integration/SKILL.md" not in uploaded_paths
 
 
+def test_upload_tree_streams_file_contents(tmp_path):
+    source = tmp_path / "large.bin"
+    source.write_bytes(b"mount content")
+    client = FakeClient()
+
+    provider = _make_provider()
+    provider._upload_tree(client, source, "/mnt/data", read_only=False)
+
+    assert client.files.write_calls == [("/mnt/data/large.bin", b"mount content")]
+    assert client.files.write_streamed == [True]
+
+
+@pytest.mark.parametrize("replacement_content", [b"123", b"12345"], ids=["smaller", "larger"])
+def test_upload_tree_rejects_file_size_changed_after_preflight(monkeypatch, tmp_path, replacement_content):
+    source = tmp_path / "small.bin"
+    source.write_bytes(b"1234")
+    replacement = tmp_path / "replacement.bin"
+    replacement.write_bytes(replacement_content)
+    original_open = Path.open
+
+    def replace_before_open(path: Path, *args, **kwargs):
+        if path == source and replacement.exists():
+            os.replace(replacement, source)
+        return original_open(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "open", replace_before_open)
+    client = FakeClient()
+
+    provider = _make_provider()
+    with pytest.raises(ValueError, match="changed during upload preflight"):
+        provider._upload_tree(client, source, "/mnt/data", read_only=False)
+
+    assert client.files.write_calls == []
+
+
+def test_upload_tree_rejects_oversized_file_before_upload(monkeypatch, tmp_path):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_FILE_SIZE", 4)
+    source = tmp_path / "large.bin"
+    source.write_bytes(b"12345")
+    client = FakeClient()
+
+    provider = _make_provider()
+    with pytest.raises(ValueError, match="exceeds the 4-byte file limit"):
+        provider._upload_tree(client, source, "/mnt/data", read_only=False)
+
+    assert client.files.write_calls == []
+
+
+def test_upload_tree_rejects_oversized_tree_before_upload(monkeypatch, tmp_path):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_FILE_SIZE", 10)
+    monkeypatch.setattr(mod, "_MAX_MOUNT_TOTAL_SIZE", 8)
+    source = tmp_path / "mount"
+    source.mkdir()
+    (source / "first.bin").write_bytes(b"12345")
+    (source / "second.bin").write_bytes(b"67890")
+    client = FakeClient()
+
+    provider = _make_provider()
+    with pytest.raises(ValueError, match="exceeds the 8-byte total limit"):
+        provider._upload_tree(client, source, "/mnt/data", read_only=False)
+
+    assert client.files.write_calls == []
+
+
+def test_upload_tree_rejects_excess_file_count_before_upload(monkeypatch, tmp_path):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_FILES", 1)
+    source = tmp_path / "mount"
+    source.mkdir()
+    (source / "first.txt").write_text("first", encoding="utf-8")
+    (source / "second.txt").write_text("second", encoding="utf-8")
+    client = FakeClient()
+
+    provider = _make_provider()
+    with pytest.raises(ValueError, match="exceeds the 1-file limit"):
+        provider._upload_tree(client, source, "/mnt/data", read_only=False)
+
+    assert client.files.write_calls == []
+
+
+def test_apply_mounts_continues_after_mount_exceeds_limit(monkeypatch, tmp_path):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_FILE_SIZE", 4)
+    monkeypatch.setattr(mod, "get_app_config", lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")))
+    oversized = tmp_path / "oversized"
+    oversized.mkdir()
+    (oversized / "large.bin").write_bytes(b"12345")
+    valid = tmp_path / "valid"
+    valid.mkdir()
+    (valid / "small.bin").write_bytes(b"1234")
+
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(oversized), container_path="/mnt/oversized", read_only=False),
+        SimpleNamespace(host_path=str(valid), container_path="/mnt/valid", read_only=False),
+    ]
+    client = FakeClient()
+
+    provider._apply_mounts(client, user_id="user-1")
+
+    assert client.files.write_calls == [("/mnt/valid/small.bin", b"1234")]
+
+
+def test_upload_tree_logs_upload_summary(caplog, tmp_path):
+    source = tmp_path / "mount"
+    source.mkdir()
+    (source / "first.txt").write_bytes(b"123")
+    (source / "second.txt").write_bytes(b"4567")
+    client = FakeClient()
+
+    provider = _make_provider()
+    with caplog.at_level("INFO"):
+        provider._upload_tree(client, source, "/mnt/data", read_only=False)
+
+    assert "source=" in caplog.text
+    assert "destination=/mnt/data" in caplog.text
+    assert "files=2" in caplog.text
+    assert "bytes=7" in caplog.text
+    assert "elapsed_ms=" in caplog.text
+
+
 def test_skill_projection_mounts_swallows_projection_failure(monkeypatch):
     """``_skill_projection_mounts`` must not raise — a projection failure used
     to propagate out of ``_apply_mounts`` before the configured-mounts loop