fix: make inbound attachments collision-safe

backend/app/channels/dingtalk.py

@@ -17,10 +17,11 @@ from app.channels.base import Channel
 from app.channels.commands import is_known_channel_command
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, 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.manager import UnsafeUploadPathError, claim_unique_filename, normalize_filename, write_upload_file_no_symlink
+from deerflow.uploads.layout import upload_virtual_path
+from deerflow.uploads.manager import UnsafeUploadPathError, normalize_filename, publish_upload_bytes
 
 logger = logging.getLogger(__name__)
 
@@ -156,9 +157,6 @@ class DingTalkChannel(Channel):
         self._incoming_messages: dict[str, Any] = {}
         self._incoming_messages_lock = threading.Lock()
         self._card_repliers: dict[str, Any] = {}
-        # Serialize inbound-file writes into the uploads directory to avoid
-        # racing writers clobbering one another (mirrors FeishuChannel).
-        self._file_write_lock = threading.Lock()
 
     @property
     def supports_streaming(self) -> bool:
@@ -630,23 +628,11 @@ class DingTalkChannel(Channel):
             safe_filename = fallback_name
 
         def _persist() -> Path:
-            # Directory prep, the uniqueness claim, and the write are blocking
-            # filesystem IO — the whole sequence stays off the event loop. The
-            # claim and the write share one lock because generated names repeat
-            # across messages ("image.png" for every picture message): without a
-            # claim a later attachment silently overwrites an earlier one whose
-            # path was already handed to the agent, and letting the claim and
-            # write interleave would resolve two attachments to the same free name.
+            # Directory prep and collision-safe publication are blocking
+            # filesystem IO, so the whole sequence stays off the event loop.
             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()
-            with self._file_write_lock:
-                seen = {entry.name for entry in uploads_dir.iterdir() if entry.is_file()}
-                unique_name = claim_unique_filename(safe_filename, seen)
-                # write_upload_file_no_symlink refuses a symlinked destination:
-                # uploads dirs can be mounted into local sandboxes, so a sandbox
-                # process could otherwise redirect this privileged write outside
-                # the bucket.
-                return write_upload_file_no_symlink(uploads_dir, unique_name, content)
+            return publish_upload_bytes(uploads_dir, safe_filename, content)
 
         try:
             resolved_target = await asyncio.to_thread(_persist)
@@ -654,7 +640,7 @@ class DingTalkChannel(Channel):
             logger.exception("[DingTalk] failed to persist downloaded file: %s", safe_filename)
             return ""
 
-        virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}"
+        virtual_path = upload_virtual_path(resolved_target.name)
 
         try:
             sandbox_provider = get_sandbox_provider()

backend/app/channels/manager.py

@@ -882,22 +882,20 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
     if not msg.files:
         return []
 
+    from deerflow.uploads.layout import upload_virtual_path
     from deerflow.uploads.manager import (
         UnsafeUploadPathError,
-        claim_unique_filename,
         ensure_uploads_dir,
         normalize_filename,
-        write_upload_file_no_symlink,
+        publish_upload_bytes,
     )
 
-    def _prepare_uploads_dir() -> tuple[Path, set[str]]:
-        # Worker thread: ensure_uploads_dir's mkdir and the iterdir enumeration are
-        # blocking filesystem IO that must stay off the event loop.
-        target = ensure_uploads_dir(thread_id, user_id=user_id)
-        existing = {entry.name for entry in target.iterdir() if entry.is_file()}
-        return target, existing
+    def _prepare_uploads_dir() -> Path:
+        # Worker thread: directory creation is blocking filesystem IO that must
+        # stay off the event loop.
+        return ensure_uploads_dir(thread_id, user_id=user_id)
 
-    uploads_dir, seen_names = await asyncio.to_thread(_prepare_uploads_dir)
+    uploads_dir = await asyncio.to_thread(_prepare_uploads_dir)
 
     created: list[dict[str, Any]] = []
     file_reader = INBOUND_FILE_READERS.get(msg.channel_name, _read_http_inbound_file)
@@ -940,7 +938,7 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                 filename = f"{msg.thread_ts or 'msg'}_{idx}{ext}"
 
             try:
-                safe_name = claim_unique_filename(normalize_filename(filename), seen_names)
+                safe_name = normalize_filename(filename)
             except ValueError:
                 logger.warning(
                     "[Manager] skipping inbound file with unsafe filename: channel=%s, file=%r",
@@ -949,21 +947,21 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                 )
                 continue
 
-            dest = uploads_dir / safe_name
             try:
-                dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data)
+                dest = await asyncio.to_thread(publish_upload_bytes, uploads_dir, safe_name, data)
             except UnsafeUploadPathError:
                 logger.warning("[Manager] skipping inbound file with unsafe destination: %s", safe_name)
                 continue
             except Exception:
-                logger.exception("[Manager] failed to write inbound file: %s", dest)
+                logger.exception("[Manager] failed to write inbound file: %s", safe_name)
                 continue
 
+            actual_name = dest.name
             created.append(
                 {
-                    "filename": safe_name,
+                    "filename": actual_name,
                     "size": len(data),
-                    "path": f"/mnt/user-data/uploads/{safe_name}",
+                    "path": upload_virtual_path(actual_name),
                     "is_image": ftype == "image",
                 }
             )

backend/tests/blocking_io/test_channels_ingest.py

@@ -1,10 +1,9 @@
 """Regression anchor: ingesting inbound channel files must not block the event loop.
 
 ``ChannelManager``'s ``_ingest_inbound_files`` ensures the thread uploads
-directory (``mkdir``), enumerates it (``iterdir`` / ``is_file``) to de-duplicate
-filenames, and writes each downloaded attachment to disk
-(``write_upload_file_no_symlink``) — all blocking filesystem IO. The async
-function offloads the directory prep and every per-file write via
+directory (``mkdir``), stages each downloaded attachment, and publishes it
+without replacing an existing name — all blocking filesystem IO. The async
+function offloads the directory prep and every per-file publication via
 ``asyncio.to_thread`` while keeping the genuinely async network read
 (``file_reader``) on the loop. If any of that regresses back onto the event
 loop, the strict Blockbuster gate raises ``BlockingError`` and this test fails.

backend/tests/blocking_io/test_dingtalk_receive_file.py

@@ -1,11 +1,12 @@
 """Regression anchor: DingTalk ``receive_file`` must not block the event loop.
 
 ``_receive_single_file`` prepares the thread directories, resolves the uploads
-dir, scans it for the uniqueness claim, and writes the attachment — all blocking
-filesystem IO that must run inside ``asyncio.to_thread`` (and sandbox sync must
-go through ``acquire_async`` + an offloaded ``update_file``). This anchor drives
-the real ``receive_file`` under the strict Blockbuster gate; if any of that
-regresses back onto the event loop, Blockbuster raises ``BlockingError``.
+dir, stages the attachment, and publishes it without replacing an existing name
+— all blocking filesystem IO that must run inside ``asyncio.to_thread`` (and
+sandbox sync must go through ``acquire_async`` + an offloaded ``update_file``).
+This anchor drives the real ``receive_file`` under the strict Blockbuster gate;
+if any of that regresses back onto the event loop, Blockbuster raises
+``BlockingError``.
 
 The ``Paths`` construction is offloaded only because ``Paths.__init__`` resolves
 paths synchronously; the surface under test (``receive_file``'s persist path) is

backend/tests/test_channel_file_attachments.py

@@ -282,7 +282,7 @@ class TestInboundFileIngestion:
         assert (uploads_dir / "report.pdf").read_bytes() == b"pdf bytes"
         assert "_content" not in msg.files[0]
 
-    def test_rejects_preexisting_symlink_destination(self, tmp_path):
+    def test_renames_around_preexisting_symlink_destination(self, tmp_path):
         from app.channels import manager
 
         uploads_dir = tmp_path / "uploads"
@@ -307,11 +307,19 @@ class TestInboundFileIngestion:
         ):
             result = _run(manager._ingest_inbound_files("thread-1", msg))
 
-        assert result == []
+        assert result == [
+            {
+                "filename": "victim_1.txt",
+                "size": len(b"attacker data"),
+                "path": "/mnt/user-data/uploads/victim_1.txt",
+                "is_image": False,
+            }
+        ]
         assert not outside_file.exists()
         assert (uploads_dir / "victim.txt").is_symlink()
+        assert (uploads_dir / "victim_1.txt").read_bytes() == b"attacker data"
 
-    def test_rejects_dangling_symlink_destination(self, tmp_path):
+    def test_renames_around_dangling_symlink_destination(self, tmp_path):
         from app.channels import manager
 
         uploads_dir = tmp_path / "uploads"
@@ -336,9 +344,17 @@ class TestInboundFileIngestion:
         ):
             result = _run(manager._ingest_inbound_files("thread-1", msg))
 
-        assert result == []
+        assert result == [
+            {
+                "filename": "victim_1.txt",
+                "size": len(b"attacker data"),
+                "path": "/mnt/user-data/uploads/victim_1.txt",
+                "is_image": False,
+            }
+        ]
         assert not missing_target.exists()
         assert (uploads_dir / "victim.txt").is_symlink()
+        assert (uploads_dir / "victim_1.txt").read_bytes() == b"attacker data"
 
     def test_hardlinked_existing_file_is_not_overwritten(self, tmp_path):
         from app.channels import manager
@@ -378,6 +394,33 @@ class TestInboundFileIngestion:
         assert (uploads_dir / "victim.txt").read_text(encoding="utf-8") == "protected"
         assert (uploads_dir / "victim_1.txt").read_bytes() == b"new attachment data"
 
+    def test_concurrent_inbound_messages_with_same_name_preserve_all_bytes(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        payloads = [f"payload-{index}".encode() for index in range(8)]
+        messages = [
+            InboundMessage(
+                channel_name="telegram",
+                chat_id="chat-1",
+                user_id="user-1",
+                text="attachment",
+                files=[{"type": "file", "filename": "same.txt", "_content": payload}],
+            )
+            for payload in payloads
+        ]
+
+        async def run_all():
+            return await asyncio.gather(*(manager._ingest_inbound_files("thread-1", message) for message in messages))
+
+        with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir):
+            results = _run(run_all())
+
+        created = [batch[0] for batch in results]
+        assert len({item["filename"] for item in created}) == len(payloads)
+        assert {(uploads_dir / item["filename"]).read_bytes() for item in created} == set(payloads)
+
 
 # ---------------------------------------------------------------------------
 # Channel base class _on_outbound with attachments

backend/tests/test_dingtalk_channel.py

@@ -2223,6 +2223,33 @@ class TestReceiveFile:
 
         _run(go())
 
+    def test_concurrent_duplicate_document_names_preserve_all_bytes(self, tmp_path, monkeypatch):
+        async def go():
+            channel = DingTalkChannel(MessageBus(), config={})
+            uploads = tmp_path / "uploads"
+            uploads.mkdir()
+            _patch_uploads(monkeypatch, uploads)
+            payloads = [f"version-{index}".encode() for index in range(8)]
+            channel._download_by_code = AsyncMock(side_effect=payloads)
+
+            paths = await asyncio.gather(
+                *(
+                    channel._receive_single_file(
+                        f"code-{index}",
+                        "file",
+                        "quote.xlsx",
+                        "thread-1",
+                        user_id="default",
+                    )
+                    for index in range(len(payloads))
+                )
+            )
+
+            assert len(set(paths)) == len(payloads)
+            assert {(uploads / path.rsplit("/", 1)[-1]).read_bytes() for path in paths} == set(payloads)
+
+        _run(go())
+
     def test_write_does_not_follow_planted_symlink(self, tmp_path, monkeypatch):
         """A symlink planted at the destination must not be written through.
 
@@ -2250,7 +2277,9 @@ class TestReceiveFile:
             out = await channel.receive_file(msg, "t1", user_id="default")
 
             assert not outside.exists()
-            assert "[failed to load image: image.png]" in out.text
+            assert out.text == f"{VIRTUAL_PATH_PREFIX}/uploads/image_1.png"
+            assert (uploads / "image.png").is_symlink()
+            assert (uploads / "image_1.png").read_bytes() == b"PWNED"
 
         _run(go())