fix: close upload review gaps

backend/app/channels/dingtalk.py

@@ -20,9 +20,13 @@ from app.channels.message_bus import InboundMessage, InboundMessageType, Message
 from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.sandbox_provider import get_sandbox_provider
-from deerflow.uploads.async_helpers import publish_upload_bytes_leased_async, release_published_upload_async
+from deerflow.uploads.async_helpers import (
+    publish_upload_bytes_leased_async,
+    release_published_upload_async,
+    run_upload_io_cancellation_safe,
+)
 from deerflow.uploads.layout import upload_virtual_path
-from deerflow.uploads.manager import UnsafeUploadPathError, normalize_filename
+from deerflow.uploads.manager import UnsafeUploadPathError, make_upload_file_sandbox_readable, normalize_filename
 
 logger = logging.getLogger(__name__)
 
@@ -646,16 +650,18 @@ class DingTalkChannel(Channel):
 
             try:
                 sandbox_provider = get_sandbox_provider()
-                # acquire_async keeps provider lifecycle work (Docker discovery,
-                # readiness polls) off the event loop; update_file is blocking
-                # transport IO on remote sandboxes, so it is offloaded too.
-                sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-                if sandbox_id != "local":
+                if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+                    await run_upload_io_cancellation_safe(make_upload_file_sandbox_readable, publication.path)
+                else:
+                    # acquire_async keeps provider lifecycle work (Docker discovery,
+                    # readiness polls) off the event loop; update_file is blocking
+                    # transport IO on remote sandboxes, so it is offloaded too.
+                    sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
                     sandbox = sandbox_provider.get(sandbox_id)
                     if sandbox is None:
                         logger.warning("[DingTalk] sandbox %s not found after acquire, dropping attachment: %s", sandbox_id, virtual_path)
                         return ""
-                    await asyncio.to_thread(sandbox.update_file, virtual_path, content)
+                    await run_upload_io_cancellation_safe(sandbox.update_file, virtual_path, content)
             except Exception:
                 logger.exception("[DingTalk] failed to sync downloaded file into non-local sandbox: %s", virtual_path)
                 return ""

backend/app/channels/feishu.py

