fix(sandbox): release local E2B capacity when reconnect confirms the VM is gone (#6547)

When Sandbox.connect() fails with a trusted 'sandbox not found' error,
_reconnect_live_client() used to release the deployment-level slot but
still re-raise. _evict_oldest_warm() then treated that confirmed-gone
result like an unknown-state failure: it kept the eviction tombstone and
never called _end_transition_locked(), so the local transitioning slot
leaked on every retry against an already-reaped VM. Once the leaked slots
reach the configured replicas, every acquire rejects or times out with
SANDBOX_CAPACITY_EXCEEDED while active=warm=reserved=0.

Return None from _reconnect_live_client() for trusted gone errors so all
callers (eviction, warm reclaim, discovery, reconciliation) take their
existing already-gone cleanup paths, keeping local bookkeeping consistent
with the deployment-level release on the same signal. Unknown-state
errors (network, auth) still retain the slot for a later retry.

Also report the transitioning count in SandboxCapacityExceededError
details and the wait-timeout message so hidden slot usage is visible
without a restart.

Fixes #6540

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py

@@ -990,6 +990,7 @@ class E2BSandboxProvider(SandboxProvider):
                 active=len(self._sandboxes),
                 warm=len(self._warm_pool),
                 reserved=self._reserved_slots,
+                transitioning=self._transitioning_slots,
                 replicas=int(self._config["replicas"]),
                 reason=reason,
             )
@@ -1166,6 +1167,7 @@ class E2BSandboxProvider(SandboxProvider):
                             active=len(self._sandboxes),
                             warm=len(self._warm_pool),
                             reserved=self._reserved_slots,
+                            transitioning=self._transitioning_slots,
                             replicas=int(self._config["replicas"]),
                         )
 
@@ -1175,16 +1177,19 @@ class E2BSandboxProvider(SandboxProvider):
                             active=len(self._sandboxes),
                             warm=len(self._warm_pool),
                             reserved=self._reserved_slots,
+                            transitioning=self._transitioning_slots,
                             replicas=int(self._config["replicas"]),
                         )
 
                     remaining = deadline - time.monotonic()
                     if remaining <= 0:
+                        counts = f"replicas={self._config['replicas']}, active={len(self._sandboxes)}, warm={len(self._warm_pool)}, reserved={self._reserved_slots}, transitioning={self._transitioning_slots}"
                         raise SandboxCapacityExceededError(
-                            f"Timed out after {timeout}s waiting for a sandbox capacity slot (replicas={self._config['replicas']}, active={len(self._sandboxes)}, warm={len(self._warm_pool)}, reserved={self._reserved_slots})",
+                            f"Timed out after {timeout}s waiting for a sandbox capacity slot ({counts})",
                             active=len(self._sandboxes),
                             warm=len(self._warm_pool),
                             reserved=self._reserved_slots,
+                            transitioning=self._transitioning_slots,
                             replicas=int(self._config["replicas"]),
                         )
                     self._capacity_cond.wait(timeout=min(remaining, 1.0))
@@ -1917,13 +1922,18 @@ class E2BSandboxProvider(SandboxProvider):
 
         ``Sandbox.connect`` may succeed even after the E2B control plane has
         reaped the VM. Closing that host-side client before returning ``None``
