diff --git a/README.md b/README.md
index 1e5e13a6d1..f1ab1d2165 100644
--- a/README.md
+++ b/README.md
@@ -2093,6 +2093,8 @@ Content-less sub-agent final messages report `No response generated` instead of
 
 An ordinary `task` also receives a defensive snapshot of the dispatching run's current uploads. This lets eligible sub-agents use `list_uploaded_files` to find earlier-turn files without returning same-turn attachments as historical. Delayed or recovered `batch_task` workers leave this tool disabled because they have no valid turn-local upload boundary.
 
+Embedded Python uploads use the same captured content for the saved document and its Markdown conversion, even if the caller's source changes during upload.
+
 历史上传发现支持稳定续页：`list_uploaded_files` 默认每页 20 项、最多 100 项，
 将返回的 `next_cursor` 作为下一次调用的 `cursor`，保留相同过滤条件，直到末页不再
 返回游标。目录元数据或调用上下文变化时会明确要求重新枚举；详情见
@@ -2665,6 +2667,14 @@ client.clear_goal("thread-1")
 
 When continuing a thread, `values` still contains the full conversation state, while `messages-tuple` and `end.usage` cover only the current turn.
 
+Concurrent `upload_files()` calls preserve filenames absent at the start of each
+call by assigning numeric suffixes and returning the actual saved names. The same
+rule applies to new Markdown companions. Exact names already present retain
+replacement behavior. Convertible files use one private source capture through
+all filename retries and conversion, preserving the original permissions and
+timestamps. Detected staging-file substitution is rejected; staging cleanup
+failures are logged without masking the upload outcome.
+
 The HTTP Gateway accepts `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom` stream modes. Unsupported modes such as `messages` and `events`, unsupported non-default run options such as webhooks, delayed execution, or `multitask_strategy="enqueue"`, and undeclared SDK options such as checkpoint durability overrides return `422` before execution instead of being silently ignored or downgraded.
 
 All dict-returning methods are validated against Gateway Pydantic response models in CI (`TestGatewayConformance`), ensuring the embedded client stays in sync with the HTTP API schemas. See `backend/packages/harness/deerflow/client.py` for full API documentation.
diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md
index bd3856f7b5..25879c0cb1 100644
--- a/backend/packages/harness/deerflow/AGENTS.md
+++ b/backend/packages/harness/deerflow/AGENTS.md
@@ -93,6 +93,8 @@ drift.
 
 **Gateway differences**: Upload takes local `Path`, not `UploadFile`, rejects directories before copying, and reuses one conversion worker inside an active event loop. Artifacts return `(bytes, mime_type)`, not HTTP Response. Gateway alone deletes `.deer-flow/threads/{thread_id}` after LangGraph thread deletion; the client has no equivalent. `update_mcp_config()` and `update_skill()` invalidate the cached agent.
 
+Upload capture, filename claims and publication: [guide](uploads/AGENTS.md).
+
 **Tests**: `tests/test_client.py` is offline, including `TestGatewayConformance`.
 `tests/test_client_live.py` requires root `config.yaml`, valid API credentials,
 and opt-in via `make test-live` or `DEER_FLOW_RUN_LIVE_TESTS=1`. It calls real
diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py
index d467d54d06..be54b91425 100644
--- a/backend/packages/harness/deerflow/client.py
+++ b/backend/packages/harness/deerflow/client.py
@@ -82,6 +82,7 @@ from deerflow.uploads.manager import (
     get_uploads_dir,
     list_files_in_dir,
     normalize_filename,
+    publish_upload_copy,
     upload_artifact_url,
     upload_virtual_path,
 )
@@ -1777,12 +1778,15 @@ class DeerFlowClient:
                 raise FileNotFoundError(f"File not found: {f}")
             if not p.is_file():
                 raise ValueError(f"Path is not a file: {f}")
-            dest_name = claim_unique_filename(normalize_filename(p.name), seen_names)
-            resolved_files.append((p, dest_name))
+            claim_seed = normalize_filename(p.name)
+            dest_name = claim_unique_filename(claim_seed, seen_names)
+            resolved_files.append((p, dest_name, claim_seed))
             if not has_convertible_file and p.suffix.lower() in CONVERTIBLE_EXTENSIONS:
                 has_convertible_file = True
 
         uploads_dir = ensure_uploads_dir(thread_id)
+        with os.scandir(uploads_dir) as entries:
+            existing_names = {entry.name for entry in entries if entry.is_file(follow_symlinks=False)}
         uploaded_files: list[dict] = []
         skipped_files: list[str] = []
 
@@ -1802,11 +1806,26 @@ class DeerFlowClient:
         def _convert_in_thread(path: Path, output_path: Path | None = None):
             return asyncio.run(convert_file_to_markdown(path, output_path=output_path))
 
+        conversion_sources = None
         try:
-            for src_path, dest_name in resolved_files:
+            for src_path, dest_name, claim_seed in resolved_files:
                 try:
-                    dest = copy_upload_file_no_symlink(uploads_dir, dest_name, src_path)
+                    conversion_source = None
+                    if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS:
+                        if conversion_sources is None:
+                            conversion_sources = tempfile.TemporaryDirectory(prefix="deerflow-upload-source-")
+                        conversion_source = Path(conversion_sources.name) / dest_name
+                    if dest_name in existing_names:
+                        if conversion_source is not None:
+                            dest = copy_upload_file_no_symlink(uploads_dir, dest_name, src_path, snapshot_path=conversion_source)
+                        else:
+                            dest = copy_upload_file_no_symlink(uploads_dir, dest_name, src_path)
+                    else:
+                        dest = publish_upload_copy(uploads_dir, dest_name, src_path, claim_seed=claim_seed, seen_names=seen_names, snapshot_path=conversion_source)
+                        dest_name = dest.name
                 except UnsafeUploadPathError:
+                    if conversion_source is not None:
+                        conversion_source.unlink(missing_ok=True)
                     logger.warning("Skipping upload with unsafe destination: %s", dest_name)
                     skipped_files.append(dest_name)
                     continue
@@ -1829,24 +1848,25 @@ 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 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.
+                        # Convert the same private capture used to publish the
+                        # original. Neither an external source edit nor a sandbox
+                        # replacing the uploaded path can change these input bytes.
                         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, src_path, md_output).result()
+                                converted = conversion_pool.submit(_convert_in_thread, conversion_source, md_output).result()
                             else:
-                                converted = asyncio.run(convert_file_to_markdown(src_path, output_path=md_output))
+                                converted = asyncio.run(convert_file_to_markdown(conversion_source, output_path=md_output))
                             md_path = None
                             if converted is not None:
                                 # copy, not write_bytes: the companion keeps the
                                 # converter's permissions, so a sandbox running as
                                 # another uid can still read it.
