fix(sandbox): sync same-size E2B output updates (#4329)

* fix(sandbox): detect same-size E2B output updates

* test(sandbox): cover E2B output modification times

* fix(sandbox): persist E2B output sync versions

* test(sandbox): cover E2B output sync manifest

* docs(backend): document E2B sync manifest

* fix(sandbox): scope E2B sync manifests

* test(sandbox): cover E2B sync manifest lifecycle

* docs(backend): describe E2B manifest lifecycle

* fix(sandbox): fall back from empty E2B sandbox ID

* test(sandbox): cover empty E2B sandbox IDs

* docs: clarify E2B output sync ownership
This commit is contained in:
luo jiyin 2026-07-21 22:30:14 +08:00 committed by GitHub
parent da3feb3863
commit d2116d861b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 348 additions and 31 deletions

View File

@ -464,6 +464,8 @@ 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.
**ACP agent tools**:
- `invoke_acp_agent` - Invokes external ACP-compatible agents from `config.yaml`
- ACP launchers must be real ACP adapters. The standard `codex` CLI is not ACP-compatible by itself; configure a wrapper such as `npx -y @zed-industries/codex-acp` or an installed `codex-acp` binary

View File

@ -74,7 +74,7 @@ class E2BSandbox(Sandbox):
@property
def sandbox_id(self) -> str:
"""e2b-side sandbox id (different from DeerFlow's ``self.id`` cache key)."""
return getattr(self._client, "sandbox_id", self.id)
return getattr(self._client, "sandbox_id", None) or self.id
def close(self) -> None:
with self._lock:

View File

@ -26,6 +26,7 @@ from __future__ import annotations
import asyncio
import atexit
import hashlib
import json
import logging
import os
import shlex
@ -34,6 +35,7 @@ import threading
import time
import uuid
from collections import OrderedDict
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
@ -716,6 +718,52 @@ class E2BSandboxProvider(SandboxProvider):
# ── Output mirroring ────────────────────────────────────────────────
_SYNC_BACK_SUBDIRS = ("outputs", "workspace")
_SYNC_MANIFEST_NAME = ".e2b-output-sync.json"
@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."""
try:
data = json.loads(manifest_path.read_text(encoding="utf-8"))
except FileNotFoundError:
return {}, False
except (OSError, json.JSONDecodeError) as e:
logger.warning("e2b sync: failed to load manifest %s: %s", manifest_path, e)
return {}, True
if not isinstance(data, dict) or data.get("version") != 1 or not isinstance(data.get("files"), dict):
logger.warning("e2b sync: ignoring invalid manifest %s", manifest_path)
return {}, True
if data.get("sandbox_id") != sandbox_id:
logger.debug("e2b sync: ignoring manifest from another sandbox %s", manifest_path)
return {}, True
files: dict[str, dict[str, int]] = {}
for key, value in data["files"].items():
if not isinstance(key, str) or not isinstance(value, dict):
continue
required = ("remote_size", "remote_mtime_ns", "host_size", "host_mtime_ns")
if all(isinstance(value.get(field), int) for field in required):
files[key] = {field: value[field] for field in required}
return files, False
@staticmethod
def _write_sync_manifest(
manifest_path: Path,
sandbox_id: str,
files: dict[str, dict[str, int]],
) -> None:
"""Atomically store output-sync versions after host files are written."""
try:
manifest_path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = manifest_path.with_name(f"{manifest_path.name}.tmp")
tmp_path.write_text(
json.dumps({"version": 1, "sandbox_id": sandbox_id, "files": files}, sort_keys=True),
encoding="utf-8",
)
tmp_path.replace(manifest_path)
except OSError as e:
logger.warning("e2b sync: failed to write manifest %s: %s", manifest_path, e)
def _sync_outputs_to_host(
self,
@ -733,11 +781,14 @@ class E2BSandboxProvider(SandboxProvider):
local provider. The e2b VM has no shared host filesystem, so we
explicitly pull artifacts back at release time.
We only mirror files whose host-side counterpart is missing or has a
different size this gives an effective per-file dedup with a single
round-trip per release for unchanged trees, and avoids re-downloading
large generated files (e.g. PDFs, datasets) on every tool turn that
triggers a release.
We mirror files whose host-side counterpart is missing, has a different
size, or has a different remote modification time. A thread-local
manifest stores the remote version and the actual host metadata after
each write. This avoids false updates on host filesystems that round
modification times.
The remote file is the source of truth. The next sync overwrites a
host-side edit when its size or modification time differs.
Failures are logged at WARNING level but never raised: artifact
download is non-critical for sandbox lifecycle, and we already log
@ -753,13 +804,17 @@ class E2BSandboxProvider(SandboxProvider):
home_dir = sandbox.home_dir.rstrip("/") or "/home/user"
paths = get_paths()
thread_root = paths.thread_dir(thread_id, user_id=user_id) / "user-data"
thread_dir = paths.thread_dir(thread_id, user_id=user_id)
thread_root = thread_dir / "user-data"
host_targets: dict[str, Path] = {sub: thread_root / sub for sub in self._SYNC_BACK_SUBDIRS}
manifest_path = thread_dir / self._SYNC_MANIFEST_NAME
remote_sandbox_id = sandbox.sandbox_id
manifest, manifest_dirty = self._load_sync_manifest(manifest_path, remote_sandbox_id)
# Build a single shell command that lists all files in the sync dirs
# with size + path, NUL-separated for safe parsing of weird filenames.
# find -printf '%s\t%p\0' keeps us to one round-trip regardless of
# how many subdirs we mirror.
# with size, modification time, and path, NUL-separated for safe
# parsing of weird filenames. This keeps us to one round-trip
# regardless of how many subdirs we mirror.
#
# We list using the *physical* /home/user paths (the bootstrap symlink
# /mnt/user-data -> /home/user follows transparently), then translate
@ -768,7 +823,7 @@ class E2BSandboxProvider(SandboxProvider):
# that the path is under ``VIRTUAL_PATH_PREFIX`` (/mnt/user-data) and
# internally re-resolves it to /home/user via ``_resolve_path``.
find_targets = " ".join(shlex.quote(f"{home_dir}/{sub}") for sub in self._SYNC_BACK_SUBDIRS)
list_cmd = f'for d in {find_targets}; do [ -d "$d" ] && find "$d" -type f -printf \'%s\\t%p\\0\' 2>/dev/null; done'
list_cmd = f'for d in {find_targets}; do [ -d "$d" ] && find "$d" -type f -printf \'%s\\t%T@\\t%p\\0\' 2>/dev/null; done'
try:
result = client.commands.run(list_cmd)
@ -781,10 +836,13 @@ class E2BSandboxProvider(SandboxProvider):
stdout = getattr(result, "stdout", "") or ""
if not stdout:
if manifest or manifest_dirty:
self._write_sync_manifest(manifest_path, remote_sandbox_id, {})
return
synced = 0
skipped = 0
seen_manifest_keys: set[str] = set()
from .e2b_sandbox import _MAX_DOWNLOAD_SIZE
for entry in stdout.split("\0"):
@ -792,22 +850,13 @@ class E2BSandboxProvider(SandboxProvider):
if not entry:
continue
try:
size_str, remote_path = entry.split("\t", 1)
size_str, remote_mtime_str, remote_path = entry.split("\t", 2)
remote_size = int(size_str)
except ValueError:
remote_mtime_ns = int(Decimal(remote_mtime_str) * 1_000_000_000)
except (InvalidOperation, ValueError):
logger.debug("e2b sync: unparseable entry %r", entry)
continue
if remote_size > _MAX_DOWNLOAD_SIZE:
logger.warning(
"e2b sync: skipping oversize artefact %s (%d bytes > %d cap)",
remote_path,
remote_size,
_MAX_DOWNLOAD_SIZE,
)
skipped += 1
continue
# Determine which subdir this file belongs to so we can compute
# the relative path on the host side. remote_path is absolute,
# e.g. /home/user/outputs/foo/bar.pdf
@ -824,9 +873,28 @@ class E2BSandboxProvider(SandboxProvider):
if sub_match is None:
continue
_sub, host_path, virtual_path = sub_match
manifest_key = host_path.relative_to(thread_root).as_posix()
seen_manifest_keys.add(manifest_key)
if remote_size > _MAX_DOWNLOAD_SIZE:
logger.warning(
"e2b sync: skipping oversize artefact %s (%d bytes > %d cap)",
remote_path,
remote_size,
_MAX_DOWNLOAD_SIZE,
)
skipped += 1
continue
try:
if host_path.exists() and host_path.stat().st_size == remote_size:
host_stat = host_path.stat()
entry = manifest.get(manifest_key)
if entry == {
"remote_size": remote_size,
"remote_mtime_ns": remote_mtime_ns,
"host_size": host_stat.st_size,
"host_mtime_ns": host_stat.st_mtime_ns,
}:
skipped += 1
continue
except OSError:
@ -847,11 +915,29 @@ class E2BSandboxProvider(SandboxProvider):
host_path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = host_path.with_name(host_path.name + ".e2bsync.tmp")
tmp_path.write_bytes(data)
os.utime(tmp_path, ns=(remote_mtime_ns, remote_mtime_ns))
tmp_path.replace(host_path)
host_stat = host_path.stat()
manifest[manifest_key] = {
"remote_size": remote_size,
"remote_mtime_ns": remote_mtime_ns,
"host_size": host_stat.st_size,
"host_mtime_ns": host_stat.st_mtime_ns,
}
manifest_dirty = True
synced += 1
except OSError as e:
logger.warning("e2b sync: failed to write %s on host: %s", host_path, e)
stale_keys = set(manifest) - seen_manifest_keys
if stale_keys:
for key in stale_keys:
manifest.pop(key)
manifest_dirty = True
if manifest_dirty:
self._write_sync_manifest(manifest_path, remote_sandbox_id, manifest)
if synced or skipped:
logger.info(
"e2b sync: sandbox=%s thread=%s synced=%d skipped=%d",

View File

@ -3,6 +3,8 @@
from __future__ import annotations
import importlib
import json
import os
import threading
from collections import OrderedDict
from types import SimpleNamespace
@ -210,6 +212,13 @@ def test_thread_key_returns_user_thread_tuple():
assert p._thread_key("t1", "u1") == ("u1", "t1")
def test_sandbox_id_falls_back_when_client_id_is_none():
client = FakeClient(sandbox_id=None)
sandbox = _make_sandbox(client, sandbox_id="fallback-id")
assert sandbox.sandbox_id == "fallback-id"
def test_stable_seed_is_deterministic_and_user_scoped():
p = _make_provider()
s_a = p._stable_seed("t1", "u1")
@ -817,7 +826,7 @@ def _setup_paths(monkeypatch, tmp_path):
def test_sync_outputs_to_host_writes_new_files(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
listing = "13\t/home/user/outputs/random.pdf\x00"
listing = "13\t2.000000000\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(commands=cmds, files=files)
@ -830,23 +839,243 @@ def test_sync_outputs_to_host_writes_new_files(monkeypatch, tmp_path):
assert expected.read_bytes() == b"%PDF-1.4hello"
def test_sync_outputs_to_host_skips_unchanged_files(monkeypatch, tmp_path):
def test_sync_outputs_to_host_updates_changed_same_size_file(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
out_dir = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs"
out_dir.mkdir(parents=True, exist_ok=True)
target = out_dir / "random.pdf"
target.write_bytes(b"%PDF-1.4hello")
os.utime(target, ns=(1_000_000_000, 1_000_000_000))
listing = "13\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"DIFFERENT-SAME-LEN"})
listing = "13\t2.000000000\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"changed-value"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-sync-2")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
assert files.read_calls == [], "size match should skip the download round-trip"
assert files.read_calls, "same-size files must be checked for updates"
assert target.read_bytes() == b"changed-value"
assert target.stat().st_mtime_ns == 2_000_000_000
def test_sync_outputs_to_host_skips_file_when_manifest_matches(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
out_dir = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs"
out_dir.mkdir(parents=True, exist_ok=True)
target = out_dir / "random.pdf"
target.write_bytes(b"%PDF-1.4hello")
os.utime(target, ns=(2_000_000_000, 2_000_000_000))
listing = "13\t2.000000000\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"})
cmds = FakeCommandsAPI(
[
SimpleNamespace(stdout=listing, stderr="", exit_code=0),
SimpleNamespace(stdout=listing, stderr="", exit_code=0),
]
)
client = FakeClient(commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-sync-unchanged")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
files.read_calls.clear()
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
assert files.read_calls == []
assert target.read_bytes() == b"%PDF-1.4hello"
def test_sync_outputs_to_host_uses_manifest_when_host_mtime_is_rounded(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
paths = Paths(base_dir=tmp_path)
thread_dir = paths.thread_dir("t1", user_id="u1")
target = thread_dir / "user-data" / "outputs" / "random.pdf"
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(b"%PDF-1.4hello")
rounded_host_mtime_ns = 1_720_000_000_123_456_800
os.utime(target, ns=(rounded_host_mtime_ns, rounded_host_mtime_ns))
(thread_dir / ".e2b-output-sync.json").write_text(
json.dumps(
{
"version": 1,
"sandbox_id": "sb-sync-manifest",
"files": {
"outputs/random.pdf": {
"remote_size": 13,
"remote_mtime_ns": 1_720_000_000_123_456_789,
"host_size": 13,
"host_mtime_ns": rounded_host_mtime_ns,
}
},
}
),
encoding="utf-8",
)
listing = "13\t1720000000.1234567890\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(sandbox_id="sb-sync-manifest", commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-sync-manifest")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
assert files.read_calls == []
def test_sync_outputs_to_host_records_variable_precision_remote_mtime(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
listing = "13\t1720000000.1234567890\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-sync-precision")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
manifest_path = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / ".e2b-output-sync.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
assert manifest["files"]["outputs/random.pdf"]["remote_mtime_ns"] == 1_720_000_000_123_456_789
def test_sync_outputs_to_host_removes_manifest_entries_for_deleted_files(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
paths = Paths(base_dir=tmp_path)
thread_dir = paths.thread_dir("t1", user_id="u1")
thread_dir.mkdir(parents=True, exist_ok=True)
(thread_dir / ".e2b-output-sync.json").write_text(
json.dumps(
{
"version": 1,
"sandbox_id": "sb-sync-cleanup",
"files": {
"outputs/deleted.txt": {
"remote_size": 3,
"remote_mtime_ns": 1,
"host_size": 3,
"host_mtime_ns": 1,
}
},
}
),
encoding="utf-8",
)
listing = "5\t1720000000.1234567890\t/home/user/outputs/live.txt\x00"
files = FakeFilesAPI(store={"/home/user/outputs/live.txt": b"alive"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-sync-cleanup")
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 set(manifest["files"]) == {"outputs/live.txt"}
def test_sync_outputs_to_host_discards_manifest_from_another_sandbox(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
paths = Paths(base_dir=tmp_path)
thread_dir = paths.thread_dir("t1", user_id="u1")
target = thread_dir / "user-data" / "outputs" / "random.pdf"
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(b"%PDF-1.4hello")
remote_mtime_ns = 1_720_000_000_123_456_789
os.utime(target, ns=(remote_mtime_ns, remote_mtime_ns))
(thread_dir / ".e2b-output-sync.json").write_text(
json.dumps(
{
"version": 1,
"sandbox_id": "sb-old",
"files": {
"outputs/random.pdf": {
"remote_size": 13,
"remote_mtime_ns": remote_mtime_ns,
"host_size": 13,
"host_mtime_ns": remote_mtime_ns,
}
},
}
),
encoding="utf-8",
)
listing = "13\t1720000000.1234567890\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(sandbox_id="sb-new", commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-new")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
assert files.read_calls
manifest = json.loads((thread_dir / ".e2b-output-sync.json").read_text(encoding="utf-8"))
assert manifest["sandbox_id"] == "sb-new"
def test_sync_outputs_to_host_resets_empty_manifest_for_new_sandbox(monkeypatch, tmp_path):
p = _make_provider()
_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)
manifest_path = thread_dir / ".e2b-output-sync.json"
manifest_path.write_text(
json.dumps(
{
"version": 1,
"sandbox_id": "sb-old",
"files": {
"outputs/stale.txt": {
"remote_size": 1,
"remote_mtime_ns": 1,
"host_size": 1,
"host_mtime_ns": 1,
}
},
}
),
encoding="utf-8",
)
client = FakeClient(
sandbox_id="sb-new",
commands=FakeCommandsAPI([SimpleNamespace(stdout="", stderr="", exit_code=0)]),
)
sb = _make_sandbox(client, sandbox_id="sb-new")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
assert manifest == {"version": 1, "sandbox_id": "sb-new", "files": {}}
def test_sync_outputs_to_host_restores_externally_modified_file(monkeypatch, tmp_path):
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
listing = "13\t1720000000.1234567890\t/home/user/outputs/random.pdf\x00"
files = FakeFilesAPI(store={"/home/user/outputs/random.pdf": b"%PDF-1.4hello"})
cmds = FakeCommandsAPI(
[
SimpleNamespace(stdout=listing, stderr="", exit_code=0),
SimpleNamespace(stdout=listing, stderr="", exit_code=0),
]
)
client = FakeClient(commands=cmds, files=files)
sb = _make_sandbox(client, sandbox_id="sb-sync-local-change")
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
target = Paths(base_dir=tmp_path).thread_dir("t1", user_id="u1") / "user-data" / "outputs" / "random.pdf"
target.write_bytes(b"changed-value")
os.utime(target, ns=(1_720_000_001_000_000_000, 1_720_000_001_000_000_000))
p._sync_outputs_to_host(sb, thread_id="t1", user_id="u1")
assert len(files.read_calls) == 2
assert target.read_bytes() == b"%PDF-1.4hello"
@ -869,7 +1098,7 @@ def test_sync_outputs_to_host_uses_virtual_path_for_download(monkeypatch, tmp_pa
p = _make_provider()
_setup_paths(monkeypatch, tmp_path)
listing = "5\t/home/user/outputs/sub/x.txt\x00"
listing = "5\t2.000000000\t/home/user/outputs/sub/x.txt\x00"
files = FakeFilesAPI(store={"/home/user/outputs/sub/x.txt": b"hello"})
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(commands=cmds, files=files)
@ -990,7 +1219,7 @@ def test_sync_outputs_to_host_skips_oversize_files(monkeypatch, tmp_path):
_setup_paths(monkeypatch, tmp_path)
oversize = e2b_sb_mod._MAX_DOWNLOAD_SIZE + 1
listing = f"{oversize}\t/home/user/outputs/huge.bin\x00"
listing = f"{oversize}\t2.000000000\t/home/user/outputs/huge.bin\x00"
files = FakeFilesAPI() # no store entry: any read attempt would raise
cmds = FakeCommandsAPI([SimpleNamespace(stdout=listing, stderr="", exit_code=0)])
client = FakeClient(commands=cmds, files=files)