fix(uploads): convert the bytes we wrote, not the name they landed under (#5611)

* fix(uploads): convert the bytes we wrote, not the name they landed under

Document conversion re-opened the upload by name after it was already
visible in the thread's uploads directory: the Gateway converted the
committed file_path, and DeerFlowClient converted the copy it had just
placed there. That directory is writable from local and AIO sandboxes,
so a process watching it can replace the name with a symlink in the
window between the upload landing and the converter opening it. The
converter then reads whatever host file the link points at and writes
that content back into the thread as the .md companion, which the
sandbox can read. Reproduced end to end on both paths with a real xlsx:
the companion came back holding the host file's rows.

The Gateway now duplicates the descriptor of the staged file before the
link-commit, copies those bytes into a private directory outside the
uploads tree, and converts there. A descriptor cannot be redirected by
replacing a name, so the conversion input is the content this request
wrote. The client converts the caller's own source file instead of the
copy in uploads; the source is the file the caller handed in, which the
sandbox cannot reach.

Both already wrote the companion without following a symlink, so only
the read side changes. The uploads directory still receives exactly the
same files.

* docs(changelog): note upload conversion source fix (#5611)

* fix(uploads): close the conversion descriptor when staging its copy fails

Review follow-up. The private directory for the conversion copy was
created before the try that owns the duplicated descriptor, so a failure
there — a full or unwritable temporary filesystem — propagated without
closing it. The upload's own cleanup only unlinks the committed name and
releases the sandbox lease, so the descriptor stayed open for the life of
the process and kept the unlinked staged bytes allocated with it; repeated
failures accumulated both.

Directory creation now happens inside that try, and the finally removes
the directory only once it exists.

* fix(uploads): keep the conversion descriptor owned across cancellation

Review follow-up. run_file_io cannot interrupt its worker, so cancelling
the await around os.dup only abandoned the result: the duplicate was
created moments later with nothing left to close it, and it pinned the
staged bytes of an upload whose name the cleanup had already unlinked.
Cancellation after the duplication was just as leaky, because the
commit-path handler caught Exception and CancelledError is not one.

The duplication now runs as its own task, shielded from the caller's
cancellation, and closes its own result when the caller is gone by the
time the worker finishes. The commit path catches BaseException, closing
the descriptor it already owns before re-raising.

Both windows are pinned: one test stalls the duplication worker after it
allocates and cancels ingestion, the other stalls the commit so the
cancellation lands while the descriptor is owned.

* fix(uploads): drain the conversion copy so its descriptor always closes

Review follow-up. The copy worker owns the duplicated descriptor and
closes it in its own finally, but a bare await let a cancellation cancel
the executor job while it was still queued: the worker never ran, so that
finally never ran either, and the enclosing scope had already handed
ownership away and saw None. Draining also keeps a late worker from
writing into a private directory this scope has since removed.

The copy now goes through await_drained, the shield-and-drain helper the
Gateway already uses for offloads that must not be abandoned mid-flight.
Pinned by a test that holds the copy job queued, cancels ingestion, then
releases it and requires the descriptor to come back closed.

CHANGELOG.md

@@ -2785,6 +2785,13 @@ This release closes that milestone with **765 merged pull requests**.
 
 ### Security
 
+- **uploads:** Document conversion no longer re-opens the upload by name. The
+  Gateway converted the committed file and the embedded client converted the
+  copy it had just placed in the thread's uploads directory, so a sandbox that
+  replaced that name with a symlink in between had a host file converted into
+  the thread as the `.md` companion. The Gateway now converts a private copy of
+  the staged bytes, read through the descriptor it wrote, and the client
+  converts the caller's own source file. ([#5611])
 - **client:** `DeerFlowClient.upload_files` no longer writes through a
   symlink. A symlink planted in the sandbox-writable uploads directory, at an
   upload's name or its Markdown companion's name, made the embedded client
@@ -4323,3 +4330,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#5534]: https://github.com/bytedance/deer-flow/pull/5534
 [#5547]: https://github.com/bytedance/deer-flow/pull/5547
 [#5578]: https://github.com/bytedance/deer-flow/pull/5578
+[#5611]: https://github.com/bytedance/deer-flow/pull/5611

CHANGELOG_zh.md

@@ -2091,6 +2091,10 @@
 
 ### 安全
 
+- **上传：** 文档转换不再按文件名重新打开上传文件。此前 Gateway 转换的是已提交的文件，嵌入式
+  客户端转换的是刚放入线程 uploads 目录的副本，因此沙箱若在此期间把该文件名替换为符号链接，
+  宿主文件的内容就会被转换成该线程的 `.md` 配套文件。现在 Gateway 通过自己写入时持有的文件
+  描述符，转换 uploads 之外的私有副本；客户端则转换调用方提供的源文件。([#5611])
 - **客户端：** `DeerFlowClient.upload_files` 不再写穿符号链接。沙箱可写的 uploads 目录中，
   若在上传文件名或其 Markdown 配套文件名处放置符号链接，嵌入式客户端此前会覆盖链接指向的宿主
   文件并报告成功。现在该文件会被跳过并列入 `skipped_files`，`success` 为 `false`，与 Gateway
@@ -3517,3 +3521,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#5534]: https://github.com/bytedance/deer-flow/pull/5534
 [#5547]: https://github.com/bytedance/deer-flow/pull/5547
 [#5578]: https://github.com/bytedance/deer-flow/pull/5578
+[#5611]: https://github.com/bytedance/deer-flow/pull/5611

backend/app/gateway/AGENTS.md

@@ -102,7 +102,7 @@ owner-scoped assistant version selection remains enabled.
 | **Subagents** (`/api/subagents`) | Admin managed-worker CRUD and listing. |
 | **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/config/credentials` - atomically switch the caller's per-user Lark app after validating the new `app_id`/`app_secret` through the official CLI's live tenant-token probe, revoke/remove the previous OAuth tokens, and restore the prior credential tree if the switch fails; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants. Config and auth flows carry a server-issued, per-user generation persisted under the credential lock; a rejected direct switch leaves the current generation unchanged, stale completions return 409, and browser re-registration uses the same token-clearing/revocation transaction as direct credential switches. |
 | **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |
-| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); non-mounted sandbox sync uses a non-releasing request lease; `GET /list` - list; `DELETE /{filename}` - delete a regular file; a symlink is never followed and 404s like `GET /list` |
+| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); non-mounted sandbox sync uses a non-releasing request lease; `GET /list` - list; `DELETE /{filename}` - delete a regular file; a symlink 404s like `GET /list` |
 | **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - branch a completed assistant turn with a replay checkpoint; inherited titles take next-free displayed sibling suffixes, including explicit/renamed ones, while explicit titles stay unchanged. Durable `branch` admission rejects races. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Thread-scoped runtime channels (`sandbox`, `thread_data`) are not copied onto the branch: the parent's `sandbox_id` binds path mappings and the release lifecycle to the parent's workspace, so the branch lazily acquires its own sandbox instead. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380). Seeded rows are grouped into one synthetic run per inherited turn (`branch-seed-{thread_id}-{n}`, a new turn opening at every persisted human message, including an allowlisted hidden `ask_clarification` reply) because `run_id` is a turn identity to the feed's consumers, not a provenance tag: regenerating an inherited answer supersedes that row's whole `run_id` in `GET /messages/page`, so one shared id for the entire seed deleted the complete inherited history on a branch's first regenerate (#4458); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - summarize older active context, deriving memory policy and bucket from the state-producing checkpoint rather than request `agent_name`, and block while a run is in flight; unexpected failures return a generic 500 detail |
 | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content (`text/html`, `text/xml`, `application/xml`, `text/xsl`, any `+xml` type such as XHTML/SVG; `.skill` members too) is always forced as a download attachment to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update under a request lease. The outputs-only rule is `path_utils.resolve_outputs_confined_path`, shared with IM-channel attachment delivery: it collapses `..` before the prefix check and re-checks the resolved host path against the resolved outputs root, since `resolve_thread_virtual_path` only confines to `user-data/`; a percent-encoded `..` or a symlink planted in `outputs/` must not reach a sibling `uploads/` file. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). |
 | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`<think>...</think>`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing |

backend/app/gateway/upload_ingestion.py

@@ -28,7 +28,11 @@ patches keep binding to the one pipeline both callers use.
 
 from __future__ import annotations
 
+import asyncio
 import logging
+import os
+import shutil
+import tempfile
 import uuid
 from collections.abc import AsyncIterator
 from pathlib import Path
@@ -37,7 +41,7 @@ from typing import TYPE_CHECKING, Any
 from fastapi import HTTPException
 
 from deerflow.config.app_config import AppConfig
-from deerflow.utils.file_io import run_file_io
+from deerflow.utils.file_io import await_drained, run_file_io
 
 if TYPE_CHECKING:
     from fastapi import Request
@@ -48,6 +52,58 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _close_fd(fd: int | None) -> None:
+    """Close a conversion-source descriptor, tolerating an already-closed one."""
+    if fd is None:
+        return
+    try:
+        os.close(fd)
+    except OSError:
+        logger.warning("Failed to close upload conversion descriptor", exc_info=True)
+
+
+def _close_abandoned_fd(duplication: asyncio.Future) -> None:
+    """Close a descriptor whose owner was cancelled before it could take it."""
+    if duplication.cancelled() or duplication.exception() is not None:
+        return
+    _close_fd(duplication.result())
+
+
+async def _dup_for_conversion(fileno: int) -> int:
+    """Duplicate *fileno* off-thread so cancellation cannot strand the copy.
+
+    ``run_file_io`` cannot interrupt its worker: cancelling the await only
+    abandons the result, and here that result is an open descriptor nothing
+    would ever close — repeated cancellations would exhaust the Gateway's
+    descriptor limit and pin the staged bytes of every unlinked upload. The
+    duplication therefore runs as its own task, shielded from the caller's
+    cancellation, and closes itself if the caller is gone by the time the
+    worker finishes.
+    """
+    duplication = asyncio.ensure_future(run_file_io(os.dup, fileno))
+    try:
+        return await asyncio.shield(duplication)
+    except BaseException:
+        duplication.add_done_callback(_close_abandoned_fd)
+        raise
+
+
+def _copy_fd_to_path(fd: int, dest: Path) -> None:
+    """Copy the bytes behind *fd* to *dest*, then close *fd*.
+
+    Reads through the descriptor, not the name it was committed under, so the
+    copy is the content this request wrote even if the name has since been
+    replaced.
+    """
+    try:
+        os.lseek(fd, 0, os.SEEK_SET)
+        with open(dest, "wb") as out:
+            while chunk := os.read(fd, 1 << 20):
+                out.write(chunk)
+    finally:
+        _close_fd(fd)
+
+
 def _uploads() -> Any:
     """Return the uploads router module (late binding — see module docstring)."""
     from app.gateway.routers import uploads
@@ -192,6 +248,7 @@ class ThreadUploadIngestionService:
 
         file_size = 0
         upload_temp = None
+        convert_source_fd: int | None = None
         try:
             upload_temp = await run_file_io(uploads._prepare_upload_destination, self._uploads_dir, safe_filename)
             async for chunk in chunks:
@@ -202,6 +259,13 @@ class ThreadUploadIngestionService:
                 if self._total_size > self._limits.max_total_size:
                     raise HTTPException(status_code=413, detail="Total upload size too large")
                 await run_file_io(uploads._write_upload_chunk, upload_temp, chunk)
+            if self._auto_convert and Path(safe_filename).suffix.lower() in uploads.CONVERTIBLE_EXTENSIONS:
+                # Conversion must read the bytes this request staged. Once the
+                # name is committed a sandbox process can replace it with a
+                # symlink, and converting by name would then pull a host file
+                # into this thread's uploads. A descriptor on the staged inode
+                # cannot be redirected that way.
+                convert_source_fd = await _dup_for_conversion(upload_temp.handle.fileno())
             # 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).
@@ -213,10 +277,16 @@ class ThreadUploadIngestionService:
                     safe_filename = uploads.claim_unique_filename(safe_filename, self._seen_filenames)
             upload_temp = None
         except uploads.UnsafeUploadPathError as exc:
+            _close_fd(convert_source_fd)
             if upload_temp is not None:
                 await run_file_io(uploads._abort_upload_temp, upload_temp)
             raise UnsafeUploadDestinationError(safe_filename) from exc
-        except Exception:
+        except BaseException:
+            # BaseException, not Exception: a cancellation between the
+            # duplication and the commit would otherwise leave the descriptor
+            # open, and nothing downstream owns it yet.
+            _close_fd(convert_source_fd)
+            convert_source_fd = None
             if upload_temp is not None:
                 await run_file_io(uploads._abort_upload_temp, upload_temp)
             raise
@@ -237,7 +307,7 @@ class ThreadUploadIngestionService:
             file_info["original_filename"] = original_filename
         logger.info(f"Saved file: {safe_filename} ({file_size} bytes) to {file_info['path']}")
 
-        if self._auto_convert and file_path.suffix.lower() in uploads.CONVERTIBLE_EXTENSIONS:
+        if convert_source_fd is not None:
             # The companion gets the same atomic no-overwrite commit as the
             # original: staged under the hidden .part pattern, link-committed
             # with next-suffix retry — conversion can never silently truncate
@@ -246,12 +316,35 @@ class ThreadUploadIngestionService:
             provisional_md_name = Path(safe_filename).with_suffix(".md").name
             unique_md_name = uploads.claim_unique_filename(provisional_md_name, self._seen_filenames)
             md_staging = self._uploads_dir / f"{uploads.UPLOAD_STAGING_PREFIX}{uuid.uuid4().hex}{uploads.UPLOAD_STAGING_SUFFIX}"
+            # The staged bytes are copied out of the sandbox-writable tree and
+            # converted there; the uploads dir only ever receives the result.
+            # Creating that directory belongs inside the cleanup scope: it can
+            # fail on its own (a full or unwritable temporary filesystem), and
+            # the descriptor is already owned here — leaking it would also hold
+            # the unlinked staged bytes until the process exits.
+            private_dir: Path | None = None
             try:
-                md_staged = await uploads.convert_file_to_markdown(file_path, output_path=md_staging)
+                private_dir = Path(await run_file_io(tempfile.mkdtemp, "-deerflow-convert"))
+                conversion_source = private_dir / safe_filename
+                # Hand the descriptor over before the call: the copy closes it
+                # even when it fails, so this scope must not close it again and
+                # risk closing an unrelated descriptor that reused the number.
+                # await_drained, not a bare await: a cancelled await would
+                # cancel the queued executor job before its worker — and its
+                # closing finally — ever ran, and draining also keeps the
+                # worker from writing into a private directory this scope has
+                # already removed.
+                staged_fd, convert_source_fd = convert_source_fd, None
+                await await_drained(run_file_io(_copy_fd_to_path, staged_fd, conversion_source))
+                md_staged = await uploads.convert_file_to_markdown(conversion_source, output_path=md_staging)
             except Exception:
                 self._seen_filenames.discard(unique_md_name)
                 await run_file_io(md_staging.unlink, True)
                 raise
+            finally:
+                _close_fd(convert_source_fd)
+                if private_dir is not None:
+                    await run_file_io(shutil.rmtree, private_dir, True)
             if not md_staged:
                 # Conversion failed and wrote nothing (or a partial staged
                 # file, removed here): release the claim; holding it would

backend/docs/FILE_UPLOAD.md

@@ -237,6 +237,7 @@ backend/.deer-flow/threads/
 - 文件名安全性：系统会自动验证文件路径，防止目录遍历攻击
 - 删除只作用于普通文件：上传目录中的符号链接不会被跟随，删除请求按文件不存在（404）处理
 - 上传（HTTP 与嵌入式 `DeerFlowClient`）不会写穿符号链接：目标名已是符号链接的文件会被跳过并列入 `skipped_files`，转换生成的 Markdown 也不会写入同名符号链接
+- 转换读取的是本次上传写入的字节，而非落盘后的文件名：HTTP 上传在 uploads 之外的私有副本上转换，嵌入式客户端转换调用方提供的源文件，因此沙箱替换该文件名无法让宿主文件内容被转换进 uploads
 - 线程隔离：每个线程的上传文件相互隔离，无法跨线程访问
 - 自动文档转换默认关闭；如需启用，需在 `config.yaml` 中显式设置 `uploads.auto_convert_documents: true`
 

backend/packages/harness/deerflow/client.py

@@ -1687,14 +1687,18 @@ class DeerFlowClient:
                     provisional_md_name = Path(dest_name).with_suffix(".md").name
                     unique_md_name = claim_unique_filename(provisional_md_name, seen_names)
                     try:
-                        # Convert outside the sandbox-writable uploads dir, then
-                        # publish without following a symlink at the companion name.
+                        # Convert the caller's own file, not the copy that just
+                        # landed in the sandbox-writable uploads dir: a sandbox
+                        # that swaps that name for a symlink would otherwise have
+                        # a host file converted into this thread's uploads. Write
+                        # the result outside uploads too, then publish it without
+                        # following a symlink at the companion name.
                         with tempfile.TemporaryDirectory() as md_dir:
                             md_output = Path(md_dir) / unique_md_name
                             if conversion_pool is not None:
-                                converted = conversion_pool.submit(_convert_in_thread, dest, md_output).result()
+                                converted = conversion_pool.submit(_convert_in_thread, src_path, md_output).result()
                             else:
-                                converted = asyncio.run(convert_file_to_markdown(dest, output_path=md_output))
+                                converted = asyncio.run(convert_file_to_markdown(src_path, output_path=md_output))
                             md_path = None
                             if converted is not None:
                                 # copy, not write_bytes: the companion keeps the

backend/tests/test_client.py

@@ -2712,6 +2712,83 @@ class TestUploads:
             assert (uploads_dir / "a.md").read_text(encoding="utf-8") == "FROM:a.pdf"
             assert not (uploads_dir / "a_1.md").exists()
 
+    def test_upload_files_converts_the_source_not_the_landed_copy(self, client):
+        """A sandbox swapping the landed upload must not redirect conversion at a host file."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            uploads_dir = tmp_path / "uploads"
+            uploads_dir.mkdir()
+            host_file = tmp_path / "host-secret.pdf"
+            host_file.write_bytes(b"HOST SECRET")
+            pdf = tmp_path / "report.pdf"
+            pdf.write_bytes(b"pdf-bytes")
+
+            async def racing_convert(path: Path, output_path: Path | None = None) -> Path:
+                # The sandbox wins the race: the landed upload now points outside uploads.
+                landed = uploads_dir / "report.pdf"
+                if landed.exists() and not landed.is_symlink():
+                    landed.unlink()
+                    try:
+                        landed.symlink_to(host_file)
+                    except OSError as exc:
+                        if getattr(exc, "winerror", None) == 1314:
+                            pytest.skip("symlink creation requires Developer Mode or elevated privileges on Windows")
+                        raise
+                md_path = output_path if output_path is not None else path.with_suffix(".md")
+                md_path.write_bytes(b"CONVERTED:" + path.read_bytes())
+                return md_path
+
+            with (
+                patch("deerflow.client.get_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}),
+                patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=racing_convert),
+            ):
+                result = client.upload_files("thread-1", [pdf])
+
+            companion = uploads_dir / result["files"][0]["markdown_file"]
+            assert companion.read_bytes() == b"CONVERTED:pdf-bytes"
+            assert b"HOST SECRET" not in companion.read_bytes()
+
+    def test_upload_files_converts_the_source_inside_an_event_loop_too(self, client):
+        """The pooled conversion branch reads the source file as well."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            uploads_dir = tmp_path / "uploads"
+            uploads_dir.mkdir()
+            host_file = tmp_path / "host-secret.pdf"
+            host_file.write_bytes(b"HOST SECRET")
+            pdf = tmp_path / "report.pdf"
+            pdf.write_bytes(b"pdf-bytes")
+
+            async def racing_convert(path: Path, output_path: Path | None = None) -> Path:
+                landed = uploads_dir / "report.pdf"
+                if landed.exists() and not landed.is_symlink():
+                    landed.unlink()
+                    try:
+                        landed.symlink_to(host_file)
+                    except OSError as exc:
+                        if getattr(exc, "winerror", None) == 1314:
+                            pytest.skip("symlink creation requires Developer Mode or elevated privileges on Windows")
+                        raise
+                md_path = output_path if output_path is not None else path.with_suffix(".md")
+                md_path.write_bytes(b"CONVERTED:" + path.read_bytes())
+                return md_path
+
+            async def call_upload() -> dict:
+                return client.upload_files("thread-async", [pdf])
+
+            with (
+                patch("deerflow.client.get_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}),
+                patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=racing_convert),
+            ):
+                result = asyncio.run(call_upload())
+
+            companion = uploads_dir / result["files"][0]["markdown_file"]
+            assert companion.read_bytes() == b"CONVERTED:pdf-bytes"
+
     def test_upload_files_rejects_reuploading_a_file_already_in_the_thread(self, client):
         """Uploading an existing upload onto itself must not destroy its bytes."""
         with tempfile.TemporaryDirectory() as tmp:

backend/tests/test_uploads_router.py

@@ -12,6 +12,7 @@ from _router_auth_helpers import call_unwrapped, make_authed_test_app
 from fastapi import HTTPException, UploadFile
 from fastapi.testclient import TestClient
 
+from app.gateway import upload_ingestion
 from app.gateway.deps import get_config
 from app.gateway.routers import uploads
 from deerflow.sandbox.lease import get_sandbox_lease_manager
@@ -935,6 +936,316 @@ def test_upload_files_oversized_replacement_preserves_existing_regular_file(tmp_
     assert [path.name for path in thread_uploads_dir.iterdir()] == ["a.txt"]
 
 
+def test_upload_files_converts_the_bytes_it_wrote_not_the_committed_name(tmp_path):
+    """A sandbox swapping the landed upload must not redirect conversion at a host file."""
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    host_file = tmp_path / "host-secret.pdf"
+    host_file.write_bytes(b"HOST SECRET")
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+
+    async def racing_convert(file_path: Path, output_path: Path | None = None) -> Path:
+        # The sandbox wins the race: the committed name now points outside uploads.
+        landed = thread_uploads_dir / "report.pdf"
+        if landed.exists() and not landed.is_symlink():
+            landed.unlink()
+            _symlink_to_or_skip(landed, host_file)
+        md_path = output_path if output_path is not None else file_path.with_suffix(".md")
+        md_path.write_bytes(b"CONVERTED:" + file_path.read_bytes())
+        return md_path
+
+    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=racing_convert)),
+    ):
+        file = ChunkedUpload("report.pdf", [b"pdf-bytes"])
+        result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-race", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    companion = thread_uploads_dir / result.files[0].markdown_file
+    assert companion.read_bytes() == b"CONVERTED:pdf-bytes"
+    assert b"HOST SECRET" not in companion.read_bytes()
+
+
+def test_upload_files_conversion_source_survives_a_swap_before_it_is_read(tmp_path):
+    """The swap lands after the upload commits and before conversion reads it."""
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    host_file = tmp_path / "host-secret.pdf"
+    host_file.write_bytes(b"HOST SECRET")
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+    real_mkdtemp = upload_ingestion.tempfile.mkdtemp
+
+    def swap_then_mkdtemp(*args, **kwargs):
+        # Runs between the link-commit and the read of the staged bytes.
+        landed = thread_uploads_dir / "report.pdf"
+        if landed.exists() and not landed.is_symlink():
+            landed.unlink()
+            _symlink_to_or_skip(landed, host_file)
+        return real_mkdtemp(*args, **kwargs)
+
+    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_bytes(b"CONVERTED:" + file_path.read_bytes())
+        return md_path
+
+    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(upload_ingestion.tempfile, "mkdtemp", side_effect=swap_then_mkdtemp),
+    ):
+        file = ChunkedUpload("report.pdf", [b"pdf-bytes"])
+        result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-race2", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    companion = thread_uploads_dir / result.files[0].markdown_file
+    assert companion.read_bytes() == b"CONVERTED:pdf-bytes"
+    assert b"HOST SECRET" not in companion.read_bytes()
+
+
+def test_upload_files_closes_conversion_descriptor_when_private_dir_fails(tmp_path):
+    """A failed temp-dir creation must not strand the duplicated descriptor."""
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+
+    duplicated: list[int] = []
+    closed: list[int] = []
+    real_dup, real_close = os.dup, os.close
+
+    def tracking_dup(fd: int) -> int:
+        new_fd = real_dup(fd)
+        duplicated.append(new_fd)
+        return new_fd
+
+    def tracking_close(fd: int) -> None:
+        closed.append(fd)
+        real_close(fd)
+
+    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()),
+        patch.object(upload_ingestion.os, "dup", side_effect=tracking_dup),
+        patch.object(upload_ingestion.os, "close", side_effect=tracking_close),
+        patch.object(upload_ingestion.tempfile, "mkdtemp", side_effect=OSError("No space left on device")),
+    ):
+        file = ChunkedUpload("report.pdf", [b"pdf-bytes"])
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(call_unwrapped(uploads.upload_files, "thread-fd", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    assert exc_info.value.status_code == 500
+    assert len(duplicated) == 1
+    assert duplicated[0] in closed
+
+
+def test_upload_files_closes_conversion_descriptor_when_ingestion_is_cancelled(tmp_path):
+    """Cancelling mid-duplication must not strand the descriptor the worker still produces."""
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+
+    started = threading.Event()
+    release = threading.Event()
+    duplicated: list[int] = []
+    closed: list[int] = []
+    real_dup, real_close = os.dup, os.close
+
+    def slow_dup(fd: int) -> int:
+        # Allocate first, then stall: the descriptor exists while the caller
+        # is cancelled, which is exactly what must not be abandoned.
+        new_fd = real_dup(fd)
+        duplicated.append(new_fd)
+        started.set()
+        release.wait(5)
+        return new_fd
+
+    def tracking_close(fd: int) -> None:
+        closed.append(fd)
+        real_close(fd)
+
+    async def scenario() -> None:
+        service = upload_ingestion.ThreadUploadIngestionService(request=None, thread_id="thread-cancel", user_id="u", app_config=SimpleNamespace())
+        await service.open()
+
+        async def chunks():
+            yield b"pdf-bytes"
+
+        task = asyncio.create_task(service.ingest_chunks(chunks(), display_name="report.pdf"))
+        await asyncio.to_thread(started.wait, 5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        # The worker cannot be interrupted; it finishes after the cancellation.
+        release.set()
+        for _ in range(100):
+            if duplicated and duplicated[0] in closed:
+                break
+            await asyncio.sleep(0.05)
+        await service.aclose()
+
+    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()),
+        patch.object(upload_ingestion.os, "dup", side_effect=slow_dup),
+        patch.object(upload_ingestion.os, "close", side_effect=tracking_close),
+    ):
+        asyncio.run(scenario())
+
+    assert len(duplicated) == 1, "the duplication worker must have run"
+    assert duplicated[0] in closed, "the abandoned descriptor was never closed"
+
+
+def test_upload_files_closes_conversion_descriptor_when_cancelled_during_commit(tmp_path):
+    """Cancellation after the descriptor is owned must still close it."""
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+
+    started = threading.Event()
+    release = threading.Event()
+    duplicated: list[int] = []
+    closed: list[int] = []
+    real_dup, real_close = os.dup, os.close
+    real_commit = uploads._commit_upload_temp_no_overwrite
+
+    def tracking_dup(fd: int) -> int:
+        new_fd = real_dup(fd)
+        duplicated.append(new_fd)
+        return new_fd
+
+    def tracking_close(fd: int) -> None:
+        closed.append(fd)
+        real_close(fd)
+
+    def slow_commit(*args, **kwargs):
+        started.set()
+        release.wait(5)
+        return real_commit(*args, **kwargs)
+
+    async def scenario() -> None:
+        service = upload_ingestion.ThreadUploadIngestionService(request=None, thread_id="thread-cancel-commit", user_id="u", app_config=SimpleNamespace())
+        await service.open()
+
+        async def chunks():
+            yield b"pdf-bytes"
+
+        task = asyncio.create_task(service.ingest_chunks(chunks(), display_name="report.pdf"))
+        await asyncio.to_thread(started.wait, 5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        release.set()
+        for _ in range(100):
+            if duplicated and duplicated[0] in closed:
+                break
+            await asyncio.sleep(0.05)
+        await service.aclose()
+
+    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()),
+        patch.object(uploads, "_commit_upload_temp_no_overwrite", side_effect=slow_commit),
+        patch.object(upload_ingestion.os, "dup", side_effect=tracking_dup),
+        patch.object(upload_ingestion.os, "close", side_effect=tracking_close),
+    ):
+        asyncio.run(scenario())
+
+    assert len(duplicated) == 1
+    assert duplicated[0] in closed, "the owned descriptor was not closed on cancellation"
+
+
+def test_upload_files_closes_conversion_descriptor_when_cancelled_while_copy_is_queued(tmp_path):
+    """A copy job cancelled before its worker starts must not strand the descriptor."""
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+
+    duplicated: list[int] = []
+    closed: list[int] = []
+    real_dup, real_close = os.dup, os.close
+    real_run_file_io = upload_ingestion.run_file_io
+    state: dict[str, object] = {}
+
+    def tracking_dup(fd: int) -> int:
+        new_fd = real_dup(fd)
+        duplicated.append(new_fd)
+        return new_fd
+
+    def tracking_close(fd: int) -> None:
+        closed.append(fd)
+        real_close(fd)
+
+    async def queueing_run_file_io(func, *args, **kwargs):
+        if func is not upload_ingestion._copy_fd_to_path:
+            return await real_run_file_io(func, *args, **kwargs)
+        # The pool is busy: the job is queued, not running.
+        state["queued"].set()
+        await state["release"].wait()
+        # Only a job that was never cancelled reaches its worker.
+        return func(*args, **kwargs)
+
+    async def scenario() -> None:
+        state["queued"] = asyncio.Event()
+        state["release"] = asyncio.Event()
+        service = upload_ingestion.ThreadUploadIngestionService(request=None, thread_id="thread-cancel-copy", user_id="u", app_config=SimpleNamespace())
+        await service.open()
+
+        async def chunks():
+            yield b"pdf-bytes"
+
+        task = asyncio.create_task(service.ingest_chunks(chunks(), display_name="report.pdf"))
+        await asyncio.wait_for(state["queued"].wait(), 5)
+        task.cancel()
+        state["release"].set()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        for _ in range(100):
+            if duplicated and duplicated[0] in closed:
+                break
+            await asyncio.sleep(0.05)
+        await service.aclose()
+
+    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()),
+        patch.object(upload_ingestion, "run_file_io", side_effect=queueing_run_file_io),
+        patch.object(upload_ingestion.os, "dup", side_effect=tracking_dup),
+        patch.object(upload_ingestion.os, "close", side_effect=tracking_close),
+    ):
+        asyncio.run(scenario())
+
+    assert len(duplicated) == 1
+    assert duplicated[0] in closed, "the descriptor handed to the queued copy was never closed"
+
+
 def test_delete_uploaded_file_removes_generated_markdown_companion(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)