fix(gateway): offload gateway upload file IO (#3935)

* fix(gateway): offload gateway upload file IO

Move Gateway upload router filesystem work off the asyncio event loop by
using a dedicated ContextVar-preserving file IO executor. Use async sandbox
acquisition for non-mounted sandbox uploads and offload remote sandbox sync
together with host file reads.

Add blocking-IO regression coverage for upload, list, delete, and remote
sandbox sync paths.

* fix(gateway): align file IO worker env var prefix

Rename the file IO executor worker-count environment variable from
DEERFLOW_FILE_IO_WORKERS to DEER_FLOW_FILE_IO_WORKERS to match the repo's
existing runtime configuration prefix convention.

backend/AGENTS.md

@@ -136,9 +136,12 @@ Blocking-IO runtime gate (`tests/blocking_io/`):
   for #1917); `test_sqlite_lifespan.py` (locks the offload around
   SQLite path resolution plus `ensure_sqlite_parent_dir`, fix for #1912);
   `test_jsonl_run_event_store.py` (locks `JsonlRunEventStore`'s async
-  API offloading its file IO via `asyncio.to_thread`, fix #3084); and
+  API offloading its file IO via `asyncio.to_thread`);
   `test_uploads_middleware.py` (locks `UploadsMiddleware.abefore_agent`
-  offloading the uploads-directory scan off the event loop).
+  offloading the uploads-directory scan off the event loop); and
+  `test_uploads_router.py` (locks Gateway upload/list/delete endpoints
+  offloading upload directory creation, staged writes, chmod/cleanup,
+  directory scans/deletes, and remote sandbox sync off the event loop).
 - `test_gate_smoke.py` is a meta-test asserting the gate actually catches
   unoffloaded blocking IO and that the `@pytest.mark.allow_blocking_io`
   opt-out works.
