diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 72939deb4..66c37e9a4 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -22,9 +22,11 @@ from app.channels.message_bus import ( OutboundMessage, ResolvedAttachment, ) -from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths +from deerflow.config.paths import get_paths from deerflow.runtime.user_context import get_effective_user_id from deerflow.sandbox.sandbox_provider import get_sandbox_provider +from deerflow.uploads.layout import upload_virtual_path +from deerflow.uploads.manager import publish_upload_bytes logger = logging.getLogger(__name__) PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60 @@ -447,11 +449,7 @@ class FeishuChannel(Channel): 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 + return publish_upload_bytes(uploads_dir, filename, content) try: resolved_target = await asyncio.to_thread(_persist) @@ -459,7 +457,7 @@ class FeishuChannel(Channel): 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}" + virtual_path = upload_virtual_path(resolved_target.name) try: sandbox_provider = await asyncio.to_thread(get_sandbox_provider) diff --git a/backend/tests/blocking_io/test_feishu_receive_file.py b/backend/tests/blocking_io/test_feishu_receive_file.py index 59603284c..2231f79e0 100644 --- a/backend/tests/blocking_io/test_feishu_receive_file.py +++ b/backend/tests/blocking_io/test_feishu_receive_file.py @@ -124,3 +124,55 @@ async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monke 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" + + +async def test_concurrent_same_name_files_preserve_every_payload(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()) + payloads = [f"payload-{index}".encode() for index in range(8)] + channels = [_channel_with_file(payload, "report.pdf") for payload in payloads] + + results = await asyncio.gather( + *( + channel._receive_single_file( + f"message-{index}", + f"file-key-{index}", + "file", + "thread-1", + user_id="ou-user", + ) + for index, channel in enumerate(channels) + ) + ) + + assert len(set(results)) == len(payloads) + uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user") + assert {await asyncio.to_thread((uploads / result.rsplit("/", 1)[-1]).read_bytes) for result in results} == set(payloads) + + +async def test_receive_file_renames_around_planted_symlink(tmp_path, monkeypatch) -> None: + from deerflow.config.paths import Paths + + paths = await asyncio.to_thread(Paths, str(tmp_path)) + await asyncio.to_thread(paths.ensure_thread_dirs, "thread-1", user_id="ou-user") + uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user") + outside = tmp_path / "outside.pdf" + await asyncio.to_thread((uploads / "report.pdf").symlink_to, outside) + monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths) + monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: _MountedProvider()) + + result = await _channel_with_file(b"DATA", "report.pdf")._receive_single_file( + "message-1", + "file-key", + "file", + "thread-1", + user_id="ou-user", + ) + + assert result == "/mnt/user-data/uploads/report_1.pdf" + assert not await asyncio.to_thread(outside.exists) + assert await asyncio.to_thread((uploads / "report.pdf").is_symlink) + assert await asyncio.to_thread((uploads / "report_1.pdf").read_bytes) == b"DATA"