diff --git a/backend/AGENTS.md b/backend/AGENTS.md index fb2b85bcb..147b79a29 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -454,6 +454,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti **Implementations**: - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` API honours the `/mnt/user-data` contract uniformly with AIO. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Legacy global-custom mounts are gated by the same user-scoped skill discovery rule used for prompt/list visibility; providers must not infer visibility from raw directory presence alone. - `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths — it never touches the ownership store (that would be blocking IO on the event loop); ownership is published on acquire/reclaim and refreshed off the event loop by the dedicated renewal thread (`_renew_owned_leases`). Legacy global-custom mounts follow the same shared visibility helper as local and remote providers. +- `E2BSandboxProvider` (`packages/harness/deerflow/community/e2b_sandbox/`) - E2B-backed remote isolation. Acquire and the complete release transition (output sync, timeout refresh, client close, and warm-pool publication) share a per-user/thread lock, so a same-process acquire cannot discover or create a replacement while release is between active and warm states. The provider-wide registry lock is not held across remote IO. - **Cross-instance ownership store** (`aio_sandbox/ownership/`, #4206): gateway instances sharing a container backend coordinate container ownership through a pluggable lease store, selected by `sandbox.ownership.type` (`memory` | `redis`) and resolved like `stream_bridge` (`factory.py`, lazy per-branch import, `redis` optional extra, `DEER_FLOW_SANDBOX_OWNERSHIP_REDIS_URL` env escape hatch; a set `DEER_FLOW_STREAM_BRIDGE_REDIS_URL` implies a multi-instance deployment and infers `redis`). `memory` is single-instance only and declares `supports_cross_process = False`. - **A lease answers "who reaps this container", not "who may use it".** That splits the interface in two: `take()` transfers ownership on the **acquire** path (a container is deterministic per user/thread, so consecutive turns legitimately land on different instances — a conditional claim there would strand the thread until the previous lease expired), while `claim()` succeeds only if the container is unowned or already ours and gates every **adopt/reap** path. `release()` never clears a peer's lease. - **A lease carries a state, and that is what makes the destroy window safe.** `own:` = responsible for this container; `del:` = tearing it down (`claim(..., for_destroy=True)`). `take()` is refused against a `del:` lease, so a container cannot be re-acquired between a destroy path's claim and its container stop. Without the two states an unconditional `take()` would silently overwrite the destroyer's claim and the peer's stop would land on a container the new owner had already handed to an agent — i.e. #4206 again. That pairing is what replaced the previous same-host `flock` guard, which is gone; Redis makes the scope genuinely multi-instance instead of same-host. A destroyer that dies mid-stop leaves a `del:` marker that lapses with the TTL. On the acquire path a refused take raises `SandboxBeingDestroyedError`: the reuse/reclaim paths drop the container and cold-start, and the discover path propagates (falling through to create would collide with the not-yet-removed container name). 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 a59379651..92dce7f23 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 @@ -84,7 +84,8 @@ class E2BSandboxProvider(SandboxProvider): self._sandboxes: dict[str, E2BSandbox] = {} # (user_id, thread_id) -> sandbox id for fast in-process lookup. self._thread_sandboxes: dict[tuple[str, str], str] = {} - # Per-(user,thread) lock to serialise acquire() against itself. + # Per-(user,thread) lock to serialise acquire() and release() state + # transitions without holding the provider-wide lock across remote IO. self._thread_locks: dict[tuple[str, str], threading.Lock] = {} # Warm pool: released sandboxes whose remote micro-VM is still alive. # ``OrderedDict`` maintains insertion / move_to_end order for LRU. @@ -1072,6 +1073,22 @@ class E2BSandboxProvider(SandboxProvider): the warm-pool entry stays valid for at least one ``idle_timeout`` window after release. """ + with self._lock: + thread_key = next( + (key for key, sid in self._thread_sandboxes.items() if sid == sandbox_id), + None, + ) + + if thread_key is None: + self._release_internal(sandbox_id) + return + + user_id, thread_id = thread_key + with self._get_thread_lock(thread_id, user_id): + self._release_internal(sandbox_id) + + def _release_internal(self, sandbox_id: str) -> None: + """Complete one release while the thread transition lock is held.""" sandbox: E2BSandbox | None = None seed: str | None = None diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py index 9c27fe04c..8b17172b3 100644 --- a/backend/tests/test_e2b_sandbox_provider.py +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -803,6 +803,54 @@ def test_release_healthy_sandbox_parks_in_warm_pool(monkeypatch, tmp_path): assert client.timeouts_set +def test_acquire_waits_for_same_thread_release_transition(monkeypatch): + provider = _make_provider() + _install_fake_sdk(monkeypatch, provider) + client = FakeClient(sandbox_id="sb-release-race") + sandbox = _make_sandbox(client) + provider._sandboxes[sandbox.id] = sandbox + provider._thread_sandboxes[("user-1", "thread-1")] = sandbox.id + + sync_started = threading.Event() + allow_sync_to_finish = threading.Event() + + def blocking_sync(*_args, **_kwargs) -> None: + sync_started.set() + assert allow_sync_to_finish.wait(timeout=2) + + monkeypatch.setattr(provider, "_sync_outputs_to_host", blocking_sync) + early_discovery = MagicMock(return_value="discovered-too-early") + early_create = MagicMock(return_value="created-too-early") + monkeypatch.setattr(provider, "_discover_remote_sandbox", early_discovery) + monkeypatch.setattr(provider, "_create_sandbox", early_create) + + release_thread = threading.Thread(target=provider.release, args=(sandbox.id,)) + acquired: list[str] = [] + acquire_done = threading.Event() + + def acquire() -> None: + acquired.append(provider.acquire("thread-1", user_id="user-1")) + acquire_done.set() + + acquire_thread = threading.Thread(target=acquire) + release_thread.start() + assert sync_started.wait(timeout=1) + acquire_thread.start() + + try: + assert not acquire_done.wait(timeout=0.1), "acquire must wait while release is syncing outputs" + finally: + allow_sync_to_finish.set() + release_thread.join(timeout=2) + acquire_thread.join(timeout=2) + + assert not release_thread.is_alive() + assert not acquire_thread.is_alive() + assert acquired == [sandbox.id] + early_discovery.assert_not_called() + early_create.assert_not_called() + + def test_release_skips_warm_pool_when_sync_reveals_dead_vm(monkeypatch, tmp_path): p = _make_provider() _setup_paths(monkeypatch, tmp_path)