-        keeps both acquire paths from leaking a connection.
+        keeps both acquire paths from leaking a connection.  ``connect`` may
+        equally *fail* with a trusted "sandbox not found" error once the VM
+        is reaped; that confirmed-gone result also returns ``None`` so every
+        caller takes its already-gone cleanup path instead of retaining local
+        capacity for a kill that can never succeed.
         """
         try:
             client = self._reconnect_client(sandbox_cls, sandbox_id)
         except Exception as error:
             if _is_sandbox_gone_error(error):
                 self._release_deployment_sandbox(sandbox_id)
+                return None
             raise
         if self._client_alive(client):
             return client

backend/packages/harness/deerflow/sandbox/exceptions.py

@@ -87,6 +87,7 @@ class SandboxCapacityExceededError(SandboxError):
         active: int = 0,
         warm: int = 0,
         reserved: int = 0,
+        transitioning: int = 0,
         replicas: int = 0,
         retry_after_seconds: float = 5.0,
         reason: str = "capacity",
@@ -104,10 +105,13 @@ class SandboxCapacityExceededError(SandboxError):
             details["warm"] = warm
         if reserved:
             details["reserved"] = reserved
+        if transitioning:
+            details["transitioning"] = transitioning
         super().__init__(message, details)
         self.active = active
         self.warm = warm
         self.reserved = reserved
+        self.transitioning = transitioning
         self.replicas = replicas
         self.retry_after_seconds = retry_after_seconds
         self.reason = reason

backend/tests/test_e2b_sandbox_provider.py

@@ -2147,6 +2147,30 @@ def test_reclaim_warm_pool_sandbox_handles_reconnect_exception(monkeypatch):
     assert "sb-broken" not in p._warm_pool
 
 
+def test_reclaim_warm_pool_sandbox_with_missing_sandbox_frees_slot(monkeypatch):
+    """Reclaiming a confirmed-missing warm sandbox frees the transition slot (#6540).
+
+    ``Sandbox.connect`` raising a trusted not-found error is a confirmed-gone
+    result: it must not leave an eviction tombstone behind that pins the
+    transitioning slot forever.
+    """
+    p = _make_provider()
+    fake_cls = _install_fake_sdk(monkeypatch, p)
+
+    def missing_sandbox(_sid, **_kw):
+        raise RuntimeError(FakeCommandsAPI.NOT_FOUND_MSG)
+
+    fake_cls.connect_factory = missing_sandbox
+    seed = p._stable_seed("t1", "u1")
+    p._warm_pool["sb-gone"] = (seed, 12345.0)
+
+    assert p._reclaim_warm_pool_sandbox("t1", user_id="u1") is None
+    assert "sb-gone" not in p._warm_pool
+    assert p._eviction_tombstones == set()
+    assert p._remote_ops_in_progress == set()
+    assert p._transitioning_slots == 0
+
+
 def test_acquire_discards_warm_sandbox_when_bootstrap_fails(monkeypatch):
     p = _make_provider()
     fake_cls = _install_fake_sdk(monkeypatch, p)
@@ -4658,6 +4682,59 @@ def test_capacity_retries_tombstone_until_warm_vm_is_destroyed(monkeypatch):
     assert p._transitioning_slots == 0
 
 
+def test_eviction_of_missing_sandbox_releases_local_capacity(monkeypatch):
+    """A control-plane confirmed missing warm sandbox frees its local slot (#6540).
+
+    The eviction sees a trusted not-found error from ``Sandbox.connect``, so
+    the transitioning slot is released instead of being retained behind a
+    tombstone that can never clear; the next acquire must succeed.
+    """
+    p = _make_provider(replicas=1, overflow_policy="reject")
+    fake_cls = _install_fake_sdk(monkeypatch, p)
+
+    sid = p.acquire("t1", user_id="u1")
+    p.release(sid)
+
+    def missing_sandbox(_sid, **_kw):
+        raise RuntimeError(FakeCommandsAPI.NOT_FOUND_MSG)
+
+    fake_cls.connect_factory = missing_sandbox
+
+    sid2 = p.acquire("t2", user_id="u2")
+
+    assert sid2 != sid
+    assert p._eviction_tombstones == set()
+    assert p._evictions_in_progress == set()
+    assert p._transitioning_slots == 0
+
+
+def test_tombstone_retry_of_missing_sandbox_frees_retained_slot(monkeypatch):
+    """Retrying a retained tombstone against a confirmed-missing VM frees the slot exactly once."""
+    p = _make_provider(replicas=1, overflow_policy="reject")
+    fake_cls = _install_fake_sdk(monkeypatch, p)
+
+    sid = p.acquire("t1", user_id="u1")
+    p.release(sid)
+    fake_cls.connect_factory = lambda _sid, **_kw: (_ for _ in ()).throw(RuntimeError("network down"))
+
+    with pytest.raises(SandboxCapacityExceededError) as excinfo:
+        p.acquire("t2", user_id="u2")
+    # A state-unknown failure keeps the slot retained, and the error reports it.
+    assert excinfo.value.transitioning == 1
+    assert p._eviction_tombstones == {sid}
+    assert p._transitioning_slots == 1
+
+    fake_cls.connect_factory = lambda _sid, **_kw: (_ for _ in ()).throw(RuntimeError(FakeCommandsAPI.NOT_FOUND_MSG))
+    assert p._evict_oldest_warm() == sid
+
+    assert p._eviction_tombstones == set()
+    assert p._evictions_in_progress == set()
+    assert p._transitioning_slots == 0
+
+    sid2 = p.acquire("t2", user_id="u2")
+    assert sid2 != sid
+
+
 def test_tombstone_eviction_has_one_retry_owner(monkeypatch):
     """Only one thread can retry a tombstone at a time."""
     p = _make_provider()
@@ -4723,6 +4800,44 @@ def test_shutdown_during_initial_eviction_reconnect_failure_tracks_vm(monkeypatc
     assert p._evictions_in_progress == set()
 
 
+def test_shutdown_during_gone_sandbox_eviction_releases_slot_once(monkeypatch):
+    """Shutdown racing a confirmed-missing eviction leaves no retained slot.
+
+    The eviction's own cleanup must stay idempotent when shutdown has already
+    cleared the bookkeeping sets: no double slot release, no resurrected
+    tombstone.
+    """
+    p = _make_provider()
+    fake_cls = _install_fake_sdk(monkeypatch, p)
+    p._warm_pool["warm-1"] = ("seed", time.time())
+
+    reconnect_started = threading.Event()
+    allow_failure = threading.Event()
+    calls = 0
+
+    def reconnect(_sid, **_kw):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            reconnect_started.set()
+            assert allow_failure.wait(timeout=2)
+        raise RuntimeError(FakeCommandsAPI.NOT_FOUND_MSG)
+
+    fake_cls.connect_factory = reconnect
+    eviction = threading.Thread(target=p._evict_oldest_warm)
+    eviction.start()
+    assert reconnect_started.wait(timeout=1)
+
+    p.shutdown()
+    allow_failure.set()
+    eviction.join(timeout=5)
+
+    assert p._warm_pool == {}
+    assert p._eviction_tombstones == set()
+    assert p._evictions_in_progress == set()
+    assert p._transitioning_slots == 0
+
+
 def test_capacity_reset_uses_destructive_shutdown_semantics(monkeypatch):
     """reset() destroys tracked E2B resources and ends the provider."""
     p = _make_provider(replicas=1, overflow_policy="wait", acquire_timeout=30)