fix(sandbox): widen boxlite/aio_sandbox tenant hash and verify identity on reclaim (#4171)

* fix(sandbox): prevent truncated tenant ID reuse

* fix(sandbox): handle late same-tenant box registration
This commit is contained in:
Daoyuan Li 2026-07-25 18:47:57 -07:00 committed by GitHub
parent 6e6c078595
commit 5d073991c2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 515 additions and 30 deletions

View File

@ -87,6 +87,19 @@ class SandboxBeingDestroyedError(RuntimeError):
self.sandbox_id = sandbox_id
class SandboxIdentityCollisionError(RuntimeError):
"""A deterministic ID is already tracked for a different user/thread."""
def __init__(
self,
sandbox_id: str,
stored_key: tuple[str, str] | None,
requested_key: tuple[str, str],
) -> None:
super().__init__(f"sandbox ID collision for {sandbox_id}: tracked identity is {stored_key!r}, requested identity is {requested_key!r}")
self.sandbox_id = sandbox_id
def _lock_file_exclusive(lock_file) -> None:
if fcntl is not None:
fcntl.flock(lock_file, fcntl.LOCK_EX)
@ -183,6 +196,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# Containers here can be reclaimed quickly (no cold-start) or destroyed
# when replicas capacity is exhausted.
self._warm_pool: dict[str, tuple[SandboxInfo, float]] = {}
self._active_sandbox_identity: dict[str, tuple[str, str] | None] = {}
self._warm_pool_identity: dict[str, tuple[str, str] | None] = {}
# sandbox_id -> when reconciliation first saw it running with no lease.
# Gates adoption behind a recovery grace (see _adoptable_after_grace).
self._unowned_since: dict[str, float] = {}
@ -700,6 +715,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
logger.debug("Deferring container %s during reconciliation: this instance is tearing it down", info.sandbox_id)
continue
self._warm_pool[info.sandbox_id] = (info, current_time)
self._warm_pool_identity[info.sandbox_id] = None
self._unowned_since.pop(info.sandbox_id, None)
adopted += 1
logger.info(f"Adopted container {info.sandbox_id} into warm pool (age: {age:.0f}s)")
@ -728,8 +744,42 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
Includes user_id so a previously-created default-bucket sandbox cannot be
reused for an auth/channel run that should mount a user-scoped bucket.
During a mixed-version rollout, older 8-character containers are not
reused under the new 16-character identity. They remain eligible for
normal orphan cleanup while the first new-version acquire cold-starts.
"""
return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:8]
return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16]
def _assert_active_identity_available_locked(
self,
sandbox_id: str,
requested_key: tuple[str, str],
) -> None:
"""Fail closed if an active truncated ID belongs to another identity."""
if sandbox_id not in self._sandboxes and sandbox_id not in self._sandbox_infos:
return
stored_key = self._active_sandbox_identity.get(sandbox_id)
if stored_key is None:
matching_keys = [key for key, mapped_id in self._thread_sandboxes.items() if mapped_id == sandbox_id]
if len(matching_keys) == 1:
stored_key = matching_keys[0]
if stored_key != requested_key:
raise SandboxIdentityCollisionError(sandbox_id, stored_key, requested_key)
def _assert_warm_identity_available_locked(
self,
sandbox_id: str,
requested_key: tuple[str, str],
) -> None:
"""Fail closed if a warm ID changed tenants during an acquire."""
if sandbox_id not in self._warm_pool:
return
# Startup-adopted entries have unknown identity until their first reclaim.
stored_key = self._warm_pool_identity.get(sandbox_id)
if stored_key is not None and stored_key != requested_key:
raise SandboxIdentityCollisionError(sandbox_id, stored_key, requested_key)
# ── Mount helpers ────────────────────────────────────────────────────
@ -1069,8 +1119,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
sandbox = self._sandboxes.pop(sandbox_id, None)
self._sandbox_infos.pop(sandbox_id, None)
self._active_sandbox_identity.pop(sandbox_id, None)
self._last_activity.pop(sandbox_id, None)
self._warm_pool.pop(sandbox_id, None)
self._warm_pool_identity.pop(sandbox_id, None)
self._acquire_epoch.pop(sandbox_id, None)
for key, mapped_id in list(self._thread_sandboxes.items()):
if mapped_id == sandbox_id:
@ -1310,6 +1362,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
with self._lock:
if sandbox_id not in self._warm_pool:
return None
self._assert_warm_identity_available_locked(sandbox_id, key)
if self._being_torn_down_locally(sandbox_id):
# The entry deliberately stays in `_warm_pool` for the whole stop
# (so a refused claim does not lose the container), so pool
@ -1350,13 +1403,16 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# that is already stopped.
logger.info("Warm-pool sandbox %s was claimed for teardown while publishing ownership; not reclaiming it", sandbox_id)
return None
self._assert_warm_identity_available_locked(sandbox_id, key)
warm_item = self._warm_pool.pop(sandbox_id, None)
if warm_item is None:
return None
self._warm_pool_identity.pop(sandbox_id, None)
info, _ = warm_item
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
self._sandboxes[sandbox_id] = sandbox
self._sandbox_infos[sandbox_id] = info
self._active_sandbox_identity[sandbox_id] = key
self._last_activity[sandbox_id] = time.time()
self._thread_sandboxes[key] = sandbox_id
@ -1385,6 +1441,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
prevent. The window is a peer's in-flight container stop, so the
thread's next turn discovers nothing and cold-starts cleanly.
"""
key = self._thread_key(thread_id, user_id)
with self._lock:
if self._being_torn_down_locally(info.sandbox_id):
# Discovery is the fall-through once the caches miss, so it is
@ -1392,9 +1449,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# only refuse this once the reaper's `del:` claim has landed;
# until then it succeeds against our own lease.
raise SandboxBeingDestroyedError(info.sandbox_id)
self._assert_active_identity_available_locked(info.sandbox_id, key)
self._assert_warm_identity_available_locked(info.sandbox_id, key)
sandbox = AioSandbox(id=info.sandbox_id, base_url=info.sandbox_url)
key = self._thread_key(thread_id, user_id)
# Ownership first, so a failure cannot leave a tracked-but-unowned sandbox.
# There is no container to roll back (we did not create it), but the
# host-side HTTP client constructed above is ours and must not leak —
@ -1408,6 +1466,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# trip. Do not install a client for a container that reaper
# has already committed to stopping.
raise SandboxBeingDestroyedError(info.sandbox_id)
self._assert_active_identity_available_locked(info.sandbox_id, key)
self._assert_warm_identity_available_locked(info.sandbox_id, key)
# Active and warm are exclusive states, and only this insert can
# violate that: a warm entry for the same id is stale the moment
# the id becomes active. Leaving it there gives the container two
@ -1416,11 +1476,17 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# agent is actively using while `_sandboxes` still hands out its
# client.
self._warm_pool.pop(info.sandbox_id, None)
self._warm_pool_identity.pop(info.sandbox_id, None)
self._sandboxes[info.sandbox_id] = sandbox
self._sandbox_infos[info.sandbox_id] = info
self._active_sandbox_identity[info.sandbox_id] = key
self._last_activity[info.sandbox_id] = time.time()
self._thread_sandboxes[key] = info.sandbox_id
except (OwnershipBackendError, SandboxBeingDestroyedError):
except (
OwnershipBackendError,
SandboxBeingDestroyedError,
SandboxIdentityCollisionError,
):
try:
sandbox.close()
except Exception as e:
@ -1433,6 +1499,14 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
def _register_created_sandbox(self, thread_id: str | None, sandbox_id: str, info: SandboxInfo, *, user_id: str | None = None) -> str:
"""Track a newly-created sandbox in the active maps."""
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
key = (
self._thread_key(
thread_id,
self._effective_acquire_user_id(user_id),
)
if thread_id
else None
)
# Ownership first. Unlike the discover path there IS something to roll
# back: we just started this container, and an unowned running container
# is exactly what a peer's reconciliation adopts. Leaking it would hand a
@ -1441,9 +1515,34 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# mid-stop leaves a teardown marker until its TTL lapses. Roll back on
# both, or the container we just started is leaked.
try:
if key is not None:
with self._lock:
self._assert_active_identity_available_locked(sandbox_id, key)
self._assert_warm_identity_available_locked(sandbox_id, key)
self._publish_ownership(sandbox_id)
except (OwnershipBackendError, SandboxBeingDestroyedError):
logger.error("Could not publish ownership for new sandbox %s; destroying it rather than leaking an unowned container", sandbox_id)
with self._lock:
if key is not None:
self._assert_active_identity_available_locked(sandbox_id, key)
self._assert_warm_identity_available_locked(sandbox_id, key)
# Same exclusivity rule as the discover path.
self._warm_pool.pop(sandbox_id, None)
self._warm_pool_identity.pop(sandbox_id, None)
self._sandboxes[sandbox_id] = sandbox
self._sandbox_infos[sandbox_id] = info
self._active_sandbox_identity[sandbox_id] = key
self._last_activity[sandbox_id] = time.time()
if key is not None:
self._thread_sandboxes[key] = sandbox_id
except (
OwnershipBackendError,
SandboxBeingDestroyedError,
SandboxIdentityCollisionError,
):
logger.error(
"Could not register new sandbox %s; destroying it rather than leaking an untracked container",
sandbox_id,
)
try:
sandbox.close()
except Exception as e:
@ -1451,18 +1550,13 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
try:
self._backend.destroy(info)
except Exception as e:
logger.error("Failed to destroy unowned sandbox %s after ownership failure: %s", sandbox_id, e)
logger.error(
"Failed to destroy sandbox %s after registration failure: %s",
sandbox_id,
e,
)
raise
with self._lock:
# Same exclusivity rule as the discover path.
self._warm_pool.pop(sandbox_id, None)
self._sandboxes[sandbox_id] = sandbox
self._sandbox_infos[sandbox_id] = info
self._last_activity[sandbox_id] = time.time()
if thread_id:
self._thread_sandboxes[self._thread_key(thread_id, self._effective_acquire_user_id(user_id))] = sandbox_id
logger.info(f"Created sandbox {sandbox_id} for thread {thread_id} at {info.sandbox_url}")
return sandbox_id
@ -1498,6 +1592,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
sandbox = self._sandboxes.pop(sandbox_id, None)
info = self._sandbox_infos.pop(sandbox_id, None)
self._active_sandbox_identity.pop(sandbox_id, None)
thread_keys_to_remove = [key for key, sid in self._thread_sandboxes.items() if sid == sandbox_id]
for key in thread_keys_to_remove:
del self._thread_sandboxes[key]
@ -1507,6 +1602,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
info, _ = self._warm_pool.pop(sandbox_id)
else:
self._warm_pool.pop(sandbox_id, None)
self._warm_pool_identity.pop(sandbox_id, None)
return sandbox, info, True
@ -1623,7 +1719,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# The pop stays deferred relative to the *stop* (a refused or failed
# stop keeps the entry), just no longer relative to the reservation.
with self._lock:
self._warm_pool.pop(sandbox_id, None)
current = self._warm_pool.get(sandbox_id)
if current is not None and current[0] is entry:
self._warm_pool.pop(sandbox_id, None)
self._warm_pool_identity.pop(sandbox_id, None)
finally:
self._finish_local_teardown(sandbox_id)
@ -1692,6 +1791,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# Deterministic ID for thread-specific, random for anonymous
sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
if thread_id:
key = self._thread_key(thread_id, user_id)
with self._lock:
self._assert_active_identity_available_locked(sandbox_id, key)
# ── Layer 1.5: Warm pool (container still running, no cold-start) ──
reclaimed_id = self._reclaim_warm_pool_sandbox(thread_id, sandbox_id, user_id=user_id)
@ -1715,6 +1818,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
# Deterministic ID for thread-specific, random for anonymous
sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
if thread_id:
key = self._thread_key(thread_id, user_id)
with self._lock:
self._assert_active_identity_available_locked(sandbox_id, key)
# ── Layer 1.5: Warm pool (container still running, no cold-start) ──
reclaimed_id = await asyncio.to_thread(self._reclaim_warm_pool_sandbox, thread_id, sandbox_id, user_id=user_id)
@ -1909,10 +2016,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
thread_keys_to_remove = [key for key, sid in self._thread_sandboxes.items() if sid == sandbox_id]
for key in thread_keys_to_remove:
del self._thread_sandboxes[key]
active_identity = self._active_sandbox_identity.pop(sandbox_id, None)
self._last_activity.pop(sandbox_id, None)
# Park in warm pool — container keeps running
if info and sandbox_id not in self._warm_pool:
self._warm_pool[sandbox_id] = (info, time.time())
self._warm_pool_identity[sandbox_id] = thread_keys_to_remove[0] if thread_keys_to_remove else active_identity
if sandbox is not None:
# Defense-in-depth: close() already swallows its own errors; this
@ -2015,6 +2124,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
sandbox_ids = list(self._sandboxes.keys())
warm_items = list(self._warm_pool.items())
self._warm_pool.clear()
self._warm_pool_identity.clear()
self._stop_idle_checker()
# Stop renewing before destroying: the destroy paths claim ownership

View File

@ -41,6 +41,7 @@ T = TypeVar("T")
DEFAULT_IMAGE = "python:3.12-slim"
_BOX_NAME_PREFIX = "deer-flow-boxlite-"
_NO_ACTIVE_IDENTITY = object()
# DeerFlow's virtual prefixes, materialised on the box rootfs at start so the
# Sandbox file APIs (which address /mnt/user-data/...) resolve natively.
_VIRTUAL_DIRS = (
@ -51,6 +52,18 @@ _VIRTUAL_DIRS = (
)
class SandboxIdentityCollisionError(RuntimeError):
"""A deterministic ID is already tracked for a different user/thread."""
def __init__(
self,
sandbox_id: str,
stored_key: tuple[str, str] | None,
requested_key: tuple[str, str],
) -> None:
super().__init__(f"Sandbox ID collision for {sandbox_id}: active box belongs to {stored_key!r}, not {requested_key!r}")
def _import_simplebox() -> type[SimpleBox]:
"""Import BoxLite's async ``SimpleBox`` lazily.
@ -166,8 +179,13 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
Includes user_id so a box created for one user's bucket cannot be
reclaimed by another user's thread with the same thread_id.
The 8-to-16-character change intentionally causes one cold start during
a mixed-version rollout. Older 8-character boxes remain in the warm pool
until the normal orphan cleanup removes them; they are never reused
under a new 16-character identity.
"""
return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:8]
return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16]
# ── Provider ────────────────────────────────────────────────────────
@ -176,6 +194,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
self._boxes: dict[str, BoxliteBox] = {}
self._thread_boxes: dict[tuple[str, str], str] = {}
self._warm_pool: dict[str, tuple[BoxliteBox, float]] = {}
self._active_box_identity: dict[str, tuple[str, str] | None] = {}
self._warm_pool_identity: dict[str, tuple[str, str] | None] = {}
self._skip_health_check_warm_ids: set[str] = set()
self._acquire_locks: dict[str, threading.Lock] = {}
self._idle_checker_stop = threading.Event()
@ -245,7 +265,9 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
def _destroy_warm_entry(self, sandbox_id: str, entry: BoxliteBox, *, reason: str) -> None:
"""Close a removed warm-pool entry and log with context."""
with self._lock:
self._skip_health_check_warm_ids.discard(sandbox_id)
if sandbox_id not in self._warm_pool:
self._skip_health_check_warm_ids.discard(sandbox_id)
self._warm_pool_identity.pop(sandbox_id, None)
try:
entry.close()
if reason == "idle_timeout":
@ -268,6 +290,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
with self._lock:
active_box = self._boxes.pop(sandbox_id, None)
warm_entry = self._warm_pool.pop(sandbox_id, None)
self._active_box_identity.pop(sandbox_id, None)
self._warm_pool_identity.pop(sandbox_id, None)
self._skip_health_check_warm_ids.discard(sandbox_id)
for key in [k for k, sid in self._thread_boxes.items() if sid == sandbox_id]:
self._thread_boxes.pop(key, None)
@ -335,6 +359,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
box_runtime.stop()
continue
self._warm_pool[sandbox_id] = (wrapped, now)
self._warm_pool_identity[sandbox_id] = None
adopted += 1
logger.info("Adopted existing BoxLite box %s (%s) into warm pool", sandbox_id, name)
@ -348,6 +373,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
box = self._create_box(sandbox_id)
with self._lock:
self._boxes[box.id] = box
self._active_box_identity[box.id] = None
return box.id
key = self._thread_key(thread_id, user_id)
@ -358,17 +384,42 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
existing = self._thread_boxes.get(key)
if existing is not None and existing in self._boxes:
return existing
if sandbox_id in self._boxes:
owner = self._active_box_identity.get(sandbox_id)
if owner != key:
raise SandboxIdentityCollisionError(
sandbox_id,
owner,
key,
)
self._thread_boxes[key] = sandbox_id
return sandbox_id
reclaimed = self._reclaim_warm_pool(sandbox_id)
reclaimed = self._reclaim_warm_pool(sandbox_id, key)
if reclaimed is not None:
with self._lock:
self._thread_boxes[key] = reclaimed
return reclaimed
box = self._create_box(sandbox_id)
conflict: tuple[str, str] | None | object = _NO_ACTIVE_IDENTITY
with self._lock:
self._boxes[box.id] = box
self._thread_boxes[key] = box.id
if box.id in self._boxes:
conflict = self._active_box_identity.get(box.id)
if conflict == key:
self._thread_boxes[key] = box.id
else:
self._boxes[box.id] = box
self._active_box_identity[box.id] = key
self._thread_boxes[key] = box.id
if conflict is not _NO_ACTIVE_IDENTITY:
box.close()
if conflict != key:
raise SandboxIdentityCollisionError(
sandbox_id,
conflict,
key,
)
return box.id
def _create_box(self, sandbox_id: str) -> BoxliteBox:
@ -410,15 +461,19 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
close_box: BoxliteBox | None = None
with self._lock:
box = self._boxes.pop(sandbox_id, None)
for key in [k for k, sid in self._thread_boxes.items() if sid == sandbox_id]:
released_keys = [k for k, sid in self._thread_boxes.items() if sid == sandbox_id]
for key in released_keys:
self._thread_boxes.pop(key, None)
active_identity = self._active_box_identity.pop(sandbox_id, None)
if box is None:
return
if self._shutdown_called:
close_box = box
self._skip_health_check_warm_ids.discard(sandbox_id)
self._warm_pool_identity.pop(sandbox_id, None)
else:
self._warm_pool[sandbox_id] = (box, time.time())
self._warm_pool_identity[sandbox_id] = released_keys[0] if released_keys else active_identity
self._skip_health_check_warm_ids.add(sandbox_id)
if close_box is not None:
@ -427,7 +482,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
else:
logger.info("Released sandbox %s to warm pool (VM still running)", sandbox_id)
def _reclaim_warm_pool(self, sandbox_id: str) -> str | None:
def _reclaim_warm_pool(self, sandbox_id: str, expected_key: tuple[str, str]) -> str | None:
"""Try to reclaim a warm-pool box by sandbox_id.
Returns sandbox_id on success, None if not found or dead.
@ -440,6 +495,14 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
with self._lock:
if sandbox_id not in self._warm_pool:
return None
# Startup-adopted entries have unknown identity until their first reclaim.
stored_key = self._warm_pool_identity.get(sandbox_id)
if stored_key is not None and stored_key != expected_key:
raise SandboxIdentityCollisionError(
sandbox_id,
stored_key,
expected_key,
)
box, released_at = self._warm_pool[sandbox_id]
skip_eligible = sandbox_id in self._skip_health_check_warm_ids
@ -449,10 +512,21 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
# health-check round trip, but never return an adapter that this
# process already knows is closed.
with self._lock:
current = self._warm_pool.get(sandbox_id)
stored_key = self._warm_pool_identity.get(sandbox_id)
if stored_key is not None and stored_key != expected_key:
raise SandboxIdentityCollisionError(
sandbox_id,
stored_key,
expected_key,
)
if current is None or current[0] is not box:
return None
warm_entry = self._warm_pool.pop(sandbox_id, None)
if warm_entry is None:
return None # Raced with another thread
self._skip_health_check_warm_ids.discard(sandbox_id)
self._warm_pool_identity.pop(sandbox_id, None)
box, _ = warm_entry
if box.is_closed:
logger.warning("Warm-pool box %s was closed before skipped health check reclaim", sandbox_id)
@ -460,6 +534,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
else:
close_box = None
self._boxes[sandbox_id] = box
self._active_box_identity[sandbox_id] = expected_key
if close_box is not None:
close_box.close()
return None
@ -476,26 +551,60 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
if "ok" not in result:
logger.warning("Warm pool box %s health check failed: %s", sandbox_id, result)
with self._lock:
warm_entry = self._warm_pool.pop(sandbox_id, None)
current = self._warm_pool.get(sandbox_id)
stored_key = self._warm_pool_identity.get(sandbox_id)
if current is not None and current[0] is not box and stored_key is not None and stored_key != expected_key:
raise SandboxIdentityCollisionError(
sandbox_id,
stored_key,
expected_key,
)
warm_entry = self._warm_pool.pop(sandbox_id, None) if current is not None and current[0] is box else None
if warm_entry is not None:
self._warm_pool_identity.pop(sandbox_id, None)
if warm_entry is not None:
self._destroy_warm_entry(sandbox_id, warm_entry[0], reason="health_check_failed")
return None
except SandboxIdentityCollisionError:
raise
except Exception as e:
logger.warning("Warm pool box %s health check error: %s", sandbox_id, e)
with self._lock:
warm_entry = self._warm_pool.pop(sandbox_id, None)
current = self._warm_pool.get(sandbox_id)
stored_key = self._warm_pool_identity.get(sandbox_id)
if current is not None and current[0] is not box and stored_key is not None and stored_key != expected_key:
raise SandboxIdentityCollisionError(
sandbox_id,
stored_key,
expected_key,
)
warm_entry = self._warm_pool.pop(sandbox_id, None) if current is not None and current[0] is box else None
if warm_entry is not None:
self._warm_pool_identity.pop(sandbox_id, None)
if warm_entry is not None:
self._destroy_warm_entry(sandbox_id, warm_entry[0], reason="health_check_failed")
return None
# Promote from warm pool to active
with self._lock:
current = self._warm_pool.get(sandbox_id)
stored_key = self._warm_pool_identity.get(sandbox_id)
if stored_key is not None and stored_key != expected_key:
raise SandboxIdentityCollisionError(
sandbox_id,
stored_key,
expected_key,
)
if current is None or current[0] is not box:
return None
warm_entry = self._warm_pool.pop(sandbox_id, None)
if warm_entry is None:
return None # Raced with another thread
self._skip_health_check_warm_ids.discard(sandbox_id)
self._warm_pool_identity.pop(sandbox_id, None)
box, _ = warm_entry
self._boxes[sandbox_id] = box
self._active_box_identity[sandbox_id] = expected_key
logger.info("Reclaimed warm-pool box %s", sandbox_id)
return sandbox_id
@ -513,8 +622,10 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
now = time.time()
for sandbox_id, box in self._boxes.items():
self._warm_pool.setdefault(sandbox_id, (box, now))
self._warm_pool_identity.setdefault(sandbox_id, self._active_box_identity.get(sandbox_id))
self._skip_health_check_warm_ids.discard(sandbox_id)
self._boxes.clear()
self._active_box_identity.clear()
self._thread_boxes.clear()
self._acquire_locks.clear()
@ -531,6 +642,8 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
warm = [box for box, _ in self._warm_pool.values()]
self._boxes.clear()
self._warm_pool.clear()
self._active_box_identity.clear()
self._warm_pool_identity.clear()
self._thread_boxes.clear()
self._acquire_locks.clear()
self._skip_health_check_warm_ids.clear()

View File

@ -82,6 +82,8 @@ def _make_provider(tmp_path: Path):
provider._thread_locks = {}
provider._last_activity = {}
provider._warm_pool = {}
provider._active_sandbox_identity = {}
provider._warm_pool_identity = {}
provider._local_teardown = set()
provider._acquire_epoch = {}
provider._acquire_epoch_counter = 0

View File

@ -75,6 +75,7 @@ def _make_provider_with_active_sandbox(tmp_path: Path, sandbox_id: str):
provider = AioSandboxProvider.__new__(AioSandboxProvider)
provider._lock = threading.Lock()
provider._sandboxes = {sandbox_id: MagicMock()}
provider._active_sandbox_identity = {sandbox_id: ("default", "thread-1")}
provider._sandbox_infos = {
sandbox_id: SandboxInfo(
sandbox_id=sandbox_id,
@ -87,6 +88,7 @@ def _make_provider_with_active_sandbox(tmp_path: Path, sandbox_id: str):
provider._thread_locks = {}
provider._last_activity = {sandbox_id: 1.0}
provider._warm_pool = {}
provider._warm_pool_identity = {}
provider._unowned_since = {}
provider._local_teardown = set()
provider._acquire_epoch = {}

View File

@ -1,6 +1,8 @@
"""Tests for AioSandboxProvider mount helpers."""
import asyncio
import contextlib
import hashlib
import importlib
import stat
from types import SimpleNamespace
@ -11,6 +13,11 @@ import pytest
from deerflow.config.paths import Paths, join_host_path
from deerflow.runtime.user_context import reset_current_user, set_current_user
_LEGACY_COLLIDING_IDENTITIES = (
("user-9721", "thread-9721"),
("user-94361", "thread-94361"),
)
# ── ensure_thread_dirs ───────────────────────────────────────────────────────
@ -62,6 +69,8 @@ def _make_provider(tmp_path):
provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
provider._config = {"idle_timeout": 600, "replicas": 3}
provider._sandboxes = {}
provider._active_sandbox_identity = {}
provider._warm_pool_identity = {}
provider._local_teardown = set()
provider._acquire_epoch = {}
provider._acquire_epoch_counter = 0
@ -883,3 +892,101 @@ def test_create_sandbox_evicts_oldest_warm_replica_via_shared_lifecycle(tmp_path
assert "warm-oldest" not in provider._warm_pool
assert provider._warm_pool == {"warm-newest": (newest_info, 20.0)}
assert provider._sandbox_infos["created"] is created_info
def _make_tenant_isolation_provider(tmp_path, monkeypatch):
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
provider = _make_provider(tmp_path)
provider._lock = aio_mod.threading.Lock()
provider._sandboxes = {}
provider._sandbox_infos = {}
provider._thread_sandboxes = {}
provider._thread_locks = {}
provider._last_activity = {}
provider._warm_pool = {}
provider._active_sandbox_identity = {}
provider._warm_pool_identity = {}
provider._shutdown_called = False
provider._config = {"replicas": 3, "idle_timeout": 0}
create_calls = []
def _create(thread_id, sandbox_id, **kwargs):
create_calls.append((thread_id, sandbox_id, kwargs.get("user_id")))
return aio_mod.SandboxInfo(
sandbox_id=sandbox_id,
sandbox_url=f"http://sandbox-{len(create_calls)}.local",
container_name=f"deer-flow-sandbox-{sandbox_id}",
)
provider._backend = SimpleNamespace(
create=MagicMock(side_effect=_create),
destroy=MagicMock(),
discover=MagicMock(return_value=None),
is_alive=MagicMock(return_value=True),
list_running=MagicMock(return_value=[]),
)
provider._claim_ownership = MagicMock(return_value=True)
provider._held_teardown_lease = lambda _sandbox_id: contextlib.nullcontext()
monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
monkeypatch.setattr(
aio_mod.AioSandboxProvider,
"_get_extra_mounts",
lambda self, thread_id, *, user_id=None: [],
)
monkeypatch.setattr(
aio_mod,
"wait_for_sandbox_ready",
lambda _url, timeout=60: True,
)
return provider, create_calls, aio_mod
def test_aio_wider_id_separates_known_legacy_collision():
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
identity_a, identity_b = _LEGACY_COLLIDING_IDENTITIES
user_a, thread_a = identity_a
user_b, thread_b = identity_b
old_a = hashlib.sha256(f"{user_a}:{thread_a}".encode()).hexdigest()[:8]
old_b = hashlib.sha256(f"{user_b}:{thread_b}".encode()).hexdigest()[:8]
assert old_a == old_b
assert aio_mod.AioSandboxProvider._deterministic_sandbox_id(
thread_a,
user_a,
) != aio_mod.AioSandboxProvider._deterministic_sandbox_id(
thread_b,
user_b,
)
def test_aio_forced_collision_never_overwrites_active_tenant(
tmp_path,
monkeypatch,
):
provider, create_calls, aio_mod = _make_tenant_isolation_provider(
tmp_path,
monkeypatch,
)
monkeypatch.setattr(
aio_mod.AioSandboxProvider,
"_deterministic_sandbox_id",
staticmethod(lambda thread_id, user_id: "deadbeefdeadbeef"),
)
sandbox_id = provider.acquire("thread-a", user_id="user-a")
info_a = provider._sandbox_infos[sandbox_id]
provider.release(sandbox_id)
assert sandbox_id in provider._warm_pool
with pytest.raises(aio_mod.SandboxIdentityCollisionError):
provider.acquire("thread-b", user_id="user-b")
assert provider._warm_pool[sandbox_id][0] is info_a
provider._backend.destroy.assert_not_called()
assert len(create_calls) == 1
assert provider.acquire("thread-a", user_id="user-a") == sandbox_id
assert provider._sandbox_infos[sandbox_id] is info_a

View File

@ -7,6 +7,7 @@ which need a live box.
from __future__ import annotations
import asyncio
import hashlib
import logging
import sys
import threading
@ -18,6 +19,11 @@ import pytest
from deerflow.community.boxlite.box import BoxliteBox
from deerflow.community.boxlite.provider import BoxliteProvider, _import_simplebox
_LEGACY_COLLIDING_IDENTITIES = (
("user-9721", "thread-9721"),
("user-94361", "thread-94361"),
)
# ── Fake BoxLite SDK ──────────────────────────────────────────────────
@ -262,7 +268,7 @@ def test_sandbox_id_deterministic(monkeypatch):
id1 = provider._sandbox_id("thread-1", "user-a")
id2 = provider._sandbox_id("thread-1", "user-a")
assert id1 == id2
assert len(id1) == 8
assert len(id1) == 16
def test_sandbox_id_different_users(monkeypatch):
@ -460,7 +466,7 @@ def test_explicit_recent_reclaim_skip_avoids_health_check(monkeypatch):
monkeypatch.setattr(box, "execute_command", _fail_if_called)
reclaimed = provider._reclaim_warm_pool(sid)
reclaimed = provider._reclaim_warm_pool(sid, ("u1", "thread-1"))
assert reclaimed == sid
assert sid in provider._boxes
assert sid not in provider._warm_pool
@ -494,7 +500,7 @@ def test_recent_reclaim_validates_by_default(monkeypatch):
monkeypatch.setattr(box, "execute_command", _record_health_check)
reclaimed = provider._reclaim_warm_pool(sid)
reclaimed = provider._reclaim_warm_pool(sid, ("u1", "thread-1"))
assert reclaimed == sid
assert calls == 1
assert sid in provider._boxes
@ -526,7 +532,7 @@ def test_default_recent_reclaim_drops_dead_warm_box(monkeypatch):
monkeypatch.setattr(box, "execute_command", _dead_health_check)
reclaimed = provider._reclaim_warm_pool(sid)
reclaimed = provider._reclaim_warm_pool(sid, ("u1", "thread-1"))
assert reclaimed is None
assert sid not in provider._boxes
assert sid not in provider._warm_pool
@ -592,7 +598,10 @@ def test_adopted_warm_pool_box_still_health_checks(monkeypatch):
monkeypatch.setattr(adopted, "execute_command", _record_health_check)
reclaimed = provider._reclaim_warm_pool("adopted")
reclaimed = provider._reclaim_warm_pool(
"adopted",
("some-user", "some-thread"),
)
assert reclaimed == "adopted"
assert calls == 1
assert "adopted" in provider._boxes
@ -1044,3 +1053,143 @@ def test_shutdown_stops_idle_reaper_and_destroys_all_boxes(monkeypatch):
assert len(provider._boxes) == 0
assert len(provider._warm_pool) == 0
assert len(provider._thread_boxes) == 0
def test_wider_id_separates_known_legacy_collision():
identity_a, identity_b = _LEGACY_COLLIDING_IDENTITIES
user_a, thread_a = identity_a
user_b, thread_b = identity_b
old_a = hashlib.sha256(f"{user_a}:{thread_a}".encode()).hexdigest()[:8]
old_b = hashlib.sha256(f"{user_b}:{thread_b}".encode()).hexdigest()[:8]
assert old_a == old_b
assert BoxliteProvider._sandbox_id(thread_a, user_a) != BoxliteProvider._sandbox_id(
thread_b,
user_b,
)
def test_forced_collision_never_overwrites_active_tenant(monkeypatch):
monkeypatch.setattr(
"deerflow.community.boxlite.provider.get_app_config",
lambda: _stub_config(),
)
monkeypatch.setattr(
"deerflow.community.boxlite.provider._import_simplebox",
lambda: _FakeBox,
)
monkeypatch.setattr(
BoxliteProvider,
"_sandbox_id",
staticmethod(lambda thread_id, user_id: "deadbeefdeadbeef"),
)
provider = BoxliteProvider()
provider._loop.run = _fake_run
sandbox_id = provider.acquire("thread-a", user_id="user-a")
box_a = provider.get(sandbox_id)
provider.release(sandbox_id)
assert box_a is not None
assert sandbox_id in provider._warm_pool
with pytest.raises(RuntimeError, match="Sandbox ID collision"):
provider.acquire("thread-b", user_id="user-b")
assert provider._warm_pool[sandbox_id][0] is box_a
assert not box_a.is_closed
assert provider.acquire("thread-a", user_id="user-a") == sandbox_id
assert provider.get(sandbox_id) is box_a
provider.shutdown()
def test_late_same_tenant_collision_reuses_active_box(monkeypatch):
monkeypatch.setattr(
"deerflow.community.boxlite.provider.get_app_config",
lambda: _stub_config(),
)
provider = BoxliteProvider()
sandbox_id = "deadbeefdeadbeef"
key = ("user-a", "thread-a")
active = BoxliteBox(
sandbox_id,
_FakeBox(name=f"deer-flow-boxlite-{sandbox_id}"),
_fake_run,
default_env={},
)
duplicate = BoxliteBox(
sandbox_id,
_FakeBox(name=f"deer-flow-boxlite-{sandbox_id}"),
_fake_run,
default_env={},
)
monkeypatch.setattr(
BoxliteProvider,
"_sandbox_id",
staticmethod(lambda thread_id, user_id: sandbox_id),
)
def _create_box_with_late_registration(_sandbox_id):
with provider._lock:
provider._boxes[sandbox_id] = active
provider._active_box_identity[sandbox_id] = key
return duplicate
monkeypatch.setattr(provider, "_create_box", _create_box_with_late_registration)
assert provider.acquire("thread-a", user_id="user-a") == sandbox_id
assert duplicate.is_closed
assert provider.get(sandbox_id) is active
assert provider._thread_boxes[key] == sandbox_id
provider.shutdown()
def test_failed_health_check_does_not_remove_swapped_warm_entry(monkeypatch):
monkeypatch.setattr(
"deerflow.community.boxlite.provider.get_app_config",
lambda: _stub_config(),
)
provider = BoxliteProvider()
stale = BoxliteBox(
"shared-id",
_FakeBox(name="deer-flow-boxlite-shared-id"),
_fake_run,
default_env={},
)
replacement = BoxliteBox(
"shared-id",
_FakeBox(name="deer-flow-boxlite-shared-id"),
_fake_run,
default_env={},
)
provider._warm_pool["shared-id"] = (stale, time.time())
provider._warm_pool_identity["shared-id"] = ("user-a", "thread-a")
def _swap_during_health_check(*args, **kwargs):
with provider._lock:
provider._warm_pool["shared-id"] = (replacement, time.time())
provider._warm_pool_identity["shared-id"] = (
"user-b",
"thread-b",
)
return "not healthy"
monkeypatch.setattr(stale, "execute_command", _swap_during_health_check)
with pytest.raises(RuntimeError, match="Sandbox ID collision"):
provider._reclaim_warm_pool(
"shared-id",
("user-a", "thread-a"),
)
assert provider._warm_pool["shared-id"][0] is replacement
assert provider._warm_pool_identity["shared-id"] == (
"user-b",
"thread-b",
)
assert not replacement.is_closed
provider.shutdown()

View File

@ -431,6 +431,8 @@ def _make_provider_for_reconciliation(tmp_path=None, *, worker_id: str = "worker
provider._thread_locks = {}
provider._last_activity = {}
provider._warm_pool = {}
provider._active_sandbox_identity = {}
provider._warm_pool_identity = {}
provider._unowned_since = {}
provider._local_teardown = set()
provider._acquire_epoch = {}