fix: close upload rollback gaps

backend/AGENTS.md

@@ -1461,10 +1461,10 @@ 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()`
 - Every ingress stages a complete payload and atomically publishes it without replacing an existing entry. Collisions across requests, processes, HTTP, embedded, and IM adapters use `name.ext`, `name_1.ext`, `name_2.ext`; storage that cannot provide atomic no-replace publication fails explicitly.
-- Filenames containing `<`, `>`, or reserved model-context boundary markers are rejected so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed.
-- Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools.
+- Filenames containing NUL, `<`, `>`, or reserved model-context boundary markers are rejected before staging so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed.
+- Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files. Cancellation during staging creation drains the worker and aborts the returned stage before propagating. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools.
 - Generated Markdown is owned by `user-data/.upload-conversions/<actual-primary-filename>.md` and is omitted from primary upload listings. Deletion removes only the selected primary and that exact generated asset; it never guesses or deletes a legacy/user-owned `uploads/<stem>.md` sibling.
-- Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; potentially blocking name-lease acquisition uses a separate pool so waiters cannot starve release. Gateway, embedded-client, and IM ingresses share provider-aware publication: mounted providers make the exact host paths sandbox-readable; non-mounted providers acquire the sandbox and synchronize the primary plus generated conversion to their exact virtual paths. Each ingress records completed remote paths and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those paths before host rollback and lease release.
+- Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; potentially blocking name-lease acquisition uses a separate pool so waiters cannot starve release, and cold sandbox-provider construction is also offloaded. Gateway, embedded-client, and IM ingresses share provider-aware publication: mounted providers make the exact host paths sandbox-readable; non-mounted providers acquire the sandbox and synchronize the primary plus generated conversion to their exact virtual paths. Each ingress records attempted remote paths before the write can commit and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those paths before host rollback and lease release. Embedded multi-file calls retain every publication and receipt until the whole response is built, then roll back the complete batch on failure.
 - Mounted upload paths skip both sandbox acquisition and per-file synchronization. For AIO remote/provisioner deployments this requires an explicit, accurate `sandbox.thread_data_mounts: true`; omission preserves backend auto-detection.
 - Agent receives uploaded file list via `UploadsMiddleware`
 

backend/app/channels/dingtalk.py

@@ -650,7 +650,7 @@ class DingTalkChannel(Channel):
             virtual_path = upload_virtual_path(publication.path.name)
 
             try:
-                sandbox_provider = get_sandbox_provider()
+                sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
                 await make_upload_paths_available_async(
                     sandbox_provider,
                     thread_id,

backend/app/channels/manager.py

@@ -878,6 +878,16 @@ def _prepare_artifact_delivery(
     return response_text, attachments
 
 
+def _build_inbound_file_metadata(publication, virtual_path: str, data: bytes, file_type: str) -> dict[str, Any]:
+    """Build response metadata while the exact-name publication lease is active."""
+    return {
+        "filename": publication.path.name,
+        "size": len(data),
+        "path": virtual_path,
+        "is_image": file_type == "image",
+    }
+
+
 async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: str | None = None) -> list[dict[str, Any]]:
     if not msg.files:
         return []
@@ -894,7 +904,10 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
         ensure_uploads_dir,
         normalize_filename,
     )
-    from deerflow.uploads.sandbox_sync import make_upload_paths_available_async
+    from deerflow.uploads.sandbox_sync import (
+        make_upload_paths_available_async,
+        rollback_sandbox_sync_async,
+    )
 
     def _prepare_uploads_dir() -> Path:
         # Worker thread: directory creation is blocking filesystem IO that must
@@ -954,18 +967,25 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                 continue
 
             publication = None
+            sandbox_receipt = None
             try:
                 publication = await publish_upload_bytes_leased_async(uploads_dir, safe_name, data)
                 dest = publication.path
                 virtual_path = upload_virtual_path(dest.name)
                 sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
-                await make_upload_paths_available_async(
+                sandbox_receipt = await make_upload_paths_available_async(
                     sandbox_provider,
                     thread_id,
                     user_id=user_id,
                     paths=[(dest, virtual_path)],
                 )
+                created.append(_build_inbound_file_metadata(publication, virtual_path, data, ftype))
             except asyncio.CancelledError:
+                if sandbox_receipt is not None:
+                    try:
+                        await rollback_sandbox_sync_async(sandbox_receipt)
+                    except BaseException:
+                        logger.warning("[Manager] failed to roll back cancelled inbound sandbox file: %s", safe_name, exc_info=True)
                 if publication is not None:
                     try:
                         await rollback_published_upload_async(publication)
@@ -973,6 +993,11 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                         logger.warning("[Manager] failed to roll back cancelled inbound file: %s", safe_name, exc_info=True)
                 raise
             except UnsafeUploadPathError:
+                if sandbox_receipt is not None:
+                    try:
+                        await rollback_sandbox_sync_async(sandbox_receipt)
+                    except BaseException:
+                        logger.warning("[Manager] failed to roll back unsafe inbound sandbox file: %s", safe_name, exc_info=True)
                 if publication is not None:
                     try:
                         await rollback_published_upload_async(publication)
@@ -981,6 +1006,11 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                 logger.warning("[Manager] skipping inbound file with unsafe destination: %s", safe_name)
                 continue
             except Exception:
+                if sandbox_receipt is not None:
+                    try:
+                        await rollback_sandbox_sync_async(sandbox_receipt)
+                    except BaseException:
+                        logger.warning("[Manager] failed to roll back rejected inbound sandbox file: %s", safe_name, exc_info=True)
                 if publication is not None:
                     try:
                         await rollback_published_upload_async(publication)
@@ -992,16 +1022,6 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                 if publication is not None:
                     await release_published_upload_async(publication)
 
-            actual_name = dest.name
-            created.append(
-                {
-                    "filename": actual_name,
-                    "size": len(data),
-                    "path": virtual_path,
-                    "is_image": ftype == "image",
-                }
-            )
-
     return created
 
 

backend/app/gateway/routers/uploads.py

@@ -181,8 +181,8 @@ def _cleanup_published_uploads(
             logger.warning("Failed to roll back published upload after rejected request: %s", publication.path, exc_info=True)
 
 
-def _cleanup_synced_sandbox_paths(sandbox, virtual_paths: list[str]) -> None:
-    """Best-effort removal of the exact remote copies completed by this request."""
+def _cleanup_attempted_sandbox_paths(sandbox, virtual_paths: list[str]) -> None:
+    """Best-effort removal of every exact remote path attempted by this request."""
     if sandbox is None or not virtual_paths:
         return
     for virtual_path in reversed(virtual_paths):
@@ -194,12 +194,12 @@ def _cleanup_synced_sandbox_paths(sandbox, virtual_paths: list[str]) -> None:
 
 def _rollback_upload_request(
     sandbox,
-    synced_sandbox_paths: list[str],
+    attempted_sandbox_paths: list[str],
     publications: list[PublishedUpload],
     generated_paths: list[os.PathLike[str] | str],
 ) -> None:
     try:
-        _cleanup_synced_sandbox_paths(sandbox, synced_sandbox_paths)
+        _cleanup_attempted_sandbox_paths(sandbox, attempted_sandbox_paths)
     finally:
         _cleanup_published_uploads(publications, generated_paths)
 
@@ -244,6 +244,23 @@ async def _publish_staged_upload_cancellation_safe(staged: StagedUpload, filenam
         raise
 
 
+async def _create_upload_staging_file_cancellation_safe(uploads_dir: Path) -> StagedUpload:
+    """Create a staged upload without leaking it when the caller is cancelled."""
+    create_task = asyncio.create_task(run_file_io(create_upload_staging_file, uploads_dir))
+    try:
+        return await asyncio.shield(create_task)
+    except asyncio.CancelledError:
+        await wait_for_task_completion(create_task)
+        if not create_task.cancelled() and create_task.exception() is None:
+            cleanup_task = asyncio.create_task(run_file_io(abort_staged_upload, create_task.result()))
+            await wait_for_task_completion(cleanup_task)
+            try:
+                cleanup_task.result()
+            except Exception:
+                logger.warning("Failed to abort a cancelled upload staging file", exc_info=True)
+        raise
+
+
 def _make_uploaded_paths_sandbox_readable(paths: list[os.PathLike[str] | str]) -> None:
     for file_path in paths:
         _make_file_sandbox_readable(file_path)
@@ -253,11 +270,12 @@ def _sync_upload_to_sandbox(
     sandbox,
     file_path: os.PathLike[str] | str,
     virtual_path: str,
-    synced_sandbox_paths: list[str],
+    attempted_sandbox_paths: list[str],
 ) -> None:
     _make_file_sandbox_writable(file_path)
-    sandbox.update_file(virtual_path, Path(file_path).read_bytes())
-    synced_sandbox_paths.append(virtual_path)
+    data = Path(file_path).read_bytes()
+    attempted_sandbox_paths.append(virtual_path)
+    sandbox.update_file(virtual_path, data)
 
 
 def _list_uploaded_files_for_thread(thread_id: str, user_id: str) -> dict:
@@ -288,7 +306,7 @@ async def _write_upload_file_with_limits(
     file_size = 0
     upload_temp: StagedUpload | None = None
     try:
-        upload_temp = await run_file_io(create_upload_staging_file, Path(uploads_dir))
+        upload_temp = await _create_upload_staging_file_cancellation_safe(Path(uploads_dir))
         while chunk := await file.read(UPLOAD_CHUNK_SIZE):
             file_size += len(chunk)
             total_size += len(chunk)
@@ -348,10 +366,10 @@ async def upload_files(
     publications: list[PublishedUpload] = []
     generated_paths: list[Path] = []
     sandbox_sync_targets = []
-    synced_sandbox_paths: list[str] = []
+    attempted_sandbox_paths: list[str] = []
     skipped_files = []
     total_size = 0
-    sandbox_provider = get_sandbox_provider()
+    sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
     sync_to_sandbox = not _uses_thread_data_mounts(sandbox_provider)
     sandbox = None
     if sync_to_sandbox:
@@ -435,7 +453,7 @@ async def upload_files(
                     sandbox,
                     file_path,
                     virtual_path,
-                    synced_sandbox_paths,
+                    attempted_sandbox_paths,
                 )
 
         message = f"Successfully uploaded {len(uploaded_files)} file(s)"
@@ -452,7 +470,7 @@ async def upload_files(
         await _run_file_io_cancellation_safe(
             _rollback_upload_request,
             sandbox,
-            synced_sandbox_paths,
+            attempted_sandbox_paths,
             publications,
             generated_paths,
         )
@@ -462,7 +480,7 @@ async def upload_files(
         await _run_file_io_cancellation_safe(
             _rollback_upload_request,
             sandbox,
-            synced_sandbox_paths,
+            attempted_sandbox_paths,
             publications,
             generated_paths,
         )
@@ -471,7 +489,7 @@ async def upload_files(
         await _run_file_io_cancellation_safe(
             _rollback_upload_request,
             sandbox,
-            synced_sandbox_paths,
+            attempted_sandbox_paths,
             publications,
             generated_paths,
         )

backend/docs/API.md

@@ -636,7 +636,7 @@ Content-Type: multipart/form-data
 - Excel (`.xls`, `.xlsx`)
 - Word (`.doc`, `.docx`)
 
-All upload entry points publish complete payloads without replacing an existing name. Concurrent collisions are returned as `document.pdf`, `document_1.pdf`, `document_2.pdf`, and so on. A published filename remains leased through conversion, permission adjustment, sandbox synchronization, and response construction; deletion of that exact filename waits for the active lifecycle, while other filenames remain independent. Mounted providers make the exact published paths sandbox-readable; non-mounted providers receive exact private copies for Gateway, embedded-client, and IM-channel ingresses. If a non-mounted sandbox update later fails or the request is cancelled, DeerFlow removes only the remote paths created by that request before rolling back its host generations. Basenames matching the internal `.upload-*.part` staging pattern, containing `<` or `>`, or containing the reserved model-context boundary markers are rejected so every accepted model-visible filename and path can be rendered losslessly.
+All upload entry points publish complete payloads without replacing an existing name. Concurrent collisions are returned as `document.pdf`, `document_1.pdf`, `document_2.pdf`, and so on. A published filename remains leased through conversion, permission adjustment, sandbox synchronization, and response construction; deletion of that exact filename waits for the active lifecycle, while other filenames remain independent. Mounted providers make the exact published paths sandbox-readable; non-mounted providers receive exact private copies for Gateway, embedded-client, and IM-channel ingresses. If a non-mounted sandbox update later fails or the request is cancelled, DeerFlow removes every exact remote path attempted by that request before rolling back its host generations. Gateway cancellation also drains and aborts an in-flight staging creation. Basenames matching the internal `.upload-*.part` staging pattern, containing NUL, `<`, or `>`, or containing the reserved model-context boundary markers are rejected before staging so every accepted model-visible filename and path can be rendered losslessly. Embedded multi-file calls are request-atomic: a later failure rolls back every earlier host and remote generation in that call.
 
 Generated Markdown is stored outside the primary namespace and is not returned by the list endpoint. Normal conversion names are `<actual-primary-filename>.md`; if that component would exceed 255 UTF-8 bytes, the response contains a deterministic UTF-8-safe prefix plus the full SHA-256 digest and `.md`. Clients must consume the returned `markdown_*` fields rather than derive the path. Mounted AIO sandboxes use a read-only conversion mount, while Local structured file APIs reject writes through a read-only path mapping; Local host bash is outside that boundary. Non-mounted providers receive a private synchronized copy rather than the authoritative host namespace. Deleting `document.pdf` also deletes only its exact generated conversion; an independent `uploads/document.md` is preserved.
 

backend/docs/FILE_UPLOAD.md

@@ -52,7 +52,7 @@ POST /api/threads/{thread_id}/uploads
 - `virtual_path`: Agent 在沙箱中使用的虚拟路径
 - `artifact_url`: 前端通过 HTTP 访问文件的 URL
 
-所有上传入口都先完整写入同目录暂存文件，再以“不替换已有条目”的原子操作发布。同名碰撞依次命名为 `document.pdf`、`document_1.pdf`、`document_2.pdf`；响应中的 `filename` 和各路径字段始终使用实际发布名。系统内部保留 `.upload-*.part` 作为暂存命名空间；使用该模式的 basename、包含 `<` 或 `>`、或包含保留模型上下文边界标记的文件名会在创建暂存文件前被拒绝，以保证所有已接受的文件名和 Agent 可见路径都能无损呈现。
+所有上传入口都先完整写入同目录暂存文件，再以“不替换已有条目”的原子操作发布。同名碰撞依次命名为 `document.pdf`、`document_1.pdf`、`document_2.pdf`；响应中的 `filename` 和各路径字段始终使用实际发布名。系统内部保留 `.upload-*.part` 作为暂存命名空间；使用该模式的 basename、包含 NUL、`<` 或 `>`、或包含保留模型上下文边界标记的文件名会在创建暂存文件前被拒绝，以保证所有已接受的文件名和 Agent 可见路径都能无损呈现。若请求在 staging 创建尚未返回时被取消，Gateway 会等待创建结束并精确 abort 该临时文件。
 
 实际发布名会在转换、权限调整、沙箱同步和响应构造期间持有同名租约。删除该名称会等待当前生命周期完成；其他文件名仍可并发处理。跨进程协调使用 `.upload-conversions/.locks/` 下稳定保留的摘要锁文件，该目录属于内部实现，不应由 Agent 或部署脚本修改或清理。
 
@@ -183,7 +183,8 @@ read_file(path="/mnt/user-data/.upload-conversions/document.pdf.md")
 - 本地沙箱（`sandbox_id=local`）直接使用线程目录内容
 - AIO 挂载模式把 `/mnt/user-data/.upload-conversions` 单独挂载为只读；Local 的结构化文件 API 通过更具体的只读路径映射执行同一规则，但 Local 宿主机 bash 不属于该边界
 - Gateway、嵌入式 `DeerFlowClient` 和 IM 通道都会执行同一沙箱可见性步骤：挂载型 provider 调整精确发布路径的读取权限；非挂载 provider 获取沙箱后，把本次主文件及生成转换件精确同步到各自虚拟路径
-- 非挂载同步副本是沙箱私有副本；任一路径失败、后续响应构造失败或请求取消时，会精确撤销本次创建的远端路径，再回滚宿主文件
+- 非挂载同步副本是沙箱私有副本；任一路径失败（包括远端已落盘但传输随后报错）、后续响应构造失败或请求取消时，会对本次尝试的精确远端路径执行幂等撤销，再回滚宿主文件
+- 嵌入式 `DeerFlowClient.upload_files()` 以整批为事务边界：后续文件失败会逆序撤销本次调用中此前成功的所有远端副本和宿主 generation
 - 如果 Gateway 与远端沙箱保证挂载同一份线程 user-data（例如正确对齐的共享 PVC、NFS 或 hostPath），可设置 `sandbox.thread_data_mounts: true`；上传路由会跳过 sandbox acquire 和逐文件同步
 - 不确定挂载关系时应省略该配置并保留自动检测。错误地设为 `true` 会导致文件只存在于 Gateway 存储、沙箱内不可见
 

backend/packages/harness/deerflow/client.py

@@ -1499,7 +1499,7 @@ class DeerFlowClient:
         """
         validate_thread_id(thread_id)
         from deerflow.sandbox.sandbox_provider import get_sandbox_provider
-        from deerflow.uploads.sandbox_sync import make_upload_paths_available, rollback_sandbox_sync
+        from deerflow.uploads.sandbox_sync import SandboxSyncReceipt, make_upload_paths_available, rollback_sandbox_sync
         from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS
 
         # Validate all files upfront to avoid partial uploads.
@@ -1519,6 +1519,8 @@ class DeerFlowClient:
         sandbox_provider = get_sandbox_provider(app_config=self._app_config)
         effective_user_id = get_effective_user_id()
         uploaded_files: list[dict] = []
+        publications: list[PublishedUpload] = []
+        sandbox_receipts: list[SandboxSyncReceipt] = []
 
         conversion_pool = None
         if has_convertible_file:
@@ -1544,82 +1546,84 @@ class DeerFlowClient:
         try:
             for src_path in resolved_files:
                 publication = publish_upload_copy_leased(uploads_dir, src_path.name, src_path)
-                sandbox_receipt = None
-                try:
-                    dest = publication.path
-                    dest_name = dest.name
+                publications.append(publication)
+                dest = publication.path
+                dest_name = dest.name
 
-                    info: dict[str, Any] = {
-                        "filename": dest_name,
-                        "size": dest.stat().st_size,
-                        "path": str(dest),
-                        "virtual_path": upload_virtual_path(dest_name),
-                        "artifact_url": upload_artifact_url(thread_id, dest_name),
-                    }
-                    if dest_name != src_path.name:
-                        info["original_filename"] = src_path.name
+                info: dict[str, Any] = {
+                    "filename": dest_name,
+                    "size": dest.stat().st_size,
+                    "path": str(dest),
+                    "virtual_path": upload_virtual_path(dest_name),
+                    "artifact_url": upload_artifact_url(thread_id, dest_name),
+                }
+                if dest_name != src_path.name:
+                    info["original_filename"] = src_path.name
 
-                    md_path = None
-                    if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS:
-                        try:
-                            if conversion_pool is not None:
-                                md_path = conversion_pool.submit(_convert_in_thread, publication).result()
-                            else:
-                                md_path = asyncio.run(
-                                    convert_uploaded_file_to_markdown(
-                                        dest,
-                                        publication=publication,
-                                    )
+                md_path = None
+                if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS:
+                    try:
+                        if conversion_pool is not None:
+                            md_path = conversion_pool.submit(_convert_in_thread, publication).result()
+                        else:
+                            md_path = asyncio.run(
+                                convert_uploaded_file_to_markdown(
+                                    dest,
+                                    publication=publication,
                                 )
-                        except Exception:
-                            logger.warning(
-                                "Failed to convert %s to markdown",
-                                src_path.name,
-                                exc_info=True,
                             )
-                            md_path = None
+                    except Exception:
+                        logger.warning(
+                            "Failed to convert %s to markdown",
+                            src_path.name,
+                            exc_info=True,
+                        )
+                        md_path = None
 
-                        if md_path is not None:
-                            md_virtual_path = conversion_virtual_path(dest_name)
-                            info["markdown_file"] = md_path.name
-                            info["markdown_path"] = str(md_path)
-                            info["markdown_virtual_path"] = md_virtual_path
-                            info["markdown_artifact_url"] = artifact_url_for_virtual_path(thread_id, md_virtual_path)
-
-                    sandbox_paths = [(dest, upload_virtual_path(dest_name))]
                     if md_path is not None:
-                        sandbox_paths.append((md_path, conversion_virtual_path(dest_name)))
-                    sandbox_receipt = make_upload_paths_available(
+                        md_virtual_path = conversion_virtual_path(dest_name)
+                        info["markdown_file"] = md_path.name
+                        info["markdown_path"] = str(md_path)
+                        info["markdown_virtual_path"] = md_virtual_path
+                        info["markdown_artifact_url"] = artifact_url_for_virtual_path(thread_id, md_virtual_path)
+
+                sandbox_paths = [(dest, upload_virtual_path(dest_name))]
+                if md_path is not None:
+                    sandbox_paths.append((md_path, conversion_virtual_path(dest_name)))
+                sandbox_receipts.append(
+                    make_upload_paths_available(
                         sandbox_provider,
                         thread_id,
                         user_id=effective_user_id,
                         paths=sandbox_paths,
                     )
+                )
 
-                    uploaded_files.append(info)
+                uploaded_files.append(info)
+
+            return {
+                "success": True,
+                "files": uploaded_files,
+                "message": f"Successfully uploaded {len(uploaded_files)} file(s)",
+            }
+        except BaseException:
+            for receipt in reversed(sandbox_receipts):
+                try:
+                    rollback_sandbox_sync(receipt)
                 except BaseException:
-                    if sandbox_receipt is not None:
-                        try:
-                            rollback_sandbox_sync(sandbox_receipt)
-                        except BaseException:
-                            logger.warning("Failed to roll back rejected embedded sandbox upload: %s", publication.path, exc_info=True)
-                    try:
-                        rollback_published_upload(publication)
-                    except Exception:
-                        logger.warning("Failed to roll back rejected embedded upload: %s", publication.path, exc_info=True)
-                    raise
-                finally:
-                    publication.release()
+                    logger.warning("Failed to roll back rejected embedded sandbox batch", exc_info=True)
+            for publication in reversed(publications):
+                try:
+                    rollback_published_upload(publication)
+                except BaseException:
+                    logger.warning("Failed to roll back rejected embedded upload: %s", publication.path, exc_info=True)
+            raise
         finally:
+            for publication in reversed(publications):
+                publication.release()
             if conversion_pool is not None:
                 conversion_pool.shutdown(wait=True)
 
-        return {
-            "success": True,
-            "files": uploaded_files,
-            "message": f"Successfully uploaded {len(uploaded_files)} file(s)",
-        }
-
     def list_uploads(self, thread_id: str) -> dict:
         """List files in a thread's uploads directory.
 

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

@@ -90,6 +90,8 @@ def normalize_filename(filename: str) -> str:
     """
     if not filename:
         raise ValueError("Filename is empty")
+    if "\0" in filename:
+        raise ValueError(f"Filename contains NUL byte: {filename!r}")
     safe = Path(filename).name
     if not safe or safe in {".", ".."}:
         raise ValueError(f"Filename is unsafe: {filename!r}")

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

@@ -50,6 +50,16 @@ def rollback_sandbox_sync(receipt: SandboxSyncReceipt) -> None:
         _remove_remote_paths(receipt.sandbox, receipt.virtual_paths)
 
 
+async def rollback_sandbox_sync_async(receipt: SandboxSyncReceipt) -> None:
+    """Cancellation-safely remove every remote path recorded by *receipt*."""
+    cleanup_task = asyncio.create_task(
+        asyncio.to_thread(rollback_sandbox_sync, receipt),
+        name="rollback-sandbox-upload-paths",
+    )
+    await wait_for_task_completion(cleanup_task)
+    cleanup_task.result()
+
+
 def _sync_remote_paths(sandbox: Any, paths: tuple[tuple[Path, str], ...]) -> SandboxSyncReceipt:
     attempted: list[str] = []
     completed: list[str] = []
@@ -112,13 +122,8 @@ async def make_upload_paths_available_async(
     cancelled = await wait_for_task_completion(sync_task)
     receipt = sync_task.result()
     if cancelled:
-        cleanup_task = asyncio.create_task(
-            asyncio.to_thread(rollback_sandbox_sync, receipt),
-            name=f"rollback-sandbox-upload-paths:{thread_id}",
-        )
-        await wait_for_task_completion(cleanup_task)
         try:
-            cleanup_task.result()
+            await rollback_sandbox_sync_async(receipt)
         except BaseException:
             logger.warning("Failed to roll back sandbox uploads after cancellation", exc_info=True)
         raise asyncio.CancelledError

backend/tests/blocking_io/test_dingtalk_receive_file.py

@@ -65,6 +65,30 @@ async def test_receive_file_persist_does_not_block_event_loop(tmp_path, monkeypa
     assert (await asyncio.to_thread(uploaded.stat)).st_mode & 0o044 == 0o044
 
 
+async def test_receive_file_initializes_provider_off_event_loop(tmp_path, monkeypatch) -> None:
+    from app.channels.dingtalk import DingTalkChannel
+    from app.channels.message_bus import MessageBus
+    from deerflow.config.paths import Paths
+
+    paths = await asyncio.to_thread(Paths, str(tmp_path))
+    monkeypatch.setattr("app.channels.dingtalk.get_paths", lambda: paths)
+    event_loop_thread = threading.get_ident()
+    provider_threads: list[int] = []
+
+    def get_provider():
+        provider_threads.append(threading.get_ident())
+        return SimpleNamespace(uses_thread_data_mounts=True)
+
+    monkeypatch.setattr("app.channels.dingtalk.get_sandbox_provider", get_provider)
+    channel = DingTalkChannel(MessageBus(), config={})
+    channel._download_by_code = AsyncMock(return_value=b"DATA")
+
+    result = await channel._receive_single_file("dc", "file", "a.pdf", "t1", user_id="default")
+
+    assert result == "/mnt/user-data/uploads/a.pdf"
+    assert provider_threads and provider_threads[0] != event_loop_thread
+
+
 async def test_receive_file_rejects_symlinked_upload_directory(tmp_path, monkeypatch) -> None:
     from app.channels.dingtalk import DingTalkChannel
     from app.channels.message_bus import MessageBus

backend/tests/blocking_io/test_uploads_router.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 from io import BytesIO
 from pathlib import Path
 from types import SimpleNamespace
@@ -92,6 +93,29 @@ async def test_upload_endpoint_mounted_provider_does_not_block_event_loop(tmp_pa
     assert await asyncio.to_thread(target.read_bytes) == b"hello uploads"
 
 
+async def test_upload_endpoint_initializes_provider_off_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _reset_paths(tmp_path, monkeypatch)
+    event_loop_thread = threading.get_ident()
+    provider_threads: list[int] = []
+
+    def get_provider():
+        provider_threads.append(threading.get_ident())
+        return _MountedProvider()
+
+    monkeypatch.setattr(uploads, "get_sandbox_provider", get_provider)
+
+    result = await call_unwrapped(
+        uploads.upload_files,
+        "t-provider-init",
+        request=None,
+        files=[UploadFile(filename="notes.txt", file=BytesIO(b"hello"))],
+        config=SimpleNamespace(),
+    )
+
+    assert result.success is True
+    assert provider_threads and provider_threads[0] != event_loop_thread
+
+
 async def test_upload_endpoint_remote_provider_syncs_without_blocking_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     _reset_paths(tmp_path, monkeypatch)
     provider = _RemoteProvider()

backend/tests/test_channel_file_attachments.py

@@ -341,6 +341,80 @@ class TestInboundFileIngestion:
         assert updates == [(result[0]["path"], b"pdf bytes")]
         assert removals == []
 
+    def test_builds_inbound_metadata_while_name_lease_is_active(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        msg = InboundMessage(
+            channel_name="telegram",
+            chat_id="chat-1",
+            user_id="user-1",
+            text="see attachment",
+            files=[{"type": "file", "filename": "report.pdf", "_content": b"pdf bytes"}],
+        )
+
+        def build_metadata(publication, virtual_path, data, file_type):
+            assert publication.is_active
+            return {
+                "filename": publication.path.name,
+                "size": len(data),
+                "path": virtual_path,
+                "is_image": file_type == "image",
+            }
+
+        with (
+            patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir),
+            patch.object(manager, "_build_inbound_file_metadata", side_effect=build_metadata),
+        ):
+            result = _run(manager._ingest_inbound_files("thread-1", msg))
+
+        assert result[0]["filename"] == "report.pdf"
+
+    def test_metadata_failure_rolls_back_remote_and_host_publication(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        remote_files: dict[str, bytes] = {}
+
+        class Sandbox:
+            def update_file(self, path, content):
+                remote_files[path] = content
+
+            def remove_file(self, path):
+                remote_files.pop(path, None)
+
+        sandbox = Sandbox()
+
+        class Provider:
+            uses_thread_data_mounts = False
+
+            async def acquire_async(self, thread_id, user_id=None):
+                return "remote"
+
+            def get(self, sandbox_id):
+                return sandbox if sandbox_id == "remote" else None
+
+        msg = InboundMessage(
+            channel_name="telegram",
+            chat_id="chat-1",
+            user_id="user-1",
+            text="see attachment",
+            files=[{"type": "file", "filename": "report.pdf", "_content": b"pdf bytes"}],
+        )
+
+        with (
+            patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=Provider()),
+            patch.object(manager, "_build_inbound_file_metadata", side_effect=RuntimeError("metadata failed")),
+        ):
+            result = _run(manager._ingest_inbound_files("thread-1", msg))
+
+        assert result == []
+        assert list(uploads_dir.iterdir()) == []
+        assert remote_files == {}
+
     def test_cancellation_rolls_back_generic_remote_sync_and_host_publication(self, tmp_path):
         from app.channels import manager
 

backend/tests/test_client.py

@@ -2231,6 +2231,45 @@ class TestUploads:
         ]
         assert removals == []
 
+    def test_upload_files_rolls_back_the_whole_batch_when_later_sync_fails(self, client, tmp_path):
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        first = tmp_path / "first.txt"
+        second = tmp_path / "second.txt"
+        first.write_bytes(b"first")
+        second.write_bytes(b"second")
+        remote_files: dict[str, bytes] = {}
+
+        class Sandbox:
+            def update_file(self, path, content):
+                remote_files[path] = content
+                if path.endswith("second.txt"):
+                    raise RuntimeError("second sync failed after commit")
+
+            def remove_file(self, path):
+                remote_files.pop(path, None)
+
+        sandbox = Sandbox()
+
+        class Provider:
+            uses_thread_data_mounts = False
+
+            def acquire(self, thread_id, user_id=None):
+                return "remote"
+
+            def get(self, sandbox_id):
+                return sandbox if sandbox_id == "remote" else None
+
+        with (
+            patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=Provider()),
+        ):
+            with pytest.raises(RuntimeError, match="second sync failed"):
+                client.upload_files("thread-1", [first, second])
+
+        assert list(uploads_dir.iterdir()) == []
+        assert remote_files == {}
+
     def test_upload_files_across_calls_never_overwrite(self, client, tmp_path):
         uploads_dir = tmp_path / "user-data" / "uploads"
         uploads_dir.mkdir(parents=True)

backend/tests/test_uploads_manager.py

@@ -120,6 +120,10 @@ class TestNormalizeFilename:
         with pytest.raises(ValueError, match="reserved model-context token"):
             normalize_filename(filename)
 
+    def test_rejects_nul_before_any_filesystem_operation(self):
+        with pytest.raises(ValueError, match="NUL"):
+            normalize_filename("bad\0name.pdf")
+
 
 # ---------------------------------------------------------------------------
 # claim_unique_filename

backend/tests/test_uploads_router.py

@@ -177,7 +177,7 @@ def test_sandbox_sync_failure_rolls_back_published_generation(tmp_path):
     assert not (thread_uploads_dir / "notes.txt").exists()
 
 
-def test_partial_sandbox_sync_failure_removes_only_completed_remote_paths(tmp_path):
+def test_partial_sandbox_sync_failure_removes_all_attempted_remote_paths(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)
     provider = MagicMock()
@@ -214,11 +214,90 @@ def test_partial_sandbox_sync_failure_removes_only_completed_remote_paths(tmp_pa
 
     assert exc_info.value.status_code == 500
     assert synced_paths == ["/mnt/user-data/uploads/first.txt"]
-    sandbox.remove_file.assert_called_once_with("/mnt/user-data/uploads/first.txt")
+    assert [record.args[0] for record in sandbox.remove_file.call_args_list] == [
+        "/mnt/user-data/uploads/second.txt",
+        "/mnt/user-data/uploads/first.txt",
+    ]
     assert not (thread_uploads_dir / "first.txt").exists()
     assert not (thread_uploads_dir / "second.txt").exists()
 
 
+def test_post_commit_sandbox_error_removes_attempted_remote_path(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = False
+    provider.acquire_async = AsyncMock(return_value="remote-1")
+    remote_files: dict[str, bytes] = {}
+    removed_paths: list[str] = []
+
+    class Sandbox:
+        def update_file(self, virtual_path: str, data: bytes) -> None:
+            remote_files[virtual_path] = data
+            raise RuntimeError("transport failed after commit")
+
+        def remove_file(self, virtual_path: str) -> None:
+            removed_paths.append(virtual_path)
+            remote_files.pop(virtual_path, None)
+
+    provider.get.return_value = Sandbox()
+
+    with (
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=provider),
+    ):
+        with pytest.raises(HTTPException):
+            asyncio.run(
+                call_unwrapped(
+                    uploads.upload_files,
+                    "thread-remote",
+                    request=MagicMock(),
+                    files=[UploadFile(filename="notes.txt", file=BytesIO(b"payload"))],
+                    config=SimpleNamespace(),
+                )
+            )
+
+    assert remote_files == {}
+    assert removed_paths == ["/mnt/user-data/uploads/notes.txt"]
+    assert not (thread_uploads_dir / "notes.txt").exists()
+
+
+@pytest.mark.asyncio
+async def test_cancellation_during_staging_creation_drains_and_aborts(tmp_path):
+    staging_created = threading.Event()
+    allow_create_to_return = threading.Event()
+    real_create = create_upload_staging_file
+
+    def paused_create(base_dir: Path):
+        staged = real_create(base_dir)
+        staging_created.set()
+        assert allow_create_to_return.wait(5)
+        return staged
+
+    with patch.object(uploads, "create_upload_staging_file", side_effect=paused_create):
+        task = asyncio.create_task(
+            uploads._write_upload_file_with_limits(
+                UploadFile(filename="notes.txt", file=BytesIO(b"payload")),
+                uploads_dir=tmp_path,
+                display_filename="notes.txt",
+                max_single_file_size=1024,
+                max_total_size=1024,
+                total_size=0,
+            )
+        )
+        assert await asyncio.to_thread(staging_created.wait, 5)
+        try:
+            task.cancel()
+            await asyncio.sleep(0.05)
+            assert not task.done()
+        finally:
+            allow_create_to_return.set()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    assert list(tmp_path.glob(".upload-*.part")) == []
+
+
 @pytest.mark.asyncio
 async def test_cancellation_after_remote_sync_still_removes_the_completed_copy(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"