diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py
index 6e235df75..ec53eb131 100644
--- a/backend/app/channels/feishu.py
+++ b/backend/app/channels/feishu.py
@@ -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:
diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py
index 58d278b68..6ecd547fd 100644
--- a/backend/app/gateway/routers/uploads.py
+++ b/backend/app/gateway/routers/uploads.py
@@ -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")
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
index 095663367..b0c7c635b 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
@@ -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:
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/backend.py
index a1db1bf31..0ff9cb3ea 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/backend.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/backend.py
@@ -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.
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py
index 58eaf6817..12ab6b032 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/local_backend.py
@@ -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
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
index c04f4e75d..ee9848d48 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
@@ -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,
             )
diff --git a/backend/packages/harness/deerflow/sandbox/local/local_sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/local/local_sandbox_provider.py
index 9e6523457..21a7fd083 100644
--- a/backend/packages/harness/deerflow/sandbox/local/local_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/sandbox/local/local_sandbox_provider.py
@@ -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
 
diff --git a/backend/packages/harness/deerflow/sandbox/middleware.py b/backend/packages/harness/deerflow/sandbox/middleware.py
index 0597cc557..a6ebbf7ea 100644
--- a/backend/packages/harness/deerflow/sandbox/middleware.py
+++ b/backend/packages/harness/deerflow/sandbox/middleware.py
@@ -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)
diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
index 58c52beee..b58fbd87d 100644
--- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
+++ b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
@@ -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:
diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py
index 49bb2fb1b..3440b7b52 100644
--- a/backend/packages/harness/deerflow/sandbox/tools.py
+++ b/backend/packages/harness/deerflow/sandbox/tools.py
@@ -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}
 
diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py
index 974997a69..3c0aea357 100644
--- a/backend/tests/test_aio_sandbox_provider.py
+++ b/backend/tests/test_aio_sandbox_provider.py
@@ -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"
 
 
diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py
index 34afb107d..41ccd7137 100644
--- a/backend/tests/test_feishu_parser.py
+++ b/backend/tests/test_feishu_parser.py
@@ -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"})
diff --git a/backend/tests/test_local_sandbox_virtual_path_contract.py b/backend/tests/test_local_sandbox_virtual_path_contract.py
index d9ec0cbdc..be9a73656 100644
--- a/backend/tests/test_local_sandbox_virtual_path_contract.py
+++ b/backend/tests/test_local_sandbox_virtual_path_contract.py
@@ -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())
diff --git a/backend/tests/test_remote_sandbox_backend.py b/backend/tests/test_remote_sandbox_backend.py
index 3c6f60886..6550cae10 100644
--- a/backend/tests/test_remote_sandbox_backend.py
+++ b/backend/tests/test_remote_sandbox_backend.py
@@ -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
 
 
diff --git a/backend/tests/test_sandbox_middleware.py b/backend/tests/test_sandbox_middleware.py
index 188ea242a..b9ecb57a2 100644
--- a/backend/tests/test_sandbox_middleware.py
+++ b/backend/tests/test_sandbox_middleware.py
@@ -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"
 
diff --git a/backend/tests/test_sandbox_orphan_reconciliation.py b/backend/tests/test_sandbox_orphan_reconciliation.py
index b01ad5004..5ff446003 100644
--- a/backend/tests/test_sandbox_orphan_reconciliation.py
+++ b/backend/tests/test_sandbox_orphan_reconciliation.py
@@ -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):
diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py
index ff55f04c5..7fdac6252 100644
--- a/backend/tests/test_uploads_router.py
+++ b/backend/tests/test_uploads_router.py
@@ -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()