@@ -25,8 +25,13 @@ from app.channels.message_bus import (
 from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.sandbox_provider import get_sandbox_provider
-from deerflow.uploads.async_helpers import publish_upload_bytes_leased_async, release_published_upload_async
+from deerflow.uploads.async_helpers import (
+    publish_upload_bytes_leased_async,
+    release_published_upload_async,
+    run_upload_io_cancellation_safe,
+)
 from deerflow.uploads.layout import upload_virtual_path
+from deerflow.uploads.manager import make_upload_file_sandbox_readable
 
 logger = logging.getLogger(__name__)
 PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60
@@ -462,13 +467,15 @@ class FeishuChannel(Channel):
 
             try:
                 sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
-                if not getattr(sandbox_provider, "uses_thread_data_mounts", False):
+                if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+                    await run_upload_io_cancellation_safe(make_upload_file_sandbox_readable, publication.path)
+                else:
                     sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
                     sandbox = sandbox_provider.get(sandbox_id)
                     if sandbox is None:
                         logger.warning("[Feishu] sandbox not found for thread_id=%s", thread_id)
                         return f"Failed to obtain the [{type}]"
-                    await asyncio.to_thread(sandbox.update_file, virtual_path, content)
+                    await run_upload_io_cancellation_safe(sandbox.update_file, virtual_path, content)
             except Exception:
                 logger.exception("[Feishu] failed to sync resource into non-local sandbox: %s", virtual_path)
                 return f"Failed to obtain the [{type}]"

backend/packages/harness/deerflow/agents/middlewares/uploads_middleware.py

@@ -18,6 +18,7 @@ from langgraph.runtime import Runtime
 from deerflow.agents.middlewares.input_sanitization_middleware import neutralize_untrusted_tags
 from deerflow.config.paths import Paths, get_paths
 from deerflow.runtime.user_context import resolve_runtime_user_id
+from deerflow.uploads.layout import UnsafeConversionPathError, conversion_virtual_path, existing_conversion_path_for_upload
 from deerflow.uploads.manager import is_upload_staging_file
 from deerflow.utils.file_outline import extract_outline_for_file
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_to_text
@@ -88,24 +89,30 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         size_str = f"{size_kb:.1f} KB" if size_kb < 1024 else f"{size_kb / 1024:.1f} MB"
         lines.append(f"- {neutralize_untrusted_tags(file['filename'])} ({size_str})")
         lines.append(f"  Path: {neutralize_untrusted_tags(file['path'])}")
+        markdown_virtual_path = file.get("markdown_virtual_path")
+        if markdown_virtual_path:
+            lines.append(f"  Generated Markdown: {neutralize_untrusted_tags(markdown_virtual_path)}")
         if file.get("selection_reason") == "query_match":
             lines.append("  Selected because: matched the current query.")
         outline = file.get("outline") or []
         if outline:
             truncated = outline[-1].get("truncated", False)
             visible = [e for e in outline if not e.get("truncated")]
-            lines.append("  Document outline (use `read_file` with line ranges to read sections):")
+            lines.append("  Document outline (use `read_file` on the Generated Markdown path with line ranges to read sections):")
             for entry in visible:
                 lines.append(f"    L{entry['line']}: {neutralize_untrusted_tags(entry['title'])}")
             if truncated:
-                lines.append(f"    ... (showing first {len(visible)} headings; use `read_file` to explore further)")
+                lines.append(f"    ... (showing first {len(visible)} headings; use `read_file` to explore further on the Generated Markdown path)")
         else:
             preview = file.get("outline_preview") or []
             if preview:
                 lines.append("  No structural headings detected. Document begins with:")
                 for text in preview:
                     lines.append(f"    > {neutralize_untrusted_tags(text)}")
-            lines.append("  Use `grep` to search for keywords (e.g. `grep(pattern='keyword', path='/mnt/user-data/uploads/')`).")
+            if markdown_virtual_path:
+                lines.append("  Use `grep` on the Generated Markdown path to search for keywords.")
+            else:
+                lines.append("  Use `grep` to search for keywords (e.g. `grep(pattern='keyword', path='/mnt/user-data/uploads/')`).")
         lines.append("")
 
     def _select_files_for_context(
@@ -241,6 +248,17 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
             # exclude files that became historical after the previous turn.
             return {"uploaded_files": []}
 
+        if uploads_dir:
+            for file in new_files:
+                phys_path = uploads_dir / file["filename"]
+                try:
+                    generated_path = existing_conversion_path_for_upload(phys_path)
+                except UnsafeConversionPathError:
+                    logger.warning("Ignoring unsafe generated conversion for %s", file["filename"])
+                    generated_path = None
+                if generated_path is not None:
+                    file["markdown_virtual_path"] = conversion_virtual_path(file["filename"])
+
         context_files, omitted_files = self._select_files_for_context(new_files)
 
         # Attach outlines to context files

backend/packages/harness/deerflow/client.py

@@ -70,6 +70,7 @@ from deerflow.uploads.manager import (
     ensure_uploads_dir,
     get_uploads_dir,
     list_files_in_dir,
+    make_upload_file_sandbox_readable,
     publish_upload_copy_leased,
     upload_artifact_url,
     upload_virtual_path,
@@ -1553,6 +1554,7 @@ class DeerFlowClient:
                     if dest_name != src_path.name:
                         info["original_filename"] = src_path.name
 
+                    md_path = None
                     if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS:
                         try:
                             if conversion_pool is not None:
@@ -1579,6 +1581,10 @@ class DeerFlowClient:
                             info["markdown_virtual_path"] = md_virtual_path
                             info["markdown_artifact_url"] = artifact_url_for_virtual_path(thread_id, md_virtual_path)
 
+                    make_upload_file_sandbox_readable(dest)
+                    if md_path is not None:
+                        make_upload_file_sandbox_readable(md_path)
+
                     uploaded_files.append(info)
                 finally:
                     publication.release()

backend/packages/harness/deerflow/tools/builtins/list_uploaded_files_tool.py

@@ -19,6 +19,7 @@ from deerflow.agents.middlewares.input_sanitization_middleware import neutralize
 from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.tools.types import Runtime
+from deerflow.uploads.layout import UnsafeConversionPathError, conversion_virtual_path, existing_conversion_path_for_upload
 from deerflow.uploads.manager import is_upload_staging_file
 from deerflow.utils.file_outline import extract_outline_for_file
 
@@ -111,27 +112,15 @@ def _list_uploaded_files_impl(
         outline_for_all = False
         outline_filenames = set(include_outline)
 
-    # Collect historical files (sorted by mtime descending).
-    # Skip .md files that are conversion artifacts (have a same-stem non-.md sibling).
+    # Collect historical primary uploads (sorted by mtime descending). Generated
+    # conversions live outside this directory in the system-owned namespace.
     candidates: list[tuple[float, Path, int]] = []
     try:
-        # Collect file entries once to build the name set and iterate.
         entries = [e for e in os.scandir(uploads_dir) if e.is_file() and not e.is_symlink() and not is_upload_staging_file(e.name)]
-        all_names: set[str] = {e.name for e in entries}
 
         for entry in entries:
             if entry.name in current_run_filenames:
                 continue
-            # Skip .md files that are conversion artifacts of another file.
-            # Known limitation: if a user manually uploads both report.pdf and
-            # report.md, the .md is hidden as a "conversion artifact".  This is
-            # acceptable for the MVP — triggering this requires uploading files
-            # whose stems collide with converted documents, which is rare.
-            if entry.name.endswith(".md"):
-                stem = entry.name[:-3]  # remove ".md"
-                non_md_siblings = {n for n in all_names if n != entry.name and Path(n).stem == stem}
-                if non_md_siblings:
-                    continue
             stat = entry.stat()
             candidates.append((stat.st_mtime, Path(entry.path), stat.st_size))
     except OSError:
@@ -157,6 +146,13 @@ def _list_uploaded_files_impl(
             "path": neutralize_untrusted_tags(f"/mnt/user-data/uploads/{filename}"),
             "extension": neutralize_untrusted_tags(file_path.suffix),
         }
+        try:
+            generated_path = existing_conversion_path_for_upload(file_path)
+        except UnsafeConversionPathError:
+            logger.warning("Ignoring unsafe generated conversion for %s", filename)
+            generated_path = None
+        if generated_path is not None:
+            file_info["markdown_virtual_path"] = neutralize_untrusted_tags(conversion_virtual_path(filename))
 
         should_include_outline = outline_for_all or filename in outline_filenames
         if should_include_outline:

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

@@ -46,6 +46,21 @@ async def wait_for_task_completion(task: asyncio.Task) -> bool:
     return cancelled
 
 
+async def run_upload_io_cancellation_safe[**P, T](
+    func: Callable[P, T],
+    /,
+    *args: P.args,
+    **kwargs: P.kwargs,
+) -> T:
+    """Run blocking upload I/O to completion before propagating cancellation."""
+    io_task = asyncio.create_task(asyncio.to_thread(func, *args, **kwargs))
+    cancelled = await wait_for_task_completion(io_task)
+    result = io_task.result()
+    if cancelled:
+        raise asyncio.CancelledError
+    return result
+
+
 def _rollback_and_release(publication: PublishedUpload) -> None:
     try:
         rollback_published_upload(publication)

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

@@ -374,6 +374,17 @@ def publish_upload_copy(base_dir: Path, preferred_filename: str, source_path: Pa
         publication.release()
 
 
+def make_upload_file_sandbox_readable(file_path: Path) -> None:
+    """Add group/other read bits to one verified regular upload artifact."""
+    file_path = Path(file_path)
+    file_stat = os.lstat(file_path)
+    if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_nlink != 1:
+        raise UnsafeUploadPathError(f"Unsafe upload file: {file_path.name}")
+    readable_mode = stat.S_IMODE(file_stat.st_mode) | stat.S_IRGRP | stat.S_IROTH
+    chmod_kwargs = {"follow_symlinks": False} if os.chmod in os.supports_follow_symlinks else {}
+    os.chmod(file_path, readable_mode, **chmod_kwargs)
+
+
 def rollback_published_upload(publication: PublishedUpload) -> None:
     """Remove only the still-leased upload generation represented by *publication*."""
     if not publication.is_active:
@@ -383,9 +394,9 @@ def rollback_published_upload(publication: PublishedUpload) -> None:
     if not publication.identity.matches(publication.path):
         return
     owned_conversion = existing_conversion_path_for_upload(publication.path)
-    publication.path.unlink()
     if owned_conversion is not None:
         owned_conversion.unlink(missing_ok=True)
+    publication.path.unlink()
 
 
 def replace_system_owned_staged_file(staged: StagedUpload, filename: str) -> Path:
@@ -548,9 +559,9 @@ def delete_file_safe(base_dir: Path, filename: str) -> dict:
             raise UnsafeUploadPathError(f"Unsafe upload file: {safe_name}")
 
         owned_conversion = existing_conversion_path_for_upload(file_path)
-        file_path.unlink()
         if owned_conversion is not None:
             owned_conversion.unlink(missing_ok=True)
+        file_path.unlink()
     finally:
         lease.release()
 

backend/tests/blocking_io/test_dingtalk_receive_file.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 
 import asyncio
 import threading
+from contextlib import suppress
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
@@ -35,11 +36,15 @@ async def test_receive_file_persist_does_not_block_event_loop(tmp_path, monkeypa
     monkeypatch.setattr("app.channels.dingtalk.get_paths", lambda: paths)
 
     async def _acquire_async(thread_id, user_id=None):
-        return "local"
+        raise AssertionError("mounted uploads must not acquire a sandbox")
 
     monkeypatch.setattr(
         "app.channels.dingtalk.get_sandbox_provider",
-        lambda: SimpleNamespace(acquire_async=_acquire_async, get=lambda sid: None),
+        lambda: SimpleNamespace(
+            uses_thread_data_mounts=True,
+            acquire_async=_acquire_async,
+            get=lambda sid: (_ for _ in ()).throw(AssertionError("mounted uploads must not look up a sandbox")),
+        ),
     )
 
     channel = DingTalkChannel(MessageBus(), config={})
@@ -56,6 +61,8 @@ async def test_receive_file_persist_does_not_block_event_loop(tmp_path, monkeypa
 
     assert "/uploads/a.pdf" in out.text
     assert out.files == []
+    uploaded = paths.sandbox_uploads_dir("t1", user_id="default") / "a.pdf"
+    assert (await asyncio.to_thread(uploaded.stat)).st_mode & 0o044 == 0o044
 
 
 async def test_receive_file_rejects_symlinked_upload_directory(tmp_path, monkeypatch) -> None:
@@ -118,3 +125,46 @@ async def test_receive_file_holds_name_lease_through_remote_sync(tmp_path, monke
     assert await receive == "/mnt/user-data/uploads/a.pdf"
     await deletion
     assert not await asyncio.to_thread((uploads / "a.pdf").exists)
+
+
+async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lease(tmp_path, monkeypatch) -> None:
+    from app.channels.dingtalk import DingTalkChannel
+    from app.channels.message_bus import MessageBus
+    from deerflow.config.paths import Paths
+    from deerflow.uploads.manager import delete_file_safe
+
+    paths = await asyncio.to_thread(Paths, str(tmp_path))
+    monkeypatch.setattr("app.channels.dingtalk.get_paths", lambda: paths)
+    sync_started = threading.Event()
+    allow_sync = threading.Event()
+
+    class PausedSandbox:
+        def update_file(self, _path, _content):
+            sync_started.set()
+            assert allow_sync.wait(5)
+
+    async def acquire_async(_thread_id, user_id=None):
+        return "remote"
+
+    monkeypatch.setattr(
+        "app.channels.dingtalk.get_sandbox_provider",
+        lambda: SimpleNamespace(acquire_async=acquire_async, get=lambda _sandbox_id: PausedSandbox()),
+    )
+    channel = DingTalkChannel(MessageBus(), config={})
+    channel._download_by_code = AsyncMock(return_value=b"DATA")
+    receive = asyncio.create_task(channel._receive_single_file("dc", "file", "a.pdf", "t1", user_id="default"))
+    assert await asyncio.to_thread(sync_started.wait, 5)
+    uploads = paths.sandbox_uploads_dir("t1", user_id="default")
+    receive.cancel()
+    deletion = asyncio.create_task(asyncio.to_thread(delete_file_safe, uploads, "a.pdf"))
+    try:
+        await asyncio.sleep(0.05)
+        assert not receive.done()
+        assert not deletion.done()
+    finally:
+        allow_sync.set()
+        with suppress(asyncio.CancelledError):
+            await receive
+        await deletion
+
+    assert not await asyncio.to_thread((uploads / "a.pdf").exists)

backend/tests/blocking_io/test_feishu_receive_file.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import threading
+from contextlib import suppress
 from io import BytesIO
 from unittest.mock import MagicMock
 
@@ -145,6 +146,49 @@ async def test_receive_file_holds_name_lease_through_remote_sync(tmp_path, monke
     assert not await asyncio.to_thread((uploads / "report.pdf").exists)
 
 
+async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lease(tmp_path, monkeypatch) -> None:
+    from deerflow.config.paths import Paths
+    from deerflow.uploads.manager import delete_file_safe
+
+    paths = await asyncio.to_thread(Paths, str(tmp_path))
+    provider = _RemoteProvider()
+    sync_started = threading.Event()
+    allow_sync = threading.Event()
+
+    def paused_update(path: str, content: bytes) -> None:
+        sync_started.set()
+        assert allow_sync.wait(5)
+        provider.sandbox.updates.append((path, content))
+
+    provider.sandbox.update_file = paused_update
+    monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths)
+    monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: provider)
+    receive = asyncio.create_task(
+        _channel_with_file()._receive_single_file(
+            "message-1",
+            "file-key",
+            "file",
+            "thread-1",
+            user_id="ou-user",
+        )
+    )
+    assert await asyncio.to_thread(sync_started.wait, 5)
+    uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user")
+    receive.cancel()
+    deletion = asyncio.create_task(asyncio.to_thread(delete_file_safe, uploads, "report.pdf"))
+    try:
+        await asyncio.sleep(0.05)
+        assert not receive.done()
+        assert not deletion.done()
+    finally:
+        allow_sync.set()
+        with suppress(asyncio.CancelledError):
+            await receive
+        await deletion
+
+    assert not await asyncio.to_thread((uploads / "report.pdf").exists)
+
+
 async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monkeypatch) -> None:
     from deerflow.config.paths import Paths
 
@@ -163,6 +207,7 @@ async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monke
     assert result == "/mnt/user-data/uploads/report.pdf"
     uploaded = tmp_path / "users" / "ou-user" / "threads" / "thread-1" / "user-data" / "uploads" / "report.pdf"
     assert await asyncio.to_thread(uploaded.read_bytes) == b"DATA"
+    assert (await asyncio.to_thread(uploaded.stat)).st_mode & 0o044 == 0o044
 
 
 async def test_concurrent_same_name_files_preserve_every_payload(tmp_path, monkeypatch) -> None:

backend/tests/test_client.py

@@ -2182,6 +2182,7 @@ class TestUploads:
             assert "artifact_url" in result["files"][0]
             assert "message" in result
             assert (uploads_dir / "test.txt").exists()
+            assert (uploads_dir / "test.txt").stat().st_mode & 0o044 == 0o044
 
     def test_upload_files_across_calls_never_overwrite(self, client, tmp_path):
         uploads_dir = tmp_path / "user-data" / "uploads"
@@ -2313,6 +2314,7 @@ class TestUploads:
                 md_path = conversion_path_for_upload(path)
                 md_path.parent.mkdir(parents=True, exist_ok=True)
                 md_path.write_text(f"FROM:{path.name}", encoding="utf-8")
+                md_path.chmod(0o600)
                 return md_path
 
             with (
@@ -2332,6 +2334,13 @@ class TestUploads:
             assert (uploads_dir / "a.md").read_bytes() == b"USER"
             assert conversion_path_for_upload(uploads_dir / "a.docx").read_text(encoding="utf-8") == "FROM:a.docx"
             assert conversion_path_for_upload(uploads_dir / "a.pdf").read_text(encoding="utf-8") == "FROM:a.pdf"
+            for path in (
+                uploads_dir / "a.docx",
+                uploads_dir / "a.pdf",
+                conversion_path_for_upload(uploads_dir / "a.docx"),
+                conversion_path_for_upload(uploads_dir / "a.pdf"),
+            ):
+                assert path.stat().st_mode & 0o044 == 0o044
 
     def test_upload_files_failed_conversion_does_not_block_other_conversion(self, client):
         """A failed conversion does not affect another primary's owned asset."""

backend/tests/test_dingtalk_channel.py

@@ -1736,7 +1736,11 @@ def _patch_uploads(monkeypatch, uploads_dir, *, sandbox_id="local", sandbox=None
 
     monkeypatch.setattr(
         "app.channels.dingtalk.get_sandbox_provider",
-        lambda: SimpleNamespace(acquire_async=_acquire_async, get=lambda sid: sandbox),
+        lambda: SimpleNamespace(
+            uses_thread_data_mounts=sandbox_id == "local",
+            acquire_async=_acquire_async,
+            get=lambda sid: sandbox,
+        ),
     )
 
 

backend/tests/test_list_uploaded_files_tool.py

@@ -10,7 +10,7 @@ from langchain_core.messages import HumanMessage, ToolMessage
 
 from deerflow.config.paths import Paths
 from deerflow.tools.builtins.list_uploaded_files_tool import _format_omitted_summary, _list_uploaded_files_impl, _resolve_thread_id
-from deerflow.uploads.layout import conversion_path_for_upload
+from deerflow.uploads.layout import conversion_path_for_upload, conversion_virtual_path
 
 
 def _paths(tmp_path):
@@ -196,6 +196,17 @@ class TestListUploadedFiles:
         assert result["files"][0]["outline"][0]["title"] == "Heading 1"
         assert result["files"][0]["outline"][1]["title"] == "Heading 2"
 
+    def test_long_filename_result_includes_exact_generated_markdown_path(self, tmp_path):
+        uploads_dir = _uploads_dir(tmp_path)
+        filename = f"{'a' * 250}.pdf"
+        primary = uploads_dir / filename
+        primary.write_bytes(b"%PDF")
+        _write_conversion(primary, "# Heading\n")
+
+        result = _list_uploaded_files_impl(include_outline=True, runtime=_runtime(), _paths=_paths(tmp_path))
+
+        assert result["files"][0]["markdown_virtual_path"] == conversion_virtual_path(filename)
+
     def test_include_outline_list(self, tmp_path):
         uploads_dir = _uploads_dir(tmp_path)
         primary_a = uploads_dir / "a.pdf"
@@ -243,6 +254,7 @@ class TestListUploadedFiles:
         result = _list_uploaded_files_impl(include_outline=True, runtime=_runtime(), _paths=_paths(tmp_path))
 
         files = {item["filename"]: item for item in result["files"]}
+        assert set(files) == {"report.pdf", "report.md"}
         assert "outline" not in files["report.pdf"]
         assert "outline_preview" not in files["report.pdf"]
 

backend/tests/test_uploads_manager.py

@@ -319,6 +319,28 @@ class TestUploadPublication:
 
         assert (tmp_path / "report.pdf").read_bytes() == b"new"
 
+    def test_rollback_preserves_primary_when_conversion_removal_fails(self, tmp_path):
+        publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
+        owned_conversion = conversion_path_for_upload(publication.path)
+        owned_conversion.parent.mkdir(exist_ok=True)
+        owned_conversion.write_text("generated", encoding="utf-8")
+        real_unlink = Path.unlink
+
+        def fail_conversion_unlink(path, *args, **kwargs):
+            if path == owned_conversion:
+                raise OSError("cannot unlink conversion")
+            return real_unlink(path, *args, **kwargs)
+
+        try:
+            with patch.object(Path, "unlink", autospec=True, side_effect=fail_conversion_unlink):
+                with pytest.raises(OSError, match="cannot unlink conversion"):
+                    rollback_published_upload(publication)
+        finally:
+            publication.release()
+
+        assert publication.path.read_bytes() == b"old"
+        assert owned_conversion.read_text(encoding="utf-8") == "generated"
+
     def test_staging_unlink_failure_is_not_reported_as_success(self, tmp_path):
         staged = create_upload_staging_file(tmp_path)
         staged.handle.write(b"payload")
@@ -700,3 +722,25 @@ class TestDeleteFileSafe:
         assert not primary.exists()
         assert not owned.exists()
         assert legacy_or_user.read_text(encoding="utf-8") == "user markdown"
+
+    def test_delete_preserves_primary_when_conversion_removal_fails(self, tmp_path):
+        uploads = tmp_path / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        primary = uploads / "report.pdf"
+        primary.write_bytes(b"PDF")
+        owned_conversion = conversion_path_for_upload(primary)
+        owned_conversion.parent.mkdir()
+        owned_conversion.write_text("generated", encoding="utf-8")
+        real_unlink = Path.unlink
+
+        def fail_conversion_unlink(path, *args, **kwargs):
+            if path == owned_conversion:
+                raise OSError("cannot unlink conversion")
+            return real_unlink(path, *args, **kwargs)
+
+        with patch.object(Path, "unlink", autospec=True, side_effect=fail_conversion_unlink):
+            with pytest.raises(OSError, match="cannot unlink conversion"):
+                delete_file_safe(uploads, "report.pdf")
+
+        assert primary.read_bytes() == b"PDF"
+        assert owned_conversion.read_text(encoding="utf-8") == "generated"

backend/tests/test_uploads_middleware_core_logic.py

@@ -15,7 +15,7 @@ from langchain_core.messages import AIMessage, HumanMessage
 
 from deerflow.agents.middlewares.uploads_middleware import UploadsMiddleware
 from deerflow.config.paths import Paths
-from deerflow.uploads.layout import conversion_path_for_upload
+from deerflow.uploads.layout import conversion_path_for_upload, conversion_virtual_path
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_to_text
 
 THREAD_ID = "thread-abc123"
@@ -651,6 +651,22 @@ class TestBeforeAgent:
         assert "ITEM 2. RISK" in content
         assert "read_file" in content
 
+    def test_long_filename_context_includes_exact_generated_markdown_path(self, tmp_path):
+        mw = _middleware(tmp_path)
+        uploads_dir = _uploads_dir(tmp_path)
+        filename = f"{'a' * 250}.pdf"
+        primary = uploads_dir / filename
+        primary.write_bytes(b"%PDF fake")
+        _write_conversion(primary, "# Exact generated path\n")
+
+        msg = _human("summarise", files=[{"filename": filename, "size": 9, "path": f"/mnt/user-data/uploads/{filename}"}])
+        result = mw.before_agent(self._state(msg), _runtime())
+
+        assert result is not None
+        expected = conversion_virtual_path(filename)
+        assert expected in result["messages"][-1].content
+        assert result["uploaded_files"][0]["markdown_virtual_path"] == expected
+
     def test_legacy_sibling_markdown_is_not_treated_as_generated(self, tmp_path):
         mw = _middleware(tmp_path)
         uploads_dir = _uploads_dir(tmp_path)