fix(projects): preserve document upload outcomes during cleanup (#7141)

* fix(projects): preserve upload outcomes during staging cleanup

* fix(projects): guard staging discard failures

---------

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

README.md

@@ -3159,6 +3159,9 @@ from the project page's Documents section:
 - **Upload** a file (button or drag-drop, one file per request). Shelf size
   limits reuse `uploads.max_file_size` (default 50 MiB); re-uploading identical
   content returns the existing entry instead of creating a duplicate.
+  Oversized uploads return `413`; deduplicated uploads return `200`. Staging
+  cleanup failures are logged without replacing these outcomes or a missing,
+  foreign, or archived project's `404` response.
 - **List** entries with name, size, modified time, and provenance (uploaded vs.
   saved from a conversation), and preview or download any entry.
 - **Save to project** from a thread file: the read-only conversation-files

backend/packages/harness/deerflow/projects/documents.py

@@ -32,12 +32,14 @@ import threading
 import uuid
 from collections import OrderedDict
 from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from deerflow.config.paths import Paths
 from deerflow.uploads.manager import normalize_filename
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
 from deerflow.utils.file_io import await_drained, run_file_io
 from deerflow.utils.text_detection import is_text_file_by_content
@@ -143,8 +145,11 @@ def _open_staging(staging_dir: Path, staging_path: Path):
 def _close_and_unlink(handle: Any, staging_path: Path) -> None:
     try:
         handle.close()
+    except OSError:
+        with suppress(Exception):
+            logger.warning("Failed to close shelf staging file %s", staging_path, exc_info=True)
     finally:
-        staging_path.unlink(missing_ok=True)
+        cleanup_failed_atomic_write(staging_path)
 
 
 async def stage_document_bytes(paths: Paths, *, user_id: str, project_id: str, chunks: AsyncIterator[bytes] | list[bytes], max_bytes: int) -> StagedDocument:
@@ -192,7 +197,7 @@ def _place_staging(staging_path: Path, final_path: Path) -> None:
     try:
         os.replace(staging_path, final_path)
     except Exception:
-        staging_path.unlink(missing_ok=True)
+        cleanup_failed_atomic_write(staging_path)
         raise
 
 
@@ -263,7 +268,7 @@ async def add_staged_document(
         # row raced into trash (or a trashing project deletion) before the
         # re-check still owns the placed bytes — trash is recoverable, so
         # deleting the namespace would destroy a recoverable document.
-        await run_file_io(_remove_staging, staged.staging_path)
+        await run_file_io(cleanup_failed_atomic_write, staged.staging_path)
         committed: dict | None
         try:
             committed = await repo.get(document_id, include_trashed=True, user_id=user_id)
@@ -292,11 +297,11 @@ async def add_staged_document(
         raise
     if row is None:
         # Missing/foreign/archived project: nothing was placed or inserted.
-        await run_file_io(_remove_staging, staged.staging_path)
+        await run_file_io(cleanup_failed_atomic_write, staged.staging_path)
         return None
     if row["id"] != document_id:
         # Dedup hit: the existing row wins; discard the staged duplicate.
-        await run_file_io(_remove_staging, staged.staging_path)
+        await run_file_io(cleanup_failed_atomic_write, staged.staging_path)
         return row, False
     return row, True
 

backend/packages/harness/deerflow/utils/AGENTS.md

@@ -6,9 +6,11 @@ capture/fetch publication. It writes and fsyncs a hidden exclusive sibling befor
 umask; replacements preserve existing permission bits before publication. Never
 read the process umask by changing it in this threaded runtime. Behavioral tests
 for both tool imports live in `tests/test_browser_outputs_atomic.py`.
-`cleanup_failed_atomic_write` shares failure cleanup between this publisher and
-`skills/projection.py`'s private manifest writer. Call it while handling the
-original `BaseException`; pass a raw descriptor only until `fdopen` succeeds.
+`cleanup_failed_atomic_write` shares best-effort cleanup across this publisher,
+`skills/projection.py`'s private manifest writer, and project document staging.
+Use it on publication failure or after deciding to discard an owned
+temporary; pass a raw descriptor only until `fdopen` succeeds. Shelf discards
+must preserve both a rejected project's 404 and a dedup hit's 200 response.
 Callers retain their own serialization, permissions and publication policy.
 Failure cleanup must preserve the original publication exception, including
 interruptions. Descriptor close and sibling unlink are separate best-effort

backend/packages/harness/deerflow/utils/atomic_write.py

@@ -13,11 +13,12 @@ logger = logging.getLogger(__name__)
 
 
 def cleanup_failed_atomic_write(temporary: Path, fd: int | None = None) -> None:
-    """Release an owned temporary without replacing the caller's active error.
+    """Release an owned temporary without replacing the caller's outcome.
 
     Pass *fd* only while the raw descriptor is still owned by the caller;
     successful ``fdopen`` transfers that ownership to the file object.
-    Filesystem and logging-handler errors remain independent best-effort steps.
+    Also use this to discard staging after a publication decision. Filesystem
+    and logging-handler errors remain independent best-effort steps.
     """
     # Cleanup and diagnostics are best-effort: keep the original error
     # and attempt unlink even if closing an untransferred descriptor fails.

backend/tests/blocking_io/test_project_documents.py

@@ -126,20 +126,27 @@ async def test_list_content_missing_check_dispatches_off_the_loop(tmp_path, monk
         await close_engine()
 
 
-async def test_dedup_hit_staging_cleanup_dispatches_off_the_loop(tmp_path, monkeypatch) -> None:
-    """The dedup-hit branch removes the staged duplicate via the offload."""
+@pytest.mark.parametrize("outcome", ["dedup", "archived"])
+async def test_staging_discard_dispatches_off_the_loop(tmp_path, monkeypatch, outcome) -> None:
+    """Both discarded-staging branches perform best-effort cleanup off-loop."""
     calls = _spy_offload(monkeypatch)
     env = await _make_env(tmp_path, monkeypatch)
     try:
-        first = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20)
-        await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="first.txt", staged=first)
+        if outcome == "dedup":
+            first = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20)
+            await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="first.txt", staged=first)
+        else:
+            await ProjectRepository(get_session_factory()).set_status(env.project_id, "archived", user_id=_USER)
         calls.clear()
-        duplicate = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20)
-        result = await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="second.txt", staged=duplicate)
-        assert result is not None and result[1] is False
-        assert not duplicate.staging_path.exists()
-        assert "_remove_staging" in calls
-        # No placement happened for the discarded duplicate.
+        staged = await stage_document_bytes(env.paths, user_id=_USER, project_id=env.project_id, chunks=[b"same"], max_bytes=1 << 20)
+        result = await add_staged_document(env.docs, env.paths, user_id=_USER, project_id=env.project_id, name="second.txt", staged=staged)
+        if outcome == "dedup":
+            assert result is not None and result[1] is False
+        else:
+            assert result is None
+        assert not staged.staging_path.exists()
+        assert "cleanup_failed_atomic_write" in calls
+        # No placement happened for either discarded upload.
         assert "_place_staging" not in calls
     finally:
         await close_engine()

backend/tests/test_project_documents_router.py

@@ -639,3 +639,188 @@ class TestDeleteScopesToTheUrlProject:
             assert client.delete(f"/api/projects/{project_a['id']}/documents/{doc['id']}").status_code == 404
             assert client.get(f"/api/projects/{project_b['id']}/documents/{doc['id']}/content").status_code == 200
             assert client.delete(f"/api/projects/{project_b['id']}/documents/{doc['id']}").status_code == 204
+
+
+class TestStagingFailureCleanup:
+    @pytest.mark.parametrize("failure", ["none", "close", "unlink", "logging"])
+    def test_oversize_upload_keeps_413_when_cleanup_fails(self, tmp_path, monkeypatch, caplog, failure):
+        import logging
+        from pathlib import Path
+
+        from deerflow.projects import documents
+        from deerflow.utils import atomic_write
+
+        app = _build_app(tmp_path)
+        app.dependency_overrides[get_config] = lambda: SimpleNamespace(uploads={"max_file_size": 3})
+        with TestClient(app) as client:
+            pid = _create_project(client)["id"]
+            staging = get_paths().project_documents_dir("user-a", pid) / ".staging"
+            if failure == "close":
+                open_staging = documents._open_staging
+
+                def open_with_close_error(*args):
+                    handle = open_staging(*args)
+                    close = handle.close
+
+                    def fail_close():
+                        close()
+                        raise OSError("staging close failed")
+
+                    handle.close = fail_close
+                    return handle
+
+                monkeypatch.setattr(documents, "_open_staging", open_with_close_error)
+            elif failure in {"unlink", "logging"}:
+                unlink = Path.unlink
+
+                def fail_unlink(path, *args, **kwargs):
+                    if path.parent == staging:
+                        raise OSError("staging unlink failed")
+                    return unlink(path, *args, **kwargs)
+
+                monkeypatch.setattr(Path, "unlink", fail_unlink)
+                if failure == "logging":
+
+                    def fail_warning(*args, **kwargs):
+                        raise RuntimeError("logging handler failed")
+
+                    monkeypatch.setattr(atomic_write.logger, "warning", fail_warning)
+            with caplog.at_level(logging.WARNING):
+                response = _upload(client, pid, "large.txt", b"1234")
+            assert response.status_code == 413
+            assert response.json()["detail"] == "File too large: large.txt"
+            assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0
+            remaining = list(staging.iterdir())
+            if failure in {"unlink", "logging"}:
+                assert len(remaining) == 1
+                if failure == "unlink":
+                    assert str(remaining[0]) in caplog.text
+                    assert "staging unlink failed" in caplog.text
+            else:
+                assert remaining == []
+                if failure == "close":
+                    assert str(staging) in caplog.text
+                    assert "staging close failed" in caplog.text
+
+    @pytest.mark.parametrize("after_commit", [False, True])
+    def test_insert_failure_cleanup_preserves_error_or_committed_document(self, tmp_path, monkeypatch, caplog, after_commit):
+        from pathlib import Path
+
+        from deerflow.projects import documents
+
+        app = _build_app(tmp_path)
+        primary = OSError("original shelf insert failure")
+        insert = app.state.project_document_repo.insert_active
+        if after_commit:
+
+            async def insert_then_fail(*args, **kwargs):
+                await insert(*args, **kwargs)
+                raise primary
+
+            monkeypatch.setattr(app.state.project_document_repo, "insert_active", insert_then_fail)
+        else:
+
+            def fail_replace(*args, **kwargs):
+                raise primary
+
+            monkeypatch.setattr(documents.os, "replace", fail_replace)
+        with TestClient(app) as client:
+            pid = _create_project(client)["id"]
+            staging = get_paths().project_documents_dir("user-a", pid) / ".staging"
+            unlink = Path.unlink
+
+            def fail_unlink(path, *args, **kwargs):
+                if path.parent == staging:
+                    raise OSError("staging unlink failed")
+                return unlink(path, *args, **kwargs)
+
+            monkeypatch.setattr(Path, "unlink", fail_unlink)
+            if after_commit:
+                response = _upload(client, pid, "saved.txt", b"saved bytes")
+                assert response.status_code == 201
+                doc = response.json()["document"]
+                assert client.get(f"/api/projects/{pid}/documents/{doc['id']}/content").content == b"saved bytes"
+                assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 1
+            else:
+                with pytest.raises(OSError) as caught:
+                    _upload(client, pid, "saved.txt", b"saved bytes")
+                assert caught.value is primary
+                assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0
+            assert "staging unlink failed" in caplog.text
+
+    @pytest.mark.parametrize("outcome", ["dedup", "archived"])
+    @pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+    def test_staging_discard_preserves_upload_outcome(self, tmp_path, monkeypatch, caplog, outcome, failure):
+        import logging
+        from pathlib import Path
+
+        from deerflow.utils import atomic_write
+
+        app = _build_app(tmp_path)
+        payload = b"same shelf bytes"
+        with TestClient(app) as client:
+            pid = _create_project(client)["id"]
+            staging = get_paths().project_documents_dir("user-a", pid) / ".staging"
+            if outcome == "dedup":
+                first = _upload(client, pid, "first.txt", payload)
+                assert first.status_code == 201
+                original = first.json()["document"]
+            else:
+                real_stage = project_documents.stage_document_bytes
+
+                async def archive_then_stage(*args, **kwargs):
+                    # The locked insert rejects a project archived after admission.
+                    await app.state.project_repo.set_status(kwargs["project_id"], "archived")
+                    return await real_stage(*args, **kwargs)
+
+                monkeypatch.setattr(project_documents, "stage_document_bytes", archive_then_stage)
+
+            unlink = Path.unlink
+            attempts = []
+
+            def discard(path, *args, **kwargs):
+                if path.parent == staging:
+                    attempts.append(path)
+                    if failure != "none":
+                        raise PermissionError("staging discard failed")
+                return unlink(path, *args, **kwargs)
+
+            monkeypatch.setattr(Path, "unlink", discard)
+            warning = MagicMock(wraps=atomic_write.logger.warning)
+            monkeypatch.setattr(atomic_write.logger, "warning", warning)
+            # This stdlib handler raises from emit when its delayed open fails.
+            handler = logging.FileHandler(tmp_path / "missing-logs" / "gateway.log", delay=True)
+            handler.emit = MagicMock(wraps=handler.emit)
+            if failure == "logging":
+                atomic_write.logger.addHandler(handler)
+            try:
+                with caplog.at_level(logging.WARNING, logger=atomic_write.__name__):
+                    response = _upload(client, pid, "duplicate.txt", payload)
+                if outcome == "dedup":
+                    assert response.status_code == 200
+                    body = response.json()
+                    assert body["deduplicated"] is True
+                    assert body["document"]["id"] == original["id"]
+                    assert body["document"]["name"] == "first.txt"
+                    assert client.get(f"/api/projects/{pid}/documents/{original['id']}/content").content == payload
+                    assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 1
+                else:
+                    assert response.status_code == 404
+                    assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0
+                assert len(attempts) == 1
+                if failure == "none":
+                    assert list(staging.iterdir()) == []
+                    warning.assert_not_called()
+                else:
+                    assert list(staging.iterdir()) == attempts
+                    assert attempts[0].read_bytes() == payload
+                    warning.assert_called_once()
+                    assert attempts[0] in warning.call_args.args
+                    assert warning.call_args.kwargs["exc_info"] is True
+                    if failure == "unlink":
+                        assert str(attempts[0]) in caplog.text
+                        assert "staging discard failed" in caplog.text
+                assert handler.emit.call_count == int(failure == "logging")
+            finally:
+                atomic_write.logger.removeHandler(handler)
+                handler.close()