From 05e4f4f6d8d710bda3c9a9af6c46765e7c4f2d81 Mon Sep 17 00:00:00 2001 From: Aari Date: Wed, 22 Jul 2026 14:32:54 +0800 Subject: [PATCH] fix(sandbox): bound E2B output synchronization resources (#4364) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(sandbox): bound E2B output synchronization resources E2B release-time output sync pulled every changed file back from the remote VM with only a per-file size cap and no aggregate bound, so a pathological outputs tree (thousands of files, or many sub-cap files summing to gigabytes, or a slow VM) could make release download unboundedly on a hot path that runs at every agent turn end. Add three aggregate ceilings on top of the per-file cap — total bytes, file count, and a wall-clock deadline — enforced in the sync loop. When a ceiling is hit the pass stops early, logs what it dropped, and defers the rest to the next release. A truncated pass skips stale-manifest pruning so files it never reached are reconciled next time instead of being forgotten and re-downloaded. Closes #4340 * test(sandbox): pin multi-pass convergence of bounded output sync The four truncation tests each exercise a single capped pass. Add a two-pass test that locks in the invariant the design relies on for correctness: already-synced files are skipped before the budget check, so they never consume the cap and the deferred tail drains over successive releases instead of the leading files being re-downloaded every turn. A refactor that let a skipped file consume the cap would pass the single-pass tests but fail this one. --- backend/AGENTS.md | 2 +- .../e2b_sandbox/e2b_sandbox_provider.py | 49 +++++- backend/tests/test_e2b_sandbox_provider.py | 143 ++++++++++++++++++ 3 files changed, 192 insertions(+), 2 deletions(-) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index c9d8a641f..890c3aede 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -539,7 +539,7 @@ Scheduled-task runtime note: 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. -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. +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. **ACP agent tools**: - `invoke_acp_agent` - Invokes external ACP-compatible agents from `config.yaml` 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 fa4170583..a59379651 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 @@ -720,6 +720,18 @@ class E2BSandboxProvider(SandboxProvider): _SYNC_BACK_SUBDIRS = ("outputs", "workspace") _SYNC_MANIFEST_NAME = ".e2b-output-sync.json" + # Aggregate ceilings for a single release-time output-sync pass, layered on + # top of the per-file ``_MAX_DOWNLOAD_SIZE`` cap. The per-file cap bounds one + # artefact; these bound the whole pass so a pathological outputs tree + # (thousands of files, or many sub-cap files summing to gigabytes, or a slow + # VM) cannot make release download unboundedly. When a ceiling is hit the + # pass stops early, logs what it dropped, and leaves the manifest un-pruned + # so files it never reached are reconciled on the next release rather than + # being forgotten. + _MAX_SYNC_TOTAL_BYTES = 512 * 1024 * 1024 # total bytes downloaded per pass + _MAX_SYNC_FILES = 2000 # files downloaded per pass + _SYNC_DEADLINE_SECONDS = 120 # wall-clock budget per pass + @staticmethod def _load_sync_manifest(manifest_path: Path, sandbox_id: str) -> tuple[dict[str, dict[str, int]], bool]: """Load verified remote and host versions from a prior output sync.""" @@ -845,7 +857,16 @@ class E2BSandboxProvider(SandboxProvider): seen_manifest_keys: set[str] = set() from .e2b_sandbox import _MAX_DOWNLOAD_SIZE + # Aggregate budget for this pass (see the _MAX_SYNC_* class constants). + downloaded_bytes = 0 + downloaded_files = 0 + truncated_reason: str | None = None + deadline = time.monotonic() + self._SYNC_DEADLINE_SECONDS + for entry in stdout.split("\0"): + if time.monotonic() >= deadline: + truncated_reason = f"time budget {self._SYNC_DEADLINE_SECONDS}s" + break entry = entry.strip() if not entry: continue @@ -900,6 +921,13 @@ class E2BSandboxProvider(SandboxProvider): except OSError: pass + if downloaded_files >= self._MAX_SYNC_FILES: + truncated_reason = f"file count cap {self._MAX_SYNC_FILES}" + break + if downloaded_bytes + remote_size > self._MAX_SYNC_TOTAL_BYTES: + truncated_reason = f"total byte budget {self._MAX_SYNC_TOTAL_BYTES}" + break + try: data = sandbox.download_file(virtual_path) except Exception as e: @@ -910,6 +938,10 @@ class E2BSandboxProvider(SandboxProvider): e, ) continue + # Count the download against the budget regardless of the host-side + # write below: the remote round-trip is the resource being bounded. + downloaded_files += 1 + downloaded_bytes += remote_size try: host_path.parent.mkdir(parents=True, exist_ok=True) @@ -929,8 +961,13 @@ class E2BSandboxProvider(SandboxProvider): except OSError as e: logger.warning("e2b sync: failed to write %s on host: %s", host_path, e) + # A truncated pass did not observe every remote file, so + # ``seen_manifest_keys`` is incomplete; pruning "stale" entries here + # would forget files we simply never reached. Skip pruning and let the + # next release reconcile them (freshly downloaded entries are still + # written below). stale_keys = set(manifest) - seen_manifest_keys - if stale_keys: + if stale_keys and truncated_reason is None: for key in stale_keys: manifest.pop(key) manifest_dirty = True @@ -938,6 +975,16 @@ class E2BSandboxProvider(SandboxProvider): if manifest_dirty: self._write_sync_manifest(manifest_path, remote_sandbox_id, manifest) + if truncated_reason is not None: + logger.warning( + "e2b sync: sandbox=%s thread=%s truncated (%s); downloaded=%d files/%d bytes this pass, remaining artefacts deferred to next release", + sandbox.id, + thread_id, + truncated_reason, + downloaded_files, + downloaded_bytes, + ) + if synced or skipped: logger.info( "e2b sync: sandbox=%s thread=%s synced=%d skipped=%d", diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py index e8a440167..9c27fe04c 100644 --- a/backend/tests/test_e2b_sandbox_provider.py +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -1117,6 +1117,149 @@ def test_sync_outputs_to_host_is_noop_when_client_closed(): p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") +def _outputs_dir(tmp_path): + return Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs" + + +def test_sync_outputs_to_host_stops_at_file_count_cap(monkeypatch, tmp_path): + """A pass downloads at most ``_MAX_SYNC_FILES`` artefacts, deferring the rest.""" + p = _make_provider() + p._MAX_SYNC_FILES = 2 + _setup_paths(monkeypatch, tmp_path) + + names = ["a.txt", "b.txt", "c.txt", "d.txt"] + listing = "".join(f"5\t2.000000000\t/home/user/outputs/{n}\x00" for n in names) + files = FakeFilesAPI(store={f"/home/user/outputs/{n}": b"hello" for n in names}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + sb = _make_sandbox(FakeClient(commands=cmds, files=files), sandbox_id="sb-cap-files") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + out_dir = _outputs_dir(tmp_path) + written = sorted(f.name for f in out_dir.iterdir()) if out_dir.exists() else [] + assert written == ["a.txt", "b.txt"] + assert len(files.read_calls) == 2 + + +def test_sync_outputs_to_host_stops_at_total_byte_budget(monkeypatch, tmp_path): + """A pass stops before the cumulative download exceeds ``_MAX_SYNC_TOTAL_BYTES``.""" + p = _make_provider() + p._MAX_SYNC_TOTAL_BYTES = 25 # fits two 10-byte files, not a third + _setup_paths(monkeypatch, tmp_path) + + names = ["a.txt", "b.txt", "c.txt"] + listing = "".join(f"10\t2.000000000\t/home/user/outputs/{n}\x00" for n in names) + files = FakeFilesAPI(store={f"/home/user/outputs/{n}": b"0123456789" for n in names}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + sb = _make_sandbox(FakeClient(commands=cmds, files=files), sandbox_id="sb-cap-bytes") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + out_dir = _outputs_dir(tmp_path) + written = sorted(f.name for f in out_dir.iterdir()) if out_dir.exists() else [] + assert written == ["a.txt", "b.txt"] + assert len(files.read_calls) == 2 + + +def test_sync_outputs_to_host_stops_at_deadline(monkeypatch, tmp_path): + """A zero wall-clock budget aborts the pass before any download.""" + p = _make_provider() + p._SYNC_DEADLINE_SECONDS = 0 # monotonic() >= deadline on the first entry + _setup_paths(monkeypatch, tmp_path) + + listing = "5\t2.000000000\t/home/user/outputs/a.txt\x00" + files = FakeFilesAPI(store={"/home/user/outputs/a.txt": b"hello"}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + sb = _make_sandbox(FakeClient(commands=cmds, files=files), sandbox_id="sb-cap-deadline") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + out_dir = _outputs_dir(tmp_path) + assert not out_dir.exists() or list(out_dir.iterdir()) == [] + assert files.read_calls == [] + + +def test_sync_outputs_to_host_truncated_pass_preserves_stale_manifest(monkeypatch, tmp_path): + """A capped pass must not prune manifest entries it never got to inspect.""" + p = _make_provider() + p._MAX_SYNC_FILES = 1 + _setup_paths(monkeypatch, tmp_path) + thread_dir = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") + thread_dir.mkdir(parents=True, exist_ok=True) + # A prior-sync entry for a file absent from this listing. A complete pass + # would prune it as deleted; a truncated pass must leave it for next time. + (thread_dir / ".e2b-output-sync.json").write_text( + json.dumps( + { + "version": 1, + "sandbox_id": "sb-trunc", + "files": {"outputs/kept.txt": {"remote_size": 3, "remote_mtime_ns": 1, "host_size": 3, "host_mtime_ns": 1}}, + } + ), + encoding="utf-8", + ) + names = ["a.txt", "b.txt", "c.txt"] + listing = "".join(f"5\t2.000000000\t/home/user/outputs/{n}\x00" for n in names) + files = FakeFilesAPI(store={f"/home/user/outputs/{n}": b"hello" for n in names}) + cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)]) + sb = _make_sandbox(FakeClient(sandbox_id="sb-trunc", commands=cmds, files=files), sandbox_id="sb-trunc") + + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + + manifest = json.loads((thread_dir / ".e2b-output-sync.json").read_text(encoding="utf-8")) + assert "outputs/kept.txt" in manifest["files"] # un-reached entry survives + assert "outputs/a.txt" in manifest["files"] # the one download is recorded + assert len(files.read_calls) == 1 + + +def test_sync_outputs_to_host_converges_across_passes(monkeypatch, tmp_path): + """The deferred tail drains over successive passes without re-downloading + already-synced files. + + The budget check sits *below* the manifest-skip path, so a file synced in + an earlier pass is skipped before it can consume the cap on later passes. + That is what lets ``c``/``d`` finish instead of ``a``/``b`` being re-fetched + every release forever. A refactor that moved the budget check above the skip + would still pass the single-pass truncation tests but fail this one. + """ + p = _make_provider() + p._MAX_SYNC_FILES = 2 + _setup_paths(monkeypatch, tmp_path) + + names = ["a.txt", "b.txt", "c.txt", "d.txt"] + listing = "".join(f"5\t2.000000000\t/home/user/outputs/{n}\x00" for n in names) + files = FakeFilesAPI(store={f"/home/user/outputs/{n}": b"hello" for n in names}) + # One listing per pass; both passes share the same host tree, manifest and + # files API (so downloads accumulate and the manifest carries over). + cmds = FakeCommandsAPI( + [ + SimpleNamespace(stdout=listing, stderr="", exit_code=0), + SimpleNamespace(stdout=listing, stderr="", exit_code=0), + ] + ) + sb = _make_sandbox(FakeClient(commands=cmds, files=files), sandbox_id="sb-converge") + out_dir = _outputs_dir(tmp_path) + + # Pass 1: the file cap stops the pass after a, b. + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + assert sorted(f.name for f in out_dir.iterdir()) == ["a.txt", "b.txt"] + assert [r[0] for r in files.read_calls] == [ + "/home/user/outputs/a.txt", + "/home/user/outputs/b.txt", + ] + + # Pass 2: a, b are manifest hits skipped before the budget check, so the cap + # is spent draining the deferred tail c, d rather than re-downloading a, b. + p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1") + assert sorted(f.name for f in out_dir.iterdir()) == ["a.txt", "b.txt", "c.txt", "d.txt"] + assert [r[0] for r in files.read_calls] == [ + "/home/user/outputs/a.txt", + "/home/user/outputs/b.txt", + "/home/user/outputs/c.txt", + "/home/user/outputs/d.txt", + ] + + def test_download_file_uses_streaming_read_and_returns_full_bytes(): payload = b"A" * (128 * 1024) # 128 KiB — well below the cap. files = FakeFilesAPI(store={"/home/user/outputs/small.bin": payload})