fix(feishu): keep file receive off the event loop (#4627)

backend/AGENTS.md

@@ -264,6 +264,9 @@ Blocking-IO runtime gate (`tests/blocking_io/`):
   `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_feishu_receive_file.py` (locks Feishu attachment path preparation and
+  persistence plus remote sandbox acquisition/sync off the event loop, and
+  skips redundant sandbox sync when thread data is already mounted);
   `test_openviking_memory_backend.py` (locks the OpenViking backend's async
   add/context/search entrypoints offloading synchronous HTTP and watermark
   filesystem IO); and

backend/app/channels/feishu.py

@@ -424,10 +424,8 @@ class FeishuChannel(Channel):
             logger.warning("[Feishu] empty resource content: resource_key=%s, type=%s", file_key, type)
             return f"Failed to obtain the [{type}]"
 
-        paths = get_paths()
         effective_user_id = user_id or get_effective_user_id()
-        paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
-        uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve()
+        paths = await asyncio.to_thread(get_paths)
 
         ext = "png" if type == "image" else "bin"
         raw_filename = getattr(response, "file_name", "") or f"feishu_{file_key[-12:]}.{ext}"
@@ -439,30 +437,33 @@ class FeishuChannel(Channel):
             filename = f"{name_part}.{ext}"
         else:
             filename = re.sub(r"[./\\]", "_", raw_filename)
-        resolved_target = uploads_dir / filename
 
-        def down_load():
-            # use thread_lock to avoid filename conflicts when writing
+        def _persist():
+            paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
+            uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve()
+            resolved_target = uploads_dir / filename
+            # Use thread_lock to avoid filename conflicts when writing.
             with self._thread_lock:
                 resolved_target.write_bytes(content)
+            return resolved_target
 
         try:
-            await asyncio.to_thread(down_load)
+            resolved_target = await asyncio.to_thread(_persist)
         except Exception:
-            logger.exception("[Feishu] failed to persist downloaded resource: %s, type=%s", resolved_target, type)
+            logger.exception("[Feishu] failed to persist downloaded resource: %s, type=%s", filename, type)
             return f"Failed to obtain the [{type}]"
 
         virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}"
 
         try:
-            sandbox_provider = get_sandbox_provider()
-            sandbox_id = sandbox_provider.acquire(thread_id, user_id=effective_user_id)
-            if sandbox_id != "local":
+            sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
+            if not getattr(sandbox_provider, "uses_thread_data_mounts", False):
+                sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
                 sandbox = sandbox_provider.get(sandbox_id)
                 if sandbox is None:
                     logger.warning("[Feishu] sandbox not found for thread_id=%s", thread_id)
                     return f"Failed to obtain the [{type}]"
-                sandbox.update_file(virtual_path, content)
+                await asyncio.to_thread(sandbox.update_file, virtual_path, content)
         except Exception:
             logger.exception("[Feishu] failed to sync resource into non-local sandbox: %s", virtual_path)
             return f"Failed to obtain the [{type}]"

backend/tests/blocking_io/test_feishu_receive_file.py

@@ -0,0 +1,126 @@
+"""Regression anchors: Feishu ``receive_file`` must not block the event loop."""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from io import BytesIO
+from unittest.mock import MagicMock
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+
+def _channel_with_file(content: bytes = b"DATA", filename: str = "report.pdf"):
+    from app.channels.feishu import FeishuChannel
+    from app.channels.message_bus import MessageBus
+
+    channel = FeishuChannel(MessageBus(), {"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(content)
+    response.file_name = filename
+    channel._api_client = MagicMock()
+    channel._api_client.im.v1.message_resource.get.return_value = response
+    return channel
+
+
+class _RemoteSandbox:
+    def __init__(self) -> None:
+        self.updates: list[tuple[str, bytes]] = []
+        self.update_thread_id: int | None = None
+
+    def update_file(self, path: str, content: bytes) -> None:
+        self.update_thread_id = threading.get_ident()
+        self.updates.append((path, content))
+
+
+class _RemoteProvider:
+    uses_thread_data_mounts = False
+
+    def __init__(self) -> None:
+        self.sandbox = _RemoteSandbox()
+        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("Feishu receive_file must 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):
+        return self.sandbox if sandbox_id == "remote-sandbox" else None
+
+
+class _MountedProvider:
+    uses_thread_data_mounts = True
+
+    def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        raise AssertionError("mounted uploads must not acquire a sandbox")
+
+    async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
+        raise AssertionError("mounted uploads must not acquire a sandbox")
+
+    def get(self, sandbox_id: str):
+        raise AssertionError("mounted uploads must not look up a sandbox")
+
+
+async def test_receive_file_remote_sandbox_does_not_block_event_loop(tmp_path, monkeypatch) -> None:
+    from deerflow.config.paths import Paths
+
+    paths = await asyncio.to_thread(Paths, str(tmp_path))
+    provider = _RemoteProvider()
+    provider_lookup_thread_id = None
+
+    def _get_provider():
+        nonlocal provider_lookup_thread_id
+        provider_lookup_thread_id = threading.get_ident()
+        return provider
+
+    monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths)
+    monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", _get_provider)
+
+    loop_thread_id = threading.get_ident()
+    result = await _channel_with_file()._receive_single_file(
+        "message-1",
+        "file-key",
+        "file",
+        "thread-1",
+        user_id="ou-user",
+    )
+
+    assert result == "/mnt/user-data/uploads/report.pdf"
+    assert provider.acquire_async_calls == [("thread-1", "ou-user")]
+    assert provider.sandbox.updates == [(result, b"DATA")]
+    assert provider_lookup_thread_id != loop_thread_id
+    assert provider.sandbox.update_thread_id != loop_thread_id
+
+
+async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monkeypatch) -> None:
+    from deerflow.config.paths import Paths
+
+    paths = await asyncio.to_thread(Paths, str(tmp_path))
+    monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths)
+    monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: _MountedProvider())
+
+    result = await _channel_with_file()._receive_single_file(
+        "message-1",
+        "file-key",
+        "file",
+        "thread-1",
+        user_id="ou-user",
+    )
+
+    assert result == "/mnt/user-data/uploads/report.pdf"
+    uploaded = tmp_path / "users" / "ou-user" / "threads" / "thread-1" / "user-data" / "uploads" / "report.pdf"
+    assert await asyncio.to_thread(uploaded.read_bytes) == b"DATA"

backend/tests/test_feishu_parser.py

@@ -177,7 +177,9 @@ def test_feishu_receive_file_syncs_sandbox_with_explicit_user_id(tmp_path, monke
         channel._api_client.im.v1.message_resource.get.return_value = response
 
         provider = MagicMock()
-        provider.acquire.return_value = "aio-1"
+        provider.uses_thread_data_mounts = False
+        provider.acquire.side_effect = AssertionError("receive_file must use acquire_async")
+        provider.acquire_async = AsyncMock(return_value="aio-1")
         sandbox = MagicMock()
         provider.get.return_value = sandbox
 
@@ -189,7 +191,7 @@ def test_feishu_receive_file_syncs_sandbox_with_explicit_user_id(tmp_path, monke
 
         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")
+        provider.acquire_async.assert_awaited_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())