mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-22 06:00:21 +00:00
fix: align sandbox artifact mounts with channel user (#3729)
* fix: align sandbox artifact mounts with channel user * fix: clean up sandbox user-scoped ids
This commit is contained in:
Huixin615
GitHub
parent
1ac9c9edee
commit
4f192cb469
@ -412,7 +412,7 @@ class FeishuChannel(Channel):
|
||||
|
||||
try:
|
||||
sandbox_provider = get_sandbox_provider()
|
||||
sandbox_id = sandbox_provider.acquire(thread_id)
|
||||
sandbox_id = sandbox_provider.acquire(thread_id, user_id=effective_user_id)
|
||||
if sandbox_id != "local":
|
||||
sandbox = sandbox_provider.get(sandbox_id)
|
||||
if sandbox is None:
|
||||
|
||||
@ -227,10 +227,11 @@ async def upload_files(
|
||||
raise HTTPException(status_code=413, detail=f"Too many files: maximum is {limits.max_files}")
|
||||
|
||||
try:
|
||||
uploads_dir = ensure_uploads_dir(thread_id)
|
||||
effective_user_id = get_effective_user_id()
|
||||
uploads_dir = ensure_uploads_dir(thread_id, user_id=effective_user_id)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=get_effective_user_id())
|
||||
sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=effective_user_id)
|
||||
uploaded_files = []
|
||||
written_paths = []
|
||||
sandbox_sync_targets = []
|
||||
@ -245,7 +246,7 @@ async def upload_files(
|
||||
sync_to_sandbox = not _uses_thread_data_mounts(sandbox_provider)
|
||||
sandbox = None
|
||||
if sync_to_sandbox:
|
||||
sandbox_id = sandbox_provider.acquire(thread_id)
|
||||
sandbox_id = sandbox_provider.acquire(thread_id, user_id=effective_user_id)
|
||||
sandbox = sandbox_provider.get(sandbox_id)
|
||||
if sandbox is None:
|
||||
raise HTTPException(status_code=500, detail="Failed to acquire sandbox")
|
||||
|
||||
@ -133,8 +133,8 @@ class AioSandboxProvider(SandboxProvider):
|
||||
self._lock = threading.Lock()
|
||||
self._sandboxes: dict[str, AioSandbox] = {} # sandbox_id -> AioSandbox instance
|
||||
self._sandbox_infos: dict[str, SandboxInfo] = {} # sandbox_id -> SandboxInfo (for destroy)
|
||||
self._thread_sandboxes: dict[str, str] = {} # thread_id -> sandbox_id
|
||||
self._thread_locks: dict[str, threading.Lock] = {} # thread_id -> in-process lock
|
||||
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._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).
|
||||
@ -276,22 +276,30 @@ class AioSandboxProvider(SandboxProvider):
|
||||
# ── Deterministic ID ─────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _deterministic_sandbox_id(thread_id: str) -> str:
|
||||
"""Generate a deterministic sandbox ID from a thread ID.
|
||||
def _effective_acquire_user_id(user_id: str | None) -> str:
|
||||
return user_id or get_effective_user_id()
|
||||
|
||||
Ensures all processes derive the same sandbox_id for a given thread,
|
||||
enabling cross-process sandbox discovery without shared memory.
|
||||
@staticmethod
|
||||
def _thread_key(thread_id: str, user_id: str) -> tuple[str, str]:
|
||||
return (user_id, thread_id)
|
||||
|
||||
@staticmethod
|
||||
def _deterministic_sandbox_id(thread_id: str, user_id: str) -> str:
|
||||
"""Generate a deterministic sandbox ID from user/thread scope.
|
||||
|
||||
Includes user_id so a previously-created default-bucket sandbox cannot be
|
||||
reused for an auth/channel run that should mount a user-scoped bucket.
|
||||
"""
|
||||
return hashlib.sha256(thread_id.encode()).hexdigest()[:8]
|
||||
return hashlib.sha256(f"{user_id}:{thread_id}".encode()).hexdigest()[:8]
|
||||
|
||||
# ── Mount helpers ────────────────────────────────────────────────────
|
||||
|
||||
def _get_extra_mounts(self, thread_id: str | None) -> list[tuple[str, str, bool]]:
|
||||
def _get_extra_mounts(self, thread_id: str | None, *, user_id: str | None = None) -> list[tuple[str, str, bool]]:
|
||||
"""Collect all extra mounts for a sandbox (thread-specific + skills)."""
|
||||
mounts: list[tuple[str, str, bool]] = []
|
||||
|
||||
if thread_id:
|
||||
mounts.extend(self._get_thread_mounts(thread_id))
|
||||
mounts.extend(self._get_thread_mounts(thread_id, user_id=user_id))
|
||||
logger.info(f"Adding thread mounts for thread {thread_id}: {mounts}")
|
||||
|
||||
skills_mount = self._get_skills_mount()
|
||||
@ -302,7 +310,7 @@ class AioSandboxProvider(SandboxProvider):
|
||||
return mounts
|
||||
|
||||
@staticmethod
|
||||
def _get_thread_mounts(thread_id: str) -> list[tuple[str, str, bool]]:
|
||||
def _get_thread_mounts(thread_id: str, *, user_id: str | None = None) -> list[tuple[str, str, bool]]:
|
||||
"""Get volume mounts for a thread's data directories.
|
||||
|
||||
Creates directories if they don't exist (lazy initialization).
|
||||
@ -310,16 +318,16 @@ class AioSandboxProvider(SandboxProvider):
|
||||
mounted Docker socket (DooD), the host Docker daemon can resolve the paths.
|
||||
"""
|
||||
paths = get_paths()
|
||||
user_id = get_effective_user_id()
|
||||
paths.ensure_thread_dirs(thread_id, user_id=user_id)
|
||||
effective_user_id = AioSandboxProvider._effective_acquire_user_id(user_id)
|
||||
paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
|
||||
|
||||
return [
|
||||
(paths.host_sandbox_work_dir(thread_id, user_id=user_id), f"{VIRTUAL_PATH_PREFIX}/workspace", False),
|
||||
(paths.host_sandbox_uploads_dir(thread_id, user_id=user_id), f"{VIRTUAL_PATH_PREFIX}/uploads", False),
|
||||
(paths.host_sandbox_outputs_dir(thread_id, user_id=user_id), f"{VIRTUAL_PATH_PREFIX}/outputs", False),
|
||||
(paths.host_sandbox_work_dir(thread_id, user_id=effective_user_id), f"{VIRTUAL_PATH_PREFIX}/workspace", False),
|
||||
(paths.host_sandbox_uploads_dir(thread_id, user_id=effective_user_id), f"{VIRTUAL_PATH_PREFIX}/uploads", False),
|
||||
(paths.host_sandbox_outputs_dir(thread_id, user_id=effective_user_id), f"{VIRTUAL_PATH_PREFIX}/outputs", False),
|
||||
# ACP workspace: read-only inside the sandbox (lead agent reads results;
|
||||
# the ACP subprocess writes from the host side, not from within the container).
|
||||
(paths.host_acp_workspace_dir(thread_id, user_id=user_id), "/mnt/acp-workspace", True),
|
||||
(paths.host_acp_workspace_dir(thread_id, user_id=effective_user_id), "/mnt/acp-workspace", True),
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
@ -448,31 +456,34 @@ class AioSandboxProvider(SandboxProvider):
|
||||
|
||||
# ── Thread locking (in-process) ──────────────────────────────────────
|
||||
|
||||
def _get_thread_lock(self, thread_id: str) -> threading.Lock:
|
||||
"""Get or create an in-process lock for a specific thread_id."""
|
||||
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 thread_id not in self._thread_locks:
|
||||
self._thread_locks[thread_id] = threading.Lock()
|
||||
return self._thread_locks[thread_id]
|
||||
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) -> str:
|
||||
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) if thread_id else str(uuid.uuid4())[:8]
|
||||
return self._deterministic_sandbox_id(thread_id, self._effective_acquire_user_id(user_id)) if thread_id else str(uuid.uuid4())[:8]
|
||||
|
||||
def _reuse_in_process_sandbox(self, thread_id: str | None, *, post_lock: bool = False) -> str | None:
|
||||
def _reuse_in_process_sandbox(self, thread_id: str | None, *, user_id: str | None = None, post_lock: bool = False) -> str | None:
|
||||
"""Reuse an active in-process sandbox for a thread if one is still tracked."""
|
||||
if thread_id is None:
|
||||
return None
|
||||
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
key = self._thread_key(thread_id, effective_user_id)
|
||||
with self._lock:
|
||||
if thread_id not in self._thread_sandboxes:
|
||||
if key not in self._thread_sandboxes:
|
||||
return None
|
||||
|
||||
existing_id = self._thread_sandboxes[thread_id]
|
||||
existing_id = self._thread_sandboxes[key]
|
||||
if existing_id in self._sandboxes:
|
||||
info = self._sandbox_infos.get(existing_id)
|
||||
else:
|
||||
del self._thread_sandboxes[thread_id]
|
||||
del self._thread_sandboxes[key]
|
||||
return None
|
||||
|
||||
alive = self._check_tracked_sandbox_alive(existing_id, info) if info is not None else True
|
||||
@ -485,22 +496,31 @@ class AioSandboxProvider(SandboxProvider):
|
||||
return None
|
||||
|
||||
with self._lock:
|
||||
if self._thread_sandboxes.get(thread_id) != existing_id:
|
||||
if self._thread_sandboxes.get(key) != existing_id:
|
||||
return None
|
||||
if existing_id not in self._sandboxes:
|
||||
self._thread_sandboxes.pop(thread_id, None)
|
||||
self._thread_sandboxes.pop(key, None)
|
||||
return None
|
||||
|
||||
suffix = " (post-lock check)" if post_lock else ""
|
||||
logger.info(f"Reusing in-process sandbox {existing_id} for thread {thread_id}{suffix}")
|
||||
logger.info(f"Reusing in-process sandbox {existing_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
|
||||
self._last_activity[existing_id] = time.time()
|
||||
return existing_id
|
||||
|
||||
def _reclaim_warm_pool_sandbox(self, thread_id: str | None, sandbox_id: str, *, post_lock: bool = False) -> str | None:
|
||||
def _reclaim_warm_pool_sandbox(
|
||||
self,
|
||||
thread_id: str | None,
|
||||
sandbox_id: str,
|
||||
*,
|
||||
user_id: str | None = None,
|
||||
post_lock: bool = False,
|
||||
) -> str | None:
|
||||
"""Promote a warm-pool sandbox back to active tracking if available."""
|
||||
if thread_id is None:
|
||||
return None
|
||||
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
key = self._thread_key(thread_id, effective_user_id)
|
||||
with self._lock:
|
||||
if sandbox_id not in self._warm_pool:
|
||||
return None
|
||||
@ -525,29 +545,35 @@ class AioSandboxProvider(SandboxProvider):
|
||||
self._sandboxes[sandbox_id] = sandbox
|
||||
self._sandbox_infos[sandbox_id] = info
|
||||
self._last_activity[sandbox_id] = time.time()
|
||||
self._thread_sandboxes[thread_id] = sandbox_id
|
||||
self._thread_sandboxes[key] = sandbox_id
|
||||
|
||||
suffix = " (post-lock check)" if post_lock else f" at {info.sandbox_url}"
|
||||
logger.info(f"Reclaimed warm-pool sandbox {sandbox_id} for thread {thread_id}{suffix}")
|
||||
logger.info(f"Reclaimed warm-pool sandbox {sandbox_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
|
||||
return sandbox_id
|
||||
|
||||
def _recheck_cached_sandbox(self, thread_id: str, sandbox_id: str) -> str | None:
|
||||
def _recheck_cached_sandbox(self, thread_id: str, sandbox_id: str, *, user_id: str) -> str | None:
|
||||
"""Re-check in-memory caches after acquiring the cross-process file lock."""
|
||||
return self._reuse_in_process_sandbox(thread_id, post_lock=True) or self._reclaim_warm_pool_sandbox(thread_id, sandbox_id, post_lock=True)
|
||||
return self._reuse_in_process_sandbox(thread_id, user_id=user_id, post_lock=True) or self._reclaim_warm_pool_sandbox(
|
||||
thread_id,
|
||||
sandbox_id,
|
||||
user_id=user_id,
|
||||
post_lock=True,
|
||||
)
|
||||
|
||||
def _register_discovered_sandbox(self, thread_id: str, info: SandboxInfo) -> str:
|
||||
def _register_discovered_sandbox(self, thread_id: str, info: SandboxInfo, *, user_id: str) -> str:
|
||||
"""Track a sandbox discovered through the backend."""
|
||||
sandbox = AioSandbox(id=info.sandbox_id, base_url=info.sandbox_url)
|
||||
key = self._thread_key(thread_id, user_id)
|
||||
with self._lock:
|
||||
self._sandboxes[info.sandbox_id] = sandbox
|
||||
self._sandbox_infos[info.sandbox_id] = info
|
||||
self._last_activity[info.sandbox_id] = time.time()
|
||||
self._thread_sandboxes[thread_id] = info.sandbox_id
|
||||
self._thread_sandboxes[key] = info.sandbox_id
|
||||
|
||||
logger.info(f"Discovered existing sandbox {info.sandbox_id} for thread {thread_id} at {info.sandbox_url}")
|
||||
logger.info(f"Discovered existing sandbox {info.sandbox_id} for user/thread {user_id}/{thread_id} at {info.sandbox_url}")
|
||||
return info.sandbox_id
|
||||
|
||||
def _register_created_sandbox(self, thread_id: str | None, sandbox_id: str, info: SandboxInfo) -> str:
|
||||
def _register_created_sandbox(self, thread_id: str | None, sandbox_id: str, info: SandboxInfo, *, user_id: str | None = None) -> str:
|
||||
"""Track a newly-created sandbox in the active maps."""
|
||||
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
|
||||
with self._lock:
|
||||
@ -555,7 +581,7 @@ class AioSandboxProvider(SandboxProvider):
|
||||
self._sandbox_infos[sandbox_id] = info
|
||||
self._last_activity[sandbox_id] = time.time()
|
||||
if thread_id:
|
||||
self._thread_sandboxes[thread_id] = sandbox_id
|
||||
self._thread_sandboxes[self._thread_key(thread_id, self._effective_acquire_user_id(user_id))] = sandbox_id
|
||||
|
||||
logger.info(f"Created sandbox {sandbox_id} for thread {thread_id} at {info.sandbox_url}")
|
||||
return sandbox_id
|
||||
@ -581,7 +607,7 @@ class AioSandboxProvider(SandboxProvider):
|
||||
checked. This prevents a stale health-check result from deleting a
|
||||
freshly recreated sandbox with the same deterministic id.
|
||||
"""
|
||||
thread_ids_to_remove: list[str] = []
|
||||
thread_keys_to_remove: list[tuple[str, str]] = []
|
||||
|
||||
with self._lock:
|
||||
active_info = self._sandbox_infos.get(sandbox_id)
|
||||
@ -592,9 +618,9 @@ class AioSandboxProvider(SandboxProvider):
|
||||
|
||||
sandbox = self._sandboxes.pop(sandbox_id, None)
|
||||
info = self._sandbox_infos.pop(sandbox_id, None)
|
||||
thread_ids_to_remove = [tid for tid, sid in self._thread_sandboxes.items() if sid == sandbox_id]
|
||||
for tid in thread_ids_to_remove:
|
||||
del self._thread_sandboxes[tid]
|
||||
thread_keys_to_remove = [key for key, sid in self._thread_sandboxes.items() if sid == sandbox_id]
|
||||
for key in thread_keys_to_remove:
|
||||
del self._thread_sandboxes[key]
|
||||
self._last_activity.pop(sandbox_id, None)
|
||||
if info is None and sandbox_id in self._warm_pool:
|
||||
info, _ = self._warm_pool.pop(sandbox_id)
|
||||
@ -644,7 +670,7 @@ class AioSandboxProvider(SandboxProvider):
|
||||
|
||||
# ── Core: acquire / get / release / shutdown ─────────────────────────
|
||||
|
||||
def acquire(self, thread_id: str | None = None) -> str:
|
||||
def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
"""Acquire a sandbox environment and return its ID.
|
||||
|
||||
For the same thread_id, this method will return the same sandbox_id
|
||||
@ -659,31 +685,33 @@ class AioSandboxProvider(SandboxProvider):
|
||||
Returns:
|
||||
The ID of the acquired sandbox environment.
|
||||
"""
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
if thread_id:
|
||||
thread_lock = self._get_thread_lock(thread_id)
|
||||
thread_lock = self._get_thread_lock(thread_id, effective_user_id)
|
||||
with thread_lock:
|
||||
return self._acquire_internal(thread_id)
|
||||
return self._acquire_internal(thread_id, user_id=effective_user_id)
|
||||
else:
|
||||
return self._acquire_internal(thread_id)
|
||||
return self._acquire_internal(thread_id, user_id=effective_user_id)
|
||||
|
||||
async def acquire_async(self, thread_id: str | None = None) -> str:
|
||||
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.
|
||||
|
||||
Mirrors ``acquire()`` while keeping blocking backend operations off the
|
||||
event loop and using async-native readiness polling for newly created
|
||||
sandboxes.
|
||||
"""
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
if thread_id:
|
||||
thread_lock = self._get_thread_lock(thread_id)
|
||||
thread_lock = self._get_thread_lock(thread_id, effective_user_id)
|
||||
await _acquire_thread_lock_async(thread_lock)
|
||||
try:
|
||||
return await self._acquire_internal_async(thread_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)
|
||||
return await self._acquire_internal_async(thread_id, user_id=effective_user_id)
|
||||
|
||||
def _acquire_internal(self, thread_id: str | None) -> str:
|
||||
def _acquire_internal(self, thread_id: str | None, *, user_id: str) -> str:
|
||||
"""Internal sandbox acquisition with two-layer consistency.
|
||||
|
||||
Layer 1: In-process cache (fastest, covers same-process repeated access)
|
||||
@ -691,15 +719,15 @@ class AioSandboxProvider(SandboxProvider):
|
||||
sandbox_id is deterministic from thread_id so no shared state file
|
||||
is needed — any process can derive the same container name)
|
||||
"""
|
||||
cached_id = self._reuse_in_process_sandbox(thread_id)
|
||||
cached_id = self._reuse_in_process_sandbox(thread_id, user_id=user_id)
|
||||
if cached_id is not None:
|
||||
return cached_id
|
||||
|
||||
# Deterministic ID for thread-specific, random for anonymous
|
||||
sandbox_id = self._sandbox_id_for_thread(thread_id)
|
||||
sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
|
||||
|
||||
# ── Layer 1.5: Warm pool (container still running, no cold-start) ──
|
||||
reclaimed_id = self._reclaim_warm_pool_sandbox(thread_id, sandbox_id)
|
||||
reclaimed_id = self._reclaim_warm_pool_sandbox(thread_id, sandbox_id, user_id=user_id)
|
||||
if reclaimed_id is not None:
|
||||
return reclaimed_id
|
||||
|
||||
@ -708,40 +736,40 @@ class AioSandboxProvider(SandboxProvider):
|
||||
# for the same thread_id serialize here: the second process will discover
|
||||
# the container started by the first instead of hitting a name-conflict.
|
||||
if thread_id:
|
||||
return self._discover_or_create_with_lock(thread_id, sandbox_id)
|
||||
return self._discover_or_create_with_lock(thread_id, sandbox_id, user_id=user_id)
|
||||
|
||||
return self._create_sandbox(thread_id, sandbox_id)
|
||||
return self._create_sandbox(thread_id, sandbox_id, user_id=user_id)
|
||||
|
||||
async def _acquire_internal_async(self, thread_id: str | None) -> str:
|
||||
async def _acquire_internal_async(self, thread_id: str | None, *, user_id: str) -> str:
|
||||
"""Async counterpart to ``_acquire_internal``."""
|
||||
cached_id = await asyncio.to_thread(self._reuse_in_process_sandbox, thread_id)
|
||||
cached_id = await asyncio.to_thread(self._reuse_in_process_sandbox, thread_id, user_id=user_id)
|
||||
if cached_id is not None:
|
||||
return cached_id
|
||||
|
||||
# Deterministic ID for thread-specific, random for anonymous
|
||||
sandbox_id = self._sandbox_id_for_thread(thread_id)
|
||||
sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
|
||||
|
||||
# ── Layer 1.5: Warm pool (container still running, no cold-start) ──
|
||||
reclaimed_id = await asyncio.to_thread(self._reclaim_warm_pool_sandbox, thread_id, sandbox_id)
|
||||
reclaimed_id = await asyncio.to_thread(self._reclaim_warm_pool_sandbox, thread_id, sandbox_id, user_id=user_id)
|
||||
if reclaimed_id is not None:
|
||||
return reclaimed_id
|
||||
|
||||
# ── Layer 2: Backend discovery + create (protected by cross-process lock) ──
|
||||
if thread_id:
|
||||
return await self._discover_or_create_with_lock_async(thread_id, sandbox_id)
|
||||
return await self._discover_or_create_with_lock_async(thread_id, sandbox_id, user_id=user_id)
|
||||
|
||||
return await self._create_sandbox_async(thread_id, sandbox_id)
|
||||
return await self._create_sandbox_async(thread_id, sandbox_id, user_id=user_id)
|
||||
|
||||
def _discover_or_create_with_lock(self, thread_id: str, sandbox_id: str) -> str:
|
||||
def _discover_or_create_with_lock(self, thread_id: str, sandbox_id: str, *, user_id: str | None = None) -> str:
|
||||
"""Discover an existing sandbox or create a new one under a cross-process file lock.
|
||||
|
||||
The file lock serializes concurrent sandbox creation for the same thread_id
|
||||
across multiple processes, preventing container-name conflicts.
|
||||
"""
|
||||
paths = get_paths()
|
||||
user_id = get_effective_user_id()
|
||||
paths.ensure_thread_dirs(thread_id, user_id=user_id)
|
||||
lock_path = paths.thread_dir(thread_id, user_id=user_id) / f"{sandbox_id}.lock"
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
|
||||
lock_path = paths.thread_dir(thread_id, user_id=effective_user_id) / f"{sandbox_id}.lock"
|
||||
|
||||
with open(lock_path, "a", encoding="utf-8") as lock_file:
|
||||
locked = False
|
||||
@ -750,26 +778,26 @@ class AioSandboxProvider(SandboxProvider):
|
||||
locked = True
|
||||
# Re-check in-process caches under the file lock in case another
|
||||
# thread in this process won the race while we were waiting.
|
||||
cached_id = self._recheck_cached_sandbox(thread_id, sandbox_id)
|
||||
cached_id = self._recheck_cached_sandbox(thread_id, sandbox_id, user_id=effective_user_id)
|
||||
if cached_id is not None:
|
||||
return cached_id
|
||||
|
||||
# Backend discovery: another process may have created the container.
|
||||
discovered = self._backend.discover(sandbox_id)
|
||||
if discovered is not None:
|
||||
return self._register_discovered_sandbox(thread_id, discovered)
|
||||
return self._register_discovered_sandbox(thread_id, discovered, user_id=effective_user_id)
|
||||
|
||||
return self._create_sandbox(thread_id, sandbox_id)
|
||||
return self._create_sandbox(thread_id, sandbox_id, user_id=effective_user_id)
|
||||
finally:
|
||||
if locked:
|
||||
_unlock_file(lock_file)
|
||||
|
||||
async def _discover_or_create_with_lock_async(self, thread_id: str, sandbox_id: str) -> str:
|
||||
async def _discover_or_create_with_lock_async(self, thread_id: str, sandbox_id: str, *, user_id: str | None = None) -> str:
|
||||
"""Async counterpart to ``_discover_or_create_with_lock``."""
|
||||
paths = get_paths()
|
||||
user_id = get_effective_user_id()
|
||||
await asyncio.to_thread(paths.ensure_thread_dirs, thread_id, user_id=user_id)
|
||||
lock_path = paths.thread_dir(thread_id, user_id=user_id) / f"{sandbox_id}.lock"
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
await asyncio.to_thread(paths.ensure_thread_dirs, thread_id, user_id=effective_user_id)
|
||||
lock_path = paths.thread_dir(thread_id, user_id=effective_user_id) / f"{sandbox_id}.lock"
|
||||
|
||||
lock_file = await asyncio.to_thread(_open_lock_file, lock_path)
|
||||
locked = False
|
||||
@ -778,7 +806,7 @@ class AioSandboxProvider(SandboxProvider):
|
||||
locked = True
|
||||
# Re-check in-process caches under the file lock in case another
|
||||
# thread in this process won the race while we were waiting.
|
||||
cached_id = await asyncio.to_thread(self._recheck_cached_sandbox, thread_id, sandbox_id)
|
||||
cached_id = await asyncio.to_thread(self._recheck_cached_sandbox, thread_id, sandbox_id, user_id=effective_user_id)
|
||||
if cached_id is not None:
|
||||
return cached_id
|
||||
|
||||
@ -786,9 +814,9 @@ class AioSandboxProvider(SandboxProvider):
|
||||
# Docker and perform a health check; keep it off the event loop.
|
||||
discovered = await asyncio.to_thread(self._backend.discover, sandbox_id)
|
||||
if discovered is not None:
|
||||
return self._register_discovered_sandbox(thread_id, discovered)
|
||||
return self._register_discovered_sandbox(thread_id, discovered, user_id=effective_user_id)
|
||||
|
||||
return await self._create_sandbox_async(thread_id, sandbox_id)
|
||||
return await self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id)
|
||||
finally:
|
||||
if locked:
|
||||
await asyncio.to_thread(_unlock_file, lock_file)
|
||||
@ -814,7 +842,7 @@ class AioSandboxProvider(SandboxProvider):
|
||||
return None
|
||||
return oldest_id
|
||||
|
||||
def _create_sandbox(self, thread_id: str | None, sandbox_id: str) -> str:
|
||||
def _create_sandbox(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
|
||||
"""Create a new sandbox via the backend.
|
||||
|
||||
Args:
|
||||
@ -827,7 +855,8 @@ class AioSandboxProvider(SandboxProvider):
|
||||
Raises:
|
||||
RuntimeError: If sandbox creation or readiness check fails.
|
||||
"""
|
||||
extra_mounts = self._get_extra_mounts(thread_id)
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
extra_mounts = self._get_extra_mounts(thread_id, user_id=effective_user_id)
|
||||
|
||||
# Enforce replicas: only warm-pool containers count toward eviction budget.
|
||||
# Active sandboxes are in use by live threads and must not be forcibly stopped.
|
||||
@ -836,18 +865,19 @@ class AioSandboxProvider(SandboxProvider):
|
||||
evicted = self._evict_oldest_warm()
|
||||
self._log_replicas_soft_cap(replicas, sandbox_id, evicted)
|
||||
|
||||
info = self._backend.create(thread_id, sandbox_id, extra_mounts=extra_mounts or None)
|
||||
info = self._backend.create(thread_id, sandbox_id, extra_mounts=extra_mounts or None, user_id=effective_user_id)
|
||||
|
||||
# Wait for sandbox to be ready
|
||||
if not wait_for_sandbox_ready(info.sandbox_url, timeout=60):
|
||||
self._backend.destroy(info)
|
||||
raise RuntimeError(f"Sandbox {sandbox_id} failed to become ready within timeout at {info.sandbox_url}")
|
||||
|
||||
return self._register_created_sandbox(thread_id, sandbox_id, info)
|
||||
return self._register_created_sandbox(thread_id, sandbox_id, info, user_id=effective_user_id)
|
||||
|
||||
async def _create_sandbox_async(self, thread_id: str | None, sandbox_id: str) -> str:
|
||||
async def _create_sandbox_async(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
|
||||
"""Async counterpart to ``_create_sandbox``."""
|
||||
extra_mounts = await asyncio.to_thread(self._get_extra_mounts, thread_id)
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
extra_mounts = await asyncio.to_thread(self._get_extra_mounts, thread_id, user_id=effective_user_id)
|
||||
|
||||
# Enforce replicas: only warm-pool containers count toward eviction budget.
|
||||
# Active sandboxes are in use by live threads and must not be forcibly stopped.
|
||||
@ -856,14 +886,14 @@ class AioSandboxProvider(SandboxProvider):
|
||||
evicted = await asyncio.to_thread(self._evict_oldest_warm)
|
||||
self._log_replicas_soft_cap(replicas, sandbox_id, evicted)
|
||||
|
||||
info = await asyncio.to_thread(self._backend.create, thread_id, sandbox_id, extra_mounts=extra_mounts or None)
|
||||
info = await asyncio.to_thread(self._backend.create, thread_id, sandbox_id, extra_mounts=extra_mounts or None, user_id=effective_user_id)
|
||||
|
||||
# Wait for sandbox to be ready without blocking the event loop.
|
||||
if not await wait_for_sandbox_ready_async(info.sandbox_url, timeout=60):
|
||||
await asyncio.to_thread(self._backend.destroy, info)
|
||||
raise RuntimeError(f"Sandbox {sandbox_id} failed to become ready within timeout at {info.sandbox_url}")
|
||||
|
||||
return self._register_created_sandbox(thread_id, sandbox_id, info)
|
||||
return self._register_created_sandbox(thread_id, sandbox_id, info, user_id=effective_user_id)
|
||||
|
||||
def get(self, sandbox_id: str) -> Sandbox | None:
|
||||
"""Get a sandbox by ID. Updates last activity timestamp.
|
||||
@ -897,14 +927,14 @@ class AioSandboxProvider(SandboxProvider):
|
||||
"""
|
||||
info = None
|
||||
sandbox = None
|
||||
thread_ids_to_remove: list[str] = []
|
||||
thread_keys_to_remove: list[tuple[str, str]] = []
|
||||
|
||||
with self._lock:
|
||||
sandbox = self._sandboxes.pop(sandbox_id, None)
|
||||
info = self._sandbox_infos.pop(sandbox_id, None)
|
||||
thread_ids_to_remove = [tid for tid, sid in self._thread_sandboxes.items() if sid == sandbox_id]
|
||||
for tid in thread_ids_to_remove:
|
||||
del self._thread_sandboxes[tid]
|
||||
thread_keys_to_remove = [key for key, sid in self._thread_sandboxes.items() if sid == sandbox_id]
|
||||
for key in thread_keys_to_remove:
|
||||
del self._thread_sandboxes[key]
|
||||
self._last_activity.pop(sandbox_id, None)
|
||||
# Park in warm pool — container keeps running
|
||||
if info and sandbox_id not in self._warm_pool:
|
||||
|
||||
@ -74,7 +74,14 @@ class SandboxBackend(ABC):
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def create(self, thread_id: str | None, sandbox_id: str, extra_mounts: list[tuple[str, str, bool]] | None = None) -> SandboxInfo:
|
||||
def create(
|
||||
self,
|
||||
thread_id: str | None,
|
||||
sandbox_id: str,
|
||||
extra_mounts: list[tuple[str, str, bool]] | None = None,
|
||||
*,
|
||||
user_id: str | None = None,
|
||||
) -> SandboxInfo:
|
||||
"""Create/provision a new sandbox.
|
||||
|
||||
Args:
|
||||
@ -82,6 +89,7 @@ class SandboxBackend(ABC):
|
||||
sandbox_id: Deterministic sandbox identifier.
|
||||
extra_mounts: Additional volume mounts as (host_path, container_path, read_only) tuples.
|
||||
Ignored by backends that don't manage containers (e.g., remote).
|
||||
user_id: User bucket that the sandbox should mount or provision for.
|
||||
|
||||
Returns:
|
||||
SandboxInfo with connection details.
|
||||
|
||||
@ -259,13 +259,22 @@ class LocalContainerBackend(SandboxBackend):
|
||||
|
||||
# ── SandboxBackend interface ──────────────────────────────────────────
|
||||
|
||||
def create(self, thread_id: str | None, sandbox_id: str, extra_mounts: list[tuple[str, str, bool]] | None = None) -> SandboxInfo:
|
||||
def create(
|
||||
self,
|
||||
thread_id: str | None,
|
||||
sandbox_id: str,
|
||||
extra_mounts: list[tuple[str, str, bool]] | None = None,
|
||||
*,
|
||||
user_id: str | None = None,
|
||||
) -> SandboxInfo:
|
||||
"""Start a new container and return its connection info.
|
||||
|
||||
Args:
|
||||
thread_id: Thread ID for which the sandbox is being created. Useful for backends that want to organize sandboxes by thread.
|
||||
sandbox_id: Deterministic sandbox identifier (used in container name).
|
||||
extra_mounts: Additional volume mounts as (host_path, container_path, read_only) tuples.
|
||||
user_id: User bucket already reflected in extra_mounts. Accepted for
|
||||
interface compatibility with remote backends.
|
||||
|
||||
Returns:
|
||||
SandboxInfo with container details.
|
||||
@ -273,6 +282,7 @@ class LocalContainerBackend(SandboxBackend):
|
||||
Raises:
|
||||
RuntimeError: If the container fails to start.
|
||||
"""
|
||||
del user_id
|
||||
container_name = f"{self._container_prefix}-{sandbox_id}"
|
||||
|
||||
# Retry loop: if Docker rejects the port (e.g. a stale container still
|
||||
|
||||
@ -62,13 +62,15 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
thread_id: str | None,
|
||||
sandbox_id: str,
|
||||
extra_mounts: list[tuple[str, str, bool]] | None = None,
|
||||
*,
|
||||
user_id: str | None = None,
|
||||
) -> SandboxInfo:
|
||||
"""Create a sandbox Pod + Service via the provisioner.
|
||||
|
||||
Calls ``POST /api/sandboxes`` which creates a dedicated Pod +
|
||||
NodePort Service in k3s.
|
||||
"""
|
||||
return self._provisioner_create(thread_id, sandbox_id, extra_mounts)
|
||||
return self._provisioner_create(thread_id, sandbox_id, extra_mounts, user_id=user_id)
|
||||
|
||||
def destroy(self, info: SandboxInfo) -> None:
|
||||
"""Destroy a sandbox Pod + Service via the provisioner."""
|
||||
@ -132,15 +134,24 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
logger.warning("Provisioner list_running failed: %s", exc)
|
||||
return []
|
||||
|
||||
def _provisioner_create(self, thread_id: str | None, sandbox_id: str, extra_mounts: list[tuple[str, str, bool]] | None = None) -> SandboxInfo:
|
||||
def _provisioner_create(
|
||||
self,
|
||||
thread_id: str | None,
|
||||
sandbox_id: str,
|
||||
extra_mounts: list[tuple[str, str, bool]] | None = None,
|
||||
*,
|
||||
user_id: str | None = None,
|
||||
) -> SandboxInfo:
|
||||
"""POST /api/sandboxes → create Pod + Service."""
|
||||
del extra_mounts
|
||||
effective_user_id = user_id or get_effective_user_id()
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{self._provisioner_url}/api/sandboxes",
|
||||
json={
|
||||
"sandbox_id": sandbox_id,
|
||||
"thread_id": thread_id,
|
||||
"user_id": get_effective_user_id(),
|
||||
"user_id": effective_user_id,
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
@ -75,7 +75,7 @@ class LocalSandboxProvider(SandboxProvider):
|
||||
"""
|
||||
self._path_mappings = self._setup_path_mappings()
|
||||
self._generic_sandbox: LocalSandbox | None = None
|
||||
self._thread_sandboxes: OrderedDict[str, LocalSandbox] = OrderedDict()
|
||||
self._thread_sandboxes: OrderedDict[tuple[str, str], LocalSandbox] = OrderedDict()
|
||||
self._max_cached_threads = max_cached_threads
|
||||
self._lock = threading.Lock()
|
||||
|
||||
@ -185,19 +185,41 @@ class LocalSandboxProvider(SandboxProvider):
|
||||
return mappings
|
||||
|
||||
@staticmethod
|
||||
def _build_thread_path_mappings(thread_id: str) -> list[PathMapping]:
|
||||
"""Build per-thread path mappings for /mnt/user-data and /mnt/acp-workspace.
|
||||
|
||||
Resolves ``user_id`` via :func:`get_effective_user_id` (the same path
|
||||
:class:`AioSandboxProvider` uses) and ensures the backing host
|
||||
directories exist before they are mapped into the sandbox view.
|
||||
"""
|
||||
from deerflow.config.paths import get_paths
|
||||
def _effective_acquire_user_id(user_id: str | None) -> str:
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
|
||||
return user_id or get_effective_user_id()
|
||||
|
||||
@staticmethod
|
||||
def _thread_key(thread_id: str, user_id: str) -> tuple[str, str]:
|
||||
return (user_id, thread_id)
|
||||
|
||||
@staticmethod
|
||||
def _sandbox_id_for_thread(thread_id: str, user_id: str) -> str:
|
||||
return f"local:{user_id}:{thread_id}"
|
||||
|
||||
@staticmethod
|
||||
def _key_from_sandbox_id(sandbox_id: str) -> tuple[str, str] | None:
|
||||
if not sandbox_id.startswith("local:"):
|
||||
return None
|
||||
value = sandbox_id[len("local:") :]
|
||||
user_id, separator, thread_id = value.partition(":")
|
||||
if not separator or not user_id or not thread_id:
|
||||
return None
|
||||
return (user_id, thread_id)
|
||||
|
||||
@staticmethod
|
||||
def _build_thread_path_mappings(thread_id: str, *, user_id: str | None = None) -> list[PathMapping]:
|
||||
"""Build per-thread path mappings for /mnt/user-data and /mnt/acp-workspace.
|
||||
|
||||
Uses the explicitly resolved user id when provided, falling back to
|
||||
:func:`get_effective_user_id` for legacy callers.
|
||||
"""
|
||||
from deerflow.config.paths import get_paths
|
||||
|
||||
paths = get_paths()
|
||||
user_id = get_effective_user_id()
|
||||
paths.ensure_thread_dirs(thread_id, user_id=user_id)
|
||||
effective_user_id = LocalSandboxProvider._effective_acquire_user_id(user_id)
|
||||
paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
|
||||
|
||||
return [
|
||||
# Aggregate parent mapping so ``ls /mnt/user-data`` and other
|
||||
@ -207,32 +229,32 @@ class LocalSandboxProvider(SandboxProvider):
|
||||
# because ``_find_path_mapping`` sorts by container_path length.
|
||||
PathMapping(
|
||||
container_path=_USER_DATA_VIRTUAL_PREFIX,
|
||||
local_path=str(paths.sandbox_user_data_dir(thread_id, user_id=user_id)),
|
||||
local_path=str(paths.sandbox_user_data_dir(thread_id, user_id=effective_user_id)),
|
||||
read_only=False,
|
||||
),
|
||||
PathMapping(
|
||||
container_path=f"{_USER_DATA_VIRTUAL_PREFIX}/workspace",
|
||||
local_path=str(paths.sandbox_work_dir(thread_id, user_id=user_id)),
|
||||
local_path=str(paths.sandbox_work_dir(thread_id, user_id=effective_user_id)),
|
||||
read_only=False,
|
||||
),
|
||||
PathMapping(
|
||||
container_path=f"{_USER_DATA_VIRTUAL_PREFIX}/uploads",
|
||||
local_path=str(paths.sandbox_uploads_dir(thread_id, user_id=user_id)),
|
||||
local_path=str(paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id)),
|
||||
read_only=False,
|
||||
),
|
||||
PathMapping(
|
||||
container_path=f"{_USER_DATA_VIRTUAL_PREFIX}/outputs",
|
||||
local_path=str(paths.sandbox_outputs_dir(thread_id, user_id=user_id)),
|
||||
local_path=str(paths.sandbox_outputs_dir(thread_id, user_id=effective_user_id)),
|
||||
read_only=False,
|
||||
),
|
||||
PathMapping(
|
||||
container_path=_ACP_WORKSPACE_VIRTUAL_PREFIX,
|
||||
local_path=str(paths.acp_workspace_dir(thread_id, user_id=user_id)),
|
||||
local_path=str(paths.acp_workspace_dir(thread_id, user_id=effective_user_id)),
|
||||
read_only=False,
|
||||
),
|
||||
]
|
||||
|
||||
def acquire(self, thread_id: str | None = None) -> str:
|
||||
def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
"""Return a sandbox id scoped to *thread_id* (or the generic singleton).
|
||||
|
||||
- ``thread_id=None`` keeps the legacy singleton with id ``"local"`` for
|
||||
@ -254,29 +276,32 @@ class LocalSandboxProvider(SandboxProvider):
|
||||
_singleton = self._generic_sandbox
|
||||
return self._generic_sandbox.id
|
||||
|
||||
effective_user_id = self._effective_acquire_user_id(user_id)
|
||||
key = self._thread_key(thread_id, effective_user_id)
|
||||
|
||||
# Fast path under lock.
|
||||
with self._lock:
|
||||
cached = self._thread_sandboxes.get(thread_id)
|
||||
cached = self._thread_sandboxes.get(key)
|
||||
if cached is not None:
|
||||
# Mark as most-recently used so frequently-touched threads
|
||||
# survive eviction.
|
||||
self._thread_sandboxes.move_to_end(thread_id)
|
||||
self._thread_sandboxes.move_to_end(key)
|
||||
return cached.id
|
||||
|
||||
# ``_build_thread_path_mappings`` touches the filesystem
|
||||
# (``ensure_thread_dirs``); release the lock during I/O.
|
||||
new_mappings = list(self._path_mappings) + self._build_thread_path_mappings(thread_id)
|
||||
new_mappings = list(self._path_mappings) + self._build_thread_path_mappings(thread_id, user_id=effective_user_id)
|
||||
|
||||
with self._lock:
|
||||
# Re-check after the lock-free I/O: another caller may have
|
||||
# populated the cache while we were computing mappings.
|
||||
cached = self._thread_sandboxes.get(thread_id)
|
||||
cached = self._thread_sandboxes.get(key)
|
||||
if cached is None:
|
||||
cached = LocalSandbox(f"local:{thread_id}", path_mappings=new_mappings)
|
||||
self._thread_sandboxes[thread_id] = cached
|
||||
cached = LocalSandbox(self._sandbox_id_for_thread(thread_id, effective_user_id), path_mappings=new_mappings)
|
||||
self._thread_sandboxes[key] = cached
|
||||
self._evict_until_within_cap_locked()
|
||||
else:
|
||||
self._thread_sandboxes.move_to_end(thread_id)
|
||||
self._thread_sandboxes.move_to_end(key)
|
||||
return cached.id
|
||||
|
||||
def _evict_until_within_cap_locked(self) -> None:
|
||||
@ -285,10 +310,11 @@ class LocalSandboxProvider(SandboxProvider):
|
||||
Caller MUST hold ``self._lock``.
|
||||
"""
|
||||
while len(self._thread_sandboxes) > self._max_cached_threads:
|
||||
evicted_thread_id, _ = self._thread_sandboxes.popitem(last=False)
|
||||
evicted_key, _ = self._thread_sandboxes.popitem(last=False)
|
||||
logger.info(
|
||||
"Evicting LocalSandbox cache entry for thread %s (cap=%d)",
|
||||
evicted_thread_id,
|
||||
"Evicting LocalSandbox cache entry for user/thread %s/%s (cap=%d)",
|
||||
evicted_key[0],
|
||||
evicted_key[1],
|
||||
self._max_cached_threads,
|
||||
)
|
||||
|
||||
@ -302,14 +328,16 @@ class LocalSandboxProvider(SandboxProvider):
|
||||
return self._generic_sandbox
|
||||
return generic
|
||||
if isinstance(sandbox_id, str) and sandbox_id.startswith("local:"):
|
||||
thread_id = sandbox_id[len("local:") :]
|
||||
key = self._key_from_sandbox_id(sandbox_id)
|
||||
if key is None:
|
||||
return None
|
||||
with self._lock:
|
||||
cached = self._thread_sandboxes.get(thread_id)
|
||||
cached = self._thread_sandboxes.get(key)
|
||||
if cached is not None:
|
||||
# Touching a thread via ``get`` (used by tools.py to look
|
||||
# up the sandbox once per tool call) promotes it in LRU
|
||||
# order so an active thread isn't evicted under load.
|
||||
self._thread_sandboxes.move_to_end(thread_id)
|
||||
self._thread_sandboxes.move_to_end(key)
|
||||
return cached
|
||||
return None
|
||||
|
||||
|
||||
@ -12,6 +12,7 @@ from langgraph.runtime import Runtime
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.thread_state import SandboxStateField, ThreadDataState
|
||||
from deerflow.runtime.user_context import resolve_runtime_user_id
|
||||
from deerflow.sandbox import get_sandbox_provider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -48,15 +49,15 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
super().__init__()
|
||||
self._lazy_init = lazy_init
|
||||
|
||||
def _acquire_sandbox(self, thread_id: str) -> str:
|
||||
def _acquire_sandbox(self, thread_id: str, *, user_id: str) -> str:
|
||||
provider = get_sandbox_provider()
|
||||
sandbox_id = provider.acquire(thread_id)
|
||||
sandbox_id = provider.acquire(thread_id, user_id=user_id)
|
||||
logger.info(f"Acquiring sandbox {sandbox_id}")
|
||||
return sandbox_id
|
||||
|
||||
async def _acquire_sandbox_async(self, thread_id: str) -> str:
|
||||
async def _acquire_sandbox_async(self, thread_id: str, *, user_id: str) -> str:
|
||||
provider = get_sandbox_provider()
|
||||
sandbox_id = await provider.acquire_async(thread_id)
|
||||
sandbox_id = await provider.acquire_async(thread_id, user_id=user_id)
|
||||
logger.info(f"Acquiring sandbox {sandbox_id}")
|
||||
return sandbox_id
|
||||
|
||||
@ -74,7 +75,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
thread_id = (runtime.context or {}).get("thread_id")
|
||||
if thread_id is None:
|
||||
return super().before_agent(state, runtime)
|
||||
sandbox_id = self._acquire_sandbox(thread_id)
|
||||
sandbox_id = self._acquire_sandbox(thread_id, user_id=resolve_runtime_user_id(runtime))
|
||||
logger.info(f"Assigned sandbox {sandbox_id} to thread {thread_id}")
|
||||
return {"sandbox": {"sandbox_id": sandbox_id}}
|
||||
return super().before_agent(state, runtime)
|
||||
@ -91,7 +92,7 @@ class SandboxMiddleware(AgentMiddleware[SandboxMiddlewareState]):
|
||||
thread_id = (runtime.context or {}).get("thread_id")
|
||||
if thread_id is None:
|
||||
return await super().abefore_agent(state, runtime)
|
||||
sandbox_id = await self._acquire_sandbox_async(thread_id)
|
||||
sandbox_id = await self._acquire_sandbox_async(thread_id, user_id=resolve_runtime_user_id(runtime))
|
||||
logger.info(f"Assigned sandbox {sandbox_id} to thread {thread_id}")
|
||||
return {"sandbox": {"sandbox_id": sandbox_id}}
|
||||
return await super().abefore_agent(state, runtime)
|
||||
|
||||
@ -13,7 +13,7 @@ class SandboxProvider(ABC):
|
||||
needs_upload_permission_adjustment: bool = True
|
||||
|
||||
@abstractmethod
|
||||
def acquire(self, thread_id: str | None = None) -> str:
|
||||
def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
"""Acquire a sandbox environment and return its ID.
|
||||
|
||||
Returns:
|
||||
@ -21,7 +21,7 @@ class SandboxProvider(ABC):
|
||||
"""
|
||||
pass
|
||||
|
||||
async def acquire_async(self, thread_id: str | None = None) -> str:
|
||||
async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
"""Acquire a sandbox without blocking the event loop.
|
||||
|
||||
Most sandbox providers expose a synchronous lifecycle API because local
|
||||
@ -29,7 +29,7 @@ class SandboxProvider(ABC):
|
||||
this method so those blocking operations run in a worker thread instead
|
||||
of stalling the event loop.
|
||||
"""
|
||||
return await asyncio.to_thread(self.acquire, thread_id)
|
||||
return await asyncio.to_thread(self.acquire, thread_id, user_id=user_id)
|
||||
|
||||
@abstractmethod
|
||||
def get(self, sandbox_id: str) -> Sandbox | None:
|
||||
|
||||
@ -12,6 +12,7 @@ from langchain.tools import tool
|
||||
from deerflow.agents.thread_state import ThreadDataState
|
||||
from deerflow.config import get_app_config
|
||||
from deerflow.config.paths import VIRTUAL_PATH_PREFIX
|
||||
from deerflow.runtime.user_context import resolve_runtime_user_id
|
||||
from deerflow.sandbox.exceptions import (
|
||||
SandboxError,
|
||||
SandboxNotFoundError,
|
||||
@ -1110,9 +1111,9 @@ def get_thread_data(runtime: Runtime | None) -> ThreadDataState | None:
|
||||
def is_local_sandbox(runtime: Runtime | None) -> bool:
|
||||
"""Check if the current sandbox is a local sandbox.
|
||||
|
||||
Accepts both the legacy generic id ``"local"`` (acquire with no thread
|
||||
context) and the per-thread id format ``"local:{thread_id}"`` produced by
|
||||
:meth:`LocalSandboxProvider.acquire` once a thread is known.
|
||||
Accepts both the generic id ``"local"`` (acquire with no thread context)
|
||||
and the per-thread id format ``"local:{user_id}:{thread_id}"`` produced
|
||||
by :meth:`LocalSandboxProvider.acquire` once a thread is known.
|
||||
"""
|
||||
if runtime is None:
|
||||
return False
|
||||
@ -1200,7 +1201,7 @@ def ensure_sandbox_initialized(runtime: Runtime | None = None) -> Sandbox:
|
||||
raise SandboxRuntimeError("Thread ID not available in runtime context")
|
||||
|
||||
provider = get_sandbox_provider()
|
||||
sandbox_id = provider.acquire(thread_id)
|
||||
sandbox_id = provider.acquire(thread_id, user_id=resolve_runtime_user_id(runtime))
|
||||
|
||||
# Update runtime state - this persists across tool calls
|
||||
runtime.state["sandbox"] = {"sandbox_id": sandbox_id}
|
||||
@ -1245,7 +1246,7 @@ async def ensure_sandbox_initialized_async(runtime: Runtime | None = None) -> Sa
|
||||
raise SandboxRuntimeError("Thread ID not available in runtime context")
|
||||
|
||||
provider = get_sandbox_provider()
|
||||
sandbox_id = await provider.acquire_async(thread_id)
|
||||
sandbox_id = await provider.acquire_async(thread_id, user_id=resolve_runtime_user_id(runtime))
|
||||
|
||||
runtime.state["sandbox"] = {"sandbox_id": sandbox_id}
|
||||
|
||||
|
||||
@ -86,6 +86,20 @@ def test_get_thread_mounts_includes_user_data_dirs(tmp_path, monkeypatch):
|
||||
assert "/mnt/user-data/outputs" in container_paths
|
||||
|
||||
|
||||
def test_get_thread_mounts_uses_explicit_user_id(tmp_path, monkeypatch):
|
||||
"""Channel runs must mount the same user bucket used for artifact delivery."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
|
||||
monkeypatch.setattr(aio_mod, "get_effective_user_id", lambda: "default")
|
||||
|
||||
mounts = aio_mod.AioSandboxProvider._get_thread_mounts("thread-4", user_id="ou-user")
|
||||
container_paths = {container_path: host_path for host_path, container_path, _ in mounts}
|
||||
|
||||
assert container_paths["/mnt/user-data/workspace"] == str(tmp_path / "users" / "ou-user" / "threads" / "thread-4" / "user-data" / "workspace")
|
||||
assert container_paths["/mnt/user-data/uploads"] == str(tmp_path / "users" / "ou-user" / "threads" / "thread-4" / "user-data" / "uploads")
|
||||
assert container_paths["/mnt/user-data/outputs"] == str(tmp_path / "users" / "ou-user" / "threads" / "thread-4" / "user-data" / "outputs")
|
||||
|
||||
|
||||
def test_join_host_path_preserves_windows_drive_letter_style():
|
||||
base = r"C:\Users\demo\deer-flow\backend\.deer-flow"
|
||||
|
||||
@ -172,12 +186,12 @@ async def test_acquire_async_uses_async_readiness_polling(monkeypatch):
|
||||
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("sync readiness should not be used")),
|
||||
)
|
||||
|
||||
sandbox_id = await provider._create_sandbox_async("thread-async", "sandbox-async")
|
||||
sandbox_id = await provider._create_sandbox_async("thread-async", "sandbox-async", user_id="user-async")
|
||||
|
||||
assert sandbox_id == "sandbox-async"
|
||||
assert async_readiness_calls == [("http://sandbox", 60)]
|
||||
assert provider._backend.destroy.call_count == 0
|
||||
assert provider._thread_sandboxes["thread-async"] == "sandbox-async"
|
||||
assert provider._thread_sandboxes[("user-async", "thread-async")] == "sandbox-async"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -192,7 +206,7 @@ async def test_discover_or_create_with_lock_async_offloads_lock_file_open_and_cl
|
||||
provider._thread_locks = {}
|
||||
provider._warm_pool = {}
|
||||
provider._sandbox_infos = {}
|
||||
provider._thread_sandboxes = {"thread-async-lock": "sandbox-async-lock"}
|
||||
provider._thread_sandboxes = {("default", "thread-async-lock"): "sandbox-async-lock"}
|
||||
provider._sandboxes = {"sandbox-async-lock": aio_mod.AioSandbox(id="sandbox-async-lock", base_url="http://sandbox")}
|
||||
provider._last_activity = {}
|
||||
provider._lock = aio_mod.threading.Lock()
|
||||
@ -208,7 +222,7 @@ async def test_discover_or_create_with_lock_async_offloads_lock_file_open_and_cl
|
||||
|
||||
monkeypatch.setattr(aio_mod.asyncio, "to_thread", fake_to_thread)
|
||||
|
||||
sandbox_id = await provider._discover_or_create_with_lock_async("thread-async-lock", "sandbox-async-lock")
|
||||
sandbox_id = await provider._discover_or_create_with_lock_async("thread-async-lock", "sandbox-async-lock", user_id="default")
|
||||
|
||||
assert sandbox_id == "sandbox-async-lock"
|
||||
assert aio_mod._open_lock_file in to_thread_calls
|
||||
@ -246,10 +260,10 @@ 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)
|
||||
thread_lock = provider._get_thread_lock(thread_id, "default")
|
||||
thread_lock.acquire()
|
||||
|
||||
task = asyncio.create_task(provider.acquire_async(thread_id))
|
||||
task = asyncio.create_task(provider.acquire_async(thread_id, user_id="default"))
|
||||
await asyncio.sleep(0.05)
|
||||
task.cancel()
|
||||
|
||||
@ -282,18 +296,19 @@ async def test_acquire_async_cancelled_waiter_does_not_block_successor(tmp_path,
|
||||
provider._last_activity = {}
|
||||
provider._lock = aio_mod.threading.Lock()
|
||||
|
||||
async def fake_acquire_internal_async(thread_id: str | None) -> str:
|
||||
async def fake_acquire_internal_async(thread_id: str | None, *, user_id: str) -> str:
|
||||
assert thread_id == "thread-successor-lock"
|
||||
assert user_id == "default"
|
||||
await asyncio.sleep(0)
|
||||
return "sandbox-successor"
|
||||
|
||||
monkeypatch.setattr(provider, "_acquire_internal_async", fake_acquire_internal_async)
|
||||
|
||||
thread_id = "thread-successor-lock"
|
||||
thread_lock = provider._get_thread_lock(thread_id)
|
||||
thread_lock = provider._get_thread_lock(thread_id, "default")
|
||||
thread_lock.acquire()
|
||||
|
||||
cancelled_waiter = asyncio.create_task(provider.acquire_async(thread_id))
|
||||
cancelled_waiter = asyncio.create_task(provider.acquire_async(thread_id, user_id="default"))
|
||||
await asyncio.sleep(0.05)
|
||||
cancelled_waiter.cancel()
|
||||
try:
|
||||
@ -301,7 +316,7 @@ async def test_acquire_async_cancelled_waiter_does_not_block_successor(tmp_path,
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
live_waiter = asyncio.create_task(provider.acquire_async(thread_id))
|
||||
live_waiter = asyncio.create_task(provider.acquire_async(thread_id, user_id="default"))
|
||||
thread_lock.release()
|
||||
|
||||
assert await asyncio.wait_for(live_waiter, timeout=1) == "sandbox-successor"
|
||||
@ -322,7 +337,7 @@ async def test_acquire_internal_async_offloads_cached_reuse_health_check(tmp_pat
|
||||
"""Async cached reuse must keep backend health checks off the event loop."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
provider, _sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "sandbox-cached-async")
|
||||
provider._thread_sandboxes = {"thread-cached-async": "sandbox-cached-async"}
|
||||
provider._thread_sandboxes = {("default", "thread-cached-async"): "sandbox-cached-async"}
|
||||
provider._backend.is_alive = MagicMock(return_value=True)
|
||||
|
||||
to_thread_calls: list[tuple[object, tuple[object, ...]]] = []
|
||||
@ -333,7 +348,7 @@ async def test_acquire_internal_async_offloads_cached_reuse_health_check(tmp_pat
|
||||
|
||||
monkeypatch.setattr(aio_mod.asyncio, "to_thread", fake_to_thread)
|
||||
|
||||
sandbox_id = await provider._acquire_internal_async("thread-cached-async")
|
||||
sandbox_id = await provider._acquire_internal_async("thread-cached-async", user_id="default")
|
||||
|
||||
assert sandbox_id == "sandbox-cached-async"
|
||||
assert to_thread_calls == [(provider._reuse_in_process_sandbox, ("thread-cached-async",))]
|
||||
@ -372,6 +387,31 @@ def test_remote_backend_create_forwards_effective_user_id(monkeypatch):
|
||||
}
|
||||
|
||||
|
||||
def test_remote_backend_create_prefers_explicit_user_id(monkeypatch):
|
||||
"""Provisioner mode must not fall back to the ambient default for channel runs."""
|
||||
remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
|
||||
backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
|
||||
posted: dict = {}
|
||||
|
||||
class _Response:
|
||||
def raise_for_status(self):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return {"sandbox_url": "http://sandbox.local"}
|
||||
|
||||
def _post(url, json, timeout): # noqa: A002 - mirrors requests.post kwarg
|
||||
posted.update({"url": url, "json": json, "timeout": timeout})
|
||||
return _Response()
|
||||
|
||||
monkeypatch.setattr(remote_mod.requests, "post", _post)
|
||||
monkeypatch.setattr(remote_mod, "get_effective_user_id", lambda: "default")
|
||||
|
||||
backend.create("thread-42", "sandbox-42", user_id="ou-user")
|
||||
|
||||
assert posted["json"]["user_id"] == "ou-user"
|
||||
|
||||
|
||||
# ── Sandbox client teardown (#2872) ──────────────────────────────────────────
|
||||
|
||||
|
||||
@ -459,7 +499,7 @@ def test_acquire_drops_dead_cached_sandbox(tmp_path, monkeypatch):
|
||||
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 = {"thread-dead": "sandbox-dead"}
|
||||
provider._thread_sandboxes = {("default", "thread-dead"): "sandbox-dead"}
|
||||
provider._config = {"replicas": 3}
|
||||
provider._backend.is_alive = MagicMock(return_value=False)
|
||||
provider._backend.discover = MagicMock(return_value=None)
|
||||
@ -471,19 +511,19 @@ def test_acquire_drops_dead_cached_sandbox(tmp_path, monkeypatch):
|
||||
)
|
||||
)
|
||||
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_sandbox_id_for_thread", lambda _self, _thread_id: "sandbox-dead")
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_extra_mounts", lambda _self, _thread_id: [])
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_sandbox_id_for_thread", lambda _self, _thread_id, _user_id: "sandbox-dead")
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_extra_mounts", lambda _self, _thread_id, *, user_id=None: [])
|
||||
monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
|
||||
monkeypatch.setattr(aio_mod, "get_effective_user_id", lambda: None)
|
||||
monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda _url, timeout=60: True)
|
||||
|
||||
sandbox_id = provider.acquire("thread-dead")
|
||||
sandbox_id = provider.acquire("thread-dead", user_id="default")
|
||||
|
||||
assert sandbox_id == "sandbox-dead"
|
||||
sandbox.close.assert_called_once_with()
|
||||
provider._backend.destroy.assert_called_once()
|
||||
provider._backend.create.assert_called_once()
|
||||
assert provider._thread_sandboxes["thread-dead"] == "sandbox-dead"
|
||||
assert provider._thread_sandboxes[("default", "thread-dead")] == "sandbox-dead"
|
||||
assert provider._sandboxes["sandbox-dead"].base_url == "http://fresh-sandbox"
|
||||
|
||||
|
||||
@ -491,10 +531,10 @@ 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 = {"thread-transient": "sandbox-transient"}
|
||||
provider._thread_sandboxes = {("default", "thread-transient"): "sandbox-transient"}
|
||||
provider._backend.is_alive = MagicMock(side_effect=OSError("docker daemon busy"))
|
||||
|
||||
sandbox_id = provider.acquire("thread-transient")
|
||||
sandbox_id = provider.acquire("thread-transient", user_id="default")
|
||||
|
||||
assert sandbox_id == "sandbox-transient"
|
||||
sandbox.close.assert_not_called()
|
||||
@ -509,7 +549,7 @@ def test_drop_unhealthy_sandbox_skips_recreated_entry(tmp_path):
|
||||
provider._lock = aio_mod.threading.Lock()
|
||||
provider._warm_pool = {}
|
||||
provider._last_activity = {"sandbox-toctou": 1.0}
|
||||
provider._thread_sandboxes = {"thread-toctou": "sandbox-toctou"}
|
||||
provider._thread_sandboxes = {("default", "thread-toctou"): "sandbox-toctou"}
|
||||
old_info = aio_mod.SandboxInfo(sandbox_id="sandbox-toctou", sandbox_url="http://old-sandbox")
|
||||
new_info = aio_mod.SandboxInfo(sandbox_id="sandbox-toctou", sandbox_url="http://new-sandbox")
|
||||
new_sandbox = MagicMock()
|
||||
@ -523,7 +563,7 @@ def test_drop_unhealthy_sandbox_skips_recreated_entry(tmp_path):
|
||||
provider._backend.destroy.assert_not_called()
|
||||
assert provider._sandbox_infos["sandbox-toctou"] is new_info
|
||||
assert provider._sandboxes["sandbox-toctou"] is new_sandbox
|
||||
assert provider._thread_sandboxes == {"thread-toctou": "sandbox-toctou"}
|
||||
assert provider._thread_sandboxes == {("default", "thread-toctou"): "sandbox-toctou"}
|
||||
|
||||
|
||||
def test_acquire_skips_dead_warm_pool_sandbox(tmp_path, monkeypatch):
|
||||
@ -560,19 +600,19 @@ def test_acquire_skips_dead_warm_pool_sandbox(tmp_path, monkeypatch):
|
||||
),
|
||||
)
|
||||
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_sandbox_id_for_thread", lambda _self, _thread_id: "sandbox-warm-dead")
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_extra_mounts", lambda _self, _thread_id: [])
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_sandbox_id_for_thread", lambda _self, _thread_id, _user_id: "sandbox-warm-dead")
|
||||
monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_extra_mounts", lambda _self, _thread_id, *, user_id=None: [])
|
||||
monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
|
||||
monkeypatch.setattr(aio_mod, "get_effective_user_id", lambda: None)
|
||||
monkeypatch.setattr(aio_mod, "wait_for_sandbox_ready", lambda _url, timeout=60: True)
|
||||
|
||||
sandbox_id = provider.acquire("thread-warm-dead")
|
||||
sandbox_id = provider.acquire("thread-warm-dead", user_id="default")
|
||||
|
||||
assert sandbox_id == "sandbox-warm-dead"
|
||||
provider._backend.destroy.assert_called_once()
|
||||
provider._backend.create.assert_called_once()
|
||||
assert provider._warm_pool == {}
|
||||
assert provider._thread_sandboxes["thread-warm-dead"] == "sandbox-warm-dead"
|
||||
assert provider._thread_sandboxes[("default", "thread-warm-dead")] == "sandbox-warm-dead"
|
||||
assert provider._sandboxes["sandbox-warm-dead"].base_url == "http://fresh-sandbox"
|
||||
|
||||
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import asyncio
|
||||
import json
|
||||
import tempfile
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
@ -154,6 +155,46 @@ def test_feishu_receive_file_replaces_placeholders_in_order():
|
||||
_run(go())
|
||||
|
||||
|
||||
def test_feishu_receive_file_syncs_sandbox_with_explicit_user_id(tmp_path, monkeypatch):
|
||||
async def go():
|
||||
from deerflow.config.paths import Paths
|
||||
|
||||
bus = MessageBus()
|
||||
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
|
||||
channel._GetMessageResourceRequest = MagicMock()
|
||||
builder = MagicMock()
|
||||
builder.message_id.return_value = builder
|
||||
builder.file_key.return_value = builder
|
||||
builder.type.return_value = builder
|
||||
builder.build.return_value = object()
|
||||
channel._GetMessageResourceRequest.builder.return_value = builder
|
||||
|
||||
response = MagicMock()
|
||||
response.success.return_value = True
|
||||
response.file = BytesIO(b"file-bytes")
|
||||
response.file_name = "report.md"
|
||||
channel._api_client = MagicMock()
|
||||
channel._api_client.im.v1.message_resource.get.return_value = response
|
||||
|
||||
provider = MagicMock()
|
||||
provider.acquire.return_value = "aio-1"
|
||||
sandbox = MagicMock()
|
||||
provider.get.return_value = sandbox
|
||||
|
||||
monkeypatch.setattr("app.channels.feishu.get_paths", lambda: Paths(base_dir=tmp_path))
|
||||
monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: provider)
|
||||
monkeypatch.setattr("app.channels.feishu.get_effective_user_id", lambda: "default")
|
||||
|
||||
virtual_path = await channel._receive_single_file("message-1", "file-key", "file", "thread-1", user_id="ou-user")
|
||||
|
||||
assert virtual_path == "/mnt/user-data/uploads/report.md"
|
||||
assert (tmp_path / "users" / "ou-user" / "threads" / "thread-1" / "user-data" / "uploads" / "report.md").read_bytes() == b"file-bytes"
|
||||
provider.acquire.assert_called_once_with("thread-1", user_id="ou-user")
|
||||
sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/report.md", b"file-bytes")
|
||||
|
||||
_run(go())
|
||||
|
||||
|
||||
def test_feishu_on_message_extracts_image_and_file_keys():
|
||||
bus = MessageBus()
|
||||
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
|
||||
|
||||
@ -70,8 +70,12 @@ def provider(isolated_paths, tmp_path):
|
||||
|
||||
|
||||
def test_acquire_with_thread_id_returns_per_thread_id(provider):
|
||||
sandbox_id = provider.acquire("alpha")
|
||||
assert sandbox_id == "local:alpha"
|
||||
sandbox_id = provider.acquire("alpha", user_id="default")
|
||||
assert sandbox_id == "local:default:alpha"
|
||||
|
||||
|
||||
def test_acquire_with_thread_id_uses_uniform_user_scoped_id(provider):
|
||||
assert provider.acquire("alpha", user_id="alice") == "local:alice:alpha"
|
||||
|
||||
|
||||
def test_acquire_without_thread_id_remains_legacy_local_id(provider):
|
||||
@ -180,6 +184,21 @@ def test_per_thread_user_data_mapping_isolated(provider, isolated_paths):
|
||||
sbx_b.read_file("/mnt/user-data/workspace/secret.txt")
|
||||
|
||||
|
||||
def test_same_thread_different_users_are_isolated(provider):
|
||||
"""Channel/user-scoped mounts must not reuse another user's local mapping."""
|
||||
sid_alice = provider.acquire("alpha", user_id="alice")
|
||||
sid_bob = provider.acquire("alpha", user_id="bob")
|
||||
assert sid_alice != sid_bob
|
||||
|
||||
sbx_alice = provider.get(sid_alice)
|
||||
sbx_bob = provider.get(sid_bob)
|
||||
assert sbx_alice is not sbx_bob
|
||||
|
||||
sbx_alice.write_file("/mnt/user-data/outputs/report.md", "alice-only")
|
||||
with pytest.raises(FileNotFoundError):
|
||||
sbx_bob.read_file("/mnt/user-data/outputs/report.md")
|
||||
|
||||
|
||||
def test_agent_written_paths_per_thread_isolation(provider):
|
||||
"""``_agent_written_paths`` tracks files this sandbox wrote so reverse-resolve
|
||||
runs on read. The set must not leak across threads."""
|
||||
@ -203,7 +222,7 @@ def test_get_returns_cached_instance_for_known_id(provider):
|
||||
|
||||
|
||||
def test_get_unknown_id_returns_none(provider):
|
||||
assert provider.get("local:nonexistent") is None
|
||||
assert provider.get("local:default:nonexistent") is None
|
||||
|
||||
|
||||
def test_release_is_noop_keeps_instance_available(provider):
|
||||
@ -228,19 +247,19 @@ def test_reset_clears_both_generic_and_per_thread_caches(provider):
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
# 4. is_local_sandbox detects both legacy and per-thread ids
|
||||
# 4. is_local_sandbox detects both generic and per-thread ids
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_is_local_sandbox_accepts_both_id_formats():
|
||||
def test_is_local_sandbox_accepts_generic_and_per_thread_id_formats():
|
||||
from deerflow.sandbox.tools import is_local_sandbox
|
||||
|
||||
legacy = SimpleNamespace(state={"sandbox": {"sandbox_id": "local"}}, context={})
|
||||
per_thread = SimpleNamespace(state={"sandbox": {"sandbox_id": "local:alpha"}}, context={})
|
||||
generic = SimpleNamespace(state={"sandbox": {"sandbox_id": "local"}}, context={})
|
||||
per_thread = SimpleNamespace(state={"sandbox": {"sandbox_id": "local:default:alpha"}}, context={})
|
||||
foreign = SimpleNamespace(state={"sandbox": {"sandbox_id": "aio-12345"}}, context={})
|
||||
unset = SimpleNamespace(state={}, context={})
|
||||
|
||||
assert is_local_sandbox(legacy) is True
|
||||
assert is_local_sandbox(generic) is True
|
||||
assert is_local_sandbox(per_thread) is True
|
||||
assert is_local_sandbox(foreign) is False
|
||||
assert is_local_sandbox(unset) is False
|
||||
@ -277,7 +296,7 @@ def test_concurrent_acquire_same_thread_yields_single_instance(provider):
|
||||
|
||||
def racer():
|
||||
barrier.wait()
|
||||
sid = provider.acquire("alpha")
|
||||
sid = provider.acquire("alpha", user_id="default")
|
||||
with results_lock:
|
||||
results.append(sid)
|
||||
|
||||
@ -292,7 +311,7 @@ def test_concurrent_acquire_same_thread_yields_single_instance(provider):
|
||||
assert len(set(results)) == 1, f"Racers saw different ids: {results}"
|
||||
# …and the cache must hold exactly one instance for ``alpha``.
|
||||
assert len(provider._thread_sandboxes) == 1
|
||||
assert "alpha" in provider._thread_sandboxes
|
||||
assert ("default", "alpha") in provider._thread_sandboxes
|
||||
|
||||
|
||||
def test_concurrent_acquire_distinct_threads_yields_distinct_instances(provider):
|
||||
@ -305,7 +324,7 @@ def test_concurrent_acquire_distinct_threads_yields_distinct_instances(provider)
|
||||
|
||||
def racer(name: str):
|
||||
barrier.wait()
|
||||
sid = provider.acquire(name)
|
||||
sid = provider.acquire(name, user_id="default")
|
||||
with lock:
|
||||
sids[name] = sid
|
||||
|
||||
@ -315,8 +334,8 @@ def test_concurrent_acquire_distinct_threads_yields_distinct_instances(provider)
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert set(sids.values()) == {f"local:t{i}" for i in range(6)}
|
||||
assert set(provider._thread_sandboxes.keys()) == {f"t{i}" for i in range(6)}
|
||||
assert set(sids.values()) == {f"local:default:t{i}" for i in range(6)}
|
||||
assert set(provider._thread_sandboxes.keys()) == {("default", f"t{i}") for i in range(6)}
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
@ -336,12 +355,12 @@ def test_thread_sandbox_cache_is_bounded(isolated_paths, tmp_path):
|
||||
provider = LocalSandboxProvider(max_cached_threads=3)
|
||||
|
||||
for i in range(5):
|
||||
provider.acquire(f"t{i}")
|
||||
provider.acquire(f"t{i}", user_id="default")
|
||||
|
||||
# Only the 3 most-recent thread_ids should be retained.
|
||||
assert set(provider._thread_sandboxes.keys()) == {"t2", "t3", "t4"}
|
||||
assert provider.get("local:t0") is None
|
||||
assert provider.get("local:t4") is not None
|
||||
assert set(provider._thread_sandboxes.keys()) == {("default", "t2"), ("default", "t3"), ("default", "t4")}
|
||||
assert provider.get("local:default:t0") is None
|
||||
assert provider.get("local:default:t4") is not None
|
||||
|
||||
|
||||
def test_lru_promotes_recently_used_thread(isolated_paths, tmp_path):
|
||||
@ -355,12 +374,12 @@ def test_lru_promotes_recently_used_thread(isolated_paths, tmp_path):
|
||||
provider = LocalSandboxProvider(max_cached_threads=3)
|
||||
|
||||
for name in ["a", "b", "c"]:
|
||||
provider.acquire(name)
|
||||
provider.acquire(name, user_id="default")
|
||||
# Touch "a" via ``get`` so it becomes most-recently used.
|
||||
provider.get("local:a")
|
||||
provider.get("local:default:a")
|
||||
# Adding a fourth thread should evict "b" (the new LRU), not "a".
|
||||
provider.acquire("d")
|
||||
provider.acquire("d", user_id="default")
|
||||
|
||||
assert "a" in provider._thread_sandboxes
|
||||
assert "b" not in provider._thread_sandboxes
|
||||
assert {"a", "c", "d"} == set(provider._thread_sandboxes.keys())
|
||||
assert ("default", "a") in provider._thread_sandboxes
|
||||
assert ("default", "b") not in provider._thread_sandboxes
|
||||
assert {("default", "a"), ("default", "c"), ("default", "d")} == set(provider._thread_sandboxes.keys())
|
||||
|
||||
@ -123,19 +123,26 @@ def test_provisioner_list_skips_non_dict_sandbox_entries(monkeypatch):
|
||||
assert infos[0].sandbox_url == "http://k3s:31001"
|
||||
|
||||
|
||||
def test_create_delegates_to_provisioner_create(monkeypatch):
|
||||
@pytest.mark.parametrize("expected_user_id", [None, "owner-1"])
|
||||
def test_create_delegates_to_provisioner_create(monkeypatch, expected_user_id):
|
||||
backend = RemoteSandboxBackend("http://provisioner:8002")
|
||||
expected = SandboxInfo(sandbox_id="abc123", sandbox_url="http://k3s:31001")
|
||||
|
||||
def mock_create(thread_id: str, sandbox_id: str, extra_mounts=None):
|
||||
def mock_create(thread_id: str, sandbox_id: str, extra_mounts=None, *, user_id=None):
|
||||
assert thread_id == "thread-1"
|
||||
assert sandbox_id == "abc123"
|
||||
assert extra_mounts == [("/host", "/container", False)]
|
||||
assert user_id == expected_user_id
|
||||
return expected
|
||||
|
||||
monkeypatch.setattr(backend, "_provisioner_create", mock_create)
|
||||
|
||||
result = backend.create("thread-1", "abc123", extra_mounts=[("/host", "/container", False)])
|
||||
result = backend.create(
|
||||
"thread-1",
|
||||
"abc123",
|
||||
extra_mounts=[("/host", "/container", False)],
|
||||
user_id=expected_user_id,
|
||||
)
|
||||
assert result == expected
|
||||
|
||||
|
||||
|
||||
@ -22,9 +22,11 @@ from deerflow.sandbox.tools import ls_tool
|
||||
class _SyncProvider(SandboxProvider):
|
||||
def __init__(self) -> None:
|
||||
self.thread_ids: list[str | None] = []
|
||||
self.user_ids: list[str | None] = []
|
||||
|
||||
def acquire(self, thread_id: str | None = None) -> str:
|
||||
def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
self.thread_ids.append(thread_id)
|
||||
self.user_ids.append(user_id)
|
||||
return "sync-sandbox"
|
||||
|
||||
def get(self, sandbox_id: str) -> Sandbox | None:
|
||||
@ -72,14 +74,17 @@ class _SandboxStub(Sandbox):
|
||||
class _AsyncOnlyProvider(SandboxProvider):
|
||||
def __init__(self) -> None:
|
||||
self.thread_ids: list[str | None] = []
|
||||
self.user_ids: list[str | None] = []
|
||||
self.released_ids: list[str] = []
|
||||
self.sandbox = _SandboxStub("async-sandbox")
|
||||
|
||||
def acquire(self, thread_id: str | None = None) -> str:
|
||||
def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
del user_id
|
||||
raise AssertionError("async middleware should not call sync acquire")
|
||||
|
||||
async def acquire_async(self, thread_id: str | None = None) -> str:
|
||||
async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
|
||||
self.thread_ids.append(thread_id)
|
||||
self.user_ids.append(user_id)
|
||||
return "async-sandbox"
|
||||
|
||||
def get(self, sandbox_id: str) -> Sandbox | None:
|
||||
@ -105,9 +110,9 @@ async def test_provider_default_acquire_async_offloads_sync_acquire(monkeypatch:
|
||||
provider = _SyncProvider()
|
||||
calls: list[tuple[object, tuple[object, ...]]] = []
|
||||
|
||||
async def fake_to_thread(func, /, *args):
|
||||
calls.append((func, args))
|
||||
return func(*args)
|
||||
async def fake_to_thread(func, /, *args, **kwargs):
|
||||
calls.append((func, args, kwargs))
|
||||
return func(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(asyncio, "to_thread", fake_to_thread)
|
||||
|
||||
@ -115,7 +120,8 @@ async def test_provider_default_acquire_async_offloads_sync_acquire(monkeypatch:
|
||||
|
||||
assert sandbox_id == "sync-sandbox"
|
||||
assert provider.thread_ids == ["thread-1"]
|
||||
assert calls == [(provider.acquire, ("thread-1",))]
|
||||
assert provider.user_ids == [None]
|
||||
assert calls == [(provider.acquire, ("thread-1",), {"user_id": None})]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -125,12 +131,13 @@ async def test_abefore_agent_uses_async_provider_acquire() -> None:
|
||||
try:
|
||||
middleware = SandboxMiddleware(lazy_init=False)
|
||||
|
||||
result = await middleware.abefore_agent({}, Runtime(context={"thread_id": "thread-2"}))
|
||||
result = await middleware.abefore_agent({}, Runtime(context={"thread_id": "thread-2", "user_id": "owner-2"}))
|
||||
finally:
|
||||
reset_sandbox_provider()
|
||||
|
||||
assert result == {"sandbox": {"sandbox_id": "async-sandbox"}}
|
||||
assert provider.thread_ids == ["thread-2"]
|
||||
assert provider.user_ids == ["owner-2"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -169,7 +176,7 @@ async def test_default_lazy_tool_acquisition_uses_async_provider() -> None:
|
||||
try:
|
||||
runtime = ToolRuntime(
|
||||
state={},
|
||||
context={"thread_id": "thread-lazy"},
|
||||
context={"thread_id": "thread-lazy", "user_id": "owner-lazy"},
|
||||
config={"configurable": {}},
|
||||
stream_writer=lambda _: None,
|
||||
tools=[],
|
||||
@ -183,6 +190,7 @@ async def test_default_lazy_tool_acquisition_uses_async_provider() -> None:
|
||||
|
||||
assert result == "/mnt/user-data/workspace/file.txt"
|
||||
assert provider.thread_ids == ["thread-lazy"]
|
||||
assert provider.user_ids == ["owner-lazy"]
|
||||
assert runtime.state["sandbox"] == {"sandbox_id": "async-sandbox"}
|
||||
assert runtime.context["sandbox_id"] == "async-sandbox"
|
||||
|
||||
|
||||
@ -28,7 +28,8 @@ def test_backend_list_running_default_returns_empty():
|
||||
from deerflow.community.aio_sandbox.backend import SandboxBackend
|
||||
|
||||
class StubBackend(SandboxBackend):
|
||||
def create(self, thread_id, sandbox_id, extra_mounts=None):
|
||||
def create(self, thread_id, sandbox_id, extra_mounts=None, *, user_id=None):
|
||||
del thread_id, sandbox_id, extra_mounts, user_id
|
||||
pass
|
||||
|
||||
def destroy(self, info):
|
||||
|
||||
@ -279,13 +279,15 @@ def test_upload_files_acquires_non_local_sandbox_before_writing(tmp_path):
|
||||
sandbox = MagicMock()
|
||||
provider.get.return_value = sandbox
|
||||
|
||||
def acquire_before_writes(thread_id: str) -> str:
|
||||
def acquire_before_writes(thread_id: str, *, user_id: str | None = None) -> str:
|
||||
assert list(thread_uploads_dir.iterdir()) == []
|
||||
assert user_id == "owner-upload"
|
||||
return "aio-1"
|
||||
|
||||
provider.acquire.side_effect = acquire_before_writes
|
||||
|
||||
with (
|
||||
patch.object(uploads, "get_effective_user_id", return_value="owner-upload"),
|
||||
patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
|
||||
patch.object(uploads, "get_sandbox_provider", return_value=provider),
|
||||
):
|
||||
@ -293,7 +295,7 @@ def test_upload_files_acquires_non_local_sandbox_before_writing(tmp_path):
|
||||
result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace()))
|
||||
|
||||
assert result.success is True
|
||||
provider.acquire.assert_called_once_with("thread-aio")
|
||||
provider.acquire.assert_called_once_with("thread-aio", user_id="owner-upload")
|
||||
sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/notes.txt", b"hello uploads")
|
||||
|
||||
|
||||
@ -393,6 +395,7 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_total_size_exceeds_li
|
||||
provider.get.return_value = sandbox
|
||||
|
||||
with (
|
||||
patch.object(uploads, "get_effective_user_id", return_value="owner-upload"),
|
||||
patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
|
||||
patch.object(uploads, "get_sandbox_provider", return_value=provider),
|
||||
patch.object(uploads, "_get_upload_limits", return_value=uploads.UploadLimits(max_files=10, max_file_size=10, max_total_size=5)),
|
||||
@ -405,7 +408,7 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_total_size_exceeds_li
|
||||
asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=files, config=SimpleNamespace()))
|
||||
|
||||
assert exc_info.value.status_code == 413
|
||||
provider.acquire.assert_called_once_with("thread-aio")
|
||||
provider.acquire.assert_called_once_with("thread-aio", user_id="owner-upload")
|
||||
provider.get.assert_called_once_with("aio-1")
|
||||
sandbox.update_file.assert_not_called()
|
||||
|
||||
@ -421,6 +424,7 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_conversion_fails(tmp_
|
||||
provider.get.return_value = sandbox
|
||||
|
||||
with (
|
||||
patch.object(uploads, "get_effective_user_id", return_value="owner-upload"),
|
||||
patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
|
||||
patch.object(uploads, "get_sandbox_provider", return_value=provider),
|
||||
patch.object(uploads, "_auto_convert_documents_enabled", return_value=True),
|
||||
@ -431,7 +435,7 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_conversion_fails(tmp_
|
||||
asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace()))
|
||||
|
||||
assert exc_info.value.status_code == 500
|
||||
provider.acquire.assert_called_once_with("thread-aio")
|
||||
provider.acquire.assert_called_once_with("thread-aio", user_id="owner-upload")
|
||||
provider.get.assert_called_once_with("aio-1")
|
||||
sandbox.update_file.assert_not_called()
|
||||
assert not (thread_uploads_dir / "report.pdf").exists()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user