fix(uploads): preserve link errors during staged cleanup (#5756)

* fix(uploads): preserve link errors during staged cleanup

* Keep published uploads successful when staging cleanup fails

* docs: keep upload guidance within chain budget

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1784,6 +1784,11 @@ can set `sandbox.thread_data_mounts: true` to skip that per-upload sandbox
 acquire and sync. Leave the field unset for automatic detection; setting it
 incorrectly can make uploaded files unavailable inside the sandbox.
 
+Failed upload commits report the original error even if temporary-file cleanup
+also fails, for example because of a Windows sharing violation.
+Once the file is published, a temporary-file cleanup failure is logged without
+failing the upload; hidden staging files are left for the startup sweep.
+
 This is the difference between a chatbot with tool access and an agent with an actual execution environment.
 
 ```

backend/AGENTS.md

@@ -338,7 +338,7 @@ Outlines use ATX syntax (1–6 hashes, space/tab separator, ≤3 leading spaces)
 - One conversion worker per request when called from an active event loop
 - 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 within one request get `_N` suffixes to prevent overwrites.
-- 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 uploads stage `.upload-*.part` files, hidden from upload listings, agent context, and sandbox listings/searches. After size validation, publication is atomic; staged-name cleanup logs errors and leaves leftovers for startup sweep.
 - 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.
 - Mounted uploads skip sandbox acquire/sync. AIO remote/provisioner requires accurate `sandbox.thread_data_mounts: true`; omission keeps backend auto-detection.
 - `UploadsMiddleware` caps outline titles at 200 characters and previews at 2000 including markers. Titles use `original_user_content`, not upload-prefixed content; attachment-only titles use a sanitized, bounded filename or count.

backend/app/gateway/routers/uploads.py

@@ -253,15 +253,15 @@ def _link_staged_no_overwrite(
     happens AFTER the atomic failure: an existing non-regular file (a planted
     symlink — symlinks are excluded from the seeded listing, so one could
     only come from outside) stays an unsafe destination; anything else is a
-    plain collision to retry. Any other failure removes the staged file and
-    propagates; success unlinks it. Staging and destination are co-located in
-    the uploads dir, so the hard link is always same-filesystem.
+    plain collision to retry. Any other failure removes the staged file
+    best-effort and propagates; success also removes it best-effort. Staging and destination
+    are co-located in the uploads dir, so the hard link is always same-filesystem.
 
     ``unlink_staged=False`` publishes the link but leaves the staged name in
     place, for a caller that still holds a descriptor on the staged inode and
     therefore must remove it itself. Windows refuses to remove a file that
     has an open handle, so the removal cannot happen here in that case; the
-    caller owns the staged path from the moment this returns.
+    caller retains ownership of the staged path on both success and failure.
     """
     file_path = _pure_destination(uploads_dir, display_filename)
     try:
@@ -274,13 +274,11 @@ def _link_staged_no_overwrite(
             pass  # The winner vanished between link and lstat — plain retry.
         raise
     except Exception:
-        try:
-            os.unlink(staged_path)
-        except FileNotFoundError:
-            pass
+        if unlink_staged:
+            _remove_staged_file(staged_path)
         raise
     if unlink_staged:
-        os.unlink(staged_path)
+        _remove_staged_file(staged_path)
     return file_path
 
 
@@ -296,8 +294,9 @@ def _commit_upload_temp_no_overwrite(
     Same no-overwrite contract as :func:`_link_staged_no_overwrite`:
     :class:`FileExistsError` leaves the staged part in place for a
     next-suffix retry (the handle's second ``close`` is idempotent); any
-    other failure removes it. ``unlink_staged=False`` hands the staged name
-    back to the caller, which still holds a descriptor on its inode.
+    other failure removes it best-effort. With ``unlink_staged=False``, the
+    caller retains the staged name on success and failure until it releases
+    its descriptor on the inode.
     """
     upload_temp.handle.close()
     return _link_staged_no_overwrite(upload_temp.temp_path, uploads_dir, display_filename, unlink_staged=unlink_staged)

backend/tests/test_uploads_router.py

@@ -1,4 +1,5 @@
 import asyncio
+import errno
 import os
 import stat
 import threading
@@ -443,6 +444,104 @@ def test_upload_files_removes_the_staged_part_when_the_platform_blocks_an_open_u
     assert not list(thread_uploads_dir.glob(f"{uploads.UPLOAD_STAGING_PREFIX}*{uploads.UPLOAD_STAGING_SUFFIX}"))
 
 
+@pytest.mark.parametrize("filename", ["notes.txt", "report.pdf"])
+@pytest.mark.parametrize("simulate_sharing_violation", [False, True])
+def test_upload_files_preserves_link_error_and_cleans_staging_after_closing_conversion_fd(tmp_path, monkeypatch, filename, simulate_sharing_violation):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir()
+    conversion_fds: set[int] = set()
+    duplicated_fds: list[int] = []
+    blocked_unlinks: list[str] = []
+    real_dup = upload_ingestion._dup_for_conversion
+    real_close = upload_ingestion._close_fd
+    real_unlink = os.unlink
+
+    async def dup_for_conversion(fd):
+        duplicate = await real_dup(fd)
+        conversion_fds.add(duplicate)
+        duplicated_fds.append(duplicate)
+        return duplicate
+
+    def close_fd(fd):
+        real_close(fd)
+        conversion_fds.discard(fd)
+
+    def unlink(path, *args, **kwargs):
+        # Simulate Windows sharing violations on every platform, while keeping
+        # the real conversion descriptor and the production abort path.
+        if simulate_sharing_violation and conversion_fds and str(path).endswith(uploads.UPLOAD_STAGING_SUFFIX):
+            blocked_unlinks.append(str(path))
+            raise PermissionError("conversion descriptor is still open")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(upload_ingestion, "_dup_for_conversion", dup_for_conversion)
+    monkeypatch.setattr(upload_ingestion, "_close_fd", close_fd)
+    monkeypatch.setattr(os, "unlink", unlink)
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=_mounted_provider()),
+        patch.object(uploads, "_auto_convert_documents_enabled", return_value=True),
+        patch.object(os, "link", side_effect=OSError(errno.EIO, "original link failure")),
+    ):
+        file = UploadFile(filename=filename, file=BytesIO(b"upload bytes"))
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(call_unwrapped(uploads.upload_files, "thread-link-error", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    assert exc_info.value.status_code == 500
+    assert "original link failure" in exc_info.value.detail
+    assert not blocked_unlinks
+    assert not conversion_fds
+    assert len(duplicated_fds) == (1 if filename == "report.pdf" else 0)
+    assert list(thread_uploads_dir.iterdir()) == []
+
+
+def test_upload_files_succeeds_when_published_staging_cleanup_fails(tmp_path, monkeypatch, caplog):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir()
+    real_unlink = os.unlink
+
+    def unlink(path, *args, **kwargs):
+        if str(path).endswith(uploads.UPLOAD_STAGING_SUFFIX):
+            raise PermissionError("staged cleanup failure")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(os, "unlink", unlink)
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=_mounted_provider()),
+        patch.object(uploads, "_auto_convert_documents_enabled", return_value=False),
+    ):
+        file = UploadFile(filename="notes.txt", file=BytesIO(b"published bytes"))
+        result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-cleanup-error", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    assert result.success
+    assert (thread_uploads_dir / "notes.txt").read_bytes() == b"published bytes"
+    # The hidden staging name is left for the startup sweep, not reported as a
+    # failed upload after the destination has already been published.
+    staged_paths = list(thread_uploads_dir.glob(".upload-*.part"))
+    assert len(staged_paths) == 1
+    assert "Failed to remove staged upload file" in caplog.text
+
+
+def test_link_staged_preserves_original_error_when_cleanup_fails(tmp_path):
+    staged_path = tmp_path / "upload.part"
+    staged_path.write_bytes(b"upload bytes")
+    original_error = OSError(errno.EIO, "original link failure")
+    with (
+        patch.object(os, "link", side_effect=original_error),
+        patch.object(os, "unlink", side_effect=PermissionError("cleanup failure")) as unlink,
+        pytest.raises(OSError) as exc_info,
+    ):
+        uploads._link_staged_no_overwrite(staged_path, tmp_path, "report.pdf")
+
+    assert exc_info.value is original_error
+    unlink.assert_called_once_with(staged_path)
+    assert staged_path.read_bytes() == b"upload bytes"
+    assert not (tmp_path / "report.pdf").exists()
+
+
 def test_upload_files_does_not_adjust_permissions_for_local_sandbox(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)