-                                md_path = copy_upload_file_no_symlink(uploads_dir, unique_md_name, converted)
+                                if unique_md_name in existing_names:
+                                    md_path = copy_upload_file_no_symlink(uploads_dir, unique_md_name, converted)
+                                else:
+                                    md_path = publish_upload_copy(uploads_dir, unique_md_name, converted, claim_seed=provisional_md_name, seen_names=seen_names)
+                                    unique_md_name = md_path.name
                     except UnsafeUploadPathError:
                         logger.warning("Skipping markdown companion with unsafe destination: %s", unique_md_name)
                         md_path = None
@@ -1870,10 +1890,14 @@ class DeerFlowClient:
                         # against a name this request never filled.
                         seen_names.discard(unique_md_name)
 
+                if conversion_source is not None:
+                    conversion_source.unlink(missing_ok=True)
                 uploaded_files.append(info)
         finally:
             if conversion_pool is not None:
                 conversion_pool.shutdown(wait=True)
+            if conversion_sources is not None:
+                conversion_sources.cleanup()
 
         message = f"Successfully uploaded {len(uploaded_files)} file(s)"
         if skipped_files:
diff --git a/backend/packages/harness/deerflow/uploads/AGENTS.md b/backend/packages/harness/deerflow/uploads/AGENTS.md
index 9b923916f3..350da55926 100644
--- a/backend/packages/harness/deerflow/uploads/AGENTS.md
+++ b/backend/packages/harness/deerflow/uploads/AGENTS.md
@@ -8,3 +8,23 @@ entry count at debug level. Permission and other operational errors propagate.
 
 Keep both entry and directory race coverage in
 `backend/tests/test_uploads_manager.py`.
+
+Embedded first-time uploads use `publish_upload_copy` for suffix retries from the
+normalized original name, for originals and companions. With `snapshot_path`,
+capture the opened source once before retries and retain its original `fstat`
+metadata; every attempt reads the held private capture. Replacement copies share
+this capture helper. Never copy snapshot permission bits onto the saved upload.
+Close the caller's descriptor after capture and finish conversion workers before
+removing captures. Never reopen the caller/sandbox upload path for conversion.
+Names already present in the call's initial regular-file snapshot retain
+replacement behavior; these replacements are not serialized. Apply source
+timestamps after new-file publication so cleanup cannot sweep an in-flight copy
+of an old source. New final names must expose complete copies.
+
+No-replace publication binds pre-link staging and post-link destination checks to
+the open staging descriptor. Use a no-follow hard link where supported; fallback
+identity checks narrow but do not eliminate the link-time race. Reject substituted
+regular files and symlinks, and remove a mismatched publication before reporting
+failure. Cleanup errors are logged without masking success or the original error.
+Keep publication/fallback coverage in `backend/tests/test_upload_copy_publication.py`
+and capture-through-collision cases in `backend/tests/test_client_upload_snapshot.py`.
diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py
index 7ce5fa6831..58bb1c866f 100644
--- a/backend/packages/harness/deerflow/uploads/manager.py
+++ b/backend/packages/harness/deerflow/uploads/manager.py
@@ -9,9 +9,13 @@ import logging
 import os
 import shutil
 import stat
+import tempfile
 import time
+from collections.abc import Iterator
+from contextlib import contextmanager, nullcontext
 from datetime import timedelta
 from pathlib import Path
