mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
* 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.
This commit is contained in:
Zeren Wang
GitHub
parent
bf740ffa90
commit
bb75f8d736
@ -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()
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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")
|
||||
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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)
|
||||
43
backend/packages/harness/deerflow/sandbox/identity.py
Normal file
43
backend/packages/harness/deerflow/sandbox/identity.py
Normal file
@ -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
|
||||
@ -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 = {}
|
||||
|
||||
56
backend/tests/blocking_io/test_opensandbox_acquire.py
Normal file
56
backend/tests/blocking_io/test_opensandbox_acquire.py
Normal file
@ -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()
|
||||
@ -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 = {}
|
||||
|
||||
270
backend/tests/test_acquire_serialization.py
Normal file
270
backend/tests/test_acquire_serialization.py
Normal file
@ -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)
|
||||
@ -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")
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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")
|
||||
|
||||
59
backend/tests/test_sandbox_identity.py
Normal file
59
backend/tests/test_sandbox_identity.py
Normal file
@ -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
|
||||
@ -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 = {}
|
||||
|
||||
@ -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")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user