fix(sandbox): serialize E2B release transitions (#4355)

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).

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
 

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)