+from typing import BinaryIO
 from urllib.parse import quote
 
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
@@ -370,7 +374,24 @@ def _reject_same_file(base_dir: Path, filename: str, src: Path, src_stat: os.sta
         raise shutil.SameFileError(f"{src!r} and {dest!r} are the same file")
 
 
-def copy_upload_file_no_symlink(base_dir: Path, filename: str, src: Path) -> Path:
+@contextmanager
+def _open_upload_copy_source(base_dir: Path, filename: str, src: Path, *, snapshot_path: Path | None = None) -> Iterator[tuple[BinaryIO, os.stat_result]]:
+    """Hold one source identity and, when requested, capture its bytes once."""
+    with open(snapshot_path, "w+b") if snapshot_path is not None else nullcontext() as captured:
+        with open(src, "rb") as src_fh:
+            src_stat = os.fstat(src_fh.fileno())
+            _reject_same_file(base_dir, filename, src, src_stat)
+            if captured is None:
+                yield src_fh, src_stat
+                return
+            shutil.copyfileobj(src_fh, captured)
+            captured.seek(0)
+        # Release the caller's file before publication, so removal during a
+        # collision retry is supported on hosts that cannot unlink open files.
+        yield captured, src_stat
+
+
+def copy_upload_file_no_symlink(base_dir: Path, filename: str, src: Path, *, snapshot_path: Path | None = None) -> Path:
     """Copy *src* into an upload destination without following a destination symlink.
 
     Matches ``shutil.copy2`` for content, permission bits and timestamps, but
@@ -385,13 +406,15 @@ def copy_upload_file_no_symlink(base_dir: Path, filename: str, src: Path) -> Pat
     truncates, which would otherwise leave the caller copying an emptied file
     over itself. Re-uploading a file that already sits in the uploads
     directory takes exactly that path.
+
+    With ``snapshot_path``, first capture the opened source there and publish
+    those captured bytes. The caller owns this path in a private directory
+    outside the sandbox tree and can convert it without reopening ``src``.
     """
-    with open(src, "rb") as src_fh:
-        src_stat = os.fstat(src_fh.fileno())
-        _reject_same_file(base_dir, filename, src, src_stat)
+    with _open_upload_copy_source(base_dir, filename, src, snapshot_path=snapshot_path) as (captured, src_stat):
         dest, fh = open_upload_file_no_symlink(base_dir, filename)
         with fh:
-            shutil.copyfileobj(src_fh, fh)
+            shutil.copyfileobj(captured, fh)
             fh.flush()
             if os.chmod in os.supports_fd:
                 os.chmod(fh.fileno(), stat.S_IMODE(src_stat.st_mode))
@@ -400,6 +423,83 @@ def copy_upload_file_no_symlink(base_dir: Path, filename: str, src: Path) -> Pat
     return dest
 
 
+def _validate_upload_copy_identity(path: Path, expected: os.stat_result) -> None:
+    try:
+        current = path.lstat()
+    except (FileNotFoundError, NotADirectoryError) as exc:
+        raise UnsafeUploadPathError(f"Upload copy disappeared: {path.name}") from exc
+    if not stat.S_ISREG(current.st_mode) or not os.path.samestat(current, expected):
+        raise UnsafeUploadPathError(f"Upload copy changed during publication: {path.name}")
+
+
+def _publish_upload_copy_source(base_dir: Path, filename: str, source: BinaryIO, src_stat: os.stat_result) -> Path:
+    """Publish an opened source without replacing a concurrently claimed name."""
+    safe_name = normalize_filename(filename)
+    dest = validate_upload_destination(base_dir, safe_name, exclusive=True)
+    fd, staging_name = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=base_dir)
+    staging = Path(staging_name)
+    try:
+        with os.fdopen(fd, "wb") as staged_fh:
+            shutil.copyfileobj(source, staged_fh)
+            staged_fh.flush()
+            if os.chmod in os.supports_fd:
+                os.chmod(staged_fh.fileno(), stat.S_IMODE(src_stat.st_mode))
+            staged_stat = os.fstat(staged_fh.fileno())
+            _validate_upload_copy_identity(staging, staged_stat)
+            if os.link in os.supports_follow_symlinks:
+                os.link(staging, dest, follow_symlinks=False)
+            else:
+                os.link(staging, dest)
+            try:
+                _validate_upload_copy_identity(dest, staged_stat)
+            except (UnsafeUploadPathError, OSError):
+                try:
+                    dest.unlink(missing_ok=True)
+                except OSError:
+                    logger.warning("Failed to remove invalid upload copy: %s", dest, exc_info=True)
+                raise
+            # Keep staging timestamps fresh until publication, so the orphan
+            # sweep cannot remove an in-flight copy of an old source.
+            if os.utime in os.supports_fd:
+                os.utime(staged_fh.fileno(), ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns))
+    finally:
+        try:
+            staging.unlink(missing_ok=True)
+        except OSError:
+            logger.warning("Failed to remove upload staging file: %s", staging, exc_info=True)
+    return dest
+
+
+def copy_upload_file_no_replace(base_dir: Path, filename: str, src: Path, *, snapshot_path: Path | None = None) -> Path:
+    """Publish a complete copy without replacing a concurrently claimed name.
+
+    Staging in the upload directory makes the final hard link atomic without
+    crossing filesystems. Reject staging/published names that no longer identify
+    the open copy. Use no-follow links where supported; otherwise identity checks
+    narrow, but cannot eliminate, the link-time race. Cleanup is best effort.
+    ``snapshot_path`` captures the opened source privately before publication.
+    """
+    with _open_upload_copy_source(base_dir, filename, src, snapshot_path=snapshot_path) as (source, src_stat):
+        return _publish_upload_copy_source(base_dir, filename, source, src_stat)
+
+
+def publish_upload_copy(base_dir: Path, filename: str, src: Path, *, claim_seed: str, seen_names: set[str], snapshot_path: Path | None = None) -> Path:
+    """Retry occupied names using one opened source and its original metadata.
+
+    For conversion, capture once before the retry loop. Every publication attempt
+    reads that held private capture rather than reopening the caller's source.
+    """
+    claim_seed = normalize_filename(claim_seed)
+    seen_names.add(filename)
+    with _open_upload_copy_source(base_dir, filename, src, snapshot_path=snapshot_path) as (source, src_stat):
+        while True:
+            try:
+                source.seek(0)
+                return _publish_upload_copy_source(base_dir, filename, source, src_stat)
+            except FileExistsError:
+                filename = claim_unique_filename(claim_seed, seen_names)
+
+
 def apply_upload_sandbox_permits(file_path: os.PathLike[str] | str, extra_mode_bits: int) -> None:
     """Apply sandbox permission bits to an upload, bound to its validated inode.
 
diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md
index 09d799a026..5bf609c5c7 100644
--- a/backend/tests/AGENTS.md
+++ b/backend/tests/AGENTS.md
@@ -1,5 +1,13 @@
 # Backend Tests
 
+`test_client_upload_snapshot.py` keeps upload publication and conversion on the
+same private capture. Cover source edits/removal, both conversion branches,
+same-file rejection, metadata and cleanup; retain sandbox symlink defenses.
+Exercise both replacement and first-time publication; mutation hooks must actually
+run on both paths. Force collision retries while editing/removing the source and
+assert one source open, original metadata, preserved competing uploads, companion
+ownership and capture cleanup. Retain the real XLSX conversion coverage.
+
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
 Local nginx path fixtures convert MSYS output through the selected Git Bash's
diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py
index 55ef31f10e..fcdfb43ada 100644
--- a/backend/tests/test_client.py
+++ b/backend/tests/test_client.py
@@ -7,6 +7,7 @@ import os
 import shutil
 import stat
 import tempfile
+import threading
 import zipfile
 from enum import Enum
 from pathlib import Path
@@ -5133,6 +5134,184 @@ class TestUploadDuplicateFilenames:
     includes original_filename so the agent / caller can see what happened.
     """
 
+    @pytest.mark.parametrize("companion", [False, True])
+    def test_rejects_substituted_upload_copy(self, client, tmp_path, monkeypatch, companion):
+        from support.symlinks import symlink_or_skip
+
+        from deerflow.uploads import manager
+
+        uploads = tmp_path / "uploads"
+        uploads.mkdir()
+        source = tmp_path / ("notes.pdf" if companion else "notes.txt")
+        source.write_bytes(b"original upload")
+        outside = tmp_path / "outside.txt"
+        outside.write_bytes(b"private test fixture")
+        real_copy = manager.shutil.copyfileobj
+        copy_count = 0
+
+        def copy_and_swap(src, dst):
+            nonlocal copy_count
+            real_copy(src, dst)
+            staging = next(uploads.glob(".upload-*.part"), None)
+            if staging is None:
+                # Source capture happens outside uploads, before staging.
+                return
+            copy_count += 1
+            if copy_count == (2 if companion else 1):
+                try:
+                    staging.unlink()
+                except PermissionError:
+                    if os.name == "nt":
+                        pytest.skip("host prevents replacing an open staging file")
+                    raise
+                symlink_or_skip(staging, outside)
+
+        async def fake_convert(path, output_path=None):
+            output_path.write_bytes(b"converted content")
+            return output_path
+
+        monkeypatch.setattr(manager.shutil, "copyfileobj", copy_and_swap)
+        with (
+            patch("deerflow.client.ensure_uploads_dir", return_value=uploads),
+            patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}),
+            patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=fake_convert),
+        ):
+            result = client.upload_files("t-dup", [source])
+
+        assert outside.read_bytes() == b"private test fixture"
+        assert outside.stat().st_nlink == 1
+        assert not list(uploads.glob(".upload-*.part"))
+        if companion:
+            assert copy_count == 2
+            assert result["success"] is True
+            assert (uploads / "notes.pdf").read_bytes() == b"original upload"
+            assert "markdown_file" not in result["files"][0]
+            assert not os.path.lexists(uploads / "notes.md")
+        else:
+            assert copy_count == 1
+            assert result["success"] is False
+            assert result["files"] == []
+            assert result["skipped_files"] == ["notes.txt"]
+            assert not os.path.lexists(uploads / "notes.txt")
+
+    @pytest.mark.parametrize("count", [2, 3])
+    @pytest.mark.parametrize("suffix", [".txt", ".pdf"])
+    def test_concurrent_uploads_claim_distinct_names(self, client, tmp_path, count, suffix):
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        sources = []
+        for index in range(count):
+            directory = tmp_path / str(index)
+            directory.mkdir()
+            source = directory / f"data{suffix}"
+            source.write_bytes(f"payload-{index}".encode())
+            sources.append(source)
+
+        all_ready = threading.Barrier(count)
+        source_names = set(sources)
+        opened_sources = set()
+        opened_lock = threading.Lock()
+        real_open = open
+
+        def synchronized_open(file, mode="r", *args, **kwargs):
+            fh = real_open(file, mode, *args, **kwargs)
+            wait = False
+            if mode == "rb" and isinstance(file, (str, os.PathLike)) and Path(file) in source_names:
+                with opened_lock:
+                    if Path(file) not in opened_sources:
+                        opened_sources.add(Path(file))
+                        wait = True
+            if wait:
+                # All callers reach the real copy before any publishes its name.
+                try:
+                    all_ready.wait(timeout=15)
+                except BaseException:
+                    fh.close()
+                    raise
+            return fh
+
+        async def fake_convert(path: Path, output_path: Path | None = None) -> Path:
+            assert output_path is not None
+            output_path.write_bytes(path.read_bytes())
+            return output_path
+
+        with (
+            patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+            patch("builtins.open", side_effect=synchronized_open),
+            patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}),
+            patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=fake_convert),
+        ):
+            with concurrent.futures.ThreadPoolExecutor(max_workers=count) as pool:
+                futures = [pool.submit(client.upload_files, "t-dup", [source]) for source in sources]
+                results = [future.result(timeout=30) for future in futures]
+
+        files = [result["files"][0] for result in results]
+        names = {file["filename"] for file in files}
+        assert names == {f"data{suffix}", *(f"data_{index}{suffix}" for index in range(1, count))}
+        for source, result, file in zip(sources, results, files, strict=True):
+            assert result["success"] is True
+            assert (uploads_dir / file["filename"]).read_bytes() == source.read_bytes()
+            if file["filename"] != source.name:
+                assert file["original_filename"] == source.name
+            if suffix == ".pdf":
+                from deerflow.uploads.companions import resolve_companion
+
+                assert (uploads_dir / file["markdown_file"]).read_bytes() == source.read_bytes()
+                assert resolve_companion(uploads_dir / file["filename"]) == uploads_dir / file["markdown_file"]
+        assert len(list(uploads_dir.iterdir())) == count * (2 if suffix == ".pdf" else 1)
+
+    def test_concurrent_same_stem_companions_remain_owned(self, client, tmp_path):
+        from deerflow.uploads.companions import resolve_companion
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        sources = [tmp_path / "report.pdf", tmp_path / "report.docx"]
+        for index, source in enumerate(sources):
+            source.write_bytes(f"document-{index}".encode())
+        both_converting = threading.Barrier(2)
+
+        async def fake_convert(path: Path, output_path: Path | None = None) -> Path:
+            assert output_path is not None
+            output_path.write_bytes(path.read_bytes())
+            both_converting.wait(timeout=15)
+            return output_path
+
+        with (
+            patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf", ".docx"}),
+            patch("deerflow.utils.file_conversion.convert_file_to_markdown", side_effect=fake_convert),
+        ):
+            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+                futures = [pool.submit(client.upload_files, "t-dup", [source]) for source in sources]
+                results = [future.result(timeout=30) for future in futures]
+
+        files = [result["files"][0] for result in results]
+        assert {file["markdown_file"] for file in files} == {"report.md", "report_1.md"}
+        for source, file in zip(sources, files, strict=True):
+            assert file["filename"] == source.name
+            companion = resolve_companion(uploads_dir / source.name)
+            assert companion == uploads_dir / file["markdown_file"]
+            assert companion.read_bytes() == source.read_bytes()
+        assert len(list(uploads_dir.iterdir())) == 4
+
+    def test_sequential_upload_keeps_replacement_behavior(self, client, tmp_path):
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        first = tmp_path / "first" / "data.txt"
+        second = tmp_path / "second" / "data.txt"
+        first.parent.mkdir()
+        second.parent.mkdir()
+        first.write_bytes(b"first")
+        second.write_bytes(b"second")
+
+        with patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir):
+            client.upload_files("t-dup", [first])
+            result = client.upload_files("t-dup", [second])
+
+        assert result["files"][0]["filename"] == "data.txt"
+        assert (uploads_dir / "data.txt").read_bytes() == b"second"
+        assert len(list(uploads_dir.iterdir())) == 1
+
     def test_duplicate_filenames_auto_renamed(self, client):
         """Two files with same basename → second gets _1 suffix."""
         with tempfile.TemporaryDirectory() as tmp:
diff --git a/backend/tests/test_client_upload_snapshot.py b/backend/tests/test_client_upload_snapshot.py
new file mode 100644
index 0000000000..793c2a1a14
--- /dev/null
+++ b/backend/tests/test_client_upload_snapshot.py
@@ -0,0 +1,343 @@
+"""Embedded upload and conversion must consume the same captured bytes."""
+
+import asyncio
+import getpass
+import os
+import shutil
+import stat
+import subprocess
+import sys
+from io import BytesIO
+from pathlib import Path
+
+import pytest
+
+from deerflow.client import DeerFlowClient
+from deerflow.uploads.companions import resolve_companion
+
+
+@pytest.fixture
+def upload_case(monkeypatch, tmp_path):
+    import deerflow.client as client_module
+
+    uploads = tmp_path / "thread/user-data/uploads"
+    uploads.mkdir(parents=True)
+    monkeypatch.setattr(client_module, "ensure_uploads_dir", lambda *args, **kwargs: uploads)
+    # upload_files has no dependency on initialized instance fields.
+    return DeerFlowClient.__new__(DeerFlowClient), uploads
+
+
+def call_upload(client, source, in_loop):
+    if not in_loop:
+        return client.upload_files("snapshot-thread", [source])
+
+    async def call():
+        return client.upload_files("snapshot-thread", [source])
+
+    return asyncio.run(call())
+
+
+@pytest.mark.parametrize("in_loop", [False, True])
+@pytest.mark.parametrize("change", ["edit", "remove"])
+@pytest.mark.parametrize("existing", [False, True])
+def test_conversion_keeps_uploaded_version(upload_case, monkeypatch, tmp_path, in_loop, change, existing):
+    import deerflow.client as client_module
+    import deerflow.utils.file_conversion as conversion
+
+    client, uploads = upload_case
+    source = tmp_path / "report.pdf"
+    source.write_bytes(b"VERSION A")
+    if existing:
+        (uploads / source.name).write_bytes(b"OLD UPLOAD")
+    conversion_paths = []
+    changes = []
+
+    def wrap_copy(real_copy):
+        def copy_then_change(base, name, src, **kwargs):
+            result = real_copy(base, name, src, **kwargs)
+            if Path(src) == source:
+                changes.append(change)
+                if change == "edit":
+                    source.write_bytes(b"VERSION B")
+                else:
+                    source.unlink()
+            return result
+
+        return copy_then_change
+
+    async def convert(path, output_path=None):
+        conversion_paths.append(path)
+        output_path.write_bytes(b"CONVERTED:" + path.read_bytes())
+        return output_path
+
+    monkeypatch.setattr(client_module, "copy_upload_file_no_symlink", wrap_copy(client_module.copy_upload_file_no_symlink))
+    monkeypatch.setattr(client_module, "publish_upload_copy", wrap_copy(client_module.publish_upload_copy))
+    monkeypatch.setattr(conversion, "CONVERTIBLE_EXTENSIONS", {".pdf"})
+    monkeypatch.setattr(conversion, "convert_file_to_markdown", convert)
+    response = call_upload(client, source, in_loop)
+
+    assert response["success"]
+    assert changes == [change]
+    saved = uploads / response["files"][0]["filename"]
+    assert saved.read_bytes() == b"VERSION A"
+    companion = resolve_companion(saved)
+    assert companion is not None
+    assert companion.read_bytes() == b"CONVERTED:VERSION A"
+    assert conversion_paths[0].name == source.name
+    assert conversion_paths[0] != source
+    assert not conversion_paths[0].is_relative_to(uploads)
+    assert not conversion_paths[0].parent.exists()
+
+
+@pytest.mark.parametrize("in_loop", [False, True])
+def test_failed_conversion_keeps_original_and_cleans_snapshot(upload_case, monkeypatch, tmp_path, in_loop):
+    import deerflow.utils.file_conversion as conversion
+
+    client, uploads = upload_case
+    source = tmp_path / "report.pdf"
+    source.write_bytes(b"VERSION A")
+    paths = []
+
+    async def fail(path, output_path=None):
+        paths.append(path)
+        raise RuntimeError("conversion failed")
+
+    monkeypatch.setattr(conversion, "CONVERTIBLE_EXTENSIONS", {".pdf"})
+    monkeypatch.setattr(conversion, "convert_file_to_markdown", fail)
+    response = call_upload(client, source, in_loop)
+    assert response["success"]
+    assert "markdown_file" not in response["files"][0]
+    assert (uploads / "report.pdf").read_bytes() == b"VERSION A"
+    assert source.read_bytes() == b"VERSION A"
+    assert paths[0] != source
+    assert not paths[0].parent.exists()
+
+
+def test_convertible_same_file_is_rejected_without_data_loss(upload_case):
+    client, uploads = upload_case
+    source = uploads / "report.pdf"
+    source.write_bytes(b"IMPORTANT BYTES")
+    with pytest.raises(shutil.SameFileError):
+        client.upload_files("snapshot-thread", [source])
+    assert source.read_bytes() == b"IMPORTANT BYTES"
+
+
+def test_captured_upload_preserves_original_metadata(upload_case, monkeypatch, tmp_path):
+    import deerflow.utils.file_conversion as conversion
+
+    client, uploads = upload_case
+    source = tmp_path / "report.pdf"
+    source.write_bytes(b"VERSION A")
+    source.chmod(0o640)
+    os.utime(source, ns=(1_700_000_000_000_000_000, 1_700_000_001_000_000_000))
+    before = source.stat()
+
+    async def convert(path, output_path=None):
+        output_path.write_bytes(path.read_bytes())
+        return output_path
+
+    monkeypatch.setattr(conversion, "CONVERTIBLE_EXTENSIONS", {".pdf"})
+    monkeypatch.setattr(conversion, "convert_file_to_markdown", convert)
+    client.upload_files("snapshot-thread", [source])
+    saved = (uploads / "report.pdf").stat()
+    if os.chmod in os.supports_fd:
+        assert stat.S_IMODE(saved.st_mode) == stat.S_IMODE(before.st_mode)
+    if os.utime in os.supports_fd:
+        assert saved.st_mtime_ns == before.st_mtime_ns
+
+
+@pytest.mark.parametrize("in_loop", [False, True])
+@pytest.mark.parametrize("existing", [False, True])
+def test_real_xlsx_conversion_uses_captured_version(upload_case, monkeypatch, tmp_path, in_loop, existing):
+    from openpyxl import Workbook, load_workbook
+
+    import deerflow.client as client_module
+
+    client, uploads = upload_case
+    source = tmp_path / "report.xlsx"
+    workbook = Workbook()
+    workbook.active["A1"] = "VERSION A"
+    workbook.save(source)
+    if existing:
+        (uploads / source.name).write_bytes(b"OLD UPLOAD")
+    saves = []
+
+    def wrap_copy(real_copy):
+        def copy_then_save(base, name, src, **kwargs):
+            result = real_copy(base, name, src, **kwargs)
+            if Path(src) == source:
+                saves.append(source)
+                workbook.active["A1"] = "VERSION B"
+                workbook.save(source)
+            return result
+
+        return copy_then_save
+
+    monkeypatch.setattr(client_module, "copy_upload_file_no_symlink", wrap_copy(client_module.copy_upload_file_no_symlink))
+    monkeypatch.setattr(client_module, "publish_upload_copy", wrap_copy(client_module.publish_upload_copy))
+    response = call_upload(client, source, in_loop)
+    assert response["success"]
+    assert saves == [source]
+    saved = uploads / "report.xlsx"
+    saved_workbook = load_workbook(saved)
+    try:
+        assert saved_workbook.active["A1"].value == "VERSION A"
+    finally:
+        saved_workbook.close()
+        workbook.close()
+    companion = resolve_companion(saved)
+    assert companion is not None
+    text = companion.read_text(encoding="utf-8")
+    assert "VERSION A" in text
+    assert "VERSION B" not in text
+
+
+def test_capture_failure_preserves_existing_upload(upload_case, monkeypatch, tmp_path):
+    import deerflow.uploads.manager as manager
+
+    client, uploads = upload_case
+    source = tmp_path / "report.pdf"
+    source.write_bytes(b"NEW CONTENT")
+    existing = uploads / "report.pdf"
+    existing.write_bytes(b"EXISTING CONTENT")
+    captures = []
+    real_copyfileobj = shutil.copyfileobj
+
+    def fail_capture(src, dest, *args, **kwargs):
+        src_name, dest_name = getattr(src, "name", None), getattr(dest, "name", None)
+        if isinstance(src_name, (str, os.PathLike)) and isinstance(dest_name, (str, os.PathLike)):
+            capture = Path(dest_name)
+            if Path(src_name) == source and capture.parent.name.startswith("deerflow-upload-source-"):
+                captures.append(capture)
+                raise OSError("snapshot disk full")
+        return real_copyfileobj(src, dest, *args, **kwargs)
+
+    monkeypatch.setattr(manager.shutil, "copyfileobj", fail_capture)
+    unrelated_copy = BytesIO()
+    shutil.copyfileobj(BytesIO(b"UNRELATED CONTENT"), unrelated_copy)
+    assert unrelated_copy.getvalue() == b"UNRELATED CONTENT"
+    with pytest.raises(OSError, match="snapshot disk full"):
+        client.upload_files("snapshot-thread", [source])
+    assert existing.read_bytes() == b"EXISTING CONTENT"
+    assert source.read_bytes() == b"NEW CONTENT"
+    assert not captures[0].parent.exists()
+
+
+@pytest.mark.skipif(sys.platform != "darwin", reason="chmod +a ACL grants require macOS")
+@pytest.mark.parametrize("in_loop", [False, True])
+def test_acl_only_readable_source_uploads_and_converts(upload_case, monkeypatch, tmp_path, in_loop):
+    """Preserve the ACL-only source case identified in the #6621 review."""
+    import deerflow.utils.file_conversion as conversion
+
+    if os.geteuid() == 0:
+        pytest.skip("root bypasses source permission checks")
+    client, uploads = upload_case
+    source = tmp_path / "report.pdf"
+    source.write_bytes(b"ACL-AUTHORIZED CONTENT")
+    source.chmod(0o000)
+    saved = uploads / source.name
+    conversion_paths = []
+
+    async def convert(path, output_path=None):
+        conversion_paths.append(path)
+        output_path.write_bytes(b"CONVERTED:" + path.read_bytes())
+        return output_path
+
+    monkeypatch.setattr(conversion, "CONVERTIBLE_EXTENSIONS", {".pdf"})
+    monkeypatch.setattr(conversion, "convert_file_to_markdown", convert)
+    try:
+        subprocess.run(["chmod", "+a", f"user:{getpass.getuser()} allow read", str(source)], check=True)
+        assert stat.S_IMODE(source.stat().st_mode) == 0
+        assert source.read_bytes() == b"ACL-AUTHORIZED CONTENT"
+        response = call_upload(client, source, in_loop)
+        assert response["success"]
+        companion = resolve_companion(saved)
+        assert companion is not None
+        assert companion.read_bytes() == b"CONVERTED:ACL-AUTHORIZED CONTENT"
+        if os.chmod in os.supports_fd:
+            assert stat.S_IMODE(saved.stat().st_mode) == 0
+        # The uploaded copy preserves mode bits, but not the source ACL.
+        saved.chmod(0o600)
+        assert saved.read_bytes() == b"ACL-AUTHORIZED CONTENT"
+        assert conversion_paths[0] != source
+        assert not conversion_paths[0].is_relative_to(uploads)
+        assert not conversion_paths[0].parent.exists()
+    finally:
+        source.chmod(0o600)
+        if saved.exists():
+            saved.chmod(0o600)
+
+
+@pytest.mark.parametrize("in_loop", [False, True])
+@pytest.mark.parametrize("change", ["edit", "remove"])
+def test_collision_retries_keep_one_capture(upload_case, monkeypatch, tmp_path, in_loop, change):
+    """Retry new names without reopening the source or losing its metadata."""
+    import builtins
+
+    import deerflow.uploads.manager as manager
+    import deerflow.utils.file_conversion as conversion
+
+    client, uploads = upload_case
+    source = tmp_path / "report.pdf"
+    source.write_bytes(b"VERSION A")
+    source.chmod(0o640)
+    os.utime(source, ns=(1_700_000_000_000_000_000, 1_700_000_001_000_000_000))
+    before = source.stat()
+    real_open = builtins.open
+    real_link = manager.os.link
+    source_opens = []
+    attempts = []
+    captures = []
+
+    def record_open(path, mode="r", *args, **kwargs):
+        if mode == "rb" and isinstance(path, (str, os.PathLike)) and Path(path) == source:
+            source_opens.append(path)
+        return real_open(path, mode, *args, **kwargs)
+
+    def claim_then_link(staging, destination, **kwargs):
+        if destination.suffix == ".pdf":
+            attempts.append(destination.name)
+            if len(attempts) == 1:
+                # Another caller wins two names while this call is publishing.
+                (uploads / "report.pdf").write_bytes(b"FIRST OWNER")
+                (uploads / "report_1.pdf").write_bytes(b"SECOND OWNER")
+                if change == "edit":
+                    source.write_bytes(b"VERSION B")
+                    source.chmod(0o600)
+                    os.utime(source, ns=(1_800_000_000_000_000_000, 1_800_000_001_000_000_000))
+                else:
+                    source.unlink()
+        return real_link(staging, destination, **kwargs)
+
+    async def convert(path, output_path=None):
+        captures.append(path)
+        output_path.write_bytes(b"CONVERTED:" + path.read_bytes())
+        return output_path
+
+    monkeypatch.setattr(builtins, "open", record_open)
+    monkeypatch.setattr(manager.os, "link", claim_then_link)
+    monkeypatch.setattr(conversion, "CONVERTIBLE_EXTENSIONS", {".pdf"})
+    monkeypatch.setattr(conversion, "convert_file_to_markdown", convert)
+    response = call_upload(client, source, in_loop)
+
+    assert response["success"]
+    info = response["files"][0]
+    assert info["filename"] == "report_2.pdf"
+    assert info["original_filename"] == "report.pdf"
+    assert attempts == ["report.pdf", "report_1.pdf", "report_2.pdf"]
+    assert len(source_opens) == 1
+    saved = uploads / info["filename"]
+    assert saved.read_bytes() == b"VERSION A"
+    companion = resolve_companion(saved)
+    assert companion == uploads / "report_2.md"
+    assert companion.read_bytes() == b"CONVERTED:VERSION A"
+    assert (uploads / "report.pdf").read_bytes() == b"FIRST OWNER"
+    assert (uploads / "report_1.pdf").read_bytes() == b"SECOND OWNER"
+    if os.chmod in os.supports_fd:
+        assert stat.S_IMODE(saved.stat().st_mode) == stat.S_IMODE(before.st_mode)
+    if os.utime in os.supports_fd:
+        assert saved.stat().st_mtime_ns == before.st_mtime_ns
+    assert captures[0] != source
+    assert not captures[0].is_relative_to(uploads)
+    assert not captures[0].parent.exists()
+    assert not list(uploads.glob(".upload-*.part"))
diff --git a/backend/tests/test_upload_copy_publication.py b/backend/tests/test_upload_copy_publication.py
new file mode 100644
index 0000000000..01931fed0d
--- /dev/null
+++ b/backend/tests/test_upload_copy_publication.py
@@ -0,0 +1,201 @@
+"""Upload publication rejects substituted staging paths and preserves outcomes."""
+
+import os
+import shutil
+from pathlib import Path
+
+import pytest
+from support.symlinks import symlink_or_skip
+
+from deerflow.uploads.manager import UnsafeUploadPathError, copy_upload_file_no_replace
+
+
+@pytest.fixture
+def upload_paths(tmp_path):
+    uploads = tmp_path / "uploads"
+    uploads.mkdir()
+    source = tmp_path / "notes.txt"
+    source.write_bytes(b"uploaded content")
+    outside = tmp_path / "outside.txt"
+    outside.write_bytes(b"must not be published")
+    return uploads, source, outside
+
+
+def replace_staging(staging, outside, replacement):
+    try:
+        staging.unlink()
+    except PermissionError:
+        if os.name == "nt":
+            pytest.skip("host prevents replacing an open staging file")
+        raise
+    if replacement == "symlink":
+        symlink_or_skip(staging, outside)
+    else:
+        staging.write_bytes(b"substituted content")
+
+
+@pytest.mark.parametrize("replacement", ["symlink", "regular"])
+@pytest.mark.parametrize("nofollow", [True, False])
+def test_rejects_staging_replaced_during_copy(upload_paths, monkeypatch, replacement, nofollow):
+    uploads, source, outside = upload_paths
+    real_copy = shutil.copyfileobj
+    monkeypatch.setattr(os, "supports_follow_symlinks", {os.link} if nofollow else set())
+
+    def copy_and_swap(src, dst):
+        real_copy(src, dst)
+        staging = next(uploads.glob(".upload-*.part"))
+        replace_staging(staging, outside, replacement)
+
+    monkeypatch.setattr(shutil, "copyfileobj", copy_and_swap)
+    with pytest.raises(UnsafeUploadPathError):
+        copy_upload_file_no_replace(uploads, "notes.txt", source)
+
+    assert not os.path.lexists(uploads / "notes.txt")
+    assert outside.read_bytes() == b"must not be published"
+    assert outside.stat().st_nlink == 1
+    assert list(uploads.iterdir()) == []
+
+
+@pytest.mark.parametrize("replacement", ["symlink", "regular"])
+@pytest.mark.parametrize("nofollow", [True, False])
+def test_rejects_staging_replaced_at_link(upload_paths, monkeypatch, replacement, nofollow):
+    uploads, source, outside = upload_paths
+    real_link = os.link
+    if nofollow and real_link not in os.supports_follow_symlinks:
+        pytest.skip("host does not support no-follow hard links")
+    calls = []
+
+    def swap_and_link(staging, dest, **kwargs):
+        replace_staging(staging, outside, replacement)
+        calls.append(kwargs)
+        real_link(staging, dest, **kwargs)
+
+    monkeypatch.setattr(os, "link", swap_and_link)
+    monkeypatch.setattr(os, "supports_follow_symlinks", {swap_and_link} if nofollow else set())
+    with pytest.raises(UnsafeUploadPathError):
+        copy_upload_file_no_replace(uploads, "notes.txt", source)
+
+    assert calls == [{"follow_symlinks": False} if nofollow else {}]
+    assert not os.path.lexists(uploads / "notes.txt")
+    assert outside.read_bytes() == b"must not be published"
+    assert outside.stat().st_nlink == 1
+    assert list(uploads.iterdir()) == []
+
+
+@pytest.mark.parametrize("nofollow", [True, False])
+def test_publishes_with_supported_link_options(upload_paths, monkeypatch, nofollow):
+    uploads, source, _outside = upload_paths
+    real_link = os.link
+    if nofollow and real_link not in os.supports_follow_symlinks:
+        pytest.skip("host does not support no-follow hard links")
+    calls = []
+
+    def record_link(staging, dest, **kwargs):
+        calls.append(kwargs)
+        real_link(staging, dest, **kwargs)
+
+    monkeypatch.setattr(os, "link", record_link)
+    monkeypatch.setattr(os, "supports_follow_symlinks", {record_link} if nofollow else set())
+    dest = copy_upload_file_no_replace(uploads, "notes.txt", source)
+
+    assert dest.read_bytes() == b"uploaded content"
+    assert dest.stat().st_nlink == 1
+    assert calls == [{"follow_symlinks": False} if nofollow else {}]
+
+
+def test_cleanup_directory_swap_does_not_mask_success(upload_paths, monkeypatch, caplog):
+    uploads, source, _outside = upload_paths
+    real_link = os.link
+    planted = []
+
+    def publish_and_swap(staging, dest, **kwargs):
+        real_link(staging, dest, **kwargs)
+        try:
+            staging.unlink()
+        except PermissionError:
+            if os.name == "nt":
+                pytest.skip("host prevents replacing an open staging file")
+            raise
+        staging.mkdir()
+        planted.append(staging)
+
+    monkeypatch.setattr(os, "link", publish_and_swap)
+    dest = copy_upload_file_no_replace(uploads, "notes.txt", source)
+
+    assert dest.read_bytes() == b"uploaded content"
+    assert planted[0].is_dir()
+    assert "Failed to remove upload staging file" in caplog.text
+
+
+def test_cleanup_error_does_not_mask_copy_failure(upload_paths, monkeypatch, caplog):
+    uploads, source, _outside = upload_paths
+    real_unlink = Path.unlink
+
+    def fail_copy(src, dst):
+        dst.write(b"partial")
+        raise OSError("original copy failure")
+
+    def fail_cleanup(path, *args, **kwargs):
+        if path.name.startswith(".upload-"):
+            raise PermissionError("cleanup failure")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(shutil, "copyfileobj", fail_copy)
+    monkeypatch.setattr(Path, "unlink", fail_cleanup)
+    with pytest.raises(OSError, match="original copy failure"):
+        copy_upload_file_no_replace(uploads, "notes.txt", source)
+
+    assert not (uploads / "notes.txt").exists()
+    assert "Failed to remove upload staging file" in caplog.text
+
+
+def test_cleanup_error_does_not_mask_collision(upload_paths, monkeypatch, caplog):
+    uploads, source, _outside = upload_paths
+    dest = uploads / "notes.txt"
+    dest.write_bytes(b"keep existing upload")
+    real_unlink = Path.unlink
+
+    def fail_cleanup(path, *args, **kwargs):
+        if path.name.startswith(".upload-"):
+            raise PermissionError("cleanup failure")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", fail_cleanup)
+    with pytest.raises(FileExistsError):
+        copy_upload_file_no_replace(uploads, "notes.txt", source)
+
+    assert dest.read_bytes() == b"keep existing upload"
+    assert "Failed to remove upload staging file" in caplog.text
+
+
+@pytest.mark.parametrize("collisions", [0, 2])
+def test_publication_retry_uses_original_seed(upload_paths, collisions):
+    from deerflow.uploads.manager import publish_upload_copy
+
+    uploads, source, _outside = upload_paths
+    seen = {"notes.txt", "notes_1.txt"}
+    for index in range(1, collisions + 1):
+        (uploads / f"notes_{index}.txt").write_bytes(f"existing-{index}".encode())
+
+    dest = publish_upload_copy(uploads, "notes_1.txt", source, claim_seed="notes.txt", seen_names=seen)
+
+    assert dest.name == f"notes_{collisions + 1}.txt"
+    assert dest.read_bytes() == b"uploaded content"
+    assert dest.name in seen
+    for index in range(1, collisions + 1):
+        assert (uploads / f"notes_{index}.txt").read_bytes() == f"existing-{index}".encode()
+    assert not list(uploads.glob(".upload-*.part"))
+
+
+def test_publication_retry_preserves_noncollision_errors(upload_paths, monkeypatch):
+    from deerflow.uploads import manager
+
+    uploads, source, _outside = upload_paths
+
+    def fail_link(*args, **kwargs):
+        raise PermissionError("publication denied")
+
+    monkeypatch.setattr(manager.os, "link", fail_link)
+    with pytest.raises(PermissionError, match="publication denied"):
+        manager.publish_upload_copy(uploads, "notes.txt", source, claim_seed="notes.txt", seen_names={"notes.txt"})
+    assert list(uploads.iterdir()) == []
diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py
index 444c92403f..aac05ff141 100644
--- a/backend/tests/test_uploads_manager.py
+++ b/backend/tests/test_uploads_manager.py
@@ -17,6 +17,7 @@ from deerflow.uploads.manager import (
     apply_upload_sandbox_permits,
     claim_unique_filename,
     cleanup_stale_upload_staging_files,
+    copy_upload_file_no_replace,
     copy_upload_file_no_symlink,
     delete_file_safe,
     list_files_in_dir,
@@ -353,6 +354,64 @@ class TestCopyUploadFileNoSymlink:
 
         assert dest.read_bytes() == b"new contents"
 
+    def test_no_replace_copy_preserves_existing_file_and_cleans_stage(self, tmp_path):
+        uploads = tmp_path / "uploads"
+        uploads.mkdir()
+        dest = uploads / "notes.txt"
+        dest.write_bytes(b"old contents")
+        src = tmp_path / "notes.txt"
+        src.write_bytes(b"new contents")
+
+        with pytest.raises(FileExistsError):
+            copy_upload_file_no_replace(uploads, "notes.txt", src)
+
+        assert dest.read_bytes() == b"old contents"
+        assert sorted(path.name for path in uploads.iterdir()) == ["notes.txt"]
+
+    def test_no_replace_copy_failure_removes_stage_without_publishing(self, tmp_path):
+        uploads = tmp_path / "uploads"
+        uploads.mkdir()
+        src = tmp_path / "notes.txt"
+        src.write_bytes(b"new contents")
+
+        def fail_copy(source, staged):
+            staged.write(b"partial")
+            raise OSError("copy failed")
+
+        with patch("deerflow.uploads.manager.shutil.copyfileobj", side_effect=fail_copy):
+            with pytest.raises(OSError, match="copy failed"):
+                copy_upload_file_no_replace(uploads, "notes.txt", src)
+
+        assert list(uploads.iterdir()) == []
+
+    def test_no_replace_copy_publishes_complete_file_and_preserves_metadata(self, tmp_path):
+        uploads = tmp_path / "threads" / "t1" / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        src = tmp_path / "notes.txt"
+        src.write_bytes(b"complete contents")
+        src.chmod(0o640)
+        old_time = time.time_ns() - 3 * 24 * 60 * 60 * 1_000_000_000
+        os.utime(src, ns=(old_time, old_time))
+        source_stat = src.stat()
+        real_link = os.link
+
+        def publish(staged, dest, **kwargs):
+            assert not dest.exists()
+            assert list_files_in_dir(uploads)["files"] == []
+            assert staged.read_bytes() == b"complete contents"
+            assert cleanup_stale_upload_staging_files(tmp_path) == 0
+            real_link(staged, dest, **kwargs)
+
+        with patch("deerflow.uploads.manager.os.link", side_effect=publish):
+            dest = copy_upload_file_no_replace(uploads, "notes.txt", src)
+
+        assert dest.read_bytes() == b"complete contents"
+        assert sorted(path.name for path in uploads.iterdir()) == ["notes.txt"]
+        if os.chmod in os.supports_fd:
+            assert stat.S_IMODE(dest.stat().st_mode) == stat.S_IMODE(source_stat.st_mode)
+        if os.utime in os.supports_fd:
+            assert dest.stat().st_mtime_ns == source_stat.st_mtime_ns
+
     def test_rejects_symlink_destination(self, tmp_path):
         uploads = tmp_path / "uploads"
         uploads.mkdir()
diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md
index 0b90a0fd9d..18bdbfc023 100644
--- a/frontend/AGENTS.md
+++ b/frontend/AGENTS.md
@@ -1,10 +1,10 @@
 # AGENTS.md
 
-This file provides guidance to AI coding agents (Claude Code, Codex, and others) when working with the DeerFlow frontend. It is the source of truth; the sibling `CLAUDE.md` imports it via `@AGENTS.md`.
+This is the source of truth for DeerFlow frontend coding agents; the sibling `CLAUDE.md` imports it via `@AGENTS.md`.
 
 ## Project Overview
 
-DeerFlow Frontend is a Next.js 16 web interface for an AI agent system. It communicates with a LangGraph-based backend to provide thread-based AI conversations with streaming responses, artifacts, and a skills/tools system.
+DeerFlow's Next.js 16 frontend provides thread-based AI conversations with streaming responses, artifacts, skills, and tools through a LangGraph backend.
 
 **Stack**: Next.js 16, React 19, TypeScript 5.8, Tailwind CSS 4, pnpm 10.26.2. Requires Node.js 22+ and pnpm 10.26.2+.