fix: synchronize every upload ingress

backend/AGENTS.md

@@ -1364,7 +1364,7 @@ Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and sk
 | Uploads | `upload_files(thread_id, files)`, `list_uploads(thread_id)`, `delete_upload(thread_id, filename)` | `{"success": true, "files": [...]}`, `{"files": [...], "count": N}` |
 | Artifacts | `get_artifact(thread_id, path)` → `(bytes, mime_type)` | tuple |
 
-**Key difference from Gateway**: Upload accepts local `Path` objects instead of HTTP `UploadFile`, rejects directory paths before copying, and reuses a single worker when document conversion must run inside an active event loop. Artifact returns `(bytes, mime_type)` instead of HTTP Response. The new Gateway-only thread cleanup route deletes `.deer-flow/threads/{thread_id}` after LangGraph thread deletion; there is no matching `DeerFlowClient` method yet. `update_mcp_config()` and `update_skill()` automatically invalidate the cached agent.
+**Key difference from Gateway**: Upload accepts local `Path` objects instead of HTTP `UploadFile`, rejects directory paths before copying, and reuses a single worker when document conversion must run inside an active event loop. It still applies the shared provider-aware publication step: mounted providers make the exact paths readable, while non-mounted providers receive private primary/conversion copies and exact-path rollback. Artifact returns `(bytes, mime_type)` instead of HTTP Response. The new Gateway-only thread cleanup route deletes `.deer-flow/threads/{thread_id}` after LangGraph thread deletion; there is no matching `DeerFlowClient` method yet. `update_mcp_config()` and `update_skill()` automatically invalidate the cached agent.
 
 **Tests**: `tests/test_client.py` (offline unit tests including
 `TestGatewayConformance`), `tests/test_client_live.py` (live integration tests,
@@ -1461,9 +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.
 - 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. Non-mounted sandbox uploads acquire sandboxes with `SandboxProvider.acquire_async()` and offload `read_bytes()` plus `sandbox.update_file()` together. The route records each completed remote path and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those exact 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. 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.
 - 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

@@ -23,10 +23,11 @@ from deerflow.sandbox.sandbox_provider import get_sandbox_provider
 from deerflow.uploads.async_helpers import (
     publish_upload_bytes_leased_async,
     release_published_upload_async,
-    run_upload_io_cancellation_safe,
+    rollback_published_upload_async,
 )
 from deerflow.uploads.layout import upload_virtual_path
-from deerflow.uploads.manager import UnsafeUploadPathError, make_upload_file_sandbox_readable, normalize_filename
+from deerflow.uploads.manager import UnsafeUploadPathError, normalize_filename
+from deerflow.uploads.sandbox_sync import make_upload_paths_available_async
 
 logger = logging.getLogger(__name__)
 
@@ -650,19 +651,23 @@ class DingTalkChannel(Channel):
 
             try:
                 sandbox_provider = get_sandbox_provider()
-                if getattr(sandbox_provider, "uses_thread_data_mounts", False):
-                    await run_upload_io_cancellation_safe(make_upload_file_sandbox_readable, publication.path)
-                else:
-                    # acquire_async keeps provider lifecycle work (Docker discovery,
-                    # readiness polls) off the event loop; update_file is blocking
-                    # transport IO on remote sandboxes, so it is offloaded too.
-                    sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-                    sandbox = sandbox_provider.get(sandbox_id)
-                    if sandbox is None:
-                        logger.warning("[DingTalk] sandbox %s not found after acquire, dropping attachment: %s", sandbox_id, virtual_path)
-                        return ""
-                    await run_upload_io_cancellation_safe(sandbox.update_file, virtual_path, content)
+                await make_upload_paths_available_async(
+                    sandbox_provider,
+                    thread_id,
+                    user_id=effective_user_id,
+                    paths=[(publication.path, virtual_path)],
+                )
+            except asyncio.CancelledError:
+                try:
+                    await rollback_published_upload_async(publication)
+                except Exception:
+                    logger.warning("[DingTalk] failed to roll back cancelled attachment: %s", virtual_path, exc_info=True)
+                raise
             except Exception:
+                try:
+                    await rollback_published_upload_async(publication)
+                except Exception:
+                    logger.warning("[DingTalk] failed to roll back rejected attachment: %s", virtual_path, exc_info=True)
                 logger.exception("[DingTalk] failed to sync downloaded file into non-local sandbox: %s", virtual_path)
                 return ""
 

backend/app/channels/feishu.py

@@ -28,10 +28,10 @@ from deerflow.sandbox.sandbox_provider import get_sandbox_provider
 from deerflow.uploads.async_helpers import (
     publish_upload_bytes_leased_async,
     release_published_upload_async,
-    run_upload_io_cancellation_safe,
+    rollback_published_upload_async,
 )
 from deerflow.uploads.layout import upload_virtual_path
-from deerflow.uploads.manager import make_upload_file_sandbox_readable
+from deerflow.uploads.sandbox_sync import make_upload_paths_available_async
 
 logger = logging.getLogger(__name__)
 PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60
@@ -467,16 +467,23 @@ class FeishuChannel(Channel):
 
             try:
                 sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
-                if getattr(sandbox_provider, "uses_thread_data_mounts", False):
-                    await run_upload_io_cancellation_safe(make_upload_file_sandbox_readable, publication.path)
-                else:
-                    sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-                    sandbox = sandbox_provider.get(sandbox_id)
-                    if sandbox is None:
-                        logger.warning("[Feishu] sandbox not found for thread_id=%s", thread_id)
-                        return f"Failed to obtain the [{type}]"
-                    await run_upload_io_cancellation_safe(sandbox.update_file, virtual_path, content)
+                await make_upload_paths_available_async(
+                    sandbox_provider,
+                    thread_id,
+                    user_id=effective_user_id,
+                    paths=[(publication.path, virtual_path)],
+                )
+            except asyncio.CancelledError:
+                try:
+                    await rollback_published_upload_async(publication)
+                except Exception:
+                    logger.warning("[Feishu] failed to roll back cancelled resource: %s", virtual_path, exc_info=True)
+                raise
             except Exception:
+                try:
+                    await rollback_published_upload_async(publication)
+                except Exception:
+                    logger.warning("[Feishu] failed to roll back rejected resource: %s", virtual_path, exc_info=True)
                 logger.exception("[Feishu] failed to sync resource into non-local sandbox: %s", virtual_path)
                 return f"Failed to obtain the [{type}]"
 

backend/app/channels/manager.py

@@ -882,13 +882,19 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
     if not msg.files:
         return []
 
+    from deerflow.sandbox.sandbox_provider import get_sandbox_provider
+    from deerflow.uploads.async_helpers import (
+        publish_upload_bytes_leased_async,
+        release_published_upload_async,
+        rollback_published_upload_async,
+    )
     from deerflow.uploads.layout import upload_virtual_path
     from deerflow.uploads.manager import (
         UnsafeUploadPathError,
         ensure_uploads_dir,
         normalize_filename,
-        publish_upload_bytes,
     )
+    from deerflow.uploads.sandbox_sync import make_upload_paths_available_async
 
     def _prepare_uploads_dir() -> Path:
         # Worker thread: directory creation is blocking filesystem IO that must
@@ -947,21 +953,51 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id:
                 )
                 continue
 
+            publication = None
             try:
-                dest = await asyncio.to_thread(publish_upload_bytes, uploads_dir, safe_name, data)
+                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_provider,
+                    thread_id,
+                    user_id=user_id,
+                    paths=[(dest, virtual_path)],
+                )
+            except asyncio.CancelledError:
+                if publication is not None:
+                    try:
+                        await rollback_published_upload_async(publication)
+                    except Exception:
+                        logger.warning("[Manager] failed to roll back cancelled inbound file: %s", safe_name, exc_info=True)
+                raise
             except UnsafeUploadPathError:
+                if publication is not None:
+                    try:
+                        await rollback_published_upload_async(publication)
+                    except Exception:
+                        logger.warning("[Manager] failed to roll back unsafe inbound file: %s", safe_name, exc_info=True)
                 logger.warning("[Manager] skipping inbound file with unsafe destination: %s", safe_name)
                 continue
             except Exception:
+                if publication is not None:
+                    try:
+                        await rollback_published_upload_async(publication)
+                    except Exception:
+                        logger.warning("[Manager] failed to roll back rejected inbound file: %s", safe_name, exc_info=True)
                 logger.exception("[Manager] failed to write inbound file: %s", safe_name)
                 continue
+            finally:
+                if publication is not None:
+                    await release_published_upload_async(publication)
 
             actual_name = dest.name
             created.append(
                 {
                     "filename": actual_name,
                     "size": len(data),
-                    "path": upload_virtual_path(actual_name),
+                    "path": virtual_path,
                     "is_image": ftype == "image",
                 }
             )

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. If a non-mounted sandbox update later fails or the request is cancelled, DeerFlow removes only the remote paths already completed by that request before rolling back its host generations. Basenames matching the internal `.upload-*.part` staging pattern are rejected.
+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.
 
 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 会在创建暂存文件前被拒绝。
