fix(uploads): keep in-flight staging files out of the startup sweep (#6445)

* fix(uploads): keep in-flight staging files out of the startup sweep

`cleanup_stale_upload_staging_files` removed every `.upload-*.part` it
found at Gateway startup. That assumed a single Gateway: nothing can be
uploading while it boots, so every staging file is a crash leftover.
With several replicas sharing one home volume (Helm RWX), a replica
starting during a rolling update deleted the staging file of an upload
another replica was still writing; that upload then failed at its
`os.link` commit and lost its bytes.

The sweep now skips staging files younger than `UPLOAD_STAGING_MIN_AGE`
(24 hours, the same guard as the project-document `.staging` orphan
sweep in `projects/trash.py`). Chunk writes refresh the mtime, so an
upload in flight stays younger than the guard; an unreadable mtime keeps
the file. The startup call site is unchanged and uses the default.

Tests: the existing sweep tests backdate their orphans; new cases pin
young-kept / old-removed, the configurable `min_age`, unreadable and
vanished mtimes, and a router test that runs the sweep while an upload
is blocked mid-write and asserts the upload still commits.

* docs(changelog): link #6445

* fix(uploads): reclaim published staging aliases at any age

Review follow-up (#6445): a Gateway that dies after the commit's
`os.link` but before the staged-name removal leaves the final file and
its `.upload-*.part` alias on one inode. With only the age guard, the
next startup kept that alias for 24 hours, and because the sweep runs
only at startup, `copy_upload_file_no_symlink` (embedded
`DeerFlowClient.upload_files`) kept rejecting the destination for its
multiple links.

The sweep now stats each staging name with `os.lstat` (a `DirEntry`
stat reports `st_nlink == 0` on Windows) and removes a name with
`st_nlink > 1` at any age: a staged part only ever gains its second
link through its own commit, so such a name is a published alias, never
an upload in flight. Lone parts keep the 24 hour mtime guard. The
destination hardlink check is unchanged.

Tests cover the alias next to a young lone part, the replacement upload
succeeding afterwards, and the crash reproduced through the real upload
path with the staged-name removal patched out.

---------

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

CHANGELOG.md

@@ -804,6 +804,19 @@ This release closes that milestone with **439 merged pull requests**.
 
 ### Fixed
 
+- **uploads:** The Gateway's startup sweep of orphaned `.upload-*.part` staging
+  files now skips files younger than 24 hours. The sweep removed every staging
+  file it found, which was right for one Gateway but not for several replicas
+  sharing a home volume: a replica starting during a rolling update deleted the
+  staging file of an upload another replica was still writing, and that upload
+  then failed at its atomic commit. Chunk writes refresh the staging file's
+  mtime, so an upload in flight stays younger than the guard; a crash leftover
+  is collected by the first startup more than 24 hours later, the same guard
+  project-document staging already uses. A staging file that already shares
+  its inode with the published upload (a crash between the atomic link and the
+  staged-name removal) is still reclaimed on the next startup at any age, so
+  that destination does not fail the multi-link safety check on its next
+  replacement. ([#6445])
 - **gateway:** `GET /health/ready` now reports unready while the Redis stream
   bridge is unreachable. The bridge's Redis client connects lazily and nothing
   pinged it, so a gateway whose Redis was down started, answered `200 ready` to
@@ -9144,4 +9157,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6407]: https://github.com/bytedance/deer-flow/pull/6407
 [#6412]: https://github.com/bytedance/deer-flow/pull/6412
 [#6426]: https://github.com/bytedance/deer-flow/pull/6426
+[#6445]: https://github.com/bytedance/deer-flow/pull/6445
 [#6447]: https://github.com/bytedance/deer-flow/pull/6447

backend/AGENTS.md

@@ -346,7 +346,7 @@ Outlines use ATX syntax (1–6 hashes, space/tab separator, ≤3 leading spaces)
 - Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()`
 - Per-thread `upload-companions/`: source mtime/ctime.
 - Duplicate filenames within one request get `_N` suffixes to prevent overwrites.
-- Gateway HTTP uploads stage `.upload-*.part` files, hidden from upload listings, agent context, and sandbox listings/searches. After size validation, publication is atomic; staged-name cleanup logs errors and leaves leftovers for startup sweep.
+- Gateway HTTP uploads stage `.upload-*.part` files, hidden from upload listings, agent context, and sandbox listings/searches. After size validation, publication is atomic; staged-name cleanup logs errors and leaves leftovers for startup sweep. The sweep keeps lone `.part` files under 24h: they may be in flight on another replica; `st_nlink > 1` means a published alias, removed at any age so the multi-link check can still replace the file.
 - Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor. Non-mounted sandbox uploads acquire sandboxes with `SandboxProvider.acquire_async()` and offload `read_bytes()` plus `sandbox.update_file()` together.
 - Mounted uploads skip sandbox acquire/sync. AIO remote/provisioner requires accurate `sandbox.thread_data_mounts: true`; omission keeps backend auto-detection.
 - `UploadsMiddleware` caps outline titles at 200 characters and previews at 2000 including markers. Titles use `original_user_content`, not upload-prefixed content; attachment-only titles use a sanitized, bounded filename or count.

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

@@ -9,6 +9,8 @@ import logging
 import os
 import shutil
 import stat
+import time
+from datetime import timedelta
 from pathlib import Path
 from urllib.parse import quote
 
@@ -30,6 +32,9 @@ logger = logging.getLogger(__name__)
 
 UPLOAD_STAGING_PREFIX = ".upload-"
 UPLOAD_STAGING_SUFFIX = ".part"
+# Staging files younger than this are treated as in flight by the startup sweep.
+# Mirrors the project-document ``.staging`` orphan guard (``projects/trash.py``).
+UPLOAD_STAGING_MIN_AGE = timedelta(hours=24)
 
 _MAX_FILENAME_BYTES = 255
 
@@ -181,9 +186,35 @@ def _iter_upload_dirs(base_dir: Path):
     yield from base_dir.glob("users/*/threads/*/user-data/uploads")
 
 
-def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> int:
-    """Remove orphaned Gateway upload staging files left by a hard crash."""
+def _staging_entry_stat(entry: os.DirEntry[str]) -> os.stat_result:
+    # ``os.lstat`` rather than ``entry.stat``: on Windows a ``DirEntry`` stat leaves
+    # ``st_nlink`` at zero, and the published-alias rule in the sweep depends on it.
+    return os.lstat(entry.path)
+
+
+def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, min_age: timedelta = UPLOAD_STAGING_MIN_AGE) -> int:
+    """Remove orphaned Gateway upload staging files left by a hard crash.
+
+    A lone ``.upload-*.part`` (``st_nlink == 1``) is removed only once its mtime
+    is older than *min_age*. The uploads directories may live on a volume
+    shared by several Gateway replicas, so at startup such a file can belong to
+    an upload another replica is still writing; each chunk write refreshes its
+    mtime, which keeps it younger than the guard until it is committed or
+    abandoned.
+
+    A staging name with ``st_nlink > 1`` is removed at any age: the commit's
+    ``os.link`` already published those bytes under their final name, so the
+    staged name is only an alias left behind by a crash (or a deferred removal
+    that never ran). Keeping it would make that destination fail the
+    multi-link safety check on the next replacement upload (embedded
+    ``DeerFlowClient.upload_files`` goes through ``copy_upload_file_no_symlink``).
+    Removing the alias cannot affect an upload in flight: a staged part gains
+    its second link only through its own commit.
+
+    A file that cannot be stat'ed is kept, never removed on a guess.
+    """
     root = Path(base_dir) if base_dir is not None else get_paths().base_dir
+    cutoff = time.time() - min_age.total_seconds()
     removed = 0
     for uploads_dir in _iter_upload_dirs(root):
         if not uploads_dir.is_dir():
@@ -193,6 +224,15 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> in
                 for entry in entries:
                     if not is_upload_staging_file(entry.name) or not entry.is_file(follow_symlinks=False):
                         continue
+                    try:
+                        st = _staging_entry_stat(entry)
+                    except FileNotFoundError:
+                        continue
+                    except OSError:
+                        logger.warning("Could not stat upload staging file %s; keeping it", entry.path, exc_info=True)
+                        continue
+                    if st.st_nlink <= 1 and st.st_mtime >= cutoff:
+                        continue  # a lone, young part may still be in flight on another replica
                     try:
                         os.unlink(entry.path)
                         removed += 1

backend/tests/test_uploads_manager.py

@@ -4,6 +4,8 @@ import errno
 import os
 import shutil
 import stat
+import time
+from datetime import timedelta
 from unittest.mock import patch
 
 import pytest
@@ -19,6 +21,7 @@ from deerflow.uploads.manager import (
     list_files_in_dir,
     normalize_filename,
     validate_path_traversal,
+    validate_upload_destination,
     write_upload_file_no_symlink,
 )
 
@@ -405,6 +408,12 @@ class TestListFilesInDir:
 # ---------------------------------------------------------------------------
 
 
+def _set_age(path, age: timedelta) -> None:
+    """Backdate *path*'s mtime so the cleanup sees it as *age* old."""
+    timestamp = time.time() - age.total_seconds()
+    os.utime(path, (timestamp, timestamp))
+
+
 class TestCleanupStaleUploadStagingFiles:
     def test_removes_only_stale_staging_files_from_all_upload_layouts(self, tmp_path):
         legacy_uploads = tmp_path / "threads" / "thread-legacy" / "user-data" / "uploads"
@@ -419,6 +428,15 @@ class TestCleanupStaleUploadStagingFiles:
         (legacy_uploads / ".env").write_text("intentional dotfile")
         (legacy_uploads / ".upload-note.txt").write_text("intentional upload")
         (legacy_uploads / "draft.part").write_text("intentional upload")
+        for path in (
+            legacy_uploads / ".upload-old.part",
+            user_uploads / ".upload-new.part",
+            unrelated_uploads / ".upload-ignore.part",
+            legacy_uploads / ".env",
+            legacy_uploads / ".upload-note.txt",
+            legacy_uploads / "draft.part",
+        ):
+            _set_age(path, timedelta(days=2))
 
         removed = cleanup_stale_upload_staging_files(tmp_path)
 
@@ -430,6 +448,106 @@ class TestCleanupStaleUploadStagingFiles:
         assert (legacy_uploads / ".upload-note.txt").exists()
         assert (legacy_uploads / "draft.part").exists()
 
+    def test_keeps_staging_files_younger_than_the_default_guard(self, tmp_path):
+        """On a volume shared by several Gateway replicas, a young ``.part`` may be
+        an upload another replica is still writing; only old ones are orphans."""
+        uploads_dir = tmp_path / "users" / "owner-1" / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        fresh = uploads_dir / ".upload-fresh.part"
+        recent = uploads_dir / ".upload-recent.part"
+        old = uploads_dir / ".upload-old.part"
+        for path in (fresh, recent, old):
+            path.write_text("partial")
+        _set_age(recent, timedelta(hours=23))
+        _set_age(old, timedelta(hours=25))
+
+        removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 1
+        assert fresh.exists()
+        assert recent.exists()
+        assert not old.exists()
+
+    def test_min_age_is_configurable(self, tmp_path):
+        uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        young = uploads_dir / ".upload-young.part"
+        old = uploads_dir / ".upload-old.part"
+        for path in (young, old):
+            path.write_text("partial")
+        _set_age(young, timedelta(minutes=1))
+        _set_age(old, timedelta(minutes=10))
+
+        removed = cleanup_stale_upload_staging_files(tmp_path, min_age=timedelta(minutes=5))
+
+        assert removed == 1
+        assert young.exists()
+        assert not old.exists()
+
+    def test_keeps_staging_file_when_its_age_cannot_be_read(self, tmp_path, caplog):
+        uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        old = uploads_dir / ".upload-old.part"
+        old.write_text("partial")
+        _set_age(old, timedelta(days=2))
+
+        with (
+            patch("deerflow.uploads.manager._staging_entry_stat", side_effect=PermissionError(errno.EACCES, "denied")),
+            caplog.at_level("WARNING", logger="deerflow.uploads.manager"),
+        ):
+            removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 0
+        assert old.exists()
+        assert any("keeping it" in record.getMessage() for record in caplog.records)
+
+    def test_removes_published_alias_at_any_age_but_keeps_lone_young_part(self, tmp_path):
+        """A crash between the commit's ``os.link`` and the staged-name removal
+        leaves the final file and its ``.part`` alias sharing one inode. The
+        alias is reclaimed on the next startup regardless of age, or the
+        destination would fail the multi-link check on its next replacement;
+        a lone young part next to it is still treated as in flight."""
+        uploads_dir = tmp_path / "users" / "owner-1" / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        published = uploads_dir / "notes.txt"
+        published.write_bytes(b"published bytes")
+        alias = uploads_dir / ".upload-abc123.part"
+        try:
+            os.link(published, alias)
+        except OSError as exc:  # pragma: no cover - filesystems without hard links
+            pytest.skip(f"hard links unsupported here: {exc}")
+        in_flight = uploads_dir / ".upload-inflight.part"
+        in_flight.write_bytes(b"partial")
+        with pytest.raises(UnsafeUploadPathError, match="multiple links"):
+            validate_upload_destination(uploads_dir, "notes.txt")
+
+        removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 1
+        assert not alias.exists()
+        assert in_flight.exists()
+        assert published.read_bytes() == b"published bytes"
+        assert os.lstat(published).st_nlink == 1
+        assert validate_upload_destination(uploads_dir, "notes.txt") == published
+        replacement = tmp_path / "replacement.txt"
+        replacement.write_bytes(b"replacement bytes")
+        assert copy_upload_file_no_symlink(uploads_dir, "notes.txt", replacement) == published
+        assert published.read_bytes() == b"replacement bytes"
+
+    def test_skips_staging_file_that_vanished_before_its_age_was_read(self, tmp_path, caplog):
+        uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        (uploads_dir / ".upload-gone.part").write_text("partial")
+
+        with (
+            patch("deerflow.uploads.manager._staging_entry_stat", side_effect=FileNotFoundError),
+            caplog.at_level("WARNING", logger="deerflow.uploads.manager"),
+        ):
+            removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 0
+        assert caplog.records == []
+
 
 # ---------------------------------------------------------------------------
 # delete_file_safe

backend/tests/test_uploads_router.py

@@ -3,6 +3,8 @@ import errno
 import os
 import stat
 import threading
+import time
+from datetime import timedelta
 from io import BytesIO
 from pathlib import Path, PurePosixPath
 from types import SimpleNamespace
@@ -18,6 +20,7 @@ from app.gateway.deps import get_config
 from app.gateway.routers import uploads
 from deerflow.sandbox.lease import get_sandbox_lease_manager
 from deerflow.uploads.companions import resolve_companion
+from deerflow.uploads.manager import cleanup_stale_upload_staging_files
 
 
 class ChunkedUpload:
@@ -747,13 +750,14 @@ def test_upload_files_in_flight_upload_is_invisible_to_listings(tmp_path):
 
 def test_orphaned_staging_part_is_hidden_from_listings_and_swept(tmp_path):
     """A crashed upload leaves only a hidden ``.upload-*.part`` file: the
-    listing never exposes it and the startup cleanup removes it."""
-    from deerflow.uploads.manager import cleanup_stale_upload_staging_files
-
+    listing never exposes it and, once it is older than the in-flight guard,
+    the startup cleanup removes it."""
     thread_uploads_dir = tmp_path / "threads" / "t1" / "user-data" / "uploads"
     thread_uploads_dir.mkdir(parents=True)
     orphan = thread_uploads_dir / ".upload-crashed.part"
     orphan.write_bytes(b"partial")
+    stale = time.time() - timedelta(days=2).total_seconds()
+    os.utime(orphan, (stale, stale))
     visible = thread_uploads_dir / "kept.txt"
     visible.write_bytes(b"kept")
 
@@ -764,6 +768,93 @@ def test_orphaned_staging_part_is_hidden_from_listings_and_swept(tmp_path):
     assert visible.read_bytes() == b"kept"
 
 
+def test_startup_cleanup_reclaims_a_published_staging_alias_immediately(tmp_path):
+    """A Gateway that dies after the commit's ``os.link`` but before the staged
+    name is removed leaves ``notes.txt`` and its ``.upload-*.part`` alias on one
+    inode. The next startup must reclaim the alias at once: left in place, the
+    destination fails the multi-link check on its next replacement upload."""
+    from deerflow.uploads.manager import validate_upload_destination
+
+    thread_uploads_dir = tmp_path / "threads" / "thread-local" / "user-data" / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    provider = _mounted_provider()
+
+    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),
+        # The crash: the staged name is never removed after the link publishes the bytes.
+        patch.object(uploads, "_remove_staged_file"),
+    ):
+        file = ChunkedUpload("notes.txt", [b"hello"])
+        result = asyncio.run(call_unwrapped(uploads.upload_files, "thread-local", request=MagicMock(), files=[file], config=SimpleNamespace()))
+
+    assert result.success is True
+    names = sorted(p.name for p in thread_uploads_dir.iterdir())
+    assert names[-1] == "notes.txt" and len(names) == 2 and names[0].startswith(".upload-") and names[0].endswith(".part")
+    assert os.lstat(thread_uploads_dir / "notes.txt").st_nlink == 2
+    with pytest.raises(ValueError, match="multiple links"):
+        validate_upload_destination(thread_uploads_dir, "notes.txt")
+
+    assert cleanup_stale_upload_staging_files(tmp_path) == 1
+
+    assert [p.name for p in thread_uploads_dir.iterdir()] == ["notes.txt"]
+    assert (thread_uploads_dir / "notes.txt").read_bytes() == b"hello"
+    assert validate_upload_destination(thread_uploads_dir, "notes.txt") == thread_uploads_dir / "notes.txt"
+
+
+def test_startup_cleanup_keeps_an_in_flight_upload_on_a_shared_volume(tmp_path):
+    """Several Gateway replicas can share one home volume. A replica starting
+    while another one is still writing a ``.upload-*.part`` must not sweep it:
+    that upload would lose its staged bytes at the atomic link and fail."""
+    thread_uploads_dir = tmp_path / "threads" / "thread-local" / "user-data" / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    provider = _mounted_provider()
+    real_write = uploads._write_upload_chunk
+    write_started = threading.Event()
+    release_write = threading.Event()
+
+    def gated_write(upload_temp, chunk) -> None:
+        write_started.set()
+        assert release_write.wait(timeout=10)
+        real_write(upload_temp, chunk)
+
+    results = []
+
+    def run_upload() -> None:
+        file = ChunkedUpload("notes.txt", [b"hello"])
+        try:
+            results.append(asyncio.run(call_unwrapped(uploads.upload_files, "thread-local", request=MagicMock(), files=[file], config=SimpleNamespace())))
+        except BaseException as exc:  # noqa: BLE001 - surfaced through the assertions below
+            results.append(exc)
+
+    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, "_write_upload_chunk", side_effect=gated_write),
+    ):
+        upload_thread = threading.Thread(target=run_upload)
+        upload_thread.start()
+        assert write_started.wait(timeout=10)
+        staged = [p.name for p in thread_uploads_dir.iterdir()]
+        assert len(staged) == 1 and staged[0].startswith(".upload-") and staged[0].endswith(".part")
+
+        # Another replica starts up mid-upload and runs its staging sweep.
+        assert cleanup_stale_upload_staging_files(tmp_path) == 0
+        assert [p.name for p in thread_uploads_dir.iterdir()] == staged
+
+        release_write.set()
+        upload_thread.join(timeout=15)
+
+    assert not upload_thread.is_alive()
+    assert not isinstance(results[0], BaseException), results[0]
+    assert results[0].success is True
+    assert (thread_uploads_dir / "notes.txt").read_bytes() == b"hello"
+    assert [p.name for p in thread_uploads_dir.iterdir()] == ["notes.txt"]
+
+
 def test_upload_files_rejects_oversized_single_file_and_removes_partial_file(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)