fix(gateway): fix oversized upload replacements deleting existing files (#3822)

* fix upload rollback on size limit

* fix upload staging cleanup and listings

backend/AGENTS.md

@@ -705,6 +705,7 @@ Multi-file upload with automatic document conversion:
 - Reuses one conversion worker per request when called from an active event loop
 - 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()`
 - Duplicate filenames in a single upload request are auto-renamed with `_N` suffixes so later files do not truncate earlier files
+- Gateway HTTP uploads stage bytes as `.upload-*.part` files and atomically replace the destination only after size validation. These staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools, and swept on Gateway startup if a hard crash leaves one behind.
 - Agent receives uploaded file list via `UploadsMiddleware`
 
 See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details.

backend/app/gateway/app.py

@@ -31,6 +31,7 @@ from app.gateway.routers import (
 )
 from deerflow.config import app_config as deerflow_app_config
 from deerflow.config.app_config import apply_logging_level
+from deerflow.uploads.manager import cleanup_stale_upload_staging_files
 
 AppConfig = deerflow_app_config.AppConfig
 get_app_config = deerflow_app_config.get_app_config
@@ -207,6 +208,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.warning("tiktoken warm-up skipped", exc_info=True)
 
+    try:
+        removed_upload_staging_files = await asyncio.to_thread(cleanup_stale_upload_staging_files)
+        if removed_upload_staging_files:
+            logger.info("Removed %d stale upload staging file(s)", removed_upload_staging_files)
+    except Exception:
+        logger.warning("Upload staging file cleanup skipped", exc_info=True)
+
     # Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store)
     async with langgraph_runtime(app, startup_config):
         logger.info("LangGraph runtime initialised")

backend/app/gateway/routers/uploads.py

@@ -3,6 +3,8 @@
 import logging
 import os
 import stat
+import tempfile
+from pathlib import Path
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 from pydantic import BaseModel, Field
@@ -14,6 +16,8 @@ from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provider
 from deerflow.uploads.manager import (
+    UPLOAD_STAGING_PREFIX,
+    UPLOAD_STAGING_SUFFIX,
     PathTraversalError,
     UnsafeUploadPathError,
     claim_unique_filename,
@@ -23,9 +27,9 @@ from deerflow.uploads.manager import (
     get_uploads_dir,
     list_files_in_dir,
     normalize_filename,
-    open_upload_file_no_symlink,
     upload_artifact_url,
     upload_virtual_path,
+    validate_upload_destination,
 )
 from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
 
@@ -173,7 +177,11 @@ async def _write_upload_file_with_limits(
     total_size: int,
 ) -> tuple[os.PathLike[str] | str, int, int]:
     file_size = 0
-    file_path, fh = open_upload_file_no_symlink(uploads_dir, display_filename)
+    uploads_dir_path = Path(uploads_dir)
+    file_path = validate_upload_destination(uploads_dir_path, display_filename)
+    temp_fd, temp_path_str = tempfile.mkstemp(prefix=UPLOAD_STAGING_PREFIX, suffix=UPLOAD_STAGING_SUFFIX, dir=uploads_dir_path)
+    temp_path = Path(temp_path_str)
+    fh = os.fdopen(temp_fd, "wb")
     try:
         while chunk := await file.read(UPLOAD_CHUNK_SIZE):
             file_size += len(chunk)
@@ -186,12 +194,20 @@ async def _write_upload_file_with_limits(
     except Exception:
         fh.close()
         try:
-            os.unlink(file_path)
+            os.unlink(temp_path)
         except FileNotFoundError:
             pass
         raise
     else:
         fh.close()
+        try:
+            os.replace(temp_path, file_path)
+        except Exception:
+            try:
+                os.unlink(temp_path)
+            except FileNotFoundError:
+                pass
+            raise
     return file_path, file_size, total_size
 
 

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

@@ -12,6 +12,7 @@ from langgraph.runtime import Runtime
 
 from deerflow.config.paths import Paths, get_paths
 from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.uploads.manager import is_upload_staging_file
 from deerflow.utils.file_conversion import extract_outline
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, message_content_to_text
 
@@ -172,7 +173,7 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
             if not isinstance(f, dict):
                 continue
             filename = f.get("filename") or ""
-            if not filename or Path(filename).name != filename:
+            if not filename or Path(filename).name != filename or is_upload_staging_file(filename):
                 continue
             if uploads_dir is not None and not (uploads_dir / filename).is_file():
                 continue
@@ -234,6 +235,8 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
         historical_files: list[dict] = []
         if uploads_dir and uploads_dir.exists():
             for file_path in sorted(uploads_dir.iterdir()):
+                if is_upload_staging_file(file_path.name):
+                    continue
                 if file_path.is_file() and file_path.name not in new_filenames:
                     stat = file_path.stat()
                     outline, preview = _extract_outline_for_file(file_path)

backend/packages/harness/deerflow/sandbox/search.py

@@ -43,6 +43,7 @@ IGNORE_PATTERNS = [
     "*.log",
     "*.tmp",
     "*.temp",
+    ".upload-*.part",
     "*.bak",
     "*.cache",
     ".cache",

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

@@ -1,10 +1,14 @@
 from .manager import (
+    UPLOAD_STAGING_PREFIX,
+    UPLOAD_STAGING_SUFFIX,
     PathTraversalError,
     claim_unique_filename,
+    cleanup_stale_upload_staging_files,
     delete_file_safe,
     enrich_file_listing,
     ensure_uploads_dir,
     get_uploads_dir,
+    is_upload_staging_file,
     list_files_in_dir,
     normalize_filename,
     upload_artifact_url,
@@ -18,7 +22,11 @@ __all__ = [
     "ensure_uploads_dir",
     "normalize_filename",
     "PathTraversalError",
+    "UPLOAD_STAGING_PREFIX",
+    "UPLOAD_STAGING_SUFFIX",
     "claim_unique_filename",
+    "cleanup_stale_upload_staging_files",
+    "is_upload_staging_file",
     "validate_path_traversal",
     "list_files_in_dir",
     "delete_file_safe",

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

@@ -5,6 +5,7 @@ Both Gateway and Client delegate to these functions.
 """
 
 import errno
+import logging
 import os
 import re
 import stat
@@ -23,8 +24,12 @@ class UnsafeUploadPathError(ValueError):
     """Raised when an upload destination is not a safe regular file path."""
 
 
+logger = logging.getLogger(__name__)
+
 # thread_id must be alphanumeric, hyphens, underscores, or dots only.
 _SAFE_THREAD_ID = re.compile(r"^[a-zA-Z0-9._-]+$")
+UPLOAD_STAGING_PREFIX = ".upload-"
+UPLOAD_STAGING_SUFFIX = ".part"
 
 
 def validate_thread_id(thread_id: str) -> None:
@@ -103,6 +108,11 @@ def claim_unique_filename(name: str, seen: set[str]) -> str:
     return candidate
 
 
+def is_upload_staging_file(filename: str) -> bool:
+    """Return whether *filename* is a transient Gateway upload staging file."""
+    return filename.startswith(UPLOAD_STAGING_PREFIX) and filename.endswith(UPLOAD_STAGING_SUFFIX)
+
+
 def validate_path_traversal(path: Path, base: Path) -> None:
     """Verify that *path* is inside *base*.
 
@@ -115,6 +125,56 @@ def validate_path_traversal(path: Path, base: Path) -> None:
         raise PathTraversalError("Path traversal detected") from None
 
 
+def validate_upload_destination(base_dir: Path, filename: str) -> Path:
+    """Validate an upload destination without mutating an existing file."""
+    safe_name = normalize_filename(filename)
+    dest = base_dir / safe_name
+
+    try:
+        st = os.lstat(dest)
+    except FileNotFoundError:
+        st = None
+
+    if st is not None and not stat.S_ISREG(st.st_mode):
+        raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}")
+    if st is not None and st.st_nlink > 1:
+        raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
+
+    validate_path_traversal(dest, base_dir)
+    return dest
+
+
+def _iter_upload_dirs(base_dir: Path):
+    yield from base_dir.glob("threads/*/user-data/uploads")
+    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."""
+    root = Path(base_dir) if base_dir is not None else get_paths().base_dir
+    removed = 0
+    for uploads_dir in _iter_upload_dirs(root):
+        if not uploads_dir.is_dir():
+            continue
+        try:
+            with os.scandir(uploads_dir) as entries:
+                for entry in entries:
+                    if not is_upload_staging_file(entry.name) or not entry.is_file(follow_symlinks=False):
+                        continue
+                    try:
+                        os.unlink(entry.path)
+                        removed += 1
+                    except FileNotFoundError:
+                        pass
+                    except OSError:
+                        logger.warning("Failed to remove stale upload staging file: %s", entry.path, exc_info=True)
+        except FileNotFoundError:
+            continue
+        except OSError:
+            logger.warning("Failed to scan uploads directory for stale staging files: %s", uploads_dir, exc_info=True)
+    return removed
+
+
 def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, object]:
     """Open an upload destination for safe streaming writes.
 
@@ -128,18 +188,12 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str) -> tuple[Path, ob
     validation prevents escapes from *base_dir* in both cases.
     """
     safe_name = normalize_filename(filename)
-    dest = base_dir / safe_name
+    dest = validate_upload_destination(base_dir, safe_name)
-
     try:
         st = os.lstat(dest)
     except FileNotFoundError:
         st = None
 
-    if st is not None and not stat.S_ISREG(st.st_mode):
-        raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}")
-
-    validate_path_traversal(dest, base_dir)
-
     has_nofollow = hasattr(os, "O_NOFOLLOW")
 
     if has_nofollow:
@@ -234,6 +288,8 @@ def list_files_in_dir(directory: Path) -> dict:
     files = []
     with os.scandir(directory) as entries:
         for entry in sorted(entries, key=lambda e: e.name):
+            if is_upload_staging_file(entry.name):
+                continue
             if not entry.is_file(follow_symlinks=False):
                 continue
             st = entry.stat(follow_symlinks=False)

backend/tests/test_gateway_lifespan_shutdown.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import asyncio
 from contextlib import asynccontextmanager
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from fastapi import FastAPI
@@ -32,17 +33,18 @@ async def _run_lifespan_with_hanging_stop() -> float:
         await asyncio.sleep(3600)
 
     app = FastAPI()
+    startup_config = SimpleNamespace(log_level="INFO", memory=SimpleNamespace(token_counting="char"))
 
     fake_service = MagicMock()
     fake_service.get_status = MagicMock(return_value={})
 
-    async def fake_start():
+    async def fake_start(_startup_config):
         return fake_service
 
     close_oidc_service = AsyncMock()
 
     with (
-        patch("app.gateway.app.get_app_config"),
+        patch("app.gateway.app.get_app_config", return_value=startup_config),
         patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)),
         patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime),
         patch("app.gateway.app.auth.close_oidc_service", close_oidc_service),
@@ -70,3 +72,40 @@ def test_shutdown_is_bounded_when_channel_stop_hangs():
     assert elapsed < _SHUTDOWN_HOOK_TIMEOUT_SECONDS + 2.0, f"Lifespan shutdown took {elapsed:.2f}s; expected <= {_SHUTDOWN_HOOK_TIMEOUT_SECONDS + 2.0:.1f}s"
     # Lower bound: the wait_for should actually have waited.
     assert elapsed >= _SHUTDOWN_HOOK_TIMEOUT_SECONDS - 0.5, f"Lifespan exited too quickly ({elapsed:.2f}s); wait_for may not have been invoked."
+
+
+async def _run_lifespan_with_upload_staging_cleanup():
+    from app.gateway.app import lifespan
+
+    app = FastAPI()
+    startup_config = SimpleNamespace(log_level="INFO", memory=SimpleNamespace(token_counting="char"))
+    fake_service = MagicMock()
+    fake_service.get_status = MagicMock(return_value={})
+    cleanup_upload_staging_files = MagicMock(return_value=2)
+    close_oidc_service = AsyncMock()
+    stop_channel_service = AsyncMock()
+
+    async def fake_start(_startup_config):
+        return fake_service
+
+    with (
+        patch("app.gateway.app.get_app_config", return_value=startup_config),
+        patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)),
+        patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime),
+        patch("app.gateway.app.cleanup_stale_upload_staging_files", cleanup_upload_staging_files),
+        patch("app.gateway.app.auth.close_oidc_service", close_oidc_service),
+        patch("app.channels.service.start_channel_service", side_effect=fake_start),
+        patch("app.channels.service.stop_channel_service", stop_channel_service),
+    ):
+        async with lifespan(app):
+            pass
+
+    return cleanup_upload_staging_files, close_oidc_service, stop_channel_service
+
+
+def test_lifespan_sweeps_upload_staging_files_on_startup():
+    cleanup_upload_staging_files, close_oidc_service, stop_channel_service = asyncio.run(_run_lifespan_with_upload_staging_cleanup())
+
+    cleanup_upload_staging_files.assert_called_once_with()
+    close_oidc_service.assert_awaited_once()
+    stop_channel_service.assert_awaited_once()

backend/tests/test_sandbox_search_tools.py

@@ -459,3 +459,23 @@ def test_ls_tool_returns_empty_for_empty_directory(tmp_path, monkeypatch) -> Non
     )
 
     assert result == "(empty)"
+
+
+def test_ls_tool_filters_upload_staging_files(tmp_path, monkeypatch) -> None:
+    runtime = _make_runtime(tmp_path)
+    uploads = tmp_path / "uploads"
+    (uploads / "report.txt").write_text("ready\n", encoding="utf-8")
+    (uploads / ".upload-active.part").write_text("partial\n", encoding="utf-8")
+    (uploads / ".upload-note.txt").write_text("intentional\n", encoding="utf-8")
+
+    monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: LocalSandbox(id="local"))
+
+    result = ls_tool.func(
+        runtime=runtime,
+        description="list uploads",
+        path="/mnt/user-data/uploads",
+    )
+
+    assert "/mnt/user-data/uploads/report.txt" in result
+    assert "/mnt/user-data/uploads/.upload-note.txt" in result
+    assert ".upload-active.part" not in result

backend/tests/test_uploads_manager.py

@@ -10,6 +10,7 @@ from deerflow.uploads.manager import (
     PathTraversalError,
     UnsafeUploadPathError,
     claim_unique_filename,
+    cleanup_stale_upload_staging_files,
     delete_file_safe,
     list_files_in_dir,
     normalize_filename,
@@ -187,6 +188,49 @@ class TestListFilesInDir:
         assert result["count"] == 1
         assert result["files"][0]["filename"] == "file.txt"
 
+    def test_filters_only_upload_staging_files(self, tmp_path):
+        (tmp_path / ".env").write_text("intentional dotfile")
+        (tmp_path / ".upload-active.part").write_text("partial")
+        (tmp_path / ".upload-note.txt").write_text("intentional upload")
+        (tmp_path / "draft.part").write_text("intentional upload")
+        (tmp_path / "visible.txt").write_text("visible")
+
+        result = list_files_in_dir(tmp_path)
+
+        assert result["count"] == 4
+        assert [f["filename"] for f in result["files"]] == [".env", ".upload-note.txt", "draft.part", "visible.txt"]
+
+
+# ---------------------------------------------------------------------------
+# cleanup_stale_upload_staging_files
+# ---------------------------------------------------------------------------
+
+
+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"
+        user_uploads = tmp_path / "users" / "owner-1" / "threads" / "thread-owned" / "user-data" / "uploads"
+        unrelated_uploads = tmp_path / "misc" / "thread-other" / "user-data" / "uploads"
+        for uploads_dir in (legacy_uploads, user_uploads, unrelated_uploads):
+            uploads_dir.mkdir(parents=True)
+
+        (legacy_uploads / ".upload-old.part").write_text("legacy partial")
+        (user_uploads / ".upload-new.part").write_text("user partial")
+        (unrelated_uploads / ".upload-ignore.part").write_text("outside layout")
+        (legacy_uploads / ".env").write_text("intentional dotfile")
+        (legacy_uploads / ".upload-note.txt").write_text("intentional upload")
+        (legacy_uploads / "draft.part").write_text("intentional upload")
+
+        removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 2
+        assert not (legacy_uploads / ".upload-old.part").exists()
+        assert not (user_uploads / ".upload-new.part").exists()
+        assert (unrelated_uploads / ".upload-ignore.part").exists()
+        assert (legacy_uploads / ".env").exists()
+        assert (legacy_uploads / ".upload-note.txt").exists()
+        assert (legacy_uploads / "draft.part").exists()
+
 
 # ---------------------------------------------------------------------------
 # delete_file_safe

backend/tests/test_uploads_middleware_core_logic.py

@@ -146,6 +146,11 @@ class TestFilesFromKwargs:
         assert result is not None
         assert result[0]["size"] == 0
 
+    def test_skips_upload_staging_filenames(self, tmp_path):
+        mw = _middleware(tmp_path)
+        msg = _human("hi", files=[{"filename": ".upload-active.part", "size": 5, "path": "/mnt/user-data/uploads/.upload-active.part"}])
+        assert mw._files_from_kwargs(msg) is None
+
 
 # ---------------------------------------------------------------------------
 # _create_files_message
@@ -362,6 +367,22 @@ class TestBeforeAgent:
         assert "previous messages" in content
         assert "old.txt" in content
 
+    def test_historical_files_ignore_upload_staging_files(self, tmp_path):
+        mw = _middleware(tmp_path)
+        uploads_dir = _uploads_dir(tmp_path)
+        (uploads_dir / "old.txt").write_bytes(b"old")
+        (uploads_dir / ".upload-active.part").write_bytes(b"partial")
+        (uploads_dir / ".env").write_bytes(b"intentional")
+
+        msg = _human("go")
+        result = mw.before_agent(self._state(msg), _runtime())
+
+        assert result is not None
+        content = result["messages"][-1].content
+        assert "old.txt" in content
+        assert ".env" in content
+        assert ".upload-active.part" not in content
+
     def test_no_historical_section_when_upload_dir_is_empty(self, tmp_path):
         mw = _middleware(tmp_path)
         uploads_dir = _uploads_dir(tmp_path)

backend/tests/test_uploads_router.py

@@ -653,6 +653,38 @@ def test_upload_files_overwrites_existing_regular_file(tmp_path):
     assert existing_file.stat().st_nlink == 1
 
 
+def test_upload_files_oversized_replacement_preserves_existing_regular_file(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    existing_file = thread_uploads_dir / "a.txt"
+    existing_file.write_bytes(b"original bytes")
+
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = True
+
+    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),
+    ):
+        file = ChunkedUpload("a.txt", [b"tiny", b"x" * 8])
+
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(
+                call_unwrapped(
+                    uploads.upload_files,
+                    "thread-local",
+                    request=MagicMock(),
+                    files=[file],
+                    config=SimpleNamespace(uploads={"max_file_size": 10}),
+                )
+            )
+
+    assert exc_info.value.status_code == 413
+    assert existing_file.read_bytes() == b"original bytes"
+    assert [path.name for path in thread_uploads_dir.iterdir()] == ["a.txt"]
+
+
 def test_delete_uploaded_file_removes_generated_markdown_companion(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)