From bb75f8d736e4a4d1e7d4c881b988881ca68e2217 Mon Sep 17 00:00:00 2001 From: Zeren Wang <53075619+Vanzeren@users.noreply.github.com> Date: Sun, 30 Aug 2026 04:30:34 +0200 Subject: [PATCH] feat(sandbox): share sandbox identity derivation and acquire serialization (#4741) (#5089) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(sandbox): share sandbox identity derivation and acquire serialization (#4741) Remote providers (AIO, E2B, BoxLite, Tenki, OpenSandbox) each inlined the same sha256(user:thread)[:16] sandbox-id expression and kept per-scope lock dicts that grew unboundedly until shutdown. This extracts both mechanisms into shared components without changing provider lifecycle, ids, capacity semantics, or public tool behavior: - sandbox/identity.py: keyword-only derive_sandbox_scope_token (byte-pinned compatibility contract) + is_sandbox_scope_token; per-provider golden vectors pin current behavior including BoxLite's raw-None quirk and each provider's private user_id resolution. - sandbox/acquire_serialization.py: AcquireSerializer — per-key lock table with holder/waiter refcount reclamation, bounded dedicated executor (async waits off both the event loop and the default executor), worker-owned cancellation cleanup (no event-loop callback dependency), idempotent close(). - Each provider adopts both components; AIO/E2B key by (user_id, thread_id) with acquire and (E2B) release serialized; BoxLite/Tenki/OpenSandbox key by derived sandbox id and offload the whole sync acquire to the serializer's executor so a cancelled awaiter cannot overlap a retried same-scope body (leaked-remote-VM regression caught in review). - thread_id=None acquires stay unserialized; provider shutdown()/reset() close the serializer; E2B capacity/ledger/reconciliation and AIO ownership/flock machinery untouched. - blocking-IO anchor proves contended OpenSandbox acquire_async stays off the event loop (teeth verified red/green); AGENTS.md documents the shared components. * refactor(sandbox): address review on acquire serialization (#5089) - Replace unreachable checkin branch with an assertion: run() returns False only after abandon(), which the except handler always re-raises; the old _checkin would have double-decremented the refcount. - Document the task.cancelling() == 0 assumption in hold_async. - Drop unused thread_id/user_id kwargs from BoxLite and Tenki _acquire_scope_locked (OpenSandbox still forwards them). * fix(sandbox): preserve request ContextVars in acquire executor bridge (#5089) loop.run_in_executor() does not copy contextvars, unlike the inherited SandboxProvider.acquire_async() which used asyncio.to_thread(). The BoxLite/OpenSandbox/Tenki acquire_async bridges introduced in this PR therefore dropped the request trace id (logged as trace_id=-). Add AcquireSerializer.run_on_executor(), which copies the calling context and runs the callable through ctx.run, and route all three providers through it. Add regression tests binding request_trace_context and verifying the worker thread observes it. --- .../aio_sandbox/aio_sandbox_provider.py | 68 +---- .../deerflow/community/boxlite/provider.py | 109 +++---- .../e2b_sandbox/e2b_sandbox_provider.py | 29 +- .../community/opensandbox/provider.py | 107 +++---- .../deerflow/community/tenki/provider.py | 64 +++-- .../harness/deerflow/sandbox/AGENTS.md | 5 +- .../deerflow/sandbox/acquire_serialization.py | 219 ++++++++++++++ .../harness/deerflow/sandbox/identity.py | 43 +++ .../tests/blocking_io/test_aio_sandbox_get.py | 3 +- .../blocking_io/test_opensandbox_acquire.py | 56 ++++ .../tests/blocking_io/test_sandbox_release.py | 3 +- backend/tests/test_acquire_serialization.py | 270 ++++++++++++++++++ backend/tests/test_aio_sandbox_provider.py | 83 ++++-- backend/tests/test_boxlite_provider.py | 65 ++++- backend/tests/test_e2b_sandbox_provider.py | 10 +- backend/tests/test_opensandbox_provider.py | 44 +++ backend/tests/test_sandbox_identity.py | 59 ++++ .../test_sandbox_orphan_reconciliation.py | 3 +- backend/tests/test_tenki_provider.py | 7 + 19 files changed, 1012 insertions(+), 235 deletions(-) create mode 100644 backend/packages/harness/deerflow/sandbox/acquire_serialization.py create mode 100644 backend/packages/harness/deerflow/sandbox/identity.py create mode 100644 backend/tests/blocking_io/test_opensandbox_acquire.py create mode 100644 backend/tests/test_acquire_serialization.py create mode 100644 backend/tests/test_sandbox_identity.py 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 633c8333ea..237c944b66 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 @@ -13,7 +13,6 @@ The provider itself handles: import asyncio import atexit import contextlib -import hashlib import logging import os import signal @@ -21,7 +20,6 @@ import threading import time import uuid from collections.abc import Callable -from concurrent.futures import ThreadPoolExecutor try: import fcntl @@ -42,6 +40,8 @@ from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths, join_host_path from deerflow.integrations.lark_cli import INTEGRATION_ID as LARK_CLI_INTEGRATION_ID from deerflow.integrations.lark_cli import LARK_CLI_SANDBOX_CONFIG_DIR, LARK_CLI_SANDBOX_DATA_DIR, LARK_CLI_SANDBOX_LOCKS_DIR, LARK_CLI_SANDBOX_RUNTIME_DIR, ensure_lark_cli_credential_tree, lark_skills_installed from deerflow.runtime.user_context import get_effective_user_id +from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.sandbox.identity import derive_sandbox_scope_token from deerflow.sandbox.sandbox import Sandbox from deerflow.sandbox.sandbox_provider import SandboxProvider @@ -67,9 +67,6 @@ DEFAULT_IMAGE = "enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in DEFAULT_PORT = 8080 DEFAULT_CONTAINER_PREFIX = "deer-flow-sandbox" IDLE_CHECK_INTERVAL = _SHARED_IDLE_CHECK_INTERVAL -THREAD_LOCK_EXECUTOR_WORKERS = min(32, (os.cpu_count() or 1) + 4) -_THREAD_LOCK_EXECUTOR = ThreadPoolExecutor(max_workers=THREAD_LOCK_EXECUTOR_WORKERS, thread_name_prefix="sandbox-lock-wait") -atexit.register(_THREAD_LOCK_EXECUTOR.shutdown, wait=False, cancel_futures=True) class SandboxBeingDestroyedError(RuntimeError): @@ -121,36 +118,6 @@ def _open_lock_file(lock_path): return open(lock_path, "a", encoding="utf-8") -async def _acquire_thread_lock_async(lock: threading.Lock) -> None: - """Acquire a threading.Lock without polling or using the default executor.""" - loop = asyncio.get_running_loop() - acquire_future = loop.run_in_executor(_THREAD_LOCK_EXECUTOR, lock.acquire, True) - - try: - acquired = await asyncio.shield(acquire_future) - except asyncio.CancelledError: - acquire_future.add_done_callback(lambda task: _release_cancelled_lock_acquire(lock, task)) - raise - - if not acquired: - raise RuntimeError("Failed to acquire sandbox thread lock") - - -def _release_cancelled_lock_acquire(lock: threading.Lock, task: asyncio.Future[bool]) -> None: - """Release a lock acquired after its awaiting coroutine was cancelled.""" - if task.cancelled(): - return - - try: - acquired = task.result() - except Exception as e: - logger.warning(f"Cancelled sandbox lock acquisition finished with error: {e}") - return - - if acquired: - lock.release() - - class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): """Sandbox provider that manages containers running the AIO sandbox. @@ -189,7 +156,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._sandboxes: dict[str, AioSandbox] = {} # sandbox_id -> AioSandbox instance self._sandbox_infos: dict[str, SandboxInfo] = {} # sandbox_id -> SandboxInfo (for destroy) self._thread_sandboxes: dict[tuple[str, str], str] = {} # (user_id, thread_id) -> sandbox_id - self._thread_locks: dict[tuple[str, str], threading.Lock] = {} # (user_id, thread_id) -> in-process lock + self._acquire_serializer: AcquireSerializer[tuple[str, str]] = AcquireSerializer(thread_name_prefix="aio-sandbox-lock-wait") self._last_activity: dict[str, float] = {} # sandbox_id -> last activity timestamp # Warm pool: released sandboxes whose containers are still running. # Maps sandbox_id -> (SandboxInfo, release_timestamp). @@ -754,7 +721,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): 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()[:16] + return derive_sandbox_scope_token(user_id=user_id, thread_id=thread_id) def _assert_active_identity_available_locked( self, @@ -1306,14 +1273,6 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): # ── Thread locking (in-process) ────────────────────────────────────── - def _get_thread_lock(self, thread_id: str, user_id: str) -> threading.Lock: - """Get or create an in-process lock for a specific user/thread scope.""" - key = self._thread_key(thread_id, user_id) - with self._lock: - if key not in self._thread_locks: - self._thread_locks[key] = threading.Lock() - return self._thread_locks[key] - def _sandbox_id_for_thread(self, thread_id: str | None, user_id: str | None) -> str: """Return deterministic IDs for thread sandboxes and random IDs otherwise.""" return self._deterministic_sandbox_id(thread_id, self._effective_acquire_user_id(user_id)) if thread_id else str(uuid.uuid4())[:8] @@ -1803,11 +1762,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): """ effective_user_id = self._effective_acquire_user_id(user_id) if thread_id: - thread_lock = self._get_thread_lock(thread_id, effective_user_id) - with thread_lock: + with self._acquire_serializer.hold(self._thread_key(thread_id, effective_user_id)): return self._acquire_internal(thread_id, user_id=effective_user_id) - else: - return self._acquire_internal(thread_id, user_id=effective_user_id) + return self._acquire_internal(thread_id, user_id=effective_user_id) async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: """Acquire a sandbox environment without blocking the event loop. @@ -1818,13 +1775,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): """ effective_user_id = self._effective_acquire_user_id(user_id) if thread_id: - thread_lock = self._get_thread_lock(thread_id, effective_user_id) - await _acquire_thread_lock_async(thread_lock) - try: + async with self._acquire_serializer.hold_async(self._thread_key(thread_id, effective_user_id)): return await self._acquire_internal_async(thread_id, user_id=effective_user_id) - finally: - thread_lock.release() - return await self._acquire_internal_async(thread_id, user_id=effective_user_id) def _acquire_internal(self, thread_id: str | None, *, user_id: str) -> str: @@ -2234,6 +2186,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): # lease stuck in `del:`. self._release_ownership(sandbox_id) + def reset(self) -> None: + """Release process-local acquire workers when this instance is detached.""" + self._acquire_serializer.close() + def shutdown(self) -> None: """Shutdown all sandboxes. Thread-safe and idempotent.""" with self._lock: @@ -2271,3 +2227,5 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._ownership.close() except Exception as e: logger.warning(f"Error closing sandbox ownership store during shutdown: {e}") + + self._acquire_serializer.close() diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py index e49fecdb1e..734928fac7 100644 --- a/backend/packages/harness/deerflow/community/boxlite/provider.py +++ b/backend/packages/harness/deerflow/community/boxlite/provider.py @@ -15,17 +15,19 @@ from __future__ import annotations import asyncio import atexit -import hashlib import logging import threading import time import uuid from collections.abc import Awaitable +from functools import partial from typing import TYPE_CHECKING, Any, TypeVar from deerflow.config import get_app_config from deerflow.config.paths import VIRTUAL_PATH_PREFIX from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH +from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.sandbox.identity import derive_sandbox_scope_token from deerflow.sandbox.sandbox import Sandbox from deerflow.sandbox.sandbox_provider import SandboxProvider @@ -185,7 +187,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): 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()[:16] + return derive_sandbox_scope_token(user_id=user_id, thread_id=thread_id) # ── Provider ──────────────────────────────────────────────────────── @@ -197,7 +199,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): 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._acquire_serializer: AcquireSerializer[str] = AcquireSerializer(thread_name_prefix="boxlite-acquire-wait") self._idle_checker_stop = threading.Event() self._idle_checker_thread: threading.Thread | None = None self._shutdown_called = False @@ -243,15 +245,6 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): sandbox_id = name[len(_BOX_NAME_PREFIX) :] return sandbox_id or None - def _lock_for_sandbox(self, sandbox_id: str) -> threading.Lock: - """Return the per-sandbox acquire lock for a deterministic sandbox id.""" - with self._lock: - lock = self._acquire_locks.get(sandbox_id) - if lock is None: - lock = threading.Lock() - self._acquire_locks[sandbox_id] = lock - return lock - def _start_idle_checker(self) -> None: """Start idle cleanup when enabled; idle_timeout=0 keeps it disabled.""" if self._config["idle_timeout"] <= 0: @@ -378,49 +371,63 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): key = self._thread_key(thread_id, user_id) sandbox_id = self._sandbox_id(thread_id, user_id) - acquire_lock = self._lock_for_sandbox(sandbox_id) - with acquire_lock: - with self._lock: - 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 + with self._acquire_serializer.hold(sandbox_id): + return self._acquire_scope_locked(key, 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 + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + """Acquire without blocking the event loop. - box = self._create_box(sandbox_id) - conflict: tuple[str, str] | None | object = _NO_ACTIVE_IDENTITY - with self._lock: - 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: + The entire synchronous acquire (serializer wait + body) runs on the + serializer's dedicated executor — never the default executor. A + cancelled awaiter abandons the worker thread, which runs to + completion and releases the hold itself, so a retry serializes + behind it instead of overlapping the abandoned body. + """ + acquire = partial(self.acquire, thread_id, user_id=user_id) + return await self._acquire_serializer.run_on_executor(acquire) + + def _acquire_scope_locked(self, key: tuple[str, str], sandbox_id: str) -> str: + with self._lock: + 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, - conflict, + owner, key, ) - return box.id + self._thread_boxes[key] = sandbox_id + return 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: + 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: # Enforce replica limit: evict oldest warm-pool box if active + warm boxes are at capacity. @@ -627,7 +634,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): self._boxes.clear() self._active_box_identity.clear() self._thread_boxes.clear() - self._acquire_locks.clear() + self._acquire_serializer.close() def shutdown(self) -> None: with self._lock: @@ -645,7 +652,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): self._active_box_identity.clear() self._warm_pool_identity.clear() self._thread_boxes.clear() - self._acquire_locks.clear() + self._acquire_serializer.close() self._skip_health_check_warm_ids.clear() for box in active + warm: diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py index 01e8662e56..42e378cde8 100644 --- a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py @@ -36,7 +36,6 @@ from __future__ import annotations import asyncio import atexit -import hashlib import json import logging import os @@ -58,7 +57,9 @@ from e2b_code_interpreter import Sandbox as E2BClientSandbox from deerflow.config import get_app_config from deerflow.runtime.user_context import get_effective_user_id +from deerflow.sandbox.acquire_serialization import AcquireSerializer from deerflow.sandbox.exceptions import SandboxCapacityExceededError +from deerflow.sandbox.identity import derive_sandbox_scope_token from deerflow.sandbox.sandbox import Sandbox from deerflow.sandbox.sandbox_provider import SandboxProvider @@ -191,9 +192,9 @@ 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() and release() state + # Per-(user,thread) serializer for acquire() and release() state # transitions without holding the provider-wide lock across remote IO. - self._thread_locks: dict[tuple[str, str], threading.Lock] = {} + self._acquire_serializer: AcquireSerializer[tuple[str, str]] = AcquireSerializer(thread_name_prefix="e2b-sandbox-lock-wait") # Warm pool: released sandboxes whose remote micro-VM is still alive. # ``OrderedDict`` maintains insertion / move_to_end order for LRU. self._warm_pool: OrderedDict[str, tuple[str, float]] = OrderedDict() @@ -356,7 +357,12 @@ class E2BSandboxProvider(SandboxProvider): @staticmethod def _stable_seed(thread_id: str, user_id: str) -> str: - return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16] + """Warm-pool lookup seed derived from user/thread scope. + + For E2B this value is the warm-pool lookup seed, not the + provider-issued remote id (RFC #4741 §6). + """ + return derive_sandbox_scope_token(user_id=user_id, thread_id=thread_id) def _metadata_matches_capacity_ledger( self, @@ -412,19 +418,10 @@ class E2BSandboxProvider(SandboxProvider): sig_name, ) - def _get_thread_lock(self, thread_id: str, user_id: str) -> threading.Lock: - key = self._thread_key(thread_id, user_id) - with self._lock: - lock = self._thread_locks.get(key) - if lock is None: - lock = threading.Lock() - self._thread_locks[key] = lock - return lock - def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: effective_user_id = self._effective_acquire_user_id(user_id) if thread_id: - with self._get_thread_lock(thread_id, effective_user_id): + with self._acquire_serializer.hold(self._thread_key(thread_id, effective_user_id)): return self._acquire_internal(thread_id, user_id=effective_user_id) return self._acquire_internal(thread_id, user_id=effective_user_id) @@ -2343,7 +2340,7 @@ class E2BSandboxProvider(SandboxProvider): return user_id, thread_id = thread_key - with self._get_thread_lock(thread_id, user_id): + with self._acquire_serializer.hold(self._thread_key(thread_id, user_id)): self._release_internal(sandbox_id) def _release_internal(self, sandbox_id: str) -> None: @@ -2505,7 +2502,7 @@ class E2BSandboxProvider(SandboxProvider): self._remote_ops_in_progress.clear() self._unowned_remote_ops_in_progress.clear() self._thread_sandboxes.clear() - self._thread_locks.clear() + self._acquire_serializer.close() self._owned_sandbox_ids.clear() self._acquire_inflight.clear() self._orphan_first_seen.clear() diff --git a/backend/packages/harness/deerflow/community/opensandbox/provider.py b/backend/packages/harness/deerflow/community/opensandbox/provider.py index 8287d12dd2..676f671e94 100644 --- a/backend/packages/harness/deerflow/community/opensandbox/provider.py +++ b/backend/packages/harness/deerflow/community/opensandbox/provider.py @@ -3,7 +3,6 @@ from __future__ import annotations import atexit -import hashlib import ipaddress import logging import os @@ -11,10 +10,13 @@ import threading import time import uuid from datetime import timedelta +from functools import partial from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit from deerflow.config import get_app_config +from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.sandbox.identity import derive_sandbox_scope_token from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env from deerflow.sandbox.sandbox_provider import SandboxProvider @@ -78,7 +80,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro self._sandboxes: dict[str, OpenSandboxSandbox] = {} self._thread_sandboxes: dict[tuple[str, str], str] = {} self._warm_pool: dict[str, tuple[OpenSandboxSandbox, float]] = {} - self._acquire_locks: dict[str, threading.Lock] = {} + self._acquire_serializer: AcquireSerializer[str] = AcquireSerializer(thread_name_prefix="opensandbox-acquire-wait") self._idle_checker_stop = threading.Event() self._idle_checker_thread: threading.Thread | None = None self._shutdown_called = False @@ -167,20 +169,12 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro @staticmethod def _sandbox_id(thread_id: str, user_id: str) -> str: - return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16] + return derive_sandbox_scope_token(user_id=user_id, thread_id=thread_id) @staticmethod def _thread_key(thread_id: str, user_id: str | None) -> tuple[str, str]: return (user_id or "", thread_id) - def _lock_for_sandbox(self, sandbox_id: str) -> threading.Lock: - with self._lock: - lock = self._acquire_locks.get(sandbox_id) - if lock is None: - lock = threading.Lock() - self._acquire_locks[sandbox_id] = lock - return lock - def _start_idle_checker(self) -> None: if self._config["idle_timeout"] <= 0: return @@ -231,45 +225,60 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro key = self._thread_key(thread_id, user_id) sandbox_id = self._sandbox_id(thread_id, user_id or "") - with self._lock_for_sandbox(sandbox_id): - with self._lock: - existing = self._thread_sandboxes.get(key) - active = self._sandboxes.get(existing) if existing is not None else None - if existing is not None and active is not None: - try: - active.renew() - return existing - except Exception: - # A terminal renewal failure invokes _invalidate_sandbox, - # which removes and closes this exact client. Rebuild in the - # same acquire; transient errors leave the registry intact - # and must remain visible to the caller. - with self._lock: - invalidated = self._sandboxes.get(existing) is not active and existing not in self._warm_pool - shutting_down = self._shutdown_called - if not invalidated or not active.is_closed or shutting_down: - raise - logger.info("Rebuilding terminal OpenSandbox %s during acquire", existing) - reclaimed = self._reclaim_warm_pool(sandbox_id) - if reclaimed is not None: + with self._acquire_serializer.hold(sandbox_id): + return self._acquire_scope_locked(key, sandbox_id, thread_id=thread_id, user_id=user_id) + + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + """Acquire without blocking the event loop. + + The entire synchronous acquire (serializer wait + body) runs on the + serializer's dedicated executor — never the default executor. A + cancelled awaiter abandons the worker thread, which runs to + completion and releases the hold itself, so a retry serializes + behind it instead of overlapping the abandoned body. + """ + acquire = partial(self.acquire, thread_id, user_id=user_id) + return await self._acquire_serializer.run_on_executor(acquire) + + def _acquire_scope_locked(self, key: tuple[str, str], sandbox_id: str, *, thread_id: str, user_id: str | None) -> str: + with self._lock: + existing = self._thread_sandboxes.get(key) + active = self._sandboxes.get(existing) if existing is not None else None + if existing is not None and active is not None: + try: + active.renew() + return existing + except Exception: + # A terminal renewal failure invokes _invalidate_sandbox, + # which removes and closes this exact client. Rebuild in the + # same acquire; transient errors leave the registry intact + # and must remain visible to the caller. with self._lock: - if self._shutdown_called: - raise RuntimeError("OpenSandboxProvider shut down during acquire") - if reclaimed in self._sandboxes: - self._thread_sandboxes[key] = reclaimed - return reclaimed - sandbox = self._create_sandbox(sandbox_id, thread_id=thread_id, user_id=user_id) + invalidated = self._sandboxes.get(existing) is not active and existing not in self._warm_pool + shutting_down = self._shutdown_called + if not invalidated or not active.is_closed or shutting_down: + raise + logger.info("Rebuilding terminal OpenSandbox %s during acquire", existing) + reclaimed = self._reclaim_warm_pool(sandbox_id) + if reclaimed is not None: with self._lock: if self._shutdown_called: - destroy_after_unlock = True - else: - self._sandboxes[sandbox_id] = sandbox - self._thread_sandboxes[key] = sandbox_id - destroy_after_unlock = False - if destroy_after_unlock: - self._destroy_quietly(sandbox, context="created during shutdown") - raise RuntimeError("OpenSandboxProvider shut down during acquire") - return sandbox_id + raise RuntimeError("OpenSandboxProvider shut down during acquire") + if reclaimed in self._sandboxes: + self._thread_sandboxes[key] = reclaimed + return reclaimed + sandbox = self._create_sandbox(sandbox_id, thread_id=thread_id, user_id=user_id) + with self._lock: + if self._shutdown_called: + destroy_after_unlock = True + else: + self._sandboxes[sandbox_id] = sandbox + self._thread_sandboxes[key] = sandbox_id + destroy_after_unlock = False + if destroy_after_unlock: + self._destroy_quietly(sandbox, context="created during shutdown") + raise RuntimeError("OpenSandboxProvider shut down during acquire") + return sandbox_id def _create_sandbox(self, sandbox_id: str, *, thread_id: str | None, user_id: str | None) -> OpenSandboxSandbox: replicas, total = self._replica_count() @@ -368,7 +377,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro self._warm_pool.setdefault(sandbox_id, (sandbox, now)) self._sandboxes.clear() self._thread_sandboxes.clear() - self._acquire_locks.clear() + self._acquire_serializer.close() def shutdown(self) -> None: with self._lock: @@ -381,7 +390,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro self._sandboxes.clear() self._warm_pool.clear() self._thread_sandboxes.clear() - self._acquire_locks.clear() + self._acquire_serializer.close() for sandbox in sandboxes: self._destroy_quietly(sandbox, context="shutdown") diff --git a/backend/packages/harness/deerflow/community/tenki/provider.py b/backend/packages/harness/deerflow/community/tenki/provider.py index 355f5b8a27..2b5acce1d8 100644 --- a/backend/packages/harness/deerflow/community/tenki/provider.py +++ b/backend/packages/harness/deerflow/community/tenki/provider.py @@ -18,15 +18,17 @@ needed once this provider is selected. from __future__ import annotations import atexit -import hashlib import logging import shlex import threading import time import uuid +from functools import partial from typing import TYPE_CHECKING, Any from deerflow.config import get_app_config +from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.sandbox.identity import derive_sandbox_scope_token from deerflow.sandbox.sandbox import Sandbox, _validate_extra_env from deerflow.sandbox.sandbox_provider import SandboxProvider @@ -111,7 +113,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider gateway a hash collision would let one user reclaim another's parked sandbox. 64 bits, like community/e2b_sandbox, keeps that negligible. """ - return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:16] + return derive_sandbox_scope_token(user_id=user_id, thread_id=thread_id) # ── Provider lifecycle ─────────────────────────────────────────────── @@ -120,7 +122,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider self._sandboxes: dict[str, TenkiSandbox] = {} self._thread_sandboxes: dict[tuple[str, str], str] = {} self._warm_pool: dict[str, tuple[TenkiSandbox, float]] = {} - self._acquire_locks: dict[str, threading.Lock] = {} + self._acquire_serializer: AcquireSerializer[str] = AcquireSerializer(thread_name_prefix="tenki-acquire-wait") self._idle_checker_stop = threading.Event() self._idle_checker_thread: threading.Thread | None = None self._shutdown_called = False @@ -209,14 +211,6 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider def _sandbox_name(cls, sandbox_id: str) -> str: return f"{_SANDBOX_NAME_PREFIX}{sandbox_id}" - def _lock_for_sandbox(self, sandbox_id: str) -> threading.Lock: - with self._lock: - lock = self._acquire_locks.get(sandbox_id) - if lock is None: - lock = threading.Lock() - self._acquire_locks[sandbox_id] = lock - return lock - def _start_idle_checker(self) -> None: """Start idle cleanup when enabled; idle_timeout=0 keeps it disabled.""" if self._config["idle_timeout"] <= 0: @@ -257,24 +251,38 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider key = self._thread_key(thread_id, user_id) sandbox_id = self._sandbox_id(thread_id, user_id or "") - acquire_lock = self._lock_for_sandbox(sandbox_id) - with acquire_lock: - with self._lock: - existing = self._thread_sandboxes.get(key) - if existing is not None and existing in self._sandboxes: - return existing + with self._acquire_serializer.hold(sandbox_id): + return self._acquire_scope_locked(key, sandbox_id) - reclaimed = self._reclaim_warm_pool(sandbox_id) - if reclaimed is not None: - with self._lock: - self._thread_sandboxes[key] = reclaimed - return reclaimed + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + """Acquire without blocking the event loop. - sandbox = self._create_sandbox(sandbox_id) + The entire synchronous acquire (serializer wait + body) runs on the + serializer's dedicated executor — never the default executor. A + cancelled awaiter abandons the worker thread, which runs to + completion and releases the hold itself, so a retry serializes + behind it instead of overlapping the abandoned body. + """ + acquire = partial(self.acquire, thread_id, user_id=user_id) + return await self._acquire_serializer.run_on_executor(acquire) + + def _acquire_scope_locked(self, key: tuple[str, str], sandbox_id: str) -> str: + with self._lock: + existing = self._thread_sandboxes.get(key) + if existing is not None and existing in self._sandboxes: + return existing + + reclaimed = self._reclaim_warm_pool(sandbox_id) + if reclaimed is not None: with self._lock: - self._sandboxes[sandbox.id] = sandbox - self._thread_sandboxes[key] = sandbox.id - return sandbox.id + self._thread_sandboxes[key] = reclaimed + return reclaimed + + sandbox = self._create_sandbox(sandbox_id) + with self._lock: + self._sandboxes[sandbox.id] = sandbox + self._thread_sandboxes[key] = sandbox.id + return sandbox.id def _create_sandbox(self, sandbox_id: str) -> TenkiSandbox: # Enforce replica soft cap: evict the oldest warm sandbox if active + warm @@ -428,7 +436,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider self._warm_pool.setdefault(sandbox_id, (sandbox, now)) self._sandboxes.clear() self._thread_sandboxes.clear() - self._acquire_locks.clear() + self._acquire_serializer.close() def shutdown(self) -> None: with self._lock: @@ -444,7 +452,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider self._sandboxes.clear() self._warm_pool.clear() self._thread_sandboxes.clear() - self._acquire_locks.clear() + self._acquire_serializer.close() for sandbox in active + warm: self._close_quietly(sandbox, context="shutdown") diff --git a/backend/packages/harness/deerflow/sandbox/AGENTS.md b/backend/packages/harness/deerflow/sandbox/AGENTS.md index 98eca0775b..b9c8e772e8 100644 --- a/backend/packages/harness/deerflow/sandbox/AGENTS.md +++ b/backend/packages/harness/deerflow/sandbox/AGENTS.md @@ -2,16 +2,17 @@ **Interface**: Abstract `Sandbox` with `execute_command(command, env=None)`, `read_file`, `write_file`, `list_dir`, `glob`, and `grep`. `grep` accepts either one text file or a directory tree. The optional `env` injects per-call environment variables (request-scoped secrets — see Request-Scoped Secrets below); `LocalSandbox` merges it via `subprocess.run(env=...)` and `AioSandbox` routes env-bearing commands through the `bash.exec(env=...)` API on a fresh session. **Provider Pattern**: `SandboxProvider` with `acquire`, `acquire_async`, `get`, `release` lifecycle. Async agent/tool paths call async sandbox lifecycle hooks so Docker sandbox creation, discovery, cross-process locking, readiness polling, and release stay off the event loop. +**Shared components** (RFC #4741): remote providers derive their deterministic sandbox id through `derive_sandbox_scope_token` (`sandbox/identity.py`, keyword-only; the sha256/16-hex derivation is a compatibility contract — changing it orphans existing containers), and serialize provider-selected acquire/release transitions through `AcquireSerializer` (`sandbox/acquire_serialization.py`): per-key `threading.Lock` table with holder/waiter refcount reclamation (no unbounded per-thread lock growth), a bounded dedicated executor so async waits never touch the event loop or the default executor, worker-owned cancellation cleanup that does not depend on a cancelled event loop task resuming, and idempotent `close()` called from provider `shutdown()`/`reset()`. AIO/E2B key by `(user_id, thread_id)`; BoxLite/Tenki/OpenSandbox key by the derived sandbox id. `thread_id=None` acquires (random uuid ids) never enter the serializer. **Authorization gate** (`sandbox:execute`, RFC #4063 Phase 3): every sandbox acquisition passes through `authorize_sandbox_execution` (`deerflow/authz/sandbox_authz.py`) - a binary `authorize(principal, "sandbox", "execute", target="*")` check before `provider.acquire`. The gate lives at the single acquisition entry point (`ensure_sandbox_initialized` / `_acquire_sandbox_async` in `tools.py`, and `SandboxMiddleware.before_agent` / `abefore_agent`), so it cannot be bypassed regardless of which sandbox-dependent tool triggers it; the reuse path (sandbox already in state) skips the re-check. Deny raises `SandboxAuthorizationError` (`sandbox/exceptions.py`), which propagates out of tool execution as a friendly error `ToolMessage` ("sandbox execution is not permitted for your role") - the eager path catches it and skips acquisition instead, deferring the deny to the first sandbox-touching tool call so both paths share the same semantics. Provider errors (both `authorize()` and provider resolution) follow `authorization.fail_closed` / `fail_open`; no readable `config.yaml` or `authorization.enabled: false` makes the gate a no-op (`safe_app_config` tolerates missing config). Gateway auxiliary sync paths (uploads/artifacts routers) call `try_acquire_sandbox_for_request` (`app/gateway/authz.py`), which gates via `authorize_sandbox_for_request` and skips the sync on deny - the upload/artifact edit itself still succeeds. Tests: `tests/test_sandbox_authorization.py`. **Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved. **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`. Public, custom, legacy, and managed integration skill mappings point at stable enabled-only projection roots rather than raw skill directories. On Windows, Git Bash/MSYS argument-conversion exclusions are limited to safe non-root virtual path prefixes; do not restore a blanket conversion disable, because host-native CLI launchers need normal MSYS path conversion for their own installation paths. -- `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`). `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner backends=False), while the optional `sandbox.thread_data_mounts` boolean takes precedence for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Setting it `true` skips upload-time sandbox acquire/sync; a false positive leaves uploads unavailable to the sandbox. Local-container and hostPath-provisioner mounts use the same stable skill projection roots; PVC-backed skills remain governed by the operator-supplied PVC layout until PVC materialization is implemented. Readiness probes and `agent_sandbox` clients classify loopback/private IPs, single-label cluster hosts, and Docker/Podman internal hostnames as direct control-plane destinations and set `trust_env=False`; external FQDNs and public IPs retain environment proxy support. +- `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`). `reset()` closes the per-instance acquire serializer so replacing the singleton cannot retain its executor workers; full remote sandbox teardown remains `shutdown()`. `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner backends=False), while the optional `sandbox.thread_data_mounts` boolean takes precedence for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Setting it `true` skips upload-time sandbox acquire/sync; a false positive leaves uploads unavailable to the sandbox. Local-container and hostPath-provisioner mounts use the same stable skill projection roots; PVC-backed skills remain governed by the operator-supplied PVC layout until PVC materialization is implemented. Readiness probes and `agent_sandbox` clients classify loopback/private IPs, single-label cluster hosts, and Docker/Podman internal hostnames as direct control-plane destinations and set `trust_env=False`; external FQDNs and public IPs retain environment proxy support. - `E2BSandboxProvider` (`packages/harness/deerflow/community/e2b_sandbox/`) provides E2B remote isolation. New sandboxes receive a one-shot upload from the enabled-only public, custom, legacy, and managed integration projections. Existing E2B VMs keep their creation-time snapshot because E2B has no shared host mount. - Acquire and release share a per-user and thread lock. The provider lock does + Acquire and release share an `AcquireSerializer` hold keyed by `(user_id, thread_id)`. The serializer does not cover remote IO. `burst_limit` adds capacity only for the `burst` policy. The `wait` policy fails the turn after `acquire_timeout`. The runtime does not retry the turn automatically. E2B acquisition uses a bounded executor. diff --git a/backend/packages/harness/deerflow/sandbox/acquire_serialization.py b/backend/packages/harness/deerflow/sandbox/acquire_serialization.py new file mode 100644 index 0000000000..86fb6a98d9 --- /dev/null +++ b/backend/packages/harness/deerflow/sandbox/acquire_serialization.py @@ -0,0 +1,219 @@ +"""Per-key acquire serialization with bounded lock-table growth (RFC #4741). + +Shared component for sandbox providers. Each provider selects the key that +preserves its current collision and lifecycle semantics (``(user_id, +thread_id)`` for AIO/E2B, the derived sandbox id for BoxLite/Tenki/ +OpenSandbox); the serializer never interprets it. + +The async path runs the blocking ``threading.Lock.acquire`` on a bounded +dedicated executor (never the default executor, never the event loop). A small +handoff state makes the acquire worker release an abandoned lock itself, so +cleanup does not depend on the cancelling event loop running a done callback. +""" + +from __future__ import annotations + +import asyncio +import contextvars +import os +import threading +from collections.abc import AsyncIterator, Callable, Hashable, Iterator +from concurrent.futures import Future, ThreadPoolExecutor +from contextlib import asynccontextmanager, contextmanager +from dataclasses import dataclass, field +from functools import partial + +DEFAULT_MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4) + + +@dataclass +class _Entry: + lock: threading.Lock = field(default_factory=threading.Lock) + refs: int = 0 # holders + waiters + + +class _AsyncAcquire: + """Race-safe ownership handoff between an event loop and acquire worker.""" + + def __init__(self, entry: _Entry, cleanup: Callable[[], None]) -> None: + self._entry = entry + self._cleanup = cleanup + self._state_lock = threading.Lock() + self._abandoned = False + self._acquired = False + self._cleaned = False + + def run(self) -> bool: + self._entry.lock.acquire() + with self._state_lock: + if self._abandoned: + cleanup = self._mark_cleaned() + else: + self._acquired = True + cleanup = False + if cleanup: + self._entry.lock.release() + self._cleanup() + return not cleanup + + def abandon(self) -> None: + with self._state_lock: + self._abandoned = True + cleanup = self._acquired and self._mark_cleaned() + if cleanup: + self._acquired = False + if cleanup: + self._entry.lock.release() + self._cleanup() + + def cancel_queued(self) -> None: + """Clean up an executor job cancelled before its worker started.""" + with self._state_lock: + cleanup = self._mark_cleaned() + if cleanup: + self._cleanup() + + def worker_done(self, future: Future[bool]) -> None: + """Reclaim a checkout if executor shutdown cancelled queued work.""" + if future.cancelled(): + self.cancel_queued() + + def release(self) -> None: + with self._state_lock: + cleanup = self._acquired and self._mark_cleaned() + if cleanup: + self._acquired = False + if not cleanup: # pragma: no cover - indicates an internal ownership bug + raise RuntimeError("Serializer lock is not held") + self._entry.lock.release() + self._cleanup() + + def _mark_cleaned(self) -> bool: + if self._cleaned: + return False + self._cleaned = True + return True + + +class AcquireSerializer[KeyT: Hashable]: + """Serialize provider-selected lifecycle transitions per key.""" + + def __init__( + self, + *, + max_workers: int = DEFAULT_MAX_WORKERS, + thread_name_prefix: str = "sandbox-acquire-wait", + ) -> None: + self._table: dict[KeyT, _Entry] = {} + self._table_lock = threading.Lock() + self._closed = False + self._executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix=thread_name_prefix) + + @property + def executor(self) -> ThreadPoolExecutor: + """The dedicated bounded executor backing async waits. + + Providers may offload an entire synchronous acquire (with hold() + inside) here so a cancelled awaiter abandons the worker, not the + lock: the hold then follows the body to completion, matching the + pre-serializer to_thread bridge semantics. + """ + return self._executor + + async def run_on_executor[**P, T](self, func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T: + """Run a blocking callable on the serializer's dedicated executor. + + ``asyncio.to_thread`` copies ``ContextVar`` values automatically; raw + ``loop.run_in_executor`` does not. The providers whose ``acquire_async`` + offloads the whole synchronous acquire here replaced the inherited + ``to_thread`` bridge, so copy the calling context explicitly: without + it, request-scoped ContextVars (e.g. the trace id bound by + ``request_trace_context``) read as unset inside the worker thread. + """ + loop = asyncio.get_running_loop() + ctx = contextvars.copy_context() + call = partial(func, *args, **kwargs) + return await loop.run_in_executor(self._executor, ctx.run, call) + + @contextmanager + def hold(self, key: KeyT) -> Iterator[None]: + entry = self._checkout(key) + entry.lock.acquire() + try: + yield + finally: + entry.lock.release() + self._checkin(key, entry) + + @asynccontextmanager + async def hold_async(self, key: KeyT) -> AsyncIterator[None]: + entry = self._checkout(key) + loop = asyncio.get_running_loop() + acquire = _AsyncAcquire(entry, lambda: self._checkin(key, entry)) + try: + worker_future = self._executor.submit(acquire.run) + except RuntimeError: # executor shut down between checkout and scheduling + self._checkin(key, entry) + raise RuntimeError("AcquireSerializer is closed") from None + # concurrent.futures callbacks run in the cancelling/worker thread, so + # executor-shutdown cleanup does not depend on this loop progressing. + worker_future.add_done_callback(acquire.worker_done) + acquire_future = asyncio.wrap_future(worker_future, loop=loop) + try: + acquired = await asyncio.shield(acquire_future) + except asyncio.CancelledError: + task = asyncio.current_task() + if acquire_future.cancelled() and task is not None and task.cancelling() == 0: + # close() cancelled the queued wait via executor shutdown, not + # the caller: check in exactly once (no done callback — the + # cancelled future is already done) and report the documented + # closure error instead of a spurious CancelledError. The + # `cancelling() == 0` guard assumes this await is the task's + # only pending cancellation (no stale count left by an outer + # swallowed CancelledError). + acquire.cancel_queued() + raise RuntimeError("AcquireSerializer is closed") from None + if acquire_future.cancelled(): + acquire.cancel_queued() + else: + acquire.abandon() + raise + # run() returns False only when the worker observed _abandoned, and + # abandon() is called solely from the except handler above, which + # always re-raises — so a normal completion implies the lock is held. + # If this were reachable, the worker would already have checked in; + # an extra _checkin here would underflow the refcount, so assert. + assert acquired, "acquire worker returned without holding the lock" + try: + yield + finally: + acquire.release() + + def close(self) -> None: + """Reject new holders and release executor resources. Idempotent. + + A currently executing critical section is not invalidated: its exit + still releases its lock and reclaims its table entry. + """ + with self._table_lock: + if self._closed: + return + self._closed = True + self._executor.shutdown(wait=False, cancel_futures=True) + + def _checkout(self, key: KeyT) -> _Entry: + with self._table_lock: + if self._closed: + raise RuntimeError("AcquireSerializer is closed") + entry = self._table.get(key) + if entry is None: + entry = _Entry() + self._table[key] = entry + entry.refs += 1 + return entry + + def _checkin(self, key: KeyT, entry: _Entry) -> None: + with self._table_lock: + entry.refs -= 1 + if entry.refs == 0 and not entry.lock.locked(): + self._table.pop(key, None) diff --git a/backend/packages/harness/deerflow/sandbox/identity.py b/backend/packages/harness/deerflow/sandbox/identity.py new file mode 100644 index 0000000000..a0c689ce73 --- /dev/null +++ b/backend/packages/harness/deerflow/sandbox/identity.py @@ -0,0 +1,43 @@ +"""Shared sandbox scope-token derivation (RFC #4741). + +COMPATIBILITY CONTRACT: this derivation is a durable identity boundary. AIO, +E2B, BoxLite, Tenki, and OpenSandbox locate existing containers and VMs +through this token. Changing any property — separator, encoding, digest, +casing, or truncation length — is a breaking migration: existing remote +resources would no longer be found and would be cold-started. + +Providers resolve ``user_id`` before calling this function, and their +resolutions differ (effective-user lookup, ``""`` substitution, or raw +pass-through). That resolution stays provider-private; this module only pins +what happens to the resolved strings. +""" + +from __future__ import annotations + +import hashlib +import re + +SANDBOX_ID_VERSION = 1 + +_TOKEN_HEX_LEN = 16 +_TOKEN_RE = re.compile(r"[0-9a-f]{16}") + + +def derive_sandbox_scope_token(*, user_id: str, thread_id: str) -> str: + """Return the durable 16-lowercase-hex sandbox scope token. + + Keyword-only: every provider helper this replaces takes + ``(thread_id, user_id)`` positionally — the reverse order — and both are + plain ``str``; keyword-only call sites eliminate silent argument-order + mistakes during and after the migration. + + WARNING: changing the separator, encoding, digest, casing, or truncation + length is a breaking migration — existing containers and VMs would no + longer be found. See module docstring. + """ + return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:_TOKEN_HEX_LEN] + + +def is_sandbox_scope_token(value: object) -> bool: + """Validate token shape only; a truncated hash is not reversible.""" + return isinstance(value, str) and _TOKEN_RE.fullmatch(value) is not None diff --git a/backend/tests/blocking_io/test_aio_sandbox_get.py b/backend/tests/blocking_io/test_aio_sandbox_get.py index 5ad2f13130..07f2b1a9e3 100644 --- a/backend/tests/blocking_io/test_aio_sandbox_get.py +++ b/backend/tests/blocking_io/test_aio_sandbox_get.py @@ -73,13 +73,14 @@ def _make_provider(tmp_path: Path): """Build an ``AioSandboxProvider`` without ``__init__`` (no Docker, no threads).""" from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider from deerflow.config.sandbox_config import SandboxOwnershipConfig + from deerflow.sandbox.acquire_serialization import AcquireSerializer provider = AioSandboxProvider.__new__(AioSandboxProvider) provider._lock = threading.Lock() provider._sandboxes = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} - provider._thread_locks = {} + provider._acquire_serializer = AcquireSerializer(thread_name_prefix="aio-sandbox-lock-wait") provider._last_activity = {} provider._warm_pool = {} provider._active_sandbox_identity = {} diff --git a/backend/tests/blocking_io/test_opensandbox_acquire.py b/backend/tests/blocking_io/test_opensandbox_acquire.py new file mode 100644 index 0000000000..4ff6aea2d6 --- /dev/null +++ b/backend/tests/blocking_io/test_opensandbox_acquire.py @@ -0,0 +1,56 @@ +"""Anchor: contended OpenSandbox acquire_async must not block the event loop. + +Two concurrent acquire_async calls for the same scope serialize on the +AcquireSerializer; the loser's wait and both callers' creation path must stay +off the loop. The fake SDK boundary below performs REAL file IO inside +create(), so any regression that moves creation back onto the loop trips the +Blockbuster gate (FILE_IO rules give this anchor teeth). + +Blockbuster's default rule set is blind to ``threading.Lock.acquire`` (verified +empirically under ``detect_blocking_io_strict``), so the serializer lock wait +itself cannot be pinned this way — the lock-wait placement is covered by Task +7's ExplodingExecutor contract test. What this anchor pins instead is the +provider creation path: under contention, no blocking IO may run on the loop. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest +from blockbuster import BlockingError +from test_opensandbox_provider import _FakeRemote, _FakeSandboxClass, _install + +pytestmark = pytest.mark.asyncio + + +class _FileIOProbingSandboxClass(_FakeSandboxClass): + """Fake SDK whose create() performs real blocking file IO at the boundary.""" + + def __init__(self, probe_dir: Path) -> None: + super().__init__() + self._probe_dir = probe_dir + + def create(self, image: str, **kwargs) -> _FakeRemote: + probe = self._probe_dir / f"create-{len(self.create_calls) + 1}.probe" + probe.write_text("x" * 4096) # real blocking file IO at the SDK boundary + return super().create(image, **kwargs) + + +async def test_concurrent_acquire_async_stays_off_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + provider, _fake_sdk = _install(monkeypatch, sdk=_FileIOProbingSandboxClass(tmp_path)) + first, second = await asyncio.gather( + provider.acquire_async("thread-anchor", user_id="u-anchor"), + provider.acquire_async("thread-anchor", user_id="u-anchor"), + ) + assert first == second # same scope serialized, second caller reuses + provider.shutdown() + + +async def test_sync_acquire_on_loop_trips_the_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Meta-check (teeth): a synchronous acquire on the loop MUST be caught.""" + provider, _fake_sdk = _install(monkeypatch, sdk=_FileIOProbingSandboxClass(tmp_path)) + with pytest.raises(BlockingError): + provider.acquire("thread-anchor", user_id="u-anchor") + provider.shutdown() diff --git a/backend/tests/blocking_io/test_sandbox_release.py b/backend/tests/blocking_io/test_sandbox_release.py index 226c8e2848..ba75bd44cb 100644 --- a/backend/tests/blocking_io/test_sandbox_release.py +++ b/backend/tests/blocking_io/test_sandbox_release.py @@ -71,6 +71,7 @@ def _make_provider_with_active_sandbox(tmp_path: Path, sandbox_id: str): from deerflow.community.aio_sandbox.aio_sandbox_provider import AioSandboxProvider from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo from deerflow.config.sandbox_config import SandboxOwnershipConfig + from deerflow.sandbox.acquire_serialization import AcquireSerializer provider = AioSandboxProvider.__new__(AioSandboxProvider) provider._lock = threading.Lock() @@ -85,7 +86,7 @@ def _make_provider_with_active_sandbox(tmp_path: Path, sandbox_id: str): ) } provider._thread_sandboxes = {} - provider._thread_locks = {} + provider._acquire_serializer = AcquireSerializer(thread_name_prefix="aio-sandbox-lock-wait") provider._last_activity = {sandbox_id: 1.0} provider._warm_pool = {} provider._warm_pool_identity = {} diff --git a/backend/tests/test_acquire_serialization.py b/backend/tests/test_acquire_serialization.py new file mode 100644 index 0000000000..273e42e086 --- /dev/null +++ b/backend/tests/test_acquire_serialization.py @@ -0,0 +1,270 @@ +"""Contract tests for AcquireSerializer (RFC #4741 §5).""" + +from __future__ import annotations + +import asyncio +import threading +import time +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from deerflow.sandbox.acquire_serialization import AcquireSerializer +from deerflow.trace_context import get_current_trace_id, request_trace_context + + +class TestSyncMutualExclusion: + def test_same_key_never_overlaps(self): + serializer = AcquireSerializer() + active = 0 + max_active = 0 + counter_lock = threading.Lock() + + def worker(): + nonlocal active, max_active + with serializer.hold("k"): + with counter_lock: + active += 1 + max_active = max(max_active, active) + time.sleep(0.02) + with counter_lock: + active -= 1 + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert max_active == 1 + + def test_different_keys_proceed_concurrently(self): + serializer = AcquireSerializer() + barrier = threading.Barrier(2, timeout=2) + + def worker(key): + with serializer.hold(key): + barrier.wait() # raises BrokenBarrierError if serialized + + t1 = threading.Thread(target=worker, args=("a",)) + t2 = threading.Thread(target=worker, args=("b",)) + t1.start() + t2.start() + t1.join(timeout=5) + t2.join(timeout=5) + assert not t1.is_alive() and not t2.is_alive() + + def test_lock_table_reclaims_entries(self): + serializer = AcquireSerializer() + for i in range(50): + with serializer.hold(f"scope-{i}"): + pass + assert len(serializer._table) == 0 + + +class TestAsyncContract: + @pytest.mark.asyncio + async def test_cancellation_before_acquire_leaks_nothing(self): + serializer = AcquireSerializer() + with serializer.hold("k"): # uncontended acquire, safe on the loop + waiter = asyncio.create_task(_hold_once(serializer, "k")) + await asyncio.sleep(0.05) # let the waiter start blocking in the worker + waiter.cancel() + with pytest.raises(asyncio.CancelledError): + await waiter + # The worker acquires the now-free lock in the background; its done + # callback must release it and reclaim the entry. + async with serializer.hold_async("k"): + pass + await _wait_for_table_empty(serializer) + + @pytest.mark.asyncio + async def test_cancelled_wait_releases_without_event_loop_callback(self): + serializer = AcquireSerializer() + holder_ready = threading.Event() + release_holder = threading.Event() + + def hold_in_thread(): + with serializer.hold("k"): + holder_ready.set() + release_holder.wait(2) + + holder = threading.Thread(target=hold_in_thread) + holder.start() + assert holder_ready.wait(2) + waiter = asyncio.create_task(_hold_once(serializer, "k")) + await asyncio.sleep(0.05) + waiter.cancel() + with pytest.raises(asyncio.CancelledError): + await waiter + + # Once cancellation has been observed, cleanup must be completed by + # the acquire worker itself. Blocking this loop keeps asyncio Future + # callbacks from hiding a lock leak here. + release_holder.set() + deadline = time.monotonic() + 2 + while serializer._table and time.monotonic() < deadline: + time.sleep(0.01) + holder.join(timeout=2) + assert not holder.is_alive() + assert len(serializer._table) == 0 + + @pytest.mark.asyncio + async def test_cancellation_after_acquire_releases(self): + serializer = AcquireSerializer() + + async def run(): + async with serializer.hold_async("k"): + await asyncio.sleep(10) + + task = asyncio.create_task(run()) + await asyncio.sleep(0.05) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + async with serializer.hold_async("k"): # must not hang + pass + assert len(serializer._table) == 0 + + @pytest.mark.asyncio + async def test_async_wait_keeps_loop_alive_and_off_default_executor(self): + serializer = AcquireSerializer() + loop = asyncio.get_running_loop() + + class ExplodingExecutor(ThreadPoolExecutor): + def submit(self, *args, **kwargs): + raise AssertionError("serializer wait used the default executor") + + previous = loop._default_executor + loop.set_default_executor(ExplodingExecutor()) + try: + entered = asyncio.Event() + + async def holder(): + async with serializer.hold_async("k"): + entered.set() + await asyncio.sleep(0.2) + + h = asyncio.create_task(holder()) + await entered.wait() + w = asyncio.create_task(_hold_once(serializer, "k")) + for _ in range(5): # loop stays responsive while w waits + await asyncio.sleep(0.02) + assert not w.done() + await h + await asyncio.wait_for(w, 2) + finally: + if previous is not None: # fresh loops have no default executor yet + loop.set_default_executor(previous) + + @pytest.mark.asyncio + async def test_concurrent_async_waiters_serialize(self): + serializer = AcquireSerializer() + active = 0 + max_active = 0 + + async def worker(): + nonlocal active, max_active + async with serializer.hold_async("k"): + active += 1 + max_active = max(max_active, active) + await asyncio.sleep(0.01) + active -= 1 + + await asyncio.gather(*[worker() for _ in range(8)]) + assert max_active == 1 + + @pytest.mark.asyncio + async def test_run_on_executor_preserves_request_trace_context(self): + """run_on_executor must carry request ContextVars into the worker. + + Regression test (#5089): raw ``loop.run_in_executor`` does not copy + contextvars, so without an explicit ``copy_context`` the worker thread + reads the trace id bound by ``request_trace_context()`` as unset. + """ + serializer = AcquireSerializer() + try: + with request_trace_context("trace-5089"): + seen = await serializer.run_on_executor(get_current_trace_id) + assert seen == "trace-5089" + finally: + serializer.close() + + +async def _hold_once(serializer, key): + async with serializer.hold_async(key): + pass + + +async def _wait_for_table_empty(serializer, timeout=2.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if len(serializer._table) == 0: + return + await asyncio.sleep(0.01) + raise AssertionError("serializer lock table did not drain") + + +class TestClose: + def test_close_rejects_new_holders_and_is_idempotent(self): + serializer = AcquireSerializer() + serializer.close() + serializer.close() + with pytest.raises(RuntimeError, match="closed"): + with serializer.hold("k"): + pass + + @pytest.mark.asyncio + async def test_close_does_not_invalidate_active_critical_section(self): + serializer = AcquireSerializer() + async with serializer.hold_async("k"): + serializer.close() # must not raise, must not break exit + # New holders rejected afterwards: + with pytest.raises(RuntimeError, match="closed"): + async with serializer.hold_async("k"): + pass + + @pytest.mark.asyncio + async def test_executor_shutdown_between_checkout_and_schedule_reclaims_entry(self): + serializer = AcquireSerializer() + serializer._executor.shutdown(wait=False, cancel_futures=True) # simulates close() landing mid-window + with pytest.raises(RuntimeError, match="closed"): + async with serializer.hold_async("k"): + pass + assert len(serializer._table) == 0 + + @pytest.mark.asyncio + async def test_close_reports_closed_to_queued_waiter(self): + """A waiter whose queued acquire is cancelled by close() must see + RuntimeError("closed"), not a spurious CancelledError (#4741).""" + serializer = AcquireSerializer(max_workers=1) + blocker = threading.Event() + with serializer.hold("k"): # uncontended acquire, safe on the loop + # Occupy the single worker so the waiter's acquire future stays + # queued (pending) and is cancelled by close()'s cancel_futures. + occupied = asyncio.get_running_loop().run_in_executor(serializer.executor, blocker.wait, 5) + waiter = asyncio.create_task(_hold_once(serializer, "k")) + await asyncio.sleep(0.1) # let the waiter check out and queue its acquire + serializer.close() + with pytest.raises(RuntimeError, match="closed"): + await waiter + blocker.set() + await occupied + assert len(serializer._table) == 0 + + @pytest.mark.asyncio + async def test_close_reclaims_caller_cancelled_queued_waiter(self): + serializer = AcquireSerializer(max_workers=1) + blocker = threading.Event() + occupied = asyncio.get_running_loop().run_in_executor(serializer.executor, blocker.wait, 5) + waiter = asyncio.create_task(_hold_once(serializer, "k")) + await asyncio.sleep(0.05) + waiter.cancel() + with pytest.raises(asyncio.CancelledError): + await waiter + + # The acquire was abandoned by its caller but has not started. close() + # must not strand its checkout when cancelling the executor queue. + serializer.close() + blocker.set() + await occupied + await _wait_for_table_empty(serializer) diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index f4a2942dcb..b70238cde7 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -14,6 +14,7 @@ import pytest from deerflow.config.paths import Paths, join_host_path from deerflow.config.sandbox_config import SandboxConfig from deerflow.runtime.user_context import reset_current_user, set_current_user +from deerflow.sandbox.acquire_serialization import AcquireSerializer _LEGACY_COLLIDING_IDENTITIES = ( ("user-9721", "thread-9721"), @@ -119,6 +120,7 @@ def _make_provider(tmp_path): provider._acquire_epoch = {} provider._acquire_epoch_counter = 0 provider._acquire_inflight = {} + provider._acquire_serializer = AcquireSerializer(thread_name_prefix="aio-sandbox-lock-wait") provider._lock = MagicMock() provider._idle_checker_stop = MagicMock() provider._renewal_stop = MagicMock() @@ -347,7 +349,6 @@ async def test_acquire_async_uses_async_readiness_polling(monkeypatch): aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = _make_provider(None) provider._config = {"replicas": 3} - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -389,7 +390,6 @@ async def test_discover_or_create_with_lock_async_offloads_lock_file_open_and_cl provider, aio_mod.AioSandboxProvider, ) - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {("default", "thread-async-lock"): "sandbox-async-lock"} @@ -416,21 +416,33 @@ async def test_discover_or_create_with_lock_async_offloads_lock_file_open_and_cl @pytest.mark.anyio -async def test_acquire_thread_lock_async_uses_dedicated_executor(monkeypatch): +async def test_acquire_async_lock_wait_uses_dedicated_executor(tmp_path, monkeypatch): """Per-thread lock waits should not consume the default asyncio.to_thread pool.""" aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") - lock = aio_mod.threading.Lock() + provider = _make_provider(tmp_path) async def fail_to_thread(*_args, **_kwargs): raise AssertionError("thread-lock acquisition must not use asyncio.to_thread") monkeypatch.setattr(aio_mod.asyncio, "to_thread", fail_to_thread) - await aio_mod._acquire_thread_lock_async(lock) + async def fake_acquire_internal_async(thread_id: str | None, *, user_id: str) -> str: + await asyncio.sleep(0) + return "sandbox-lock-wait" + + monkeypatch.setattr(provider, "_acquire_internal_async", fake_acquire_internal_async) + + thread_id = "thread-lock-wait" + hold = provider._acquire_serializer.hold(provider._thread_key(thread_id, "default")) + hold.__enter__() try: - assert not lock.acquire(blocking=False) + waiter = asyncio.create_task(provider.acquire_async(thread_id, user_id="default")) + await asyncio.sleep(0.05) + assert not waiter.done() finally: - lock.release() + hold.__exit__(None, None, None) + + assert await asyncio.wait_for(waiter, timeout=1) == "sandbox-lock-wait" @pytest.mark.anyio @@ -438,7 +450,6 @@ async def test_acquire_async_cancellation_does_not_leak_thread_lock(tmp_path): """Cancelled async lock waiters must not leave the per-thread lock held.""" aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = _make_provider(tmp_path) - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -446,8 +457,9 @@ async def test_acquire_async_cancellation_does_not_leak_thread_lock(tmp_path): provider._lock = aio_mod.threading.Lock() thread_id = "thread-cancel-lock" - thread_lock = provider._get_thread_lock(thread_id, "default") - thread_lock.acquire() + key = provider._thread_key(thread_id, "default") + hold = provider._acquire_serializer.hold(key) + hold.__enter__() task = asyncio.create_task(provider.acquire_async(thread_id, user_id="default")) await asyncio.sleep(0.05) @@ -458,12 +470,10 @@ async def test_acquire_async_cancellation_does_not_leak_thread_lock(tmp_path): except asyncio.CancelledError: pass - thread_lock.release() + hold.__exit__(None, None, None) deadline = asyncio.get_running_loop().time() + 1 while asyncio.get_running_loop().time() < deadline: - acquired = thread_lock.acquire(blocking=False) - if acquired: - thread_lock.release() + if key not in provider._acquire_serializer._table: return await asyncio.sleep(0.01) @@ -475,7 +485,6 @@ async def test_acquire_async_cancelled_waiter_does_not_block_successor(tmp_path, """A cancelled waiter must not prevent the next live waiter from acquiring.""" aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = _make_provider(tmp_path) - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -491,8 +500,9 @@ async def test_acquire_async_cancelled_waiter_does_not_block_successor(tmp_path, monkeypatch.setattr(provider, "_acquire_internal_async", fake_acquire_internal_async) thread_id = "thread-successor-lock" - thread_lock = provider._get_thread_lock(thread_id, "default") - thread_lock.acquire() + key = provider._thread_key(thread_id, "default") + hold = provider._acquire_serializer.hold(key) + hold.__enter__() cancelled_waiter = asyncio.create_task(provider.acquire_async(thread_id, user_id="default")) await asyncio.sleep(0.05) @@ -503,15 +513,13 @@ async def test_acquire_async_cancelled_waiter_does_not_block_successor(tmp_path, pass live_waiter = asyncio.create_task(provider.acquire_async(thread_id, user_id="default")) - thread_lock.release() + hold.__exit__(None, None, None) assert await asyncio.wait_for(live_waiter, timeout=1) == "sandbox-successor" deadline = asyncio.get_running_loop().time() + 1 while asyncio.get_running_loop().time() < deadline: - acquired = thread_lock.acquire(blocking=False) - if acquired: - thread_lock.release() + if key not in provider._acquire_serializer._table: return await asyncio.sleep(0.01) @@ -612,7 +620,6 @@ def test_create_sandbox_requests_runtime_when_lark_installed(tmp_path, monkeypat aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = _make_provider(tmp_path) provider._config = {"replicas": 3} - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -643,7 +650,6 @@ def test_create_sandbox_requests_broker_when_active(tmp_path, monkeypatch): aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = _make_provider(tmp_path) provider._config = {"replicas": 3} - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -674,7 +680,6 @@ def test_create_sandbox_skips_runtime_when_lark_absent(tmp_path, monkeypatch): aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = _make_provider(tmp_path) provider._config = {"replicas": 3} - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -765,6 +770,24 @@ def test_shutdown_closes_all_active_sandbox_clients(tmp_path): assert provider._sandboxes == {} +@pytest.mark.asyncio +async def test_reset_closes_acquire_serializer_executor(tmp_path): + provider = _make_provider(tmp_path) + async with provider._acquire_serializer.hold_async(("alice", "thread-reset")): + pass + worker_threads = tuple(provider._acquire_serializer.executor._threads) + assert worker_threads + + provider.reset() + + for thread in worker_threads: + thread.join(timeout=2) + assert all(not thread.is_alive() for thread in worker_threads) + with pytest.raises(RuntimeError, match="closed"): + async with provider._acquire_serializer.hold_async(("alice", "thread-after-reset")): + pass + + def test_release_swallows_close_errors(tmp_path, caplog): """A failure inside sandbox.close() must not break provider release().""" provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-rel-err") @@ -790,7 +813,6 @@ def test_acquire_drops_dead_cached_sandbox(tmp_path, monkeypatch): """acquire() must replace a stale active cache entry after its container dies.""" aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-dead") - provider._thread_locks = {} provider._thread_sandboxes = {("default", "thread-dead"): "sandbox-dead"} provider._config = {"replicas": 3} provider._backend.is_alive = MagicMock(return_value=False) @@ -822,7 +844,6 @@ def test_acquire_drops_dead_cached_sandbox(tmp_path, monkeypatch): def test_acquire_keeps_cached_sandbox_when_health_check_errors(tmp_path): """Transient backend health-check errors must not destroy a tracked sandbox.""" provider, sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-transient") - provider._thread_locks = {} provider._thread_sandboxes = {("default", "thread-transient"): "sandbox-transient"} provider._backend.is_alive = MagicMock(side_effect=OSError("docker daemon busy")) @@ -863,7 +884,6 @@ def test_acquire_skips_dead_warm_pool_sandbox(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._thread_locks = {} provider._sandboxes = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -997,7 +1017,6 @@ def _make_tenant_isolation_provider(tmp_path, monkeypatch): provider._sandboxes = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} - provider._thread_locks = {} provider._last_activity = {} provider._warm_pool = {} provider._active_sandbox_identity = {} @@ -1101,7 +1120,6 @@ def _make_unready_destroy_provider(tmp_path, *, sandbox_id, base_url, monkeypatc provider = _make_provider(tmp_path) provider._lock = aio_mod.threading.Lock() provider._config = {"replicas": 3} - provider._thread_locks = {} provider._warm_pool = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} @@ -1305,3 +1323,10 @@ def test_reconcile_adopts_unready_container_when_no_teardown_is_in_flight(tmp_pa provider._reconcile_orphans() assert "adoptable" in provider._warm_pool, "reconcile must still adopt a genuinely unowned container" + + +def test_deterministic_sandbox_id_matches_shared_identity(): + from deerflow.sandbox.identity import derive_sandbox_scope_token + + aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") + assert aio_mod.AioSandboxProvider._deterministic_sandbox_id("t-1", "u-1") == derive_sandbox_scope_token(user_id="u-1", thread_id="t-1") diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index 58ac686b8f..8c00b6b7f9 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/backend/tests/test_boxlite_provider.py @@ -18,6 +18,7 @@ import pytest from deerflow.community.boxlite.box import BoxliteBox from deerflow.community.boxlite.provider import BoxliteProvider, _import_simplebox +from deerflow.trace_context import get_current_trace_id, request_trace_context _LEGACY_COLLIDING_IDENTITIES = ( ("user-9721", "thread-9721"), @@ -476,6 +477,48 @@ def test_acquire_reclaims_from_warm_pool(monkeypatch): assert sid2 not in provider._warm_pool +@pytest.mark.asyncio +async def test_acquire_async_propagates_request_trace_context(monkeypatch): + """acquire_async must carry request ContextVars into the worker thread. + + Regression test (#5089): the dedicated-executor bridge replaced the + inherited ``to_thread`` acquire (which copies contextvars). Without an + explicit ``copy_context`` the worker thread reads the trace id bound by + ``request_trace_context()`` as unset and logs ``trace_id=-``. + """ + monkeypatch.setattr( + "deerflow.community.boxlite.provider.get_app_config", + lambda: _stub_config(), + ) + monkeypatch.setattr( + "deerflow.community.boxlite.provider._import_simplebox", + lambda: _FakeBox, + ) + + provider = BoxliteProvider() + provider._loop.run = _fake_run + + seen: list[str | None] = [] + original = BoxliteProvider._acquire_scope_locked + + def spy(self, key, sandbox_id): + # Runs inside the executor worker thread, under the copied context. + seen.append(get_current_trace_id()) + return original(self, key, sandbox_id) + + monkeypatch.setattr(BoxliteProvider, "_acquire_scope_locked", spy) + + try: + with request_trace_context("trace-boxlite-5089"): + sid = await provider.acquire_async("thread-1", user_id="u1") + assert sid + assert seen == ["trace-boxlite-5089"] + finally: + # _fake_run uses asyncio.run, which cannot run inside this test's event + # loop thread; shut down on a worker thread so box close() completes. + await asyncio.to_thread(provider.shutdown) + + def test_explicit_recent_reclaim_skip_avoids_health_check(monkeypatch): """A configured skip window can reclaim recently released boxes without a ping.""" monkeypatch.setattr( @@ -873,7 +916,9 @@ def test_reset_parks_running_resources_for_later_cleanup(monkeypatch): assert provider._warm_pool[sid_active][0] is active_box assert provider._warm_pool[sid_warm][0] is warm_box assert provider._thread_boxes == {} - assert provider._acquire_locks == {} + with pytest.raises(RuntimeError, match="closed"): + with provider._acquire_serializer.hold("k"): + pass assert not active_box._closed assert not warm_box._closed assert not provider._shutdown_called @@ -1226,3 +1271,21 @@ def test_failed_health_check_does_not_remove_swapped_warm_entry(monkeypatch): ) assert not replacement.is_closed provider.shutdown() + + +def test_sandbox_id_matches_shared_identity(): + from deerflow.sandbox.identity import derive_sandbox_scope_token + + assert BoxliteProvider._sandbox_id("t-1", "u-1") == derive_sandbox_scope_token(user_id="u-1", thread_id="t-1") + + +def test_sandbox_id_none_user_quirk_pinned(): + """BoxLite passes user_id through raw; None renders as the literal "None". + + Quirk pinned per RFC #4741 §2.2 (its _thread_key uses "" instead, so the + two disagree). NOT fixed here — unifying the resolution is a separate + behavior-changing decision with its own follow-up issue. + """ + from deerflow.sandbox.identity import derive_sandbox_scope_token + + assert BoxliteProvider._sandbox_id("t-1", None) == derive_sandbox_scope_token(user_id="None", thread_id="t-1") diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py index a2e2b57df1..b806717048 100644 --- a/backend/tests/test_e2b_sandbox_provider.py +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -24,6 +24,7 @@ from deerflow.community.e2b_sandbox.capacity import ( ) from deerflow.config.paths import Paths from deerflow.config.sandbox_config import SandboxConfig +from deerflow.sandbox.acquire_serialization import AcquireSerializer from deerflow.sandbox.exceptions import SandboxCapacityExceededError # ────────────────────────────────────────────────────────────────────────────── @@ -250,7 +251,7 @@ def _make_provider(*, replicas: int = 3, idle_timeout: int = 1800, overflow_poli provider._lock = threading.Lock() provider._sandboxes = {} provider._thread_sandboxes = {} - provider._thread_locks = {} + provider._acquire_serializer = AcquireSerializer(thread_name_prefix="e2b-sandbox-lock-wait") provider._warm_pool = OrderedDict() provider._eviction_tombstones = set() provider._evictions_in_progress = set() @@ -4112,3 +4113,10 @@ def test_shutdown_during_discovery_does_not_kill_unowned_vm(monkeypatch): assert client.closed assert p._sandboxes == {} assert p._reserved_slots == 0 + + +def test_stable_seed_matches_shared_identity(): + from deerflow.sandbox.identity import derive_sandbox_scope_token + + mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") + assert mod.E2BSandboxProvider._stable_seed("t-1", "u-1") == derive_sandbox_scope_token(user_id="u-1", thread_id="t-1") diff --git a/backend/tests/test_opensandbox_provider.py b/backend/tests/test_opensandbox_provider.py index 3314db38ec..b6b5576a1a 100644 --- a/backend/tests/test_opensandbox_provider.py +++ b/backend/tests/test_opensandbox_provider.py @@ -8,6 +8,7 @@ transport, search parsing, path guards, and terminal-session eviction. from __future__ import annotations +import asyncio import errno import logging import re @@ -711,3 +712,46 @@ def test_concurrent_same_scope_acquire_creates_once(monkeypatch: pytest.MonkeyPa assert len(results) == 2 and results[0] == results[1] assert len(sdk.create_calls) == 1 provider.shutdown() + + +@pytest.mark.asyncio +async def test_cancelled_acquire_async_serializes_retry_behind_abandoned_body(monkeypatch: pytest.MonkeyPatch) -> None: + """A cancelled acquire_async abandons the body thread, not the lock. + + Regression test (#4741): the serializer hold must follow the abandoned + body to completion, so a retry for the same scope serializes behind it + instead of overlapping it and creating a duplicate, untracked remote. + """ + provider, sdk = _install(monkeypatch) + started = threading.Event() + release = threading.Event() + original_create = sdk.create + + def blocking_create(image: str, **kwargs: Any) -> _FakeRemote: + started.set() + assert release.wait(timeout=10) + return original_create(image, **kwargs) + + sdk.create = blocking_create # type: ignore[method-assign] + + first = asyncio.create_task(provider.acquire_async("thread", user_id="user")) + assert await asyncio.to_thread(started.wait, 10) # body is blocked inside create() + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + + release.set() # abandoned body runs to completion and registers + second = await provider.acquire_async("thread", user_id="user") + + expected_id = provider._sandbox_id("thread", "user") + assert len(sdk.create_calls) == 1 # no duplicate remote sandbox + assert second == expected_id + assert provider._thread_sandboxes[provider._thread_key("thread", "user")] == expected_id + provider.shutdown() + + +def test_sandbox_id_matches_shared_identity(): + from deerflow.sandbox.identity import derive_sandbox_scope_token + + assert OpenSandboxProvider._sandbox_id("t-1", "u-1") == derive_sandbox_scope_token(user_id="u-1", thread_id="t-1") + assert OpenSandboxProvider._sandbox_id("t-1", "") == derive_sandbox_scope_token(user_id="", thread_id="t-1") diff --git a/backend/tests/test_sandbox_identity.py b/backend/tests/test_sandbox_identity.py new file mode 100644 index 0000000000..471b31a56c --- /dev/null +++ b/backend/tests/test_sandbox_identity.py @@ -0,0 +1,59 @@ +"""Golden-vector tests for the shared sandbox scope token (RFC #4741). + +The expected strings below are LITERALS computed from the five providers' +current inline expressions as of 2026-08-30. They pin the compatibility +contract byte-for-byte; never recompute them from the implementation. +""" + +from deerflow.sandbox.identity import ( + SANDBOX_ID_VERSION, + derive_sandbox_scope_token, + is_sandbox_scope_token, +) + + +class TestGoldenVectors: + def test_ascii(self): + assert derive_sandbox_scope_token(user_id="alice", thread_id="thread-1") == "dc977b840cb8638e" + + def test_empty_user(self): + # Tenki/OpenSandbox resolution path: user_id or "" + assert derive_sandbox_scope_token(user_id="", thread_id="t") == "983743815e8fe2ac" + + def test_empty_thread(self): + assert derive_sandbox_scope_token(user_id="u", thread_id="") == "27e14d2b41b03178" + + def test_non_ascii(self): + assert derive_sandbox_scope_token(user_id="用户", thread_id="线程") == "cfa8a5bf41ddff8d" + + def test_literal_none_string(self): + # BoxLite quirk: user_id=None renders as the literal "None" (RFC #4741 + # §2.2; pinned, NOT fixed — unifying is a separate behavior change). + assert derive_sandbox_scope_token(user_id="None", thread_id="abc") == "c0b80df1b97d187e" + + def test_version_constant(self): + assert SANDBOX_ID_VERSION == 1 + + def test_keyword_only(self): + import inspect + + sig = inspect.signature(derive_sandbox_scope_token) + for param in sig.parameters.values(): + assert param.kind is inspect.Parameter.KEYWORD_ONLY + + +class TestShapeValidation: + def test_accepts_token(self): + token = derive_sandbox_scope_token(user_id="alice", thread_id="thread-1") + assert is_sandbox_scope_token(token) is True + + def test_rejects_wrong_shape(self): + assert is_sandbox_scope_token("") is False + assert is_sandbox_scope_token("dc977b840cb8638") is False # 15 chars + assert is_sandbox_scope_token("dc977b840cb8638ee") is False # 17 chars + assert is_sandbox_scope_token("DC977B840CB8638E") is False # uppercase + assert is_sandbox_scope_token("local:alice:thread-1") is False + + def test_rejects_non_str(self): + assert is_sandbox_scope_token(None) is False + assert is_sandbox_scope_token(123) is False diff --git a/backend/tests/test_sandbox_orphan_reconciliation.py b/backend/tests/test_sandbox_orphan_reconciliation.py index a68de083cc..48292eea8f 100644 --- a/backend/tests/test_sandbox_orphan_reconciliation.py +++ b/backend/tests/test_sandbox_orphan_reconciliation.py @@ -421,6 +421,7 @@ def _make_provider_for_reconciliation(tmp_path=None, *, worker_id: str = "worker ``tmp_path`` is accepted and ignored: ownership no longer lives on disk. """ from deerflow.config.sandbox_config import SandboxOwnershipConfig + from deerflow.sandbox.acquire_serialization import AcquireSerializer aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider) @@ -428,7 +429,7 @@ def _make_provider_for_reconciliation(tmp_path=None, *, worker_id: str = "worker provider._sandboxes = {} provider._sandbox_infos = {} provider._thread_sandboxes = {} - provider._thread_locks = {} + provider._acquire_serializer = AcquireSerializer(thread_name_prefix="aio-sandbox-lock-wait") provider._last_activity = {} provider._warm_pool = {} provider._active_sandbox_identity = {} diff --git a/backend/tests/test_tenki_provider.py b/backend/tests/test_tenki_provider.py index 77547f302f..b6c1b8d109 100644 --- a/backend/tests/test_tenki_provider.py +++ b/backend/tests/test_tenki_provider.py @@ -941,3 +941,10 @@ def test_integration_real_sandbox(monkeypatch): assert box.read_file("/mnt/user-data/workspace/it.txt") == "tenki-e2e" finally: provider.shutdown() + + +def test_sandbox_id_matches_shared_identity(): + from deerflow.sandbox.identity import derive_sandbox_scope_token + + assert TenkiSandboxProvider._sandbox_id("t-1", "u-1") == derive_sandbox_scope_token(user_id="u-1", thread_id="t-1") + assert TenkiSandboxProvider._sandbox_id("t-1", "") == derive_sandbox_scope_token(user_id="", thread_id="t-1")