@@ -766,6 +769,7 @@ Multi-file upload with automatic document conversion:
 - Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()`
 - Duplicate filenames in a single upload request are auto-renamed with `_N` suffixes so later files do not truncate earlier files
 - Gateway HTTP uploads stage bytes as `.upload-*.part` files and atomically replace the destination only after size validation. These staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools, and swept on Gateway startup if a hard crash leaves one behind.
+- Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor. Non-mounted sandbox uploads acquire sandboxes with `SandboxProvider.acquire_async()` and offload `read_bytes()` plus `sandbox.update_file()` together.
 - Agent receives uploaded file list via `UploadsMiddleware`
 
 See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details.

backend/app/gateway/routers/uploads.py

@@ -4,7 +4,9 @@ import logging
 import os
 import stat
 import tempfile
+from dataclasses import dataclass
 from pathlib import Path
+from typing import BinaryIO
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 from pydantic import BaseModel, Field
@@ -32,6 +34,7 @@ from deerflow.uploads.manager import (
     validate_upload_destination,
 )
 from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
+from deerflow.utils.file_io import run_file_io
 
 logger = logging.getLogger(__name__)
 
@@ -43,6 +46,13 @@ DEFAULT_MAX_FILE_SIZE = 50 * 1024 * 1024
 DEFAULT_MAX_TOTAL_SIZE = 100 * 1024 * 1024
 
 
+@dataclass(slots=True)
+class _UploadTempFile:
+    file_path: Path
+    temp_path: Path
+    handle: BinaryIO
+
+
 class UploadedFileInfo(BaseModel):
     """Uploaded file metadata exposed by upload and list APIs."""
 
@@ -167,6 +177,78 @@ def _cleanup_uploaded_paths(paths: list[os.PathLike[str] | str]) -> None:
             logger.warning("Failed to clean up upload path after rejected request: %s", path, exc_info=True)
 
 
+def _prepare_upload_destination(uploads_dir: os.PathLike[str] | str, display_filename: str) -> _UploadTempFile:
+    uploads_dir_path = Path(uploads_dir)
+    file_path = validate_upload_destination(uploads_dir_path, display_filename)
+    temp_fd, temp_path_str = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=uploads_dir_path)
+    temp_path = Path(temp_path_str)
+    try:
+        handle = os.fdopen(temp_fd, "wb")
+    except Exception:
+        try:
+            os.close(temp_fd)
+        except OSError:
+            pass
+        try:
+            os.unlink(temp_path)
+        except FileNotFoundError:
+            pass
+        raise
+    return _UploadTempFile(file_path=file_path, temp_path=temp_path, handle=handle)
+
+
+def _write_upload_chunk(upload_temp: _UploadTempFile, chunk: bytes) -> None:
+    upload_temp.handle.write(chunk)
+
+
+def _abort_upload_temp(upload_temp: _UploadTempFile) -> None:
+    try:
+        upload_temp.handle.close()
+    finally:
+        try:
+            os.unlink(upload_temp.temp_path)
+        except FileNotFoundError:
+            pass
+
+
+def _commit_upload_temp(upload_temp: _UploadTempFile) -> None:
+    upload_temp.handle.close()
+    try:
+        os.replace(upload_temp.temp_path, upload_temp.file_path)
+    except Exception:
+        try:
+            os.unlink(upload_temp.temp_path)
+        except FileNotFoundError:
+            pass
+        raise
+
+
+def _make_uploaded_paths_sandbox_readable(paths: list[os.PathLike[str] | str]) -> None:
+    for file_path in paths:
+        _make_file_sandbox_readable(file_path)
+
+
+def _sync_upload_to_sandbox(sandbox, file_path: os.PathLike[str] | str, virtual_path: str) -> None:
+    _make_file_sandbox_writable(file_path)
+    sandbox.update_file(virtual_path, Path(file_path).read_bytes())
+
+
+def _list_uploaded_files_for_thread(thread_id: str, user_id: str) -> dict:
+    uploads_dir = get_uploads_dir(thread_id, user_id=user_id)
+    result = list_files_in_dir(uploads_dir)
+    enrich_file_listing(result, thread_id)
+
+    sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=user_id)
+    for f in result["files"]:
+        f["path"] = str(sandbox_uploads / f["filename"])
+    return result
+
+
+def _delete_uploaded_file_for_thread(thread_id: str, filename: str, user_id: str) -> dict:
+    uploads_dir = get_uploads_dir(thread_id, user_id=user_id)
+    return delete_file_safe(uploads_dir, filename, convertible_extensions=CONVERTIBLE_EXTENSIONS)
+
+
 async def _write_upload_file_with_limits(
     file: UploadFile,
     *,
@@ -177,12 +259,9 @@ async def _write_upload_file_with_limits(
     total_size: int,
 ) -> tuple[os.PathLike[str] | str, int, int]:
     file_size = 0
-    uploads_dir_path = Path(uploads_dir)
-    file_path = validate_upload_destination(uploads_dir_path, display_filename)
-    temp_fd, temp_path_str = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=uploads_dir_path)
-    temp_path = Path(temp_path_str)
-    fh = os.fdopen(temp_fd, "wb")
+    upload_temp: _UploadTempFile | None = None
     try:
+        upload_temp = await run_file_io(_prepare_upload_destination, uploads_dir, display_filename)
         while chunk := await file.read(UPLOAD_CHUNK_SIZE):
             file_size += len(chunk)
             total_size += len(chunk)
@@ -190,24 +269,15 @@ async def _write_upload_file_with_limits(
                 raise HTTPException(status_code=413, detail=f"File too large: {display_filename}")
             if total_size > max_total_size:
                 raise HTTPException(status_code=413, detail="Total upload size too large")
-            fh.write(chunk)
+            await run_file_io(_write_upload_chunk, upload_temp, chunk)
+
+        await run_file_io(_commit_upload_temp, upload_temp)
+        file_path = upload_temp.file_path
+        upload_temp = None
     except Exception:
-        fh.close()
-        try:
-            os.unlink(temp_path)
-        except FileNotFoundError:
-            pass
+        if upload_temp is not None:
+            await run_file_io(_abort_upload_temp, upload_temp)
         raise
-    else:
-        fh.close()
-        try:
-            os.replace(temp_path, file_path)
-        except Exception:
-            try:
-                os.unlink(temp_path)
-            except FileNotFoundError:
-                pass
-            raise
     return file_path, file_size, total_size
 
 
@@ -244,10 +314,10 @@ async def upload_files(
 
     try:
         effective_user_id = get_effective_user_id()
-        uploads_dir = ensure_uploads_dir(thread_id, user_id=effective_user_id)
+        uploads_dir = await run_file_io(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=effective_user_id)
+    sandbox_uploads = uploads_dir
     uploaded_files = []
     written_paths = []
     sandbox_sync_targets = []
@@ -262,7 +332,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, user_id=effective_user_id)
+        sandbox_id = await sandbox_provider.acquire_async(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")
@@ -325,7 +395,7 @@ async def upload_files(
             uploaded_files.append(file_info)
 
         except HTTPException as e:
-            _cleanup_uploaded_paths(written_paths)
+            await run_file_io(_cleanup_uploaded_paths, written_paths)
             raise e
         except UnsafeUploadPathError as e:
             logger.warning("Skipping upload with unsafe destination %s: %s", file.filename, e)
@@ -333,7 +403,7 @@ async def upload_files(
             continue
         except Exception as e:
             logger.error(f"Failed to upload {file.filename}: {e}")
-            _cleanup_uploaded_paths(written_paths)
+            await run_file_io(_cleanup_uploaded_paths, written_paths)
             raise HTTPException(status_code=500, detail=f"Failed to upload {file.filename}: {str(e)}")
 
     # Uploaded files are created with 0o600 permissions (owner read/write only).
@@ -343,13 +413,11 @@ async def upload_files(
     # directory is bind-mounted into the container or synced via
     # sandbox.update_file.  Always add group/other read bits so every sandbox
     # configuration can read the uploaded content.
-    for file_path in written_paths:
-        _make_file_sandbox_readable(file_path)
+    await run_file_io(_make_uploaded_paths_sandbox_readable, written_paths)
 
     if sync_to_sandbox:
         for file_path, virtual_path in sandbox_sync_targets:
-            _make_file_sandbox_writable(file_path)
-            sandbox.update_file(virtual_path, file_path.read_bytes())
+            await run_file_io(_sync_upload_to_sandbox, sandbox, file_path, virtual_path)
 
     message = f"Successfully uploaded {len(uploaded_files)} file(s)"
     if skipped_files:
@@ -379,16 +447,9 @@ async def get_upload_limits(
 async def list_uploaded_files(thread_id: str, request: Request) -> UploadListResponse:
     """List all files in a thread's uploads directory."""
     try:
-        uploads_dir = get_uploads_dir(thread_id)
+        result = await run_file_io(_list_uploaded_files_for_thread, thread_id, get_effective_user_id())
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
-    result = list_files_in_dir(uploads_dir)
-    enrich_file_listing(result, thread_id)
-
-    # Gateway additionally includes the sandbox-relative path.
-    sandbox_uploads = get_paths().sandbox_uploads_dir(thread_id, user_id=get_effective_user_id())
-    for f in result["files"]:
-        f["path"] = str(sandbox_uploads / f["filename"])
 
     return UploadListResponse(**result)
 
@@ -398,15 +459,13 @@ async def list_uploaded_files(thread_id: str, request: Request) -> UploadListRes
 async def delete_uploaded_file(thread_id: str, filename: str, request: Request) -> dict:
     """Delete a file from a thread's uploads directory."""
     try:
-        uploads_dir = get_uploads_dir(thread_id)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    try:
-        return delete_file_safe(uploads_dir, filename, convertible_extensions=CONVERTIBLE_EXTENSIONS)
+        return await run_file_io(_delete_uploaded_file_for_thread, thread_id, filename, get_effective_user_id())
     except FileNotFoundError:
         raise HTTPException(status_code=404, detail=f"File not found: {filename}")
     except PathTraversalError:
         raise HTTPException(status_code=400, detail="Invalid path")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
         logger.error(f"Failed to delete {filename}: {e}")
         raise HTTPException(status_code=500, detail=f"Failed to delete {filename}: {str(e)}")

backend/packages/harness/deerflow/utils/file_io.py

@@ -0,0 +1,51 @@
+"""Dedicated async offload helper for filesystem work."""
+
+from __future__ import annotations
+
+import asyncio
+import atexit
+import contextvars
+import functools
+import logging
+import os
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+
+logger = logging.getLogger(__name__)
+
+
+def _default_file_io_workers() -> int:
+    raw = os.getenv("DEER_FLOW_FILE_IO_WORKERS")
+    if raw:
+        try:
+            workers = int(raw)
+            if workers > 0:
+                return workers
+        except ValueError:
+            pass
+        logger.warning("Invalid DEER_FLOW_FILE_IO_WORKERS value; using default file IO worker count")
+    return min(32, (os.cpu_count() or 1) + 4)
+
+
+_FILE_IO_EXECUTOR = ThreadPoolExecutor(max_workers=_default_file_io_workers(), thread_name_prefix="file-io")
+
+
+def _shutdown_file_io_executor() -> None:
+    _FILE_IO_EXECUTOR.shutdown(wait=False, cancel_futures=True)
+
+
+atexit.register(_shutdown_file_io_executor)
+
+
+async def run_file_io[**P, T](func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T:
+    """Run blocking filesystem-oriented work on the dedicated file IO pool.
+
+    ``asyncio.to_thread`` copies ``ContextVar`` values automatically; raw
+    ``loop.run_in_executor`` does not. Copy the current context explicitly so
+    user-scoped helpers such as ``get_effective_user_id()`` keep working inside
+    the worker thread.
+    """
+    loop = asyncio.get_running_loop()
+    ctx = contextvars.copy_context()
+    call = functools.partial(func, *args, **kwargs)
+    return await loop.run_in_executor(_FILE_IO_EXECUTOR, ctx.run, call)

backend/tests/blocking_io/test_uploads_router.py

@@ -0,0 +1,135 @@
+"""Regression anchor: uploads router must not block the event loop."""
+
+from __future__ import annotations
+
+import asyncio
+from io import BytesIO
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from _router_auth_helpers import call_unwrapped
+from fastapi import UploadFile
+
+from app.gateway.routers import uploads
+from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.uploads.manager import ensure_uploads_dir, get_uploads_dir
+
+pytestmark = pytest.mark.asyncio
+
+
+class _SandboxRecorder:
+    def __init__(self) -> None:
+        self.updates: list[tuple[str, bytes]] = []
+
+    def update_file(self, path: str, content: bytes) -> None:
+        self.updates.append((path, content))
+
+
+class _MountedProvider:
+    uses_thread_data_mounts = True
+
+    def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        raise AssertionError("mounted upload path must not acquire a sandbox")
+
+    async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        raise AssertionError("mounted upload path must not acquire a sandbox")
+
+    def get(self, sandbox_id: str):
+        raise AssertionError("mounted upload path must not read a sandbox")
+
+
+class _RemoteProvider:
+    uses_thread_data_mounts = False
+
+    def __init__(self) -> None:
+        self.sandbox = _SandboxRecorder()
+        self.acquire_async_calls: list[tuple[str | None, str | None]] = []
+
+    def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        raise AssertionError("upload route should use acquire_async")
+
+    async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        self.acquire_async_calls.append((thread_id, user_id))
+        return "remote-sandbox"
+
+    def get(self, sandbox_id: str):
+        if sandbox_id == "remote-sandbox":
+            return self.sandbox
+        return None
+
+
+def _reset_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+
+    import deerflow.config.paths as paths_mod
+
+    monkeypatch.setattr(paths_mod, "_paths", None)
+
+
+async def _thread_uploads_dir(thread_id: str, *, user_id: str | None = None) -> Path:
+    user_id = user_id or get_effective_user_id()
+    return await asyncio.to_thread(ensure_uploads_dir, thread_id, user_id=user_id)
+
+
+async def test_upload_endpoint_mounted_provider_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _reset_paths(tmp_path, monkeypatch)
+    provider = _MountedProvider()
+    monkeypatch.setattr(uploads, "get_sandbox_provider", lambda: provider)
+
+    result = await call_unwrapped(
+        uploads.upload_files,
+        "t-mounted",
+        request=None,
+        files=[UploadFile(filename="notes.txt", file=BytesIO(b"hello uploads"))],
+        config=SimpleNamespace(),
+    )
+
+    user_id = get_effective_user_id()
+    target = await asyncio.to_thread(lambda: get_uploads_dir("t-mounted", user_id=user_id) / "notes.txt")
+    assert result.success is True
+    assert result.files[0].filename == "notes.txt"
+    assert await asyncio.to_thread(target.read_bytes) == b"hello uploads"
+
+
+async def test_upload_endpoint_remote_provider_syncs_without_blocking_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _reset_paths(tmp_path, monkeypatch)
+    provider = _RemoteProvider()
+    monkeypatch.setattr(uploads, "get_sandbox_provider", lambda: provider)
+    monkeypatch.setattr(uploads, "get_effective_user_id", lambda: "owner-upload")
+
+    result = await call_unwrapped(
+        uploads.upload_files,
+        "t-remote",
+        request=None,
+        files=[UploadFile(filename="report.txt", file=BytesIO(b"remote bytes"))],
+        config=SimpleNamespace(),
+    )
+
+    assert result.success is True
+    assert provider.acquire_async_calls == [("t-remote", "owner-upload")]
+    assert provider.sandbox.updates == [("/mnt/user-data/uploads/report.txt", b"remote bytes")]
+
+
+async def test_list_uploaded_files_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _reset_paths(tmp_path, monkeypatch)
+    uploads_dir = await _thread_uploads_dir("t-list")
+    await asyncio.to_thread((uploads_dir / "notes.txt").write_bytes, b"hello")
+
+    result = await call_unwrapped(uploads.list_uploaded_files, "t-list", request=None)
+
+    assert result.count == 1
+    assert result.files[0].filename == "notes.txt"
+    assert result.files[0].size == len(b"hello")
+
+
+async def test_delete_uploaded_file_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _reset_paths(tmp_path, monkeypatch)
+    uploads_dir = await _thread_uploads_dir("t-delete")
+    target = uploads_dir / "notes.txt"
+    await asyncio.to_thread(target.write_bytes, b"delete me")
+
+    result = await call_unwrapped(uploads.delete_uploaded_file, "t-delete", "notes.txt", request=None)
+
+    assert result == {"success": True, "message": "Deleted notes.txt"}
+    assert not await asyncio.to_thread(target.exists)

backend/tests/support/detectors/thread_boundaries.py

@@ -78,6 +78,11 @@ EXACT_CALL_RULES: dict[str, _CallRule] = {
         "ASYNC_THREAD_OFFLOAD",
         "Offloads synchronous work from an async context into a worker thread.",
     ),
+    "deerflow.utils.file_io.run_file_io": _CallRule(
+        "INFO",
+        "ASYNC_FILE_IO_OFFLOAD",
+        "Offloads filesystem-oriented work from an async context into the dedicated file IO thread pool.",
+    ),
     "asyncio.new_event_loop": _CallRule(
         "WARN",
         "NEW_EVENT_LOOP",

backend/tests/test_detect_thread_boundaries.py

@@ -20,6 +20,7 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path):
         import threading
         import time
         from concurrent.futures import ThreadPoolExecutor
+        from deerflow.utils.file_io import run_file_io
         from langchain.tools import tool
         from langchain_core.tools import StructuredTool
 
@@ -29,6 +30,7 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path):
 
         async def handler(model):
             await asyncio.to_thread(str, "x")
+            await run_file_io(str, "y")
             model.invoke("blocking")
             time.sleep(1)
 
@@ -53,6 +55,7 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path):
     assert async_tool_finding.function == "async_tool"
     assert async_tool_finding.async_context is True
     assert "ASYNC_THREAD_OFFLOAD" in categories
+    assert "ASYNC_FILE_IO_OFFLOAD" in categories
     assert "SYNC_INVOKE_IN_ASYNC" in categories
     assert "BLOCKING_CALL_IN_ASYNC" in categories
     assert "SYNC_ASYNC_BRIDGE" in categories

backend/tests/test_file_io.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import contextvars
+import threading
+
+import pytest
+
+from deerflow.utils.file_io import run_file_io
+
+
+@pytest.mark.anyio
+async def test_run_file_io_propagates_contextvars_to_worker() -> None:
+    marker: contextvars.ContextVar[str] = contextvars.ContextVar("marker", default="missing")
+    marker.set("owner-1")
+
+    def read_marker() -> tuple[str, str]:
+        return marker.get(), threading.current_thread().name
+
+    value, thread_name = await run_file_io(read_marker)
+
+    assert value == "owner-1"
+    assert thread_name.startswith("file-io")
+
+
+@pytest.mark.anyio
+async def test_run_file_io_passes_args_and_kwargs() -> None:
+    def join_values(prefix: str, *, suffix: str) -> str:
+        return f"{prefix}:{suffix}"
+
+    assert await run_file_io(join_values, "left", suffix="right") == "left:right"

backend/tests/test_uploads_router.py

@@ -36,6 +36,15 @@ def _mounted_provider() -> MagicMock:
     return provider
 
 
+def _symlink_to_or_skip(link_path: Path, target_path: Path) -> None:
+    try:
+        link_path.symlink_to(target_path)
+    except OSError as exc:
+        if getattr(exc, "winerror", None) == 1314:
+            pytest.skip("Windows symlink privilege is not available")
+        raise
+
+
 def test_upload_files_writes_thread_storage_and_skips_local_sandbox_sync(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)
@@ -178,7 +187,8 @@ def test_upload_files_syncs_non_local_sandbox_and_marks_markdown_file(tmp_path):
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = False
-    provider.acquire.return_value = "aio-1"
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(return_value="aio-1")
     sandbox = MagicMock()
     provider.get.return_value = sandbox
 
@@ -216,7 +226,8 @@ def test_upload_files_makes_non_local_files_sandbox_writable(tmp_path):
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = False
-    provider.acquire.return_value = "aio-1"
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(return_value="aio-1")
     sandbox = MagicMock()
     provider.get.return_value = sandbox
 
@@ -284,7 +295,8 @@ def test_upload_files_acquires_non_local_sandbox_before_writing(tmp_path):
         assert user_id == "owner-upload"
         return "aio-1"
 
-    provider.acquire.side_effect = acquire_before_writes
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(side_effect=acquire_before_writes)
 
     with (
         patch.object(uploads, "get_effective_user_id", return_value="owner-upload"),
@@ -295,7 +307,8 @@ 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", user_id="owner-upload")
+    provider.acquire.assert_not_called()
+    provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload")
     sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/notes.txt", b"hello uploads")
 
 
@@ -305,7 +318,8 @@ def test_upload_files_fails_before_writing_when_non_local_sandbox_unavailable(tm
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = False
-    provider.acquire.side_effect = RuntimeError("sandbox unavailable")
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(side_effect=RuntimeError("sandbox unavailable"))
     file = ChunkedUpload("notes.txt", [b"hello uploads"])
 
     with (
@@ -317,6 +331,7 @@ def test_upload_files_fails_before_writing_when_non_local_sandbox_unavailable(tm
 
     assert list(thread_uploads_dir.iterdir()) == []
     assert file.read_calls == []
+    provider.acquire.assert_not_called()
     provider.get.assert_not_called()
 
 
@@ -390,7 +405,8 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_total_size_exceeds_li
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = False
-    provider.acquire.return_value = "aio-1"
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(return_value="aio-1")
     sandbox = MagicMock()
     provider.get.return_value = sandbox
 
@@ -408,7 +424,8 @@ 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", user_id="owner-upload")
+    provider.acquire.assert_not_called()
+    provider.acquire_async.assert_awaited_once_with("thread-aio", user_id="owner-upload")
     provider.get.assert_called_once_with("aio-1")
     sandbox.update_file.assert_not_called()
 
@@ -419,7 +436,8 @@ def test_upload_files_does_not_sync_non_local_sandbox_when_conversion_fails(tmp_
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = False
-    provider.acquire.return_value = "aio-1"
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(return_value="aio-1")
     sandbox = MagicMock()
     provider.get.return_value = sandbox
 
@@ -435,7 +453,8 @@ 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", user_id="owner-upload")
+    provider.acquire.assert_not_called()
+    provider.acquire_async.assert_awaited_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()
@@ -559,7 +578,7 @@ def test_upload_files_rejects_preexisting_symlink_destination(tmp_path):
     thread_uploads_dir.mkdir(parents=True)
     outside_file = tmp_path / "outside.txt"
     outside_file.write_text("protected", encoding="utf-8")
-    (thread_uploads_dir / "victim.txt").symlink_to(outside_file)
+    _symlink_to_or_skip(thread_uploads_dir / "victim.txt", outside_file)
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = True
@@ -584,7 +603,7 @@ def test_upload_files_rejects_dangling_symlink_destination(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)
     missing_target = tmp_path / "missing-target.txt"
-    (thread_uploads_dir / "victim.txt").symlink_to(missing_target)
+    _symlink_to_or_skip(thread_uploads_dir / "victim.txt", missing_target)
 
     provider = MagicMock()
     provider.uses_thread_data_mounts = True