diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index f3f7df088..39f75d212 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -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 diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py index fa0981be4..e49fecdb1 100644 --- a/backend/packages/harness/deerflow/community/boxlite/provider.py +++ b/backend/packages/harness/deerflow/community/boxlite/provider.py @@ -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() diff --git a/backend/tests/blocking_io/test_aio_sandbox_get.py b/backend/tests/blocking_io/test_aio_sandbox_get.py index 4f6e11761..5ad2f1313 100644 --- a/backend/tests/blocking_io/test_aio_sandbox_get.py +++ b/backend/tests/blocking_io/test_aio_sandbox_get.py @@ -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 diff --git a/backend/tests/blocking_io/test_sandbox_release.py b/backend/tests/blocking_io/test_sandbox_release.py index 2dcd4b6b7..226c8e284 100644 --- a/backend/tests/blocking_io/test_sandbox_release.py +++ b/backend/tests/blocking_io/test_sandbox_release.py @@ -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 = {} diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index 4da31b581..6e08d5146 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -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 diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index dff07e888..f163285dd 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/backend/tests/test_boxlite_provider.py @@ -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() diff --git a/backend/tests/test_sandbox_orphan_reconciliation.py b/backend/tests/test_sandbox_orphan_reconciliation.py index a0a2aa7b8..a68de083c 100644 --- a/backend/tests/test_sandbox_orphan_reconciliation.py +++ b/backend/tests/test_sandbox_orphan_reconciliation.py @@ -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 = {}