fix(sandbox): bound aggregate E2B mount upload work (#4842)

* fix(sandbox): bound aggregate E2B mount upload work

* fix(sandbox): preserve mount guards on upload failure

* fix(sandbox): cover mount preflight with deadline

* refactor(sandbox): clarify mount deadline checks

* refactor(sanbox): deduplicate mount deadline reason

* fix(sandbox): evaluate mount deadline reason lazily

README.md

@@ -1137,6 +1137,12 @@ creates while Redis or initial inventory is unavailable. Run Redis with persiste
 E2B acquisition uses a bounded executor. Waiting acquisitions do not use the
 default asyncio executor.
 
+Each E2B mount upload pass accepts at most 512 MiB and 2,000 files. The pass
+also has a cooperative 120-second deadline. Skill projections and configured
+mounts share these limits. The provider checks the deadline before each mount
+and during directory preflight. The deadline stops new file uploads after it
+expires. It does not interrupt active filesystem or E2B SDK calls.
+
 An E2B VM keeps its slot until E2B confirms destruction. This rule covers
 create and reclaim operations. Discovery can find a VM from another Gateway.
 Shutdown closes an unowned discovery client without destroying its VM.

backend/packages/harness/deerflow/AGENTS.md

@@ -87,8 +87,17 @@ Each mount has these fixed limits:
 - 512 MiB for all files.
 - 2,000 files.
 
+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
+each mount, during directory preflight, and before each SDK write. The deadline
+does not interrupt active filesystem or E2B SDK calls.
+
 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.
+
+A stopped pass logs its limit reason and elapsed time. It reports attempted and completed upload totals separately.

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

@@ -99,10 +99,41 @@ 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.
+# These limits bound one E2B mount.
 _MAX_MOUNT_FILE_SIZE = 100 * 1024 * 1024
 _MAX_MOUNT_TOTAL_SIZE = 512 * 1024 * 1024
 _MAX_MOUNT_FILES = 2000
+# These limits bound all uploads during one sandbox creation pass.
+_MAX_MOUNT_PASS_TOTAL_BYTES = 512 * 1024 * 1024
+_MAX_MOUNT_PASS_FILES = 2000
+# Deadline checks stop preflight work and new writes. Active SDK writes finish.
+_MOUNT_PASS_DEADLINE_SECONDS = 120
+
+
+def _mount_deadline_reason() -> str:
+    return f"time budget {_MOUNT_PASS_DEADLINE_SECONDS}s"
+
+
+class _MountPassLimitExceeded(Exception):
+    """Stop the current mount upload pass at its aggregate resource limit."""
+
+
+@dataclass
+class _MountUploadBudget:
+    deadline: float
+    attempted_bytes: int = 0
+    attempted_files: int = 0
+    completed_bytes: int = 0
+    completed_files: int = 0
+
+    @property
+    def expired(self) -> bool:
+        return time.monotonic() >= self.deadline
+
+    def check_deadline(self) -> None:
+        if self.expired:
+            raise _MountPassLimitExceeded(_mount_deadline_reason())
+
 
 # Metadata keys we attach to every sandbox so we can discover ours via
 # ``Sandbox.list(query={...})`` from any gateway process.
@@ -1769,6 +1800,21 @@ class E2BSandboxProvider(SandboxProvider):
             return []
 
     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)
+
+        def warn_pass_stopped(reason: str) -> None:
+            elapsed_ms = int((time.monotonic() - started_at) * 1000)
+            logger.warning(
+                "e2b mount upload pass stopped: reason=%s attempted_files=%d attempted_bytes=%d completed_files=%d completed_bytes=%d elapsed_ms=%d",
+                reason,
+                budget.attempted_files,
+                budget.attempted_bytes,
+                budget.completed_files,
+                budget.completed_bytes,
+                elapsed_ms,
+            )
+
         effective_user_id = user_id or get_effective_user_id()
         projection_mounts = self._skill_projection_mounts(effective_user_id)
         configured_mounts = self._config.get("mounts") or []
@@ -1791,6 +1837,9 @@ class E2BSandboxProvider(SandboxProvider):
             mounts.append((host_path, container_path, read_only))
 
         for host_path, container_path, read_only in mounts:
+            if budget.expired:
+                warn_pass_stopped(_mount_deadline_reason())
+                break
             if not host_path.exists():
                 logger.warning("Skipping e2b mount: host_path %s does not exist", host_path)
                 continue
