fix(channels): preserve inbound files on concurrent uploads (#6476)

* fix(channels): preserve inbound files on concurrent uploads

Claim IM attachment destinations exclusively and retry filename collisions.
Preserve default replacement behavior and Gateway conversion-link compatibility.

* fix(channels): claim embedded upload destinations exclusively

Retry competing publications in Feishu and DingTalk without replacing their bytes.
Cover mounted and remote sandbox paths, and keep returned filenames consistent.

---------

Signed-off-by: LingYi <lingyi01525@gmail.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -867,6 +867,11 @@ DeerFlow can also expose user-owned IM channel connections in the workspace UI.
 | DingTalk | Stream Push (WebSocket) | Moderate |
 | Buzz | Nostr relay (WebSocket, NIP-42) | Moderate |
 
+Attachments saved by the shared IM ingestion pipeline or Feishu/DingTalk's
+embedded downloads keep distinct filenames, including when concurrent uploads
+choose the same name. The final filename is passed to the agent and used for
+sandbox sync; an existing conversation file is not overwritten.
+
 **Configuration in `config.yaml`:**
 
 Discord's `channels.discord.allowed_guilds` accepts one positive numeric guild

backend/app/channels/AGENTS.md

@@ -45,7 +45,7 @@ Cursor writes drain before cancellation; token expiry drains cursor/auth clearin
 
 **Owner-scoped file storage**: inbound files, uploads, and output artifacts are staged under the DeerFlow owner's bucket so they land where the agent run reads/writes (`users/{user_id}/threads/{thread_id}/user-data/{uploads,outputs}`). `ChannelManager._handle_chat` resolves the storage owner once via `_channel_storage_user_id(msg)` (sanitized owner id, falling back to `safe(msg.user_id)` for unbound auth-enabled channels — mirroring `_resolve_run_params`'s run identity; `None` only when no identity is available) and threads it as the `user_id=` kwarg through the file pipeline:
 - `Channel.receive_file(msg, thread_id, user_id=...)` — owner-bound channels persist downloaded files under the owner's bucket instead of the default bucket
-- `FeishuChannel._receive_single_file(...)` / `DingTalkChannel._receive_single_file(...)` — normalize provider filenames, claim a collision-free basename and write it through `write_upload_file_no_symlink` under the same channel lock; the returned basename drives both the agent-visible virtual path and non-local sandbox sync
+- `FeishuChannel._receive_single_file(...)` / `DingTalkChannel._receive_single_file(...)` — normalize filenames and claim/write through `write_upload_file_no_symlink(..., exclusive=True)`, retrying `FileExistsError` under the channel lock; the final basename drives agent-visible paths and sandbox sync
 - `sandbox_files.py` — non-mounted Feishu/DingTalk syncs acquire unique execution holders, drain blocking `update_file` workers across repeated cancellation, and release only after the last sandbox operation, so a parallel run cannot close the shared client mid-upload. Feishu requests provider release after the final holder exits, parking standalone temporary sandboxes; DingTalk retains its non-releasing holder behavior.
 - `_ingest_inbound_files(...)` and the underlying `ensure_uploads_dir` / `get_uploads_dir` — owner-scoped via the same kwarg
 - `_resolve_attachments` / `_prepare_artifact_delivery` — resolve output artifacts from the bound owner's bucket through `app.gateway.path_utils.resolve_outputs_confined_path`, the same outputs-only rule the artifact editor uses, so a sibling `uploads/`/`workspace/` path or a symlink planted in `outputs/` is skipped with a warning

backend/app/channels/dingtalk.py

@@ -672,22 +672,19 @@ class DingTalkChannel(Channel):
 
         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.
+            # filesystem IO — the whole sequence stays off the event loop.
+            # Exclusive creation also protects against publishers outside this
+            # channel instance's lock.
             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()}
-                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)
+                while True:
+                    unique_name = claim_unique_filename(safe_filename, seen)
+                    try:
+                        return write_upload_file_no_symlink(uploads_dir, unique_name, content, exclusive=True)
+                    except FileExistsError:
+                        continue
 
         try:
             resolved_target = await asyncio.to_thread(_persist)

backend/app/channels/feishu.py

@@ -495,8 +495,12 @@ class FeishuChannel(Channel):
             uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve()
             with self._thread_lock:
                 seen = {entry.name for entry in uploads_dir.iterdir()}
-                unique_name = claim_unique_filename(safe_filename, seen)
-                return write_upload_file_no_symlink(uploads_dir, unique_name, content)
+                while True:
+                    unique_name = claim_unique_filename(safe_filename, seen)
+                    try:
+                        return write_upload_file_no_symlink(uploads_dir, unique_name, content, exclusive=True)
+                    except FileExistsError:
+                        continue
 
         try:
             resolved_target = await asyncio.to_thread(_persist)

backend/app/channels/manager.py

@@ -1188,7 +1188,14 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
 
             dest = uploads_dir / safe_name
             try:
-                dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data)
+                while True:
+                    try:
+                        dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data, exclusive=True)
+                        break
+                    except FileExistsError:
+                        # Another upload can claim a name after the directory
+                        # snapshot, including while this attachment downloads.
+                        safe_name = claim_unique_filename(safe_name, seen_names)
                 # Root-written 0o600 files are unreadable to the non-root
                 # sandbox; grant group/other read like the HTTP upload path.
                 await asyncio.to_thread(_make_inbound_file_sandbox_readable, dest)

backend/packages/harness/deerflow/uploads/manager.py

@@ -163,8 +163,8 @@ def validate_path_traversal(path: Path, base: Path) -> None:
         raise PathTraversalError("Path traversal detected") from None
 
 
-def validate_upload_destination(base_dir: Path, filename: str) -> Path:
-    """Validate an upload destination without mutating an existing file."""
+def validate_upload_destination(base_dir: Path, filename: str, *, exclusive: bool = False) -> Path:
+    """Validate a destination; exclusive creation treats regular hardlinks as collisions."""
     safe_name = normalize_filename(filename)
     dest = base_dir / safe_name
 
@@ -175,7 +175,7 @@ def validate_upload_destination(base_dir: Path, filename: str) -> Path:
 
     if st is not None and not stat.S_ISREG(st.st_mode):
         raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}")
-    if st is not None and st.st_nlink > 1:
+    if not exclusive and st is not None and st.st_nlink > 1:
         raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
 
     validate_path_traversal(dest, base_dir)
@@ -248,7 +248,7 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, mi
     return removed
 
 
-def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, object]:
+def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: bool = False) -> tuple[Path, object]:
     """Open an upload destination for safe streaming writes.
 
     Upload directories may be mounted into local sandboxes. A sandbox process can
@@ -259,9 +259,13 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob
     and ``fstat`` validation after ``open()`` to reduce the TOCTOU window; this does
     not eliminate all races but makes exploitation significantly harder. Path-traversal
     validation prevents escapes from *base_dir* in both cases.
+
+    ``exclusive=True`` atomically claims a new filename. An existing regular
+    file raises ``FileExistsError`` without truncation so the caller can retry
+    under another name. The default retains replacement semantics.
     """
     safe_name = normalize_filename(filename)
-    dest = validate_upload_destination(base_dir, safe_name)
+    dest = validate_upload_destination(base_dir, safe_name, exclusive=exclusive)
     try:
         st = os.lstat(dest)
     except FileNotFoundError:
@@ -272,6 +276,8 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob
     if has_nofollow:
         # POSIX: O_NOFOLLOW makes open() fail with ELOOP if dest is a symlink.
         flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW
+        if exclusive:
+            flags |= os.O_EXCL
         if hasattr(os, "O_NONBLOCK"):
             flags |= os.O_NONBLOCK
 
@@ -299,10 +305,12 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob
     # Note: a narrow race window remains between the pre-open lstat and open(); the
     # path-traversal check mitigates escapes from base_dir but cannot prevent an
     # attacker who can atomically replace dest with a symlink after the check.
