fix(projects): check the project before staging shelf uploads (#7040)

* fix(projects): check the project before staging shelf uploads

The shelf upload and save-from-thread routes staged the body under
users/{user_id}/projects/{project_id}/ before checking the project. A
missing, foreign, or archived id left an empty {project_id}/documents/
directory that the retention sweep keeps forever, and an id longer than
one path component failed with ENAMETOOLONG (500).

Both routes now require an owned, active project first and answer the
same 404 without writing. The shelf insert still re-checks under the
project row lock.

* test(projects): pin the shelf insert's re-check behind the early project check

With the early check in place, no route test reached the insert's
row-lock re-check any more. Archive the project between the check and
the insert and assert both routes still answer 404 and leave no staging
file.

* docs(changelog): reference #7040 for the shelf upload project check

* docs(changelog): drop the whitespace-only line after the zh #7040 entry

---------

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

CHANGELOG.md

@@ -1265,6 +1265,16 @@ This release closes that milestone with **621 merged pull requests**.
 
 ### Fixed
 
+- **projects:** A shelf upload or save-from-thread into a project the caller
+  cannot write no longer leaves a directory behind or answers `500`. Both
+  routes staged the body under `users/{user_id}/projects/{project_id}/` before
+  checking the project, so a missing, foreign, or archived id kept an empty
+  `{project_id}/documents/` directory that the retention sweep never removes,
+  and an id longer than one path component failed with `ENAMETOOLONG`. They
+  now check for an owned, active project first and answer the same `404`
+  without writing; the shelf insert still re-checks under the project row lock.
+  ([#7040])
+
 - **workflows:** A script workflow waiting on a long child Agent run no longer
   fails with `Worker turn limit` after about 209 seconds regardless of
   `timeout_seconds`. The supervisor re-sent an empty frame to the QuickJS worker
@@ -11849,3 +11859,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6930]: https://github.com/bytedance/deer-flow/pull/6930
 [#6988]: https://github.com/bytedance/deer-flow/pull/6988
 [#6991]: https://github.com/bytedance/deer-flow/pull/6991
+[#7040]: https://github.com/bytedance/deer-flow/pull/7040

CHANGELOG_zh.md

@@ -1015,6 +1015,13 @@
 
 ### 修复
 
+- **项目：** 向调用者无权写入的项目上传书架文档或从对话保存文件时，不再遗留目录，
+  也不再返回 `500`。这两个路由此前先把请求体暂存到
+  `users/{user_id}/projects/{project_id}/` 下才检查项目，因此不存在、属于他人或已归档
+  的 id 会留下一个回收站对账永远不会清理的空 `{project_id}/documents/` 目录，而超过单个
+  路径分量长度的 id 会因 `ENAMETOOLONG` 失败。现在会先确认项目归调用者所有且处于活跃
+  状态，不写任何内容即返回同样的 `404`；书架插入仍会在项目行锁下再次检查。 ([#7040])
+
 - **浏览器：** `browser_screenshot` 不再替换线程 outputs 目录中已存在的文件。此前以
   `report.png` 保存的截图会覆盖用户自己的 `report.png`、更早的截图，或同一秒内生成的
   另一张默认命名截图；现在会像 `web_capture` 一样改用下一个空闲名称（`report-1.png`、
@@ -9422,3 +9429,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6930]: https://github.com/bytedance/deer-flow/pull/6930
 [#6988]: https://github.com/bytedance/deer-flow/pull/6988
 [#6991]: https://github.com/bytedance/deer-flow/pull/6991
+[#7040]: https://github.com/bytedance/deer-flow/pull/7040

backend/app/gateway/routers/project_documents.py

@@ -137,6 +137,20 @@ async def _require_project(request: Request, project_id: str) -> dict:
     return row
 
 
+async def _require_active_project(request: Request, project_id: str) -> dict:
+    """Owned ``status='active'`` project row, else 404 — checked before staging.
+
+    Staging creates ``users/{user_id}/projects/{project_id}/`` on disk, so a
+    missing, foreign, or archived project (or an id no row can carry) must be
+    rejected before any byte is written. The shelf insert re-checks under the
+    project row lock; this is the early exit, not the authority (§8.4).
+    """
+    row = await _require_project(request, project_id)
+    if row.get("status") != "active":
+        raise _not_found()
+    return row
+
+
 async def _stream_upload(file: UploadFile) -> AsyncIterator[bytes]:
     while chunk := await file.read(UPLOAD_CHUNK_SIZE):
         yield chunk
@@ -178,9 +192,9 @@ async def upload_project_document(
 ) -> ProjectDocumentUploadResponse:
     """Shelf exactly one file (§17.2): 201 created, or 200 on a content dedup hit.
 
-    Active projects only — the shelf insert locks the project row with
-    ``status='active'`` inside its transaction, so an archived/foreign/missing
-    project is the same 404 (§8.4, §11). The display name defaults to the
+    Active projects only — an archived/foreign/missing project is the same
+    404 (§8.4, §11), rejected before the body is staged and re-checked by the
+    shelf insert under the project row lock. The display name defaults to the
     multipart filename; both go through the same validation (400 when empty,
     separator-bearing, or over 255 UTF-8 bytes). Size reuses
     ``uploads.max_file_size`` (413, mirroring the uploads router).
@@ -190,6 +204,7 @@ async def upload_project_document(
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc))
 
+    await _require_active_project(request, project_id)
     user_id = get_effective_user_id()
     paths = get_paths()
     max_file_size = _get_upload_limit(config, "max_file_size", DEFAULT_MAX_FILE_SIZE, legacy_key="max_single_file_size")
@@ -287,9 +302,9 @@ async def promote_thread_file_to_shelf(
     The source is never moved — the shelf gets a project-owned snapshot with
     provenance (``source_thread_id``/``source_kind``/``source_name``), so
     deleting the thread cannot affect it. ``shelf_name`` defaults to the
-    source name and follows upload-name validation. The shelf insert locks
-    the project row with ``status='active'``, so an archived/foreign/missing
-    project is the same 404 as a source that fails confinement (§8.4, §11).
+    source name and follows upload-name validation. An archived/foreign/missing
+    project is the same 404 as a source that fails confinement (§8.4, §11),
+    rejected before staging and re-checked by the shelf insert's row lock.
     Response mirrors upload: 201 created, 200 on a content dedup hit (the
     existing row keeps its own name and provenance, §10.9).
     """
@@ -307,6 +322,7 @@ async def promote_thread_file_to_shelf(
     if source is None:
         raise _not_found()
 
+    await _require_active_project(request, project_id)
     max_file_size = _get_upload_limit(config, "max_file_size", DEFAULT_MAX_FILE_SIZE, legacy_key="max_single_file_size")
     try:
         staged = await stage_document_bytes(paths, user_id=user_id, project_id=project_id, chunks=read_file_chunks(source, chunk_size=UPLOAD_CHUNK_SIZE), max_bytes=max_file_size)

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

@@ -156,7 +156,8 @@ async def stage_document_bytes(paths: Paths, *, user_id: str, project_id: str, c
     the shelf deliberately reuses it instead of growing a second knob with
     the same meaning (§6.4). Overflow raises :class:`ShelfUploadTooLargeError`
     (the route maps it to 413, mirroring the uploads router) and leaves no
-    staging file behind.
+    staging file behind. Staging creates the project's directory tree, so
+    callers reject a missing, foreign, or archived project first.
     """
     staging_path = _new_staging_path(paths, user_id=user_id, project_id=project_id)
     handle = await run_file_io(_open_staging, staging_path.parent, staging_path)

backend/tests/test_project_documents_promotion.py

@@ -324,6 +324,47 @@ class TestFromThread:
 
             assert _promote(client, pid, thread_id="thread-1", kind="upload", name="real.txt").status_code == 404
 
+    @pytest.mark.parametrize("case", ["missing", "foreign", "archived", "overlong"])
+    def test_rejected_promotion_stages_nothing_and_creates_no_project_dir(self, tmp_path, case):
+        app = _build_app(tmp_path)
+        with TestClient(app) as client:
+            _seed_thread(app, "thread-1")
+            _thread_file("thread-1", "upload", "real.txt", b"real")
+            if case == "missing":
+                pid = "no-such-project"
+            elif case == "foreign":
+                pid = client.post("/api/projects", json={"name": "B"}, headers=_as_user("user-b")).json()["id"]
+            elif case == "archived":
+                pid = _create_project(client)["id"]
+                assert client.post(f"/api/projects/{pid}/archive").status_code == 200
+            else:
+                # Charset-safe but longer than one path component: 404, not 500.
+                pid = "a" * 300
+            with patch.object(project_documents, "stage_document_bytes", new_callable=AsyncMock, wraps=project_documents.stage_document_bytes) as stage:
+                assert _promote(client, pid, thread_id="thread-1", kind="upload", name="real.txt").status_code == 404
+            stage.assert_not_awaited()
+            projects_root = get_paths().user_projects_dir(_USER)
+            assert not projects_root.exists() or list(projects_root.iterdir()) == []
+
+    def test_project_archived_after_the_early_check_is_still_404_and_unstaged(self, tmp_path):
+        app = _build_app(tmp_path)
+        real_stage = project_documents.stage_document_bytes
+
+        async def _archive_then_stage(*args, **kwargs):
+            # Lose the race: the project is archived between the early check
+            # and the shelf insert, whose row lock is the authority.
+            await app.state.project_repo.set_status(kwargs["project_id"], "archived")
+            return await real_stage(*args, **kwargs)
+
+        with TestClient(app) as client:
+            pid = _create_project(client)["id"]
+            _seed_thread(app, "thread-1")
+            _thread_file("thread-1", "upload", "real.txt", b"real")
+            with patch.object(project_documents, "stage_document_bytes", side_effect=_archive_then_stage):
+                assert _promote(client, pid, thread_id="thread-1", kind="upload", name="real.txt").status_code == 404
+            assert list((get_paths().project_documents_dir(_USER, pid) / ".staging").iterdir()) == []
+            assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0
+
     def test_promote_rejects_a_symlinked_kind_dir_escaping_thread_storage(self, tmp_path):
         """A planted outputs symlink pointing outside thread storage must not
         become the trusted base: the source never resolves (404) and the

backend/tests/test_project_documents_router.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
 import pytest
@@ -21,6 +21,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
 from app.gateway.authz import AuthContext, Permissions
 from app.gateway.deps import get_config
 from app.gateway.routers import project_documents, projects
+from deerflow.config.paths import get_paths
 from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
 from deerflow.persistence.projects import ProjectDocumentRepository, ProjectRepository
 from deerflow.persistence.thread_meta import ThreadMetaRepository
@@ -299,6 +300,43 @@ class TestFailClosed:
             assert client.get(f"/api/projects/{pid}/documents/{doc['id']}/content", headers=foreign).status_code == 404
             assert client.delete(f"/api/projects/{pid}/documents/{doc['id']}", headers=foreign).status_code == 404
 
+    @pytest.mark.parametrize("case", ["missing", "foreign", "archived", "overlong"])
+    def test_rejected_upload_stages_nothing_and_creates_no_project_dir(self, tmp_path, case):
+        app = _build_app(tmp_path)
+        with TestClient(app) as client:
+            if case == "missing":
+                pid = "no-such-project"
+            elif case == "foreign":
+                pid = client.post("/api/projects", json={"name": "B"}, headers=_as_user("user-b")).json()["id"]
+            elif case == "archived":
+                pid = _create_project(client)["id"]
+                assert client.post(f"/api/projects/{pid}/archive").status_code == 200
+            else:
+                # Charset-safe but longer than one path component: 404, not 500.
+                pid = "a" * 300
+            with patch.object(project_documents, "stage_document_bytes", new_callable=AsyncMock, wraps=project_documents.stage_document_bytes) as stage:
+                assert _upload(client, pid, "a.txt", b"body").status_code == 404
+            stage.assert_not_awaited()
+            projects_root = get_paths().user_projects_dir("user-a")
+            assert not projects_root.exists() or list(projects_root.iterdir()) == []
+
+    def test_project_archived_after_the_early_check_is_still_404_and_unstaged(self, tmp_path):
+        app = _build_app(tmp_path)
+        real_stage = project_documents.stage_document_bytes
+
+        async def _archive_then_stage(*args, **kwargs):
+            # Lose the race: the project is archived between the early check
+            # and the shelf insert, whose row lock is the authority.
+            await app.state.project_repo.set_status(kwargs["project_id"], "archived")
+            return await real_stage(*args, **kwargs)
+
+        with TestClient(app) as client:
+            pid = _create_project(client)["id"]
+            with patch.object(project_documents, "stage_document_bytes", side_effect=_archive_then_stage):
+                assert _upload(client, pid, "a.txt", b"body").status_code == 404
+            assert list((get_paths().project_documents_dir("user-a", pid) / ".staging").iterdir()) == []
+            assert client.get(f"/api/projects/{pid}/documents").json()["total"] == 0
+
     def test_503_when_document_repo_unavailable(self, tmp_path):
         app = _build_app(tmp_path, document_repo=None)
         with TestClient(app) as client: