fix(uploads): drain conversion workers before source cleanup (#5999)

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

backend/README.md

@@ -148,6 +148,11 @@ FastAPI application providing REST endpoints for frontend integration:
 | `DELETE /api/threads/{id}` | Delete DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail |
 | `GET /api/threads/{id}/artifacts/{path}` | Serve generated artifacts |
 
+Cancelling an upload waits for an already-running document conversion worker to
+finish before removing its temporary source. This prevents cleanup from deleting
+a file that the converter is still reading; cancellation can therefore take as
+long as that conversion.
+
 Converted-upload ownership records live in each thread's `upload-companions/`
 directory, outside the sandbox-mounted `user-data/` tree. Older conversions
 without a record remain separate Markdown uploads and no longer provide an

backend/packages/harness/deerflow/utils/AGENTS.md

@@ -39,6 +39,14 @@ boundary for artifacts and project documents. Keep platform aliases such as
 Windows' `image/svg` aligned with their standard active type (`image/svg+xml`),
 and preserve the generic `+xml` rule.
 
+### Conversion Worker Ownership
+
+`convert_file_to_markdown` drains large-file conversion workers across caller
+cancellation before returning control to source-file cleanup. Keep conversion
+inside `await_drained`, including when the worker fails or cancellation repeats;
+upload ingestion owns a temporary source that it removes in `finally`. Coverage
+lives in `tests/test_file_conversion_cancellation.py`.
+
 ### Host Path Portability
 
 `host_paths.py` rejects Windows device names for host-visible creation paths on

backend/packages/harness/deerflow/utils/file_conversion.py

@@ -166,7 +166,9 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N
         file_size = (await run_file_io(file_path.stat)).st_size
 
         if file_size > _ASYNC_THRESHOLD_BYTES:
-            text = await asyncio.to_thread(_do_convert, file_path, pdf_converter)
+            # Cancellation must not release the caller's source-file ownership
+            # while the converter is still reading it in the worker thread.
+            text = await await_drained(asyncio.to_thread(_do_convert, file_path, pdf_converter))
         else:
             text = _do_convert(file_path, pdf_converter)
 

backend/tests/test_file_conversion_cancellation.py

@@ -0,0 +1,73 @@
+"""A cancelled conversion must retain its source until its worker exits."""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+
+import pytest
+
+from deerflow.utils import file_conversion
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("conversion_fails", [False, True], ids=["success", "failure"])
+async def test_cancelled_conversion_drains_worker_before_source_cleanup(tmp_path, monkeypatch, conversion_fails):
+    source = tmp_path / "large.pdf"
+    source.write_bytes(b"x" * (file_conversion._ASYNC_THRESHOLD_BYTES + 1))
+    output = tmp_path / "converted.md"
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+    source_read = threading.Event()
+    cleanup_before_finish = []
+    loop = asyncio.get_running_loop()
+    worker_started = asyncio.Event()
+
+    def convert(path, _converter):
+        started.set()
+        loop.call_soon_threadsafe(worker_started.set)
+        try:
+            assert release.wait(timeout=10), "test did not release conversion worker"
+            # A converter may reopen/read the source throughout conversion.
+            assert path.read_bytes().startswith(b"x")
+            source_read.set()
+            if conversion_fails:
+                raise ValueError("synthetic converter failure")
+            return "# Converted document"
+        finally:
+            finished.set()
+
+    monkeypatch.setattr(file_conversion, "_get_pdf_converter", lambda: "auto")
+    monkeypatch.setattr(file_conversion, "_do_convert", convert)
+
+    async def ingest():
+        try:
+            return await file_conversion.convert_file_to_markdown(source, output_path=output)
+        finally:
+            # UploadIngestionSession removes its private source in a finally.
+            cleanup_before_finish.append(not finished.is_set())
+            source.unlink(missing_ok=True)
+
+    task = asyncio.create_task(ingest())
+    try:
+        await asyncio.wait_for(worker_started.wait(), timeout=10)
+        assert started.is_set()
+        task.cancel()
+        # Yield scheduler turns, not a wall-clock delay, so cancellation reaches
+        # the suspended await while the worker is held by the release event.
+        await asyncio.sleep(0)
+        task.cancel()
+        await asyncio.sleep(0)
+        assert not cleanup_before_finish, "caller cleaned the source while the converter was still running"
+        assert not task.done()
+    finally:
+        release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert await asyncio.to_thread(finished.wait, 10)
+
+    assert cleanup_before_finish == [False]
+    assert source_read.is_set()
+    assert not source.exists()
+    assert not output.exists()