-    if st is not None and st.st_nlink > 1:
+    if not exclusive and st is not None and st.st_nlink > 1:
         raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
 
     flags = os.O_WRONLY | os.O_CREAT
+    if exclusive:
+        flags |= os.O_EXCL
     if hasattr(os, "O_BINARY"):
         flags |= os.O_BINARY
 
@@ -313,7 +321,7 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob
 
     if pre_open_st is not None and not stat.S_ISREG(pre_open_st.st_mode):
         raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}")
-    if pre_open_st is not None and pre_open_st.st_nlink > 1:
+    if not exclusive and pre_open_st is not None and pre_open_st.st_nlink > 1:
         raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
 
     try:
@@ -336,9 +344,9 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob
     return dest, fh
 
 
-def write_upload_file_no_symlink(base_dir: Path, filename: str, data: bytes) -> Path:
+def write_upload_file_no_symlink(base_dir: Path, filename: str, data: bytes, *, exclusive: bool = False) -> Path:
     """Write upload bytes without following a pre-existing destination symlink."""
-    dest, fh = open_upload_file_no_symlink(base_dir, filename)
+    dest, fh = open_upload_file_no_symlink(base_dir, filename, exclusive=exclusive)
     with fh:
         fh.write(data)
     return dest

backend/tests/test_channel_file_attachments.py

@@ -282,6 +282,105 @@ class TestResolveAttachments:
 
 
 class TestInboundFileIngestion:
+    def test_concurrent_same_name_downloads_preserve_both_payloads(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        payloads = [b"first attachment" * 127, b"second attachment" * 91]
+
+        async def scenario():
+            # Each reader starts after its call's directory snapshot. Hold
+            # both downloads there so neither snapshot can see the other file.
+            downloads_ready = asyncio.Event()
+            arrivals = 0
+
+            async def reader(file_info, client):
+                nonlocal arrivals
+                arrivals += 1
+                if arrivals == 2:
+                    downloads_ready.set()
+                await asyncio.wait_for(downloads_ready.wait(), timeout=5)
+                return payloads[file_info["index"]]
+
+            messages = [InboundMessage(channel_name="test-channel", chat_id="chat-1", user_id="user-1", text="attachment", files=[{"filename": "report.txt", "index": index}]) for index in range(2)]
+            with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}):
+                tasks = [asyncio.create_task(manager._ingest_inbound_files("thread-1", msg)) for msg in messages]
+                try:
+                    return await asyncio.wait_for(asyncio.gather(*tasks), timeout=10)
+                finally:
+                    downloads_ready.set()
+                    for task in tasks:
+                        if not task.done():
+                            task.cancel()
+                    await asyncio.gather(*tasks, return_exceptions=True)
+
+        results = asyncio.run(scenario())
+        assert all(len(batch) == 1 for batch in results)
+        names = [batch[0]["filename"] for batch in results]
+        assert len(set(names)) == 2
+        for batch, payload in zip(results, payloads, strict=True):
+            saved = batch[0]
+            assert (uploads_dir / saved["filename"]).read_bytes() == payload
+            assert saved["path"] == f"/mnt/user-data/uploads/{saved['filename']}"
+            assert saved["size"] == len(payload)
+
+    def test_file_created_during_download_is_not_overwritten(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        prior_payload = b"another request published this file"
+        new_payload = b"this attachment arrived later"
+
+        async def reader(file_info, client):
+            # A second uploader wins both the original and first suffix
+            # after this ingestion has taken its directory snapshot.
+            for name in ("report.txt", "report_1.txt"):
+                await asyncio.to_thread((uploads_dir / name).write_bytes, prior_payload)
+            return new_payload
+
+        msg = InboundMessage(channel_name="test-channel", chat_id="chat-1", user_id="user-1", text="attachment", files=[{"filename": "report.txt"}])
+        with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}):
+            result = asyncio.run(manager._ingest_inbound_files("thread-1", msg))
+
+        assert len(result) == 1
+        assert result[0]["filename"] not in {"report.txt", "report_1.txt"}
+        assert (uploads_dir / result[0]["filename"]).read_bytes() == new_payload
+        for name in ("report.txt", "report_1.txt"):
+            assert (uploads_dir / name).read_bytes() == prior_payload
+
+    def test_gateway_conversion_staging_link_is_a_filename_collision(self, tmp_path):
+        from app.channels import manager
+        from app.gateway.routers.uploads import _link_staged_no_overwrite
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        gateway_payload = b"gateway upload still being converted"
+        inbound_payload = b"independent IM attachment"
+
+        def publish_gateway_upload():
+            staged = uploads_dir / ".upload-fixture.part"
+            staged.write_bytes(gateway_payload)
+            # Gateway retains the staged link while a conversion descriptor
+            # owns this inode. This is an ordinary upload, not an unsafe path.
+            return _link_staged_no_overwrite(staged, uploads_dir, "report.pdf", unlink_staged=False)
+
+        async def reader(file_info, client):
+            published = await asyncio.to_thread(publish_gateway_upload)
+            assert (await asyncio.to_thread(published.stat)).st_nlink == 2
+            return inbound_payload
+
+        msg = InboundMessage(channel_name="test-channel", chat_id="chat-1", user_id="user-1", text="attachment", files=[{"filename": "report.pdf"}])
+        with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}):
+            result = asyncio.run(manager._ingest_inbound_files("thread-1", msg))
+
+        assert len(result) == 1
+        assert result[0]["filename"] != "report.pdf"
+        assert (uploads_dir / result[0]["filename"]).read_bytes() == inbound_payload
+        assert (uploads_dir / "report.pdf").read_bytes() == gateway_payload
+        assert (uploads_dir / ".upload-fixture.part").read_bytes() == gateway_payload
+
     def test_case_variant_names_keep_both_inbound_attachments(self, tmp_path):
         from app.channels import manager
 
@@ -1479,3 +1578,78 @@ class TestWeChatDownloadGuardLabels:
             assert channel._stage_downloaded_file("photo.bin", b"x") is None
 
         assert "no state directory configured" in caplog.text
+
+
+@pytest.mark.parametrize("channel_name", ["feishu", "dingtalk"])
+@pytest.mark.parametrize("mounted", [False, True])
+def test_embedded_upload_retries_claim_collision(tmp_path, monkeypatch, channel_name, mounted):
+    """A publisher winning after the filename claim must keep its own bytes."""
+    import importlib
+    from io import BytesIO
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+
+    from deerflow.config.paths import Paths
+    from deerflow.uploads.manager import write_upload_file_no_symlink
+
+    paths = Paths(str(tmp_path))
+    paths.ensure_thread_dirs("thread-1", user_id="owner")
+    uploads = paths.sandbox_uploads_dir("thread-1", user_id="owner")
+    module = importlib.import_module(f"app.channels.{channel_name}")
+    monkeypatch.setattr(module, "get_paths", lambda: paths)
+    synced = []
+
+    class Provider:
+        uses_thread_data_mounts = mounted
+
+        async def acquire_async(self, thread_id=None, *, user_id=None):
+            assert not mounted
+            assert (thread_id, user_id) == ("thread-1", "owner")
+            return "remote-sandbox"
+
+        def get(self, sandbox_id):
+            return SimpleNamespace(update_file=lambda path, content: synced.append((path, content)), release_command_scope=lambda scope: None)
+
+        def release(self, sandbox_id):
+            pass
+
+    provider = Provider()
+    monkeypatch.setattr(module, "get_sandbox_provider", lambda: provider)
+    claim = module.claim_unique_filename
+    other_files = {}
+
+    def publish_after_claim(name, seen):
+        chosen = claim(name, seen)
+        if len(other_files) < 2:
+            content = f"other-publisher-{len(other_files)}".encode()
+            write_upload_file_no_symlink(uploads, chosen, content, exclusive=True)
+            other_files[chosen] = content
+        return chosen
+
+    monkeypatch.setattr(module, "claim_unique_filename", publish_after_claim)
+    payload = b"CHANNEL ATTACHMENT"
+    if channel_name == "feishu":
+        channel = module.FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"})
+        channel._GetMessageResourceRequest = MagicMock()
+        response = MagicMock()
+        response.success.return_value = True
+        response.file = BytesIO(payload)
+        response.file_name = "report.pdf"
+        channel._api_client = MagicMock()
+        channel._api_client.im.v1.message_resource.get.return_value = response
+        files = [{"file_key": "file-key"}]
+    else:
+        channel = module.DingTalkChannel(MessageBus(), config={})
+        channel._download_by_code = AsyncMock(return_value=payload)
+        files = [{"type": "file", "download_code": "file-key", "filename": "report.pdf"}]
+    message = InboundMessage(channel_name=channel_name, chat_id="chat-1", user_id="sender", thread_ts="message-1", text="[file]", files=files)
+
+    result = _run(channel.receive_file(message, "thread-1", user_id="owner"))
+
+    for name, content in other_files.items():
+        assert (uploads / name).read_bytes() == content
+    assert set(other_files) == {"report.pdf", "report_1.pdf"}
+    assert (uploads / "report_2.pdf").read_bytes() == payload
+    advertised = "/mnt/user-data/uploads/report_2.pdf"
+    assert advertised in result.text
+    assert synced == ([] if mounted else [(advertised, payload)])

backend/tests/test_feishu_parser.py

@@ -427,12 +427,12 @@ def test_feishu_receive_file_path_traversal_failure_is_per_attachment(tmp_path,
         real_write = feishu_module.write_upload_file_no_symlink
         call_count = 0
 
-        def flaky_write(base_dir, filename, data):
+        def flaky_write(base_dir, filename, data, *, exclusive=False):
             nonlocal call_count
             call_count += 1
             if call_count == 1:
                 raise PathTraversalError("Path traversal detected")
-            return real_write(base_dir, filename, data)
+            return real_write(base_dir, filename, data, exclusive=exclusive)
 
         monkeypatch.setattr(feishu_module, "write_upload_file_no_symlink", flaky_write)
         msg = InboundMessage(

backend/tests/test_uploads_manager.py

@@ -233,6 +233,45 @@ class TestValidatePathTraversal:
 
 
 class TestWriteUploadFileNoSymlink:
+    @pytest.mark.parametrize("without_nofollow", [False, True])
+    def test_exclusive_write_treats_existing_hardlink_as_collision(self, tmp_path, monkeypatch, without_nofollow):
+        if without_nofollow:
+            monkeypatch.delattr(os, "O_NOFOLLOW", raising=False)
+        source = tmp_path / "source.txt"
+        source.write_bytes(b"existing upload")
+        os.link(source, tmp_path / "notes.txt")
+
+        with pytest.raises(FileExistsError):
+            write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True)
+        # The default replacement path still rejects hardlinks outright.
+        with pytest.raises(UnsafeUploadPathError, match="multiple links"):
+            write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement")
+
+        assert source.read_bytes() == b"existing upload"
+        assert (tmp_path / "notes.txt").read_bytes() == b"existing upload"
+
+    @pytest.mark.parametrize("without_nofollow", [False, True])
+    def test_exclusive_write_preserves_existing_file(self, tmp_path, monkeypatch, without_nofollow):
+        if without_nofollow:
+            monkeypatch.delattr(os, "O_NOFOLLOW", raising=False)
+        dest = tmp_path / "notes.txt"
+        dest.write_bytes(b"previous upload")
+
+        with pytest.raises(FileExistsError):
+            write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True)
+
+        assert dest.read_bytes() == b"previous upload"
+
+    @pytest.mark.parametrize("without_nofollow", [False, True])
+    def test_exclusive_write_creates_new_file(self, tmp_path, monkeypatch, without_nofollow):
+        if without_nofollow:
+            monkeypatch.delattr(os, "O_NOFOLLOW", raising=False)
+
+        dest = write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True)
+
+        assert dest == tmp_path / "notes.txt"
+        assert dest.read_bytes() == b"new attachment"
+
     def test_writes_new_file(self, tmp_path):
         dest = write_upload_file_no_symlink(tmp_path, "notes.txt", b"hello")