github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-25 07:28:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sandbox): bound E2B output synchronization resources (#4364)

Browse Source 
* 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.
This commit is contained in:
Aari 2026-07-22 14:32:54 +08:00 committed by GitHub
parent e225ad57d7
commit 05e4f4f6d8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 192 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/AGENTS.md
View File

@ -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`

49
backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py
View File

@ -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",

143
backend/tests/test_e2b_sandbox_provider.py
View File

@ -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})