fix(uploads): remove the staged .part after releasing its conversion descriptor (#5740)

The Gateway's staged upload commit links the `.part` file to its final name and
then unlinks the staged name. The conversion path duplicates a descriptor on the
staged inode *before* that commit (deliberately: it is what keeps conversion
reading the bytes this request wrote rather than whatever the committed name
points at afterwards), so the staged file still has an open handle when the
commit tries to remove it.

POSIX allows that; Windows does not. Uploading a convertible document
(`uploads.auto_convert_documents: true`, a PDF/DOCX/PPTX/XLSX upload) on a
Windows host therefore failed with a 500:

    Failed to upload report.pdf: [WinError 32] The process cannot access the
    file because it is being used by another process

Split publishing from removing the staged name: `_commit_upload_temp_no_overwrite`
now takes `unlink_staged`, and the ingestion service keeps ownership of the
staged path whenever it still holds the conversion descriptor, removing it once
that descriptor is released. The same best-effort removal replaces the
`os.unlink` in `_abort_upload_temp`, where an abandoned duplication worker can
hold the inode open and turn a cancellation into a secondary permission error.

Both removals are best-effort: a staged name that cannot be removed yet is
already hidden from every upload listing, and the Gateway sweeps leftovers on
startup.

Verified on Windows (Python 3.12):
- the new regression test reproduces the sharing violation portably (it pins the
  staged name while the descriptor lives) and fails on main with the exact 500
  above; it passes here
- `pytest tests/test_uploads_router.py tests/test_project_documents_promotion.py
  tests/blocking_io/test_project_documents_promotion.py` -> 88 passed
  (16 of those failed on main on this host)
- `ruff check` / `ruff format --check` clean

Co-authored-by: Shxiao101 <Shxiao101@users.noreply.github.com>

backend/app/gateway/routers/uploads.py

@@ -233,7 +233,13 @@ def _prepare_upload_destination(uploads_dir: os.PathLike[str] | str, display_fil
     return _UploadTempFile(file_path=file_path, temp_path=temp_path, handle=handle)
 
 
-def _link_staged_no_overwrite(staged_path: Path, uploads_dir: os.PathLike[str] | str, display_filename: str) -> Path:
+def _link_staged_no_overwrite(
+    staged_path: Path,
+    uploads_dir: os.PathLike[str] | str,
+    display_filename: str,
+    *,
+    unlink_staged: bool = True,
+) -> Path:
     """Worker: publish *staged_path* under *display_filename* atomically, never overwriting.
 
     The ``os.link`` itself is the whole no-overwrite guard: it fails with
@@ -250,6 +256,12 @@ def _link_staged_no_overwrite(staged_path: Path, uploads_dir: os.PathLike[str] |
     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.
+
+    ``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.
     """
     file_path = _pure_destination(uploads_dir, display_filename)
     try:
@@ -267,20 +279,44 @@ def _link_staged_no_overwrite(staged_path: Path, uploads_dir: os.PathLike[str] |
         except FileNotFoundError:
             pass
         raise
-    os.unlink(staged_path)
+    if unlink_staged:
+        os.unlink(staged_path)
     return file_path
 
 
-def _commit_upload_temp_no_overwrite(upload_temp: _UploadTempFile, uploads_dir: os.PathLike[str] | str, display_filename: str) -> Path:
+def _commit_upload_temp_no_overwrite(
+    upload_temp: _UploadTempFile,
+    uploads_dir: os.PathLike[str] | str,
+    display_filename: str,
+    *,
+    unlink_staged: bool = True,
+) -> Path:
     """Worker: close the staged handle and publish the ``.part`` atomically via ``os.link``.
 
     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.
+    other failure removes it. ``unlink_staged=False`` hands the staged name
+    back to the caller, which still holds a descriptor on its inode.
     """
     upload_temp.handle.close()
-    return _link_staged_no_overwrite(upload_temp.temp_path, uploads_dir, display_filename)
+    return _link_staged_no_overwrite(upload_temp.temp_path, uploads_dir, display_filename, unlink_staged=unlink_staged)
+
+
+def _remove_staged_file(staged_path: os.PathLike[str] | str) -> None:
+    """Worker: remove a staged ``.part`` name whose inode is no longer held open.
+
+    The deferred half of ``unlink_staged=False``. Removal is best-effort: the
+    name is already unreachable through the uploads listing, and failing an
+    otherwise-committed upload over leftover staging bytes would be worse than
+    leaving them for the startup sweep.
+    """
+    try:
+        os.unlink(staged_path)
+    except FileNotFoundError:
+        pass
+    except OSError:
+        logger.warning("Failed to remove staged upload file: %s", staged_path, exc_info=True)
 
 
 def _write_upload_chunk(upload_temp: _UploadTempFile, chunk: bytes) -> None:
@@ -291,10 +327,11 @@ def _abort_upload_temp(upload_temp: _UploadTempFile) -> None:
     try:
         upload_temp.handle.close()
     finally:
-        try:
-            os.unlink(upload_temp.temp_path)
-        except FileNotFoundError:
-            pass
+        # Best-effort, not ``os.unlink``: an abandoned duplication worker can
+        # still hold this inode open (Windows refuses to remove it then), and
+        # raising here would replace the caller's cancellation or original
+        # failure with a secondary permission error.
+        _remove_staged_file(upload_temp.temp_path)
 
 
 def _make_uploaded_paths_sandbox_readable(paths: list[os.PathLike[str] | str]) -> None:

backend/app/gateway/upload_ingestion.py

@@ -249,6 +249,7 @@ class ThreadUploadIngestionService:
         file_size = 0
         upload_temp = None
         convert_source_fd: int | None = None
+        deferred_staged_path: Path | None = None
         try:
             upload_temp = await run_file_io(uploads._prepare_upload_destination, self._uploads_dir, safe_filename)
             async for chunk in chunks:
@@ -269,12 +270,26 @@ class ThreadUploadIngestionService:
             # Link-commit with collision retry: the FileExistsError arm
             # leaves the staged part in place for the retry under the next
             # suffix (the handle's second close is idempotent).
+            #
+            # The conversion descriptor above keeps the staged inode open, and
+            # Windows refuses to remove a file that still has an open handle.
+            # Publishing is therefore split from removing the staged name:
+            # this request keeps ownership of the staged path and removes it
+            # once the descriptor is released below.
             while True:
                 try:
-                    file_path = await run_file_io(uploads._commit_upload_temp_no_overwrite, upload_temp, self._uploads_dir, safe_filename)
+                    file_path = await run_file_io(
+                        uploads._commit_upload_temp_no_overwrite,
+                        upload_temp,
+                        self._uploads_dir,
+                        safe_filename,
+                        unlink_staged=convert_source_fd is None,
+                    )
                     break
                 except FileExistsError:
                     safe_filename = uploads.claim_unique_filename(safe_filename, self._seen_filenames)
+            if convert_source_fd is not None:
+                deferred_staged_path = upload_temp.temp_path
             upload_temp = None
         except uploads.UnsafeUploadPathError as exc:
             _close_fd(convert_source_fd)
@@ -343,6 +358,10 @@ class ThreadUploadIngestionService:
                 raise
             finally:
                 _close_fd(convert_source_fd)
+                if deferred_staged_path is not None:
+                    # The descriptor that pinned the staged inode is released
+                    # by now, so the deferred staged-name removal can land.
+                    await run_file_io(uploads._remove_staged_file, deferred_staged_path)
                 if private_dir is not None:
                     await run_file_io(shutil.rmtree, private_dir, True)
             if not md_staged:

backend/tests/test_uploads_router.py

@@ -372,6 +372,77 @@ def test_upload_files_makes_non_local_files_sandbox_writable(tmp_path):
     make_writable.assert_any_call(thread_uploads_dir / "report.md")
 
 
+def test_upload_files_removes_the_staged_part_when_the_platform_blocks_an_open_unlink(tmp_path, monkeypatch):
+    """A staged ``.part`` that is still open must neither fail the upload nor leak.
+
+    Windows refuses to remove a file that still has an open handle, and the
+    conversion path deliberately holds a descriptor on the staged inode across
+    the link commit (that descriptor is what keeps conversion reading the bytes
+    this request wrote). Reproduce the resulting sharing violation portably by
+    pinning the staged name for as long as the descriptor lives: the commit's
+    own removal attempt would raise ``PermissionError``, and the deferred one
+    after the descriptor is released succeeds.
+    """
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = False
+    provider.acquire.side_effect = AssertionError("upload route should use acquire_async")
+    provider.acquire_async = AsyncMock(return_value="aio-1")
+    sandbox = MagicMock()
+    provider.get.return_value = sandbox
+
+    async def fake_convert(file_path: Path, output_path: Path | None = None) -> Path:
+        md_path = output_path if output_path is not None else file_path.with_suffix(".md")
+        md_path.write_text("converted", encoding="utf-8")
+        return md_path
+
+    pinned: set[str] = set()
+    real_unlink = os.unlink
+    real_prepare_upload_destination = uploads._prepare_upload_destination
+    real_copy_fd_to_path = upload_ingestion._copy_fd_to_path
+
+    def prepare_upload_destination(uploads_dir, display_filename):
+        upload_temp = real_prepare_upload_destination(uploads_dir, display_filename)
+        pinned.add(str(upload_temp.temp_path))
+        return upload_temp
+
+    def copy_fd_to_path(fd: int, dest: Path) -> None:
+        try:
+            real_copy_fd_to_path(fd, dest)
+        finally:
+            # The descriptor is gone: the staged inode is removable again.
+            pinned.clear()
+
+    def unlink(path, *args, **kwargs):
+        if str(path) in pinned:
+            raise PermissionError(32, "The process cannot access the file because it is being used by another process")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(uploads, "_prepare_upload_destination", prepare_upload_destination)
+    monkeypatch.setattr(upload_ingestion, "_copy_fd_to_path", copy_fd_to_path)
+    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=provider),
+        patch.object(uploads, "_auto_convert_documents_enabled", return_value=True),
+        patch.object(uploads, "convert_file_to_markdown", AsyncMock(side_effect=fake_convert)),
+        patch.object(uploads, "_make_file_sandbox_writable"),
+    ):
+        file = UploadFile(filename="report.pdf", file=BytesIO(b"pdf-bytes"))
+        result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-aio", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    assert result.success is True
+    assert result.files[0].filename == "report.pdf"
+    assert result.files[0].markdown_file == "report.md"
+    assert (thread_uploads_dir / "report.pdf").read_bytes() == b"pdf-bytes"
+    assert (thread_uploads_dir / "report.md").read_text(encoding="utf-8") == "converted"
+    assert not list(thread_uploads_dir.glob(f"{uploads.UPLOAD_STAGING_PREFIX}*{uploads.UPLOAD_STAGING_SUFFIX}"))
+
+
 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)