fix: bind conversions to upload generations

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

@@ -1,4 +1,3 @@
-from .conversion import convert_uploaded_file_to_markdown
 from .errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError
 from .layout import (
     UPLOAD_CONVERSIONS_DIRNAME,
@@ -6,6 +5,7 @@ from .layout import (
     UnsafeConversionPathError,
     artifact_url_for_virtual_path,
     conversion_dir_for_uploads,
+    conversion_filename_for_upload,
     conversion_path_for_upload,
     conversion_virtual_path,
     ensure_conversion_dir,
@@ -44,12 +44,22 @@ from .manager import (
     validate_thread_id,
 )
 
+
+def __getattr__(name: str):
+    if name == "convert_uploaded_file_to_markdown":
+        from .conversion import convert_uploaded_file_to_markdown
+
+        return convert_uploaded_file_to_markdown
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
 __all__ = [
     "get_uploads_dir",
     "UPLOAD_CONVERSIONS_DIRNAME",
     "UPLOAD_LOCKS_DIRNAME",
     "UnsafeConversionPathError",
     "conversion_dir_for_uploads",
+    "conversion_filename_for_upload",
     "conversion_path_for_upload",
     "conversion_virtual_path",
     "artifact_url_for_virtual_path",

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

@@ -1,34 +1,178 @@
 """Safe publication of Markdown generated from primary uploads."""
 
+import asyncio
+import logging
+import os
+import stat
+from dataclasses import dataclass
 from pathlib import Path
 
+from deerflow.uploads.errors import UnsafeUploadPathError
 from deerflow.uploads.layout import (
     UnsafeConversionPathError,
     conversion_path_for_upload,
     ensure_conversion_dir,
 )
+from deerflow.uploads.lease import UploadIdentity, UploadNameLease
 from deerflow.uploads.manager import (
+    PublishedUpload,
+    StagedUpload,
     abort_staged_upload,
     create_upload_staging_file,
     replace_system_owned_staged_file,
 )
 from deerflow.utils.file_conversion import convert_file_to_markdown
 
+logger = logging.getLogger(__name__)
 
-async def convert_uploaded_file_to_markdown(upload_path: Path) -> Path | None:
-    """Convert one primary upload and atomically publish its owned Markdown."""
-    conversion_dir = ensure_conversion_dir(upload_path.parent)
-    target = conversion_path_for_upload(upload_path)
-    staged = create_upload_staging_file(conversion_dir)
-    staged.handle.close()
+
+@dataclass(slots=True)
+class _PreparedConversion:
+    publication: PublishedUpload
+    staged: StagedUpload
+    target: Path
+    release_publication: bool
+
+
+def _abort_stage_without_masking(staged: StagedUpload) -> None:
     try:
-        result = await convert_file_to_markdown(upload_path, output_path=staged.path)
-        if result is None:
-            abort_staged_upload(staged)
-            return None
-        if Path(result) != staged.path:
-            raise UnsafeConversionPathError("Converter returned an unexpected output path")
-        return replace_system_owned_staged_file(staged, target.name)
-    except Exception:
         abort_staged_upload(staged)
+    except BaseException:
+        logger.warning("Failed to clean up upload conversion staging file: %s", staged.path, exc_info=True)
+
+
+def _release_publication_without_masking(publication: PublishedUpload) -> None:
+    try:
+        publication.release()
+    except BaseException:
+        logger.warning("Failed to release upload conversion lease: %s", publication.lease.lock_path, exc_info=True)
+
+
+def _validate_publication(upload_path: Path, publication: PublishedUpload) -> None:
+    if publication.path != upload_path:
+        raise UnsafeUploadPathError("Upload publication does not match the conversion source")
+    if publication.lease.filename != upload_path.name or publication.lease.uploads_dir != upload_path.parent:
+        raise UnsafeUploadPathError("Upload publication lease does not match the conversion source")
+    if not publication.is_active:
+        raise UnsafeUploadPathError("Upload publication lease was already released")
+    if not publication.identity.matches(upload_path):
+        raise UnsafeUploadPathError("Upload generation changed before conversion")
+    upload_stat = os.lstat(upload_path)
+    if not stat.S_ISREG(upload_stat.st_mode) or upload_stat.st_nlink != 1:
+        raise UnsafeUploadPathError("Upload conversion source is not an exclusive regular file")
+
+
+def _discard_prepared_conversion(prepared: _PreparedConversion) -> None:
+    _abort_stage_without_masking(prepared.staged)
+    if prepared.release_publication:
+        _release_publication_without_masking(prepared.publication)
+
+
+def _prepare_conversion(upload_path: Path, publication: PublishedUpload | None) -> _PreparedConversion:
+    release_publication = publication is None
+    if publication is None:
+        lease = UploadNameLease.acquire(upload_path.parent, upload_path.name)
+        try:
+            publication = PublishedUpload(
+                path=upload_path,
+                identity=UploadIdentity.from_path(upload_path),
+                lease=lease,
+            )
+        except BaseException:
+            try:
+                lease.release()
+            except BaseException:
+                logger.warning("Failed to release upload conversion lease: %s", lease.lock_path, exc_info=True)
+            raise
+
+    staged: StagedUpload | None = None
+    try:
+        _validate_publication(upload_path, publication)
+        conversion_dir = ensure_conversion_dir(upload_path.parent)
+        target = conversion_path_for_upload(upload_path)
+        staged = create_upload_staging_file(conversion_dir)
+        staged.handle.close()
+        return _PreparedConversion(
+            publication=publication,
+            staged=staged,
+            target=target,
+            release_publication=release_publication,
+        )
+    except BaseException:
+        if staged is not None:
+            _abort_stage_without_masking(staged)
+        if release_publication:
+            _release_publication_without_masking(publication)
         raise
+
+
+def _publish_prepared_conversion(prepared: _PreparedConversion, result: Path) -> Path:
+    _validate_publication(prepared.publication.path, prepared.publication)
+    if result != prepared.staged.path:
+        raise UnsafeConversionPathError("Converter returned an unexpected output path")
+    return replace_system_owned_staged_file(prepared.staged, prepared.target.name)
+
+
+async def _prepare_conversion_cancellation_safe(
+    upload_path: Path,
+    publication: PublishedUpload | None,
+) -> _PreparedConversion:
+    prepare_task = asyncio.create_task(
+        asyncio.to_thread(_prepare_conversion, upload_path, publication),
+        name=f"prepare-upload-conversion:{upload_path.name}",
+    )
+    try:
+        return await asyncio.shield(prepare_task)
+    except asyncio.CancelledError:
+        while not prepare_task.done():
+            try:
+                await asyncio.shield(prepare_task)
+            except asyncio.CancelledError:
+                continue
+            except Exception:
+                break
+        if not prepare_task.cancelled() and prepare_task.exception() is None:
+            cleanup_task = asyncio.create_task(
+                asyncio.to_thread(_discard_prepared_conversion, prepare_task.result()),
+                name=f"discard-upload-conversion:{upload_path.name}",
+            )
+            while not cleanup_task.done():
+                try:
+                    await asyncio.shield(cleanup_task)
+                except asyncio.CancelledError:
+                    continue
+        raise
+
+
+async def _run_cleanup_off_thread(function, *args) -> None:
+    cleanup_task = asyncio.create_task(asyncio.to_thread(function, *args))
+    while not cleanup_task.done():
+        try:
+            await asyncio.shield(cleanup_task)
+        except asyncio.CancelledError:
+            continue
+    cleanup_task.result()
+
+
+async def convert_uploaded_file_to_markdown(
+    upload_path: Path,
+    *,
+    publication: PublishedUpload | None = None,
+) -> Path | None:
+    """Convert one primary generation and atomically publish its owned Markdown."""
+    prepared = await _prepare_conversion_cancellation_safe(upload_path, publication)
+    stage_consumed = False
+    try:
+        result = await convert_file_to_markdown(upload_path, output_path=prepared.staged.path)
+        if result is None:
+            await _run_cleanup_off_thread(_abort_stage_without_masking, prepared.staged)
+            stage_consumed = True
+            return None
+        converted = await asyncio.to_thread(_publish_prepared_conversion, prepared, Path(result))
+        stage_consumed = True
+        return converted
+    finally:
+        if not stage_consumed:
+            await _run_cleanup_off_thread(_abort_stage_without_masking, prepared.staged)
+        if prepared.release_publication:
+            await _run_cleanup_off_thread(_release_publication_without_masking, prepared.publication)

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

@@ -1,5 +1,6 @@
 """Path and URL layout helpers for primary uploads and generated assets."""
 
+import hashlib
 import os
 import stat
 from pathlib import Path
@@ -20,9 +21,27 @@ def conversion_dir_for_uploads(uploads_dir: Path) -> Path:
     return uploads_dir.parent / UPLOAD_CONVERSIONS_DIRNAME
 
 
+def _truncate_utf8(value: str, max_bytes: int) -> str:
+    encoded = value.encode("utf-8")
+    if len(encoded) <= max_bytes:
+        return value
+    return encoded[:max_bytes].decode("utf-8", errors="ignore")
+
+
+def conversion_filename_for_upload(filename: str) -> str:
+    """Return a deterministic generated-Markdown component within 255 bytes."""
+    desired = f"{filename}.md"
+    if len(desired.encode("utf-8")) <= 255:
+        return desired
+    digest = hashlib.sha256(filename.encode("utf-8")).hexdigest()
+    marker = f".{digest}.md"
+    prefix = _truncate_utf8(filename, 255 - len(marker.encode("utf-8")))
+    return f"{prefix}{marker}"
+
+
 def conversion_path_for_upload(upload_path: Path) -> Path:
     """Return the generated Markdown path owned by one primary upload."""
-    return conversion_dir_for_uploads(upload_path.parent) / f"{upload_path.name}.md"
+    return conversion_dir_for_uploads(upload_path.parent) / conversion_filename_for_upload(upload_path.name)
 
 
 def validate_conversion_dir(uploads_dir: Path) -> Path | None:
@@ -93,7 +112,7 @@ def upload_virtual_path(filename: str) -> str:
 
 def conversion_virtual_path(filename: str) -> str:
     """Build the sandbox virtual path for an upload's generated Markdown."""
-    return f"{VIRTUAL_PATH_PREFIX}/{UPLOAD_CONVERSIONS_DIRNAME}/{filename}.md"
+    return f"{VIRTUAL_PATH_PREFIX}/{UPLOAD_CONVERSIONS_DIRNAME}/{conversion_filename_for_upload(filename)}"
 
 
 def artifact_url_for_virtual_path(thread_id: str, virtual_path: str) -> str:

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

@@ -20,6 +20,7 @@ from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.uploads.errors import AtomicUploadPublishError, PathTraversalError, UnsafeUploadPathError
 from deerflow.uploads.layout import (
     UPLOAD_CONVERSIONS_DIRNAME,
+    _truncate_utf8,
     artifact_url_for_virtual_path,
     existing_conversion_path_for_upload,
     upload_virtual_path,
@@ -192,13 +193,6 @@ def _validate_staged_upload(staged: StagedUpload) -> None:
         raise UnsafeUploadPathError("Upload staging path is not an exclusive regular file")
 
 
-def _truncate_utf8(value: str, max_bytes: int) -> str:
-    encoded = value.encode("utf-8")
-    if len(encoded) <= max_bytes:
-        return value
-    return encoded[:max_bytes].decode("utf-8", errors="ignore")
-
-
 def _filename_candidates(name: str) -> Iterator[str]:
     """Yield collision candidates that stay within the filename byte limit."""
     yield name

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

@@ -8,8 +8,8 @@ PDF conversion strategy (auto mode):
      total when page count is unavailable), treat as image-based and fall back to MarkItDown.
   3. If pymupdf4llm is not installed, use MarkItDown directly (existing behaviour).
 
-Large files (> ASYNC_THRESHOLD_BYTES) are converted in a thread pool via
-asyncio.to_thread() to avoid blocking the event loop (fixes #1569).
+Parser work and generated-Markdown writes run in a worker thread so document
+conversion never blocks an async caller's event loop.
 
 No FastAPI or HTTP dependencies — pure utility functions.
 """
@@ -39,11 +39,6 @@ CONVERTIBLE_EXTENSIONS = {
     ".docx",
 }
 
-# Files larger than this threshold are converted in a background thread.
-# Small files complete in < 1s synchronously; spawning a thread adds unnecessary
-# scheduling overhead for them.
-_ASYNC_THRESHOLD_BYTES = 1 * 1024 * 1024  # 1 MB
-
 # If pymupdf4llm produces fewer characters *per page* than this threshold,
 # the PDF is likely image-based or encrypted — fall back to MarkItDown.
 # Rationale: normal text PDFs yield 200-2000 chars/page; image-based PDFs
@@ -140,12 +135,18 @@ def _do_convert(file_path: Path, pdf_converter: str) -> str:
     return _convert_with_markitdown(file_path)
 
 
+def _convert_file_to_markdown_sync(file_path: Path, output_path: Path | None) -> Path:
+    pdf_converter = _get_pdf_converter()
+    text = _do_convert(file_path, pdf_converter)
+    md_path = output_path if output_path is not None else file_path.with_suffix(".md")
+    md_path.write_text(text, encoding="utf-8")
+    return md_path
+
+
 async def convert_file_to_markdown(file_path: Path, output_path: Path | None = None) -> Path | None:
-    """Convert a supported document file to Markdown.
+    """Convert a supported document file to Markdown off the event loop.
 
     PDF files are handled with a two-converter strategy (see module docstring).
-    Large files (> 1 MB) are offloaded to a thread pool to avoid blocking the
-    event loop.
 
     Args:
         file_path: Path to the file to convert.
@@ -158,18 +159,9 @@ async def convert_file_to_markdown(file_path: Path, output_path: Path | None = N
         Path to the generated .md file, or None if conversion failed.
     """
     try:
-        pdf_converter = _get_pdf_converter()
-        file_size = file_path.stat().st_size
+        md_path = await asyncio.to_thread(_convert_file_to_markdown_sync, file_path, output_path)
 
-        if file_size > _ASYNC_THRESHOLD_BYTES:
-            text = await asyncio.to_thread(_do_convert, file_path, pdf_converter)
-        else:
-            text = _do_convert(file_path, pdf_converter)
-
-        md_path = output_path if output_path is not None else file_path.with_suffix(".md")
-        md_path.write_text(text, encoding="utf-8")
-
-        logger.info("Converted %s to markdown: %s (%d chars)", file_path.name, md_path.name, len(text))
+        logger.info("Converted %s to markdown: %s", file_path.name, md_path.name)
         return md_path
     except Exception as e:
         logger.error("Failed to convert %s to markdown: %s", file_path.name, e)

backend/tests/blocking_io/test_upload_conversion.py

@@ -0,0 +1,27 @@
+"""Regression anchor: upload conversion lifecycle must not block the event loop."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from deerflow.uploads.conversion import convert_uploaded_file_to_markdown
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_real_upload_conversion_lifecycle_does_not_block_event_loop(tmp_path, monkeypatch):
+    uploads = tmp_path / "user-data" / "uploads"
+    await asyncio.to_thread(uploads.mkdir, parents=True)
+    source = uploads / "report.pdf"
+    await asyncio.to_thread(source.write_bytes, b"PDF")
+    monkeypatch.setattr(
+        "deerflow.utils.file_conversion._do_convert",
+        lambda path, converter: "# converted",
+    )
+
+    result = await convert_uploaded_file_to_markdown(source)
+
+    assert result is not None
+    assert await asyncio.to_thread(result.read_text, encoding="utf-8") == "# converted"

backend/tests/test_file_conversion.py

@@ -8,7 +8,6 @@ from types import ModuleType
 from unittest.mock import MagicMock, patch
 
 from deerflow.utils.file_conversion import (
-    _ASYNC_THRESHOLD_BYTES,
     _MIN_CHARS_PER_PAGE,
     MAX_OUTLINE_ENTRIES,
     _do_convert,
@@ -239,23 +238,25 @@ class TestGetPdfConverter:
 
 
 class TestConvertFileToMarkdown:
-    def test_small_file_runs_synchronously(self, tmp_path):
-        """Small files (< 1 MB) are converted in the event loop thread."""
+    def test_small_file_is_offloaded_to_thread(self, tmp_path):
+        """Small files are offloaded so parser and write IO cannot block the event loop."""
         pdf = tmp_path / "small.pdf"
         pdf.write_bytes(b"%PDF-1.4 " + b"x" * 100)  # well under 1 MB
 
+        async def fake_to_thread(fn, *args, **kwargs):
+            return fn(*args, **kwargs)
+
         with (
             patch("deerflow.utils.file_conversion._get_pdf_converter", return_value="auto"),
             patch(
                 "deerflow.utils.file_conversion._do_convert",
                 return_value="# Small PDF",
             ) as mock_convert,
-            patch("asyncio.to_thread") as mock_thread,
+            patch("asyncio.to_thread", side_effect=fake_to_thread) as mock_thread,
         ):
             md_path = _run(convert_file_to_markdown(pdf))
 
-        # asyncio.to_thread must NOT have been called
-        mock_thread.assert_not_called()
+        mock_thread.assert_called_once()
         mock_convert.assert_called_once()
         assert md_path == pdf.with_suffix(".md")
         assert md_path.read_text() == "# Small PDF"
@@ -263,8 +264,7 @@ class TestConvertFileToMarkdown:
     def test_large_file_offloaded_to_thread(self, tmp_path):
         """Large files (> 1 MB) are offloaded via asyncio.to_thread."""
         pdf = tmp_path / "large.pdf"
-        # Write slightly more than the threshold
-        pdf.write_bytes(b"%PDF-1.4 " + b"x" * (_ASYNC_THRESHOLD_BYTES + 1))
+        pdf.write_bytes(b"%PDF-1.4 " + b"x" * (1024 * 1024 + 1))
 
         async def fake_to_thread(fn, *args, **kwargs):
             return fn(*args, **kwargs)

backend/tests/test_upload_conversion.py

@@ -1,12 +1,100 @@
 """Tests for publication of system-owned Markdown upload conversions."""
 
+import asyncio
+import hashlib
 from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
 import pytest
 
 from deerflow.uploads.conversion import convert_uploaded_file_to_markdown
-from deerflow.uploads.layout import conversion_path_for_upload
+from deerflow.uploads.layout import (
+    conversion_filename_for_upload,
+    conversion_path_for_upload,
+    conversion_virtual_path,
+    existing_conversion_path_for_upload,
+)
+from deerflow.uploads.manager import delete_file_safe, publish_upload_bytes, publish_upload_bytes_leased
+
+
+@pytest.mark.asyncio
+async def test_delete_and_reupload_cannot_receive_old_conversion(tmp_path):
+    uploads = tmp_path / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    publication = publish_upload_bytes_leased(uploads, "report.pdf", b"OLD")
+    converter_started = asyncio.Event()
+    allow_converter = asyncio.Event()
+
+    async def paused_convert(source, output_path=None):
+        converter_started.set()
+        await allow_converter.wait()
+        output_path.write_text("FROM OLD", encoding="utf-8")
+        return output_path
+
+    with patch("deerflow.uploads.conversion.convert_file_to_markdown", side_effect=paused_convert):
+        conversion = asyncio.create_task(convert_uploaded_file_to_markdown(publication.path, publication=publication))
+        await converter_started.wait()
+        deletion = asyncio.create_task(asyncio.to_thread(delete_file_safe, uploads, "report.pdf"))
+        await asyncio.sleep(0.05)
+        assert not deletion.done()
+        allow_converter.set()
+        await conversion
+        publication.release()
+        await deletion
+
+    replacement = publish_upload_bytes(uploads, "report.pdf", b"NEW")
+    assert replacement.read_bytes() == b"NEW"
+    assert existing_conversion_path_for_upload(replacement) is None
+
+
+@pytest.mark.asyncio
+async def test_cancelled_waiter_does_not_leak_name_lease_or_conversion_stage(tmp_path):
+    uploads = tmp_path / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    publication = publish_upload_bytes_leased(uploads, "report.pdf", b"OLD")
+    conversion = asyncio.create_task(convert_uploaded_file_to_markdown(publication.path))
+    await asyncio.sleep(0.05)
+
+    conversion.cancel()
+    await asyncio.sleep(0.05)
+    assert not conversion.done()
+    publication.release()
+
+    with pytest.raises(asyncio.CancelledError):
+        await conversion
+
+    delete_file_safe(uploads, "report.pdf")
+    replacement = publish_upload_bytes(uploads, "report.pdf", b"NEW")
+    assert replacement.read_bytes() == b"NEW"
+    conversion_dir = uploads.parent / ".upload-conversions"
+    assert not list(conversion_dir.glob(".upload-*.part"))
+
+
+@pytest.mark.parametrize("byte_length", [252, 253, 254, 255])
+def test_long_conversion_filename_fits_component_limit(byte_length, tmp_path):
+    filename = "a" * (byte_length - 4) + ".pdf"
+    upload = tmp_path / "uploads" / filename
+    target = conversion_path_for_upload(upload)
+
+    assert len(target.name.encode("utf-8")) <= 255
+    assert target.name == conversion_filename_for_upload(filename)
+    assert conversion_virtual_path(filename).endswith(f"/{target.name}")
+    assert target.name.endswith(".md")
+
+
+def test_255_byte_conversion_name_uses_full_digest():
+    filename = "a" * 251 + ".pdf"
+    digest = hashlib.sha256(filename.encode("utf-8")).hexdigest()
+
+    assert conversion_filename_for_upload(filename) == f"{'a' * 187}.{digest}.md"
+
+
+def test_multibyte_long_conversion_name_is_utf8_safe():
+    filename = "é" * 125 + ".pdf"
+    converted = conversion_filename_for_upload(filename)
+
+    assert len(converted.encode("utf-8")) <= 255
+    assert converted.endswith(".md")
 
 
 @pytest.mark.asyncio

docs/superpowers/plans/2026-08-06-upload-review-remediation.md

@@ -322,7 +322,7 @@ def test_long_conversion_filename_fits_component_limit(byte_length, tmp_path):
     assert target.name.endswith(".md")
 ```
 
-Use literal expected output for the 255-byte case: 186 `a` bytes, a dot, the full SHA-256
+Use literal expected output for the 255-byte case: 187 `a` bytes, a dot, the full SHA-256
 of the original filename, and `.md`.
 
 - [ ] **Step 2: Write a strict blocking-I/O regression**