+所有上传入口都先完整写入同目录暂存文件，再以“不替换已有条目”的原子操作发布。同名碰撞依次命名为 `document.pdf`、`document_1.pdf`、`document_2.pdf`；响应中的 `filename` 和各路径字段始终使用实际发布名。系统内部保留 `.upload-*.part` 作为暂存命名空间；使用该模式的 basename、包含 `<` 或 `>`、或包含保留模型上下文边界标记的文件名会在创建暂存文件前被拒绝，以保证所有已接受的文件名和 Agent 可见路径都能无损呈现。
 
 实际发布名会在转换、权限调整、沙箱同步和响应构造期间持有同名租约。删除该名称会等待当前生命周期完成；其他文件名仍可并发处理。跨进程协调使用 `.upload-conversions/.locks/` 下稳定保留的摘要锁文件，该目录属于内部实现，不应由 Agent 或部署脚本修改或清理。
 
@@ -182,7 +182,8 @@ read_file(path="/mnt/user-data/.upload-conversions/document.pdf.md")
 - 先写入 `backend/.deer-flow/threads/{thread_id}/user-data/uploads/` 作为权威存储
 - 本地沙箱（`sandbox_id=local`）直接使用线程目录内容
 - AIO 挂载模式把 `/mnt/user-data/.upload-conversions` 单独挂载为只读；Local 的结构化文件 API 通过更具体的只读路径映射执行同一规则，但 Local 宿主机 bash 不属于该边界
-- 默认情况下，非本地沙箱通过 `acquire_async` 获取后，再额外同步到 `/mnt/user-data/uploads/*`，确保运行时可见；同步副本是沙箱私有副本，失败时只回滚本次已完成同步的精确路径
+- Gateway、嵌入式 `DeerFlowClient` 和 IM 通道都会执行同一沙箱可见性步骤：挂载型 provider 调整精确发布路径的读取权限；非挂载 provider 获取沙箱后，把本次主文件及生成转换件精确同步到各自虚拟路径
+- 非挂载同步副本是沙箱私有副本；任一路径失败、后续响应构造失败或请求取消时，会精确撤销本次创建的远端路径，再回滚宿主文件
 - 如果 Gateway 与远端沙箱保证挂载同一份线程 user-data（例如正确对齐的共享 PVC、NFS 或 hostPath），可设置 `sandbox.thread_data_mounts: true`；上传路由会跳过 sandbox acquire 和逐文件同步
 - 不确定挂载关系时应省略该配置并保留自动检测。错误地设为 `true` 会导致文件只存在于 Gateway 存储、沙箱内不可见
 

backend/packages/harness/deerflow/client.py

@@ -70,8 +70,8 @@ from deerflow.uploads.manager import (
     ensure_uploads_dir,
     get_uploads_dir,
     list_files_in_dir,
-    make_upload_file_sandbox_readable,
     publish_upload_copy_leased,
+    rollback_published_upload,
     upload_artifact_url,
     upload_virtual_path,
 )