@@ -1809,7 +1858,10 @@ class E2BSandboxProvider(SandboxProvider):
                 logger.debug("make_dir(%s) failed (continuing): %s", container_path, e)
 
             try:
-                self._upload_tree(client, host_path, container_path, read_only)
+                self._upload_tree(client, host_path, container_path, read_only, budget=budget)
+            except _MountPassLimitExceeded as e:
+                warn_pass_stopped(str(e))
+                break
             except Exception as e:
                 logger.warning("Failed to upload mount %s -> %s: %s", host_path, container_path, e)
 
@@ -2097,6 +2149,8 @@ class E2BSandboxProvider(SandboxProvider):
         src: Path,
         dest_dir: str,
         read_only: bool,
+        *,
+        budget: _MountUploadBudget | None = None,
     ) -> None:
         """Recursively upload ``src`` into ``dest_dir`` inside the sandbox."""
         started_at = time.monotonic()
@@ -2106,6 +2160,8 @@ class E2BSandboxProvider(SandboxProvider):
 
         def add_file(path: Path, target: str) -> None:
             nonlocal total_size
+            if budget is not None:
+                budget.check_deadline()
             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")
@@ -2120,31 +2176,49 @@ class E2BSandboxProvider(SandboxProvider):
             add_file(src, f"{dest_dir}/{src.name}")
         else:
             for path in src.rglob("*"):
+                if budget is not None:
+                    budget.check_deadline()
                 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:
+        upload_attempted = False
-            try:
+        try:
-                make_dir = getattr(client.files, "make_dir", None)
+            for path, target, expected_size in files:
-                if callable(make_dir):
+                if budget is not None:
-                    parent = target.rsplit("/", 1)[0]
+                    budget.check_deadline()
-                    if parent and parent != dest_dir:
+                if budget is not None and budget.attempted_files >= _MAX_MOUNT_PASS_FILES:
-                        make_dir(parent)
+                    raise _MountPassLimitExceeded(f"file count cap {_MAX_MOUNT_PASS_FILES}")
-            except Exception:
+                if budget is not None and budget.attempted_bytes + expected_size > _MAX_MOUNT_PASS_TOTAL_BYTES:
-                pass
+                    raise _MountPassLimitExceeded(f"total byte budget {_MAX_MOUNT_PASS_TOTAL_BYTES}")
-            with path.open("rb") as fh:
+                try:
-                actual_size = os.fstat(fh.fileno()).st_size
+                    make_dir = getattr(client.files, "make_dir", None)
-                if actual_size != expected_size:
+                    if callable(make_dir):
-                    raise ValueError(f"Mount file {path} changed during upload preflight")
+                        parent = target.rsplit("/", 1)[0]
-                client.files.write(target, fh)
+                        if parent and parent != dest_dir:
-        if read_only:
+                            make_dir(parent)
-            try:
+                except Exception:
-                chmod_target = files[0][1] if source_is_file else dest_dir
+                    pass
-                chmod_flag = "" if source_is_file else "-R "
+                with path.open("rb") as fh:
-                client.commands.run(f"chmod {chmod_flag}a-w {shlex.quote(chmod_target)}")
+                    actual_size = os.fstat(fh.fileno()).st_size
-            except Exception:
+                    if actual_size != expected_size:
-                pass
+                        raise ValueError(f"Mount file {path} changed during upload preflight")
+                    upload_attempted = True
+                    if budget is not None:
+                        budget.attempted_files += 1
+                        budget.attempted_bytes += expected_size
+                    client.files.write(target, fh)
+                if budget is not None:
+                    budget.completed_files += 1
+                    budget.completed_bytes += expected_size
+        finally:
+            if read_only and upload_attempted:
+                try:
+                    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",

backend/tests/test_e2b_sandbox_provider.py

@@ -471,6 +471,303 @@ def test_apply_mounts_continues_after_mount_exceeds_limit(monkeypatch, tmp_path)
     assert client.files.write_calls == [("/mnt/valid/small.bin", b"1234")]
 
 
+def test_apply_mounts_bounds_total_bytes_across_mounts(monkeypatch, tmp_path, caplog):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_PASS_TOTAL_BYTES", 7)
+    monkeypatch.setattr(mod, "get_app_config", lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")))
+    first = tmp_path / "first"
+    first.mkdir()
+    (first / "first.bin").write_bytes(b"1234")
+    second = tmp_path / "second"
+    second.mkdir()
+    (second / "second.bin").write_bytes(b"5678")
+
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(first), container_path="/mnt/first", read_only=False),
+        SimpleNamespace(host_path=str(second), container_path="/mnt/second", read_only=False),
+    ]
+    client = FakeClient()
+
+    with caplog.at_level("WARNING"):
+        provider._apply_mounts(client, user_id="user-1")
+
+    assert client.files.write_calls == [("/mnt/first/first.bin", b"1234")]
+    assert "total byte budget 7" in caplog.text
+    assert "attempted_files=1" in caplog.text
+    assert "attempted_bytes=4" in caplog.text
+
+
+def test_apply_mounts_bounds_total_files_across_mounts(monkeypatch, tmp_path, caplog):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_PASS_FILES", 1)
+    monkeypatch.setattr(mod, "get_app_config", lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")))
+    first = tmp_path / "first"
+    first.mkdir()
+    (first / "first.txt").write_text("first", encoding="utf-8")
+    second = tmp_path / "second"
+    second.mkdir()
+    (second / "second.txt").write_text("second", encoding="utf-8")
+
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(first), container_path="/mnt/first", read_only=False),
+        SimpleNamespace(host_path=str(second), container_path="/mnt/second", read_only=False),
+    ]
+    client = FakeClient()
+
+    with caplog.at_level("WARNING"):
+        provider._apply_mounts(client, user_id="user-1")
+
+    assert client.files.write_calls == [("/mnt/first/first.txt", b"first")]
+    assert "file count cap 1" in caplog.text
+    assert "attempted_files=1" in caplog.text
+
+
+def test_read_only_mount_remains_read_only_when_pass_limit_stops_mid_mount(monkeypatch, tmp_path):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_PASS_FILES", 1)
+    monkeypatch.setattr(
+        mod,
+        "get_app_config",
+        lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")),
+    )
+    source = tmp_path / "read-only"
+    source.mkdir()
+    (source / "first.txt").write_text("first", encoding="utf-8")
+    (source / "second.txt").write_text("second", encoding="utf-8")
+
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(source), container_path="/mnt/read-only", read_only=True),
+    ]
+    client = FakeClient()
+
+    provider._apply_mounts(client, user_id="user-1")
+
+    assert len(client.files.write_calls) == 1
+    assert "chmod -R a-w /mnt/read-only" in client.commands.calls
+
+
+def test_read_only_mount_is_not_chmodded_when_no_upload_starts(monkeypatch, tmp_path):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_PASS_FILES", 0)
+    monkeypatch.setattr(
+        mod,
+        "get_app_config",
+        lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")),
+    )
+    source = tmp_path / "read-only"
+    source.mkdir()
+    (source / "file.txt").write_text("content", encoding="utf-8")
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(source), container_path="/mnt/read-only", read_only=True),
+    ]
+    client = FakeClient()
+
+    provider._apply_mounts(client, user_id="user-1")
+
+    assert client.files.write_calls == []
+    assert client.commands.calls == []
+
+
+def test_failed_write_consumes_aggregate_upload_budget(monkeypatch, tmp_path, caplog):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MAX_MOUNT_PASS_TOTAL_BYTES", 4)
+    monkeypatch.setattr(
+        mod,
+        "get_app_config",
+        lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")),
+    )
+
+    class FailFirstWriteAPI(FakeFilesAPI):
+        def write(self, path: str, content: Any) -> None:
+            super().write(path, content)
+            if len(self.write_calls) == 1:
+                raise RuntimeError("response lost after upload")
+
+    first = tmp_path / "first"
+    first.mkdir()
+    (first / "first.bin").write_bytes(b"1234")
+    second = tmp_path / "second"
+    second.mkdir()
+    (second / "second.bin").write_bytes(b"5")
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(first), container_path="/mnt/first", read_only=False),
+        SimpleNamespace(host_path=str(second), container_path="/mnt/second", read_only=False),
+    ]
+    client = FakeClient(files=FailFirstWriteAPI())
+
+    with caplog.at_level("WARNING"):
+        provider._apply_mounts(client, user_id="user-1")
+
+    assert client.files.write_calls == [("/mnt/first/first.bin", b"1234")]
+    assert "attempted_files=1" in caplog.text
+    assert "attempted_bytes=4" in caplog.text
+    assert "completed_files=0" in caplog.text
+    assert "completed_bytes=0" in caplog.text
+
+
+def test_apply_mounts_deadline_stops_before_next_file(monkeypatch, tmp_path, caplog):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MOUNT_PASS_DEADLINE_SECONDS", 1)
+    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] = 2.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()
+    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 client.files.write_calls[0] in {
+        ("/mnt/data/first.txt", b"first"),
+        ("/mnt/data/second.txt", b"second"),
+    }
+    assert "time budget 1s" in caplog.text
+    assert "attempted_files=1" in caplog.text
+
+
+def test_apply_mounts_deadline_stops_directory_preflight(monkeypatch, tmp_path, caplog):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MOUNT_PASS_DEADLINE_SECONDS", 1)
+    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])
+    source = tmp_path / "mount"
+    source.mkdir()
+    first = source / "first.txt"
+    first.write_text("first", encoding="utf-8")
+    second = source / "second.txt"
+    second.write_text("second", encoding="utf-8")
+    original_is_file = Path.is_file
+    inspected: list[Path] = []
+
+    def slow_rglob(path: Path, pattern: str):
+        assert path == source
+        assert pattern == "*"
+        yield first
+        clock[0] = 2.0
+        yield second
+
+    def record_is_file(path: Path) -> bool:
+        inspected.append(path)
+        return original_is_file(path)
+
+    monkeypatch.setattr(Path, "rglob", slow_rglob)
+    monkeypatch.setattr(Path, "is_file", record_is_file)
+    provider = _make_provider()
+    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()
+
+    with caplog.at_level("WARNING"):
+        provider._apply_mounts(client, user_id="user-1")
+
+    assert first in inspected
+    assert second not in inspected
+    assert client.files.write_calls == []
+    assert "time budget 1s" in caplog.text
+
+
+def test_apply_mounts_deadline_stops_before_next_mount_preflight(monkeypatch, tmp_path, caplog):
+    mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
+    monkeypatch.setattr(mod, "_MOUNT_PASS_DEADLINE_SECONDS", 1)
+    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] = 2.0
+
+    first = tmp_path / "first"
+    first.mkdir()
+    (first / "first.txt").write_text("first", encoding="utf-8")
+    second = tmp_path / "second"
+    second.mkdir()
+    (second / "second.txt").write_text("second", encoding="utf-8")
+    original_is_file = Path.is_file
+    inspected: list[Path] = []
+
+    def record_is_file(path: Path) -> bool:
+        inspected.append(path)
+        return original_is_file(path)
+
+    monkeypatch.setattr(Path, "is_file", record_is_file)
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(first), container_path="/mnt/first", read_only=False),
+        SimpleNamespace(host_path=str(second), container_path="/mnt/second", read_only=False),
+    ]
+    client = FakeClient(files=DeadlineFilesAPI())
+
+    with caplog.at_level("WARNING"):
+        provider._apply_mounts(client, user_id="user-1")
+
+    assert first in inspected
+    assert second not in inspected
+    assert client.files.write_calls == [("/mnt/first/first.txt", b"first")]
+    assert "time budget 1s" 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)
+    monkeypatch.setattr(mod, "get_app_config", lambda: SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills")))
+    projection = tmp_path / "projection"
+    projection.mkdir()
+    (projection / "SKILL.md").write_text("skill", encoding="utf-8")
+    configured = tmp_path / "configured"
+    configured.mkdir()
+    (configured / "notes.txt").write_text("notes", encoding="utf-8")
+
+    provider = _make_provider()
+    monkeypatch.setattr(provider, "_skill_projection_mounts", lambda _user_id: [(projection, "/mnt/skills/public", True)])
+    provider._config["mounts"] = [
+        SimpleNamespace(host_path=str(configured), container_path="/mnt/configured", read_only=False),
+    ]
+    client = FakeClient()
+
+    provider._apply_mounts(client, user_id="user-1")
+
+    assert client.files.write_calls == [("/mnt/skills/public/SKILL.md", b"skill")]
+
+
 def test_upload_tree_logs_upload_summary(caplog, tmp_path):
     source = tmp_path / "mount"
     source.mkdir()