@@ -1498,6 +1498,8 @@ class DeerFlowClient:
             ValueError: If any supplied path exists but is not a regular file.
         """
         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.utils.file_conversion import CONVERTIBLE_EXTENSIONS
 
         # Validate all files upfront to avoid partial uploads.
@@ -1514,6 +1516,8 @@ class DeerFlowClient:
                 has_convertible_file = True
 
         uploads_dir = ensure_uploads_dir(thread_id)
+        sandbox_provider = get_sandbox_provider(app_config=self._app_config)
+        effective_user_id = get_effective_user_id()
         uploaded_files: list[dict] = []
 
         conversion_pool = None
@@ -1540,6 +1544,7 @@ 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
@@ -1581,11 +1586,28 @@ class DeerFlowClient:
                             info["markdown_virtual_path"] = md_virtual_path
                             info["markdown_artifact_url"] = artifact_url_for_virtual_path(thread_id, md_virtual_path)
 
-                    make_upload_file_sandbox_readable(dest)
+                    sandbox_paths = [(dest, upload_virtual_path(dest_name))]
                     if md_path is not None:
-                        make_upload_file_sandbox_readable(md_path)
+                        sandbox_paths.append((md_path, conversion_virtual_path(dest_name)))
+                    sandbox_receipt = make_upload_paths_available(
+                        sandbox_provider,
+                        thread_id,
+                        user_id=effective_user_id,
+                        paths=sandbox_paths,
+                    )
 
                     uploaded_files.append(info)
+                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()
         finally:

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

@@ -76,11 +76,13 @@ _default_sandbox_provider: SandboxProvider | None = None
 _provider_lock = threading.Lock()
 
 
-def get_sandbox_provider(**kwargs) -> SandboxProvider:
+def get_sandbox_provider(*, app_config=None, **kwargs) -> SandboxProvider:
     """Get the sandbox provider singleton.
 
     Returns a cached singleton instance. Use `reset_sandbox_provider()` to clear
     the cache, or `shutdown_sandbox_provider()` to properly shutdown and clear.
+    Embedded callers may pass their already-resolved ``app_config`` so cold
+    singleton construction uses the same configuration as the caller.
 
     Returns:
         A sandbox provider instance.
@@ -95,7 +97,7 @@ def get_sandbox_provider(**kwargs) -> SandboxProvider:
     # Cold start. Resolve + construct outside the lock: the import and the
     # provider constructor are plugin code and must not run under a non-reentrant
     # lock. The construction may race another caller; we reconcile under the lock.
-    config = get_app_config()
+    config = app_config or get_app_config()
     cls = resolve_class(config.sandbox.use, SandboxProvider)
     provider = cls(**kwargs)
 

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

@@ -105,3 +105,13 @@ async def release_published_upload_async(publication: PublishedUpload) -> None:
     release_task.result()
     if cancelled:
         raise asyncio.CancelledError
+
+
+async def rollback_published_upload_async(publication: PublishedUpload) -> None:
+    """Roll back a publication off-thread, draining repeated cancellation."""
+    rollback_task = asyncio.create_task(
+        asyncio.to_thread(rollback_published_upload, publication),
+        name=f"rollback-upload:{publication.path.name}",
+    )
+    await wait_for_task_completion(rollback_task)
+    rollback_task.result()

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

@@ -97,6 +97,8 @@ def normalize_filename(filename: str) -> str:
     # but they indicate a Windows-style path that should be stripped or rejected.
     if "\\" in safe:
         raise ValueError(f"Filename contains backslash: {filename!r}")
+    if "<" in safe or ">" in safe or "--- BEGIN USER INPUT ---" in safe or "--- END USER INPUT ---" in safe:
+        raise ValueError(f"Filename contains reserved model-context token: {filename!r}")
     if len(safe.encode("utf-8")) > 255:
         raise ValueError(f"Filename too long: {len(safe)} chars")
     if is_upload_staging_file(safe):

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

@@ -0,0 +1,125 @@
+"""Provider-aware publication of authoritative upload paths to sandboxes."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from deerflow.uploads.async_helpers import run_upload_io_cancellation_safe, wait_for_task_completion
+from deerflow.uploads.manager import make_upload_file_sandbox_readable
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class SandboxSyncReceipt:
+    """Remote paths created by one completed sandbox synchronization."""
+
+    sandbox: Any | None
+    virtual_paths: tuple[str, ...] = ()
+
+
+def _make_paths_readable(paths: tuple[tuple[Path, str], ...]) -> None:
+    for physical_path, _ in paths:
+        make_upload_file_sandbox_readable(physical_path)
+
+
+def _remove_remote_paths(sandbox: Any, virtual_paths: tuple[str, ...]) -> None:
+    first_error: BaseException | None = None
+    seen: set[str] = set()
+    for virtual_path in reversed(virtual_paths):
+        if virtual_path in seen:
+            continue
+        seen.add(virtual_path)
+        try:
+            sandbox.remove_file(virtual_path)
+        except BaseException as exc:
+            if first_error is None:
+                first_error = exc
+            logger.warning("Failed to remove synchronized sandbox upload: %s", virtual_path, exc_info=True)
+    if first_error is not None:
+        raise first_error
+
+
+def rollback_sandbox_sync(receipt: SandboxSyncReceipt) -> None:
+    """Remove every exact remote path recorded by *receipt*."""
+    if receipt.sandbox is not None and receipt.virtual_paths:
+        _remove_remote_paths(receipt.sandbox, receipt.virtual_paths)
+
+
+def _sync_remote_paths(sandbox: Any, paths: tuple[tuple[Path, str], ...]) -> SandboxSyncReceipt:
+    attempted: list[str] = []
+    completed: list[str] = []
+    try:
+        for physical_path, virtual_path in paths:
+            attempted.append(virtual_path)
+            sandbox.update_file(virtual_path, physical_path.read_bytes())
+            completed.append(virtual_path)
+    except BaseException:
+        try:
+            _remove_remote_paths(sandbox, tuple(attempted))
+        except BaseException:
+            pass
+        raise
+    return SandboxSyncReceipt(sandbox=sandbox, virtual_paths=tuple(completed))
+
+
+def make_upload_paths_available(
+    sandbox_provider: Any,
+    thread_id: str,
+    *,
+    user_id: str | None,
+    paths: list[tuple[Path, str]],
+) -> SandboxSyncReceipt:
+    """Synchronously make exact host upload paths available to one provider."""
+    sync_paths = tuple((Path(path), virtual_path) for path, virtual_path in paths)
+    if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+        _make_paths_readable(sync_paths)
+        return SandboxSyncReceipt(sandbox=None)
+
+    sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
+    sandbox = sandbox_provider.get(sandbox_id)
+    if sandbox is None:
+        raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
+    return _sync_remote_paths(sandbox, sync_paths)
+
+
+async def make_upload_paths_available_async(
+    sandbox_provider: Any,
+    thread_id: str,
+    *,
+    user_id: str | None,
+    paths: list[tuple[Path, str]],
+) -> SandboxSyncReceipt:
+    """Cancellation-safely expose exact upload paths to a mounted or remote sandbox."""
+    sync_paths = tuple((Path(path), virtual_path) for path, virtual_path in paths)
+    if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+        await run_upload_io_cancellation_safe(_make_paths_readable, sync_paths)
+        return SandboxSyncReceipt(sandbox=None)
+
+    sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=user_id)
+    sandbox = sandbox_provider.get(sandbox_id)
+    if sandbox is None:
+        raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
+
+    sync_task = asyncio.create_task(
+        asyncio.to_thread(_sync_remote_paths, sandbox, sync_paths),
+        name=f"sync-upload-paths:{thread_id}",
+    )
+    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()
+        except BaseException:
+            logger.warning("Failed to roll back sandbox uploads after cancellation", exc_info=True)
+        raise asyncio.CancelledError
+    return receipt

backend/tests/blocking_io/test_channels_ingest.py

@@ -39,6 +39,10 @@ async def test_ingest_inbound_files_does_not_block_event_loop(tmp_path: Path, mo
         return b"payload-bytes"
 
     monkeypatch.setattr(mgr, "_read_http_inbound_file", _fake_reader)
+    monkeypatch.setattr(
+        "deerflow.sandbox.sandbox_provider.get_sandbox_provider",
+        lambda: type("MountedProvider", (), {"uses_thread_data_mounts": True})(),
+    )
 
     msg = InboundMessage(
         channel_name="unit-test-channel",  # absent from INBOUND_FILE_READERS -> default reader

backend/tests/blocking_io/test_dingtalk_receive_file.py

@@ -131,7 +131,6 @@ async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lea
     from app.channels.dingtalk import DingTalkChannel
     from app.channels.message_bus import MessageBus
     from deerflow.config.paths import Paths
-    from deerflow.uploads.manager import delete_file_safe
 
     paths = await asyncio.to_thread(Paths, str(tmp_path))
     monkeypatch.setattr("app.channels.dingtalk.get_paths", lambda: paths)
@@ -139,16 +138,24 @@ async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lea
     allow_sync = threading.Event()
 
     class PausedSandbox:
+        def __init__(self):
+            self.removals = []
+
         def update_file(self, _path, _content):
             sync_started.set()
             assert allow_sync.wait(5)
 
+        def remove_file(self, path):
+            self.removals.append(path)
+
+    sandbox = PausedSandbox()
+
     async def acquire_async(_thread_id, user_id=None):
         return "remote"
 
     monkeypatch.setattr(
         "app.channels.dingtalk.get_sandbox_provider",
-        lambda: SimpleNamespace(acquire_async=acquire_async, get=lambda _sandbox_id: PausedSandbox()),
+        lambda: SimpleNamespace(acquire_async=acquire_async, get=lambda _sandbox_id: sandbox),
     )
     channel = DingTalkChannel(MessageBus(), config={})
     channel._download_by_code = AsyncMock(return_value=b"DATA")
@@ -156,15 +163,13 @@ async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lea
     assert await asyncio.to_thread(sync_started.wait, 5)
     uploads = paths.sandbox_uploads_dir("t1", user_id="default")
     receive.cancel()
-    deletion = asyncio.create_task(asyncio.to_thread(delete_file_safe, uploads, "a.pdf"))
     try:
         await asyncio.sleep(0.05)
         assert not receive.done()
-        assert not deletion.done()
     finally:
         allow_sync.set()
         with suppress(asyncio.CancelledError):
             await receive
-        await deletion
 
     assert not await asyncio.to_thread((uploads / "a.pdf").exists)
+    assert sandbox.removals == ["/mnt/user-data/uploads/a.pdf"]

backend/tests/blocking_io/test_feishu_receive_file.py

@@ -38,12 +38,16 @@ def _channel_with_file(content: bytes = b"DATA", filename: str = "report.pdf"):
 class _RemoteSandbox:
     def __init__(self) -> None:
         self.updates: list[tuple[str, bytes]] = []
+        self.removals: list[str] = []
         self.update_thread_id: int | None = None
 
     def update_file(self, path: str, content: bytes) -> None:
         self.update_thread_id = threading.get_ident()
         self.updates.append((path, content))
 
+    def remove_file(self, path: str) -> None:
+        self.removals.append(path)
+
 
 class _RemoteProvider:
     uses_thread_data_mounts = False
@@ -148,7 +152,6 @@ async def test_receive_file_holds_name_lease_through_remote_sync(tmp_path, monke
 
 async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lease(tmp_path, monkeypatch) -> None:
     from deerflow.config.paths import Paths
-    from deerflow.uploads.manager import delete_file_safe
 
     paths = await asyncio.to_thread(Paths, str(tmp_path))
     provider = _RemoteProvider()
@@ -175,18 +178,44 @@ async def test_receive_file_cancellation_drains_remote_sync_before_releasing_lea
     assert await asyncio.to_thread(sync_started.wait, 5)
     uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user")
     receive.cancel()
-    deletion = asyncio.create_task(asyncio.to_thread(delete_file_safe, uploads, "report.pdf"))
     try:
         await asyncio.sleep(0.05)
         assert not receive.done()
-        assert not deletion.done()
     finally:
         allow_sync.set()
         with suppress(asyncio.CancelledError):
             await receive
-        await deletion
 
     assert not await asyncio.to_thread((uploads / "report.pdf").exists)
+    assert provider.sandbox.removals == ["/mnt/user-data/uploads/report.pdf"]
+
+
+async def test_receive_file_sync_failure_rolls_back_host_and_remote_copy(tmp_path, monkeypatch) -> None:
+    from deerflow.config.paths import Paths
+
+    paths = await asyncio.to_thread(Paths, str(tmp_path))
+    provider = _RemoteProvider()
+
+    def failed_update(path: str, content: bytes) -> None:
+        provider.sandbox.updates.append((path, content))
+        raise OSError("sync failed")
+
+    provider.sandbox.update_file = failed_update
+    monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths)
+    monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: provider)
+
+    result = await _channel_with_file()._receive_single_file(
+        "message-1",
+        "file-key",
+        "file",
+        "thread-1",
+        user_id="ou-user",
+    )
+
+    uploads = paths.sandbox_uploads_dir("thread-1", user_id="ou-user")
+    assert result == "Failed to obtain the [file]"
+    assert not await asyncio.to_thread((uploads / "report.pdf").exists)
+    assert provider.sandbox.removals == ["/mnt/user-data/uploads/report.pdf"]
 
 
 async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monkeypatch) -> None:

backend/tests/test_channel_file_attachments.py

@@ -4,9 +4,14 @@ from __future__ import annotations
 
 import asyncio
 import os
+import threading
+from contextlib import suppress
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 from app.channels.base import Channel
 from app.channels.message_bus import InboundMessage, MessageBus, OutboundMessage, ResolvedAttachment
 
@@ -255,6 +260,13 @@ class TestResolveAttachments:
 
 
 class TestInboundFileIngestion:
+    @pytest.fixture(autouse=True)
+    def _mounted_sandbox_provider(self, monkeypatch):
+        monkeypatch.setattr(
+            "deerflow.sandbox.sandbox_provider.get_sandbox_provider",
+            lambda: SimpleNamespace(uses_thread_data_mounts=True),
+        )
+
     def test_consumes_inline_channel_bytes_without_exposing_them_downstream(self, tmp_path):
         from app.channels import manager
 
@@ -268,7 +280,11 @@ class TestInboundFileIngestion:
             files=[{"type": "file", "filename": "report.pdf", "_content": b"pdf bytes"}],
         )
 
-        with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir):
+        provider = SimpleNamespace(uses_thread_data_mounts=True)
+        with (
+            patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=provider),
+        ):
             result = _run(manager._ingest_inbound_files("thread-1", msg))
 
         assert result == [
@@ -280,8 +296,106 @@ class TestInboundFileIngestion:
             }
         ]
         assert (uploads_dir / "report.pdf").read_bytes() == b"pdf bytes"
+        assert (uploads_dir / "report.pdf").stat().st_mode & 0o044 == 0o044
         assert "_content" not in msg.files[0]
 
+    def test_syncs_inbound_file_to_non_mounted_sandbox(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        updates: list[tuple[str, bytes]] = []
+        removals: list[str] = []
+
+        class Sandbox:
+            def update_file(self, path, content):
+                updates.append((path, content))
+
+            def remove_file(self, path):
+                removals.append(path)
+
+        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()),
+        ):
+            result = _run(manager._ingest_inbound_files("thread-1", msg))
+
+        assert result[0]["path"] == "/mnt/user-data/uploads/report.pdf"
+        assert updates == [(result[0]["path"], b"pdf bytes")]
+        assert removals == []
+
+    def test_cancellation_rolls_back_generic_remote_sync_and_host_publication(self, tmp_path):
+        from app.channels import manager
+
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        sync_started = threading.Event()
+        allow_sync = threading.Event()
+        removals: list[str] = []
+
+        class Sandbox:
+            def update_file(self, _path, _content):
+                sync_started.set()
+                assert allow_sync.wait(5)
+
+            def remove_file(self, path):
+                removals.append(path)
+
+        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"}],
+        )
+
+        async def run_cancelled_ingest():
+            with (
+                patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=Provider()),
+            ):
+                task = asyncio.create_task(manager._ingest_inbound_files("thread-1", msg))
+                assert await asyncio.to_thread(sync_started.wait, 5)
+                task.cancel()
+                await asyncio.sleep(0.05)
+                assert not task.done()
+                allow_sync.set()
+                with suppress(asyncio.CancelledError):
+                    await task
+
+        _run(run_cancelled_ingest())
+
+        assert not (uploads_dir / "report.pdf").exists()
+        assert removals == ["/mnt/user-data/uploads/report.pdf"]
+
     def test_renames_around_preexisting_symlink_destination(self, tmp_path):
         from app.channels import manager
 

backend/tests/test_channels.py

@@ -882,6 +882,10 @@ class TestChannelManager:
 
         paths = Paths(tmp_path)
         monkeypatch.setattr("deerflow.uploads.manager.get_paths", lambda: paths)
+        monkeypatch.setattr(
+            "deerflow.sandbox.sandbox_provider.get_sandbox_provider",
+            lambda: SimpleNamespace(uses_thread_data_mounts=True),
+        )
 
         async def read_file(file_info, client):
             del file_info, client

backend/tests/test_client.py

@@ -56,6 +56,7 @@ def mock_app_config():
     config.database.checkpoint_channel_mode = "full"
     config.database.checkpoint_delta.snapshot_frequency = 10
     config.authorization = AuthorizationConfig(enabled=False)
+    config.sandbox.use = "deerflow.sandbox.local:LocalSandboxProvider"
     return config
 
 
@@ -2184,6 +2185,52 @@ class TestUploads:
             assert (uploads_dir / "test.txt").exists()
             assert (uploads_dir / "test.txt").stat().st_mode & 0o044 == 0o044
 
+    def test_upload_files_syncs_primary_and_conversion_to_non_mounted_sandbox(self, client, tmp_path):
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        source = tmp_path / "report.pdf"
+        source.write_bytes(b"PDF")
+        updates: list[tuple[str, bytes]] = []
+        removals: list[str] = []
+
+        class Sandbox:
+            def update_file(self, path, content):
+                updates.append((path, content))
+
+            def remove_file(self, path):
+                removals.append(path)
+
+        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
+
+        async def fake_convert(path: Path, *, publication=None) -> Path:
+            assert publication is not None
+            conversion = conversion_path_for_upload(path)
+            conversion.parent.mkdir(parents=True, exist_ok=True)
+            conversion.write_bytes(b"MARKDOWN")
+            return conversion
+
+        with (
+            patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.utils.file_conversion.CONVERTIBLE_EXTENSIONS", {".pdf"}),
+            patch("deerflow.client.convert_uploaded_file_to_markdown", side_effect=fake_convert),
+            patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=Provider()),
+        ):
+            result = client.upload_files("thread-1", [source])
+
+        info = result["files"][0]
+        assert updates == [
+            (info["virtual_path"], b"PDF"),
+            (info["markdown_virtual_path"], b"MARKDOWN"),
+        ]
+        assert removals == []
+
     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

@@ -108,6 +108,18 @@ class TestNormalizeFilename:
         with pytest.raises(ValueError, match="unsafe"):
             normalize_filename(".")
 
+    @pytest.mark.parametrize(
+        "filename",
+        [
+            "paper<system>.pdf",
+            "report--- BEGIN USER INPUT ---draft.pdf",
+            "report--- END USER INPUT ---draft.pdf",
+        ],
+    )
+    def test_rejects_names_that_cannot_be_exposed_losslessly_to_the_agent(self, filename):
+        with pytest.raises(ValueError, match="reserved model-context token"):
+            normalize_filename(filename)
+
 
 # ---------------------------------------------------------------------------
 # claim_unique_filename