diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index e6da5b715..e7d080045 100644
--- a/backend/AGENTS.md
+++ b/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. 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.
+**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/users/{user_id}/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,12 +1461,12 @@ 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.
-- Exact-name generation leases use a portable NFC-plus-casefold coordination key with Win32 trailing-dot/space folding, so filesystem aliases cannot bypass an active generation; filenames that Windows cannot represent losslessly are rejected before staging, while exact legacy POSIX basenames remain deletable after upgrade. Legacy deletion has a POSIX-only lease mode for literal backslashes and components made entirely from dots/spaces; it does not weaken new-upload validation. The original filename remains the published name. Publication tries each candidate lease without blocking and treats a busy canonical key as a collision, so same-batch and inverse concurrent batches advance to a UTF-8-bounded `_N` candidate instead of deadlocking while retaining earlier generations; pathological long suffixes fall back to truncating the complete basename. Deletion rejects an inode that moved outside the requested name's lease. Primary deletion and rollback atomically move the selected directory entry into an inode-tagged transaction directory under the protected conversion namespace before revalidating its identity, so a pathname replacement between check and removal is restored rather than unlinked. Startup recovery restores a crash-abandoned tombstone to its recorded basename, or to a bounded visible recovery name if that basename was recreated; it never treats the tombstone as an ordinary disposable upload stage. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result. Embedded-client finalization attempts every lease release and conversion-pool shutdown independently, logging cleanup failures rather than changing an already-committed response or stranding later leases.
+- Exact-name generation leases use a portable NFC-plus-casefold coordination key with Win32 trailing-dot/space folding, so filesystem aliases cannot bypass an active generation; filenames that Windows cannot represent losslessly are rejected before staging, while exact legacy POSIX basenames remain deletable after upgrade. Legacy deletion has a POSIX-only lease mode for literal backslashes and components made entirely from dots/spaces; it does not weaken new-upload validation. The original filename remains the published name. Publication tries each candidate lease without blocking and treats a busy canonical key as a collision, so same-batch and inverse concurrent batches advance to a UTF-8-bounded `_N` candidate instead of deadlocking while retaining earlier generations; pathological long suffixes fall back to truncating the complete basename. Deletion rejects an inode that moved outside the requested name's lease. Primary deletion and rollback atomically move the selected directory entry into an inode-and-intent-tagged transaction directory under the protected conversion namespace before revalidating its identity, so a pathname replacement between check and removal is restored rather than unlinked. Startup recovery restores a crash-abandoned user-deletion tombstone to its recorded basename (or a bounded visible recovery name if that basename was recreated), but completes a failed-upload rollback tombstone's deletion; neither is treated as an ordinary disposable upload stage. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result. Embedded-client finalization attempts every lease release and conversion-pool shutdown independently, logging cleanup failures rather than changing an already-committed response or stranding later leases.
 - 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, while `.upload-delete-<inode>-<nonce>.part/` transaction directories are recovered rather than swept. 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.
+- 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, while `.upload-delete-<restore|discard>-<inode>-<nonce>.part/` transaction directories are resolved according to their persisted crash intent. Legacy intent-less deletion transactions retain restore behavior. 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 holds the generation lease while it removes an explicitly synchronized sandbox primary/conversion and then the authoritative host paths; a remote failure is reported and leaves the host primary intact. It never guesses or deletes a legacy/user-owned `uploads/<stem>.md` sibling. Outline extraction opens one descriptor, verifies its `fstat` against the current exclusive regular directory entry, and uses that same descriptor for both outline and preview reads.
 - Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; only operations that may block waiting for a name lease use the separate lease-wait pool. Work needed by an existing lease holder and non-blocking publication stays on the general pool, so waiters cannot starve conversion, rollback, or release. 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; the command fallback requires a per-call unpredictable exact success trailer. Embedded multi-file calls retain every publication and receipt until the whole response is built, then roll back the complete batch on failure. WeChat download publication uses the cancellation-safe async lease adapter, so cancellation drains and rolls back a publication worker that completes late.
-- 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` plus a Provisioner advertising the current mount contract; unavailable or legacy capability results use explicit synchronization and are retried without a Gateway restart. A confirmed legacy peer may create thread sandboxes only for the `default` no-auth user, because its primary user-data mount cannot isolate equal thread IDs across authenticated users; authenticated rolling deployments upgrade the Provisioner first. The Provisioner validates the request before its idempotent fast path, labels Pods with hashed user/thread identity and contract version, stores the exact identity plus a mount-signature annotation, and compares the live Pod specification before reuse. Every remote acquisition path, including active-cache and warm-pool reuse, deliberately replays the idempotent create request instead of trusting GET discovery or a liveness response; a Gateway that negotiated v2 also rejects any create response that does not echo the exact current contract, user, and thread. `SANDBOX_MOUNT_CONTRACT_VERSION` changes deterministic AIO sandbox IDs so an older container cannot satisfy the new thread acquisition. Reconciliation may still enumerate/adopt the old ID for orphan cleanup; it is not selected for the new identity.
+- 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` plus a Provisioner advertising the current mount contract; every upload/delete/artifact mode decision refreshes an expired negotiation off the event loop, and rechecks the cached mode after acquisition before attempting explicit synchronization. Unavailable or legacy capability results use explicit synchronization and are retried without a Gateway restart. A confirmed legacy peer may create thread sandboxes only for the `default` no-auth user, because its primary user-data mount cannot isolate equal thread IDs across authenticated users; authenticated rolling deployments upgrade the Provisioner first. The Provisioner validates the request before its idempotent fast path, labels Pods with hashed user/thread identity and contract version, stores the exact identity plus a mount-signature annotation, and compares the live Pod specification before reuse. Every remote acquisition path, including active-cache and warm-pool reuse, deliberately replays the idempotent create request instead of trusting GET discovery or a liveness response; a Gateway freezes the negotiated contract for each create and, when that snapshot requires v2, rejects any response that does not echo the exact current contract, user, and thread. The negotiated mount-contract version namespaces deterministic AIO sandbox IDs so a legacy container cannot satisfy a v2 acquisition; active reuse also replaces its client when validation returns a changed service URL. Reconciliation may still enumerate/adopt the old ID for orphan cleanup; it is not selected for the new identity.
 - Agent receives uploaded file list via `UploadsMiddleware`
 
 See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details.
diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py
index 08f12537f..75d0cb270 100644
--- a/backend/app/gateway/routers/artifacts.py
+++ b/backend/app/gateway/routers/artifacts.py
@@ -22,7 +22,11 @@ from app.gateway.path_utils import resolve_thread_virtual_path
 from deerflow.config.paths import make_safe_user_id
 from deerflow.runtime import ConflictError, ThreadOperationKind
 from deerflow.runtime.user_context import get_effective_user_id
-from deerflow.sandbox.sandbox_provider import get_sandbox_provider
+from deerflow.sandbox.sandbox_provider import (
+    get_sandbox_provider,
+    sandbox_provider_uses_thread_data_mounts,
+    sandbox_provider_uses_thread_data_mounts_async,
+)
 from deerflow.utils.thread_id import ThreadId
 
 logger = logging.getLogger(__name__)
@@ -463,11 +467,15 @@ async def update_artifact(
             updated = _encode_artifact_update(body.content)
 
             sandbox_provider = get_sandbox_provider()
-            if not bool(getattr(sandbox_provider, "uses_thread_data_mounts", False)):
+            if not await sandbox_provider_uses_thread_data_mounts_async(sandbox_provider):
                 sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-                sandbox = sandbox_provider.get(sandbox_id)
-                if sandbox is None:
-                    raise RuntimeError("Failed to acquire sandbox for artifact update")
+                if not sandbox_provider_uses_thread_data_mounts(
+                    sandbox_provider,
+                    refresh=False,
+                ):
+                    sandbox = sandbox_provider.get(sandbox_id)
+                    if sandbox is None:
+                        raise RuntimeError("Failed to acquire sandbox for artifact update")
 
             try:
                 if sandbox is not None:
diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py
index b8ee3ef9f..f2c8f9b63 100644
--- a/backend/app/gateway/routers/uploads.py
+++ b/backend/app/gateway/routers/uploads.py
@@ -14,7 +14,11 @@ from app.gateway.deps import get_config
 from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
-from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provider
+from deerflow.sandbox.sandbox_provider import (
+    get_sandbox_provider,
+    sandbox_provider_uses_thread_data_mounts,
+    sandbox_provider_uses_thread_data_mounts_async,
+)
 from deerflow.uploads.async_helpers import run_upload_lease_io, wait_for_task_completion
 from deerflow.uploads.conversion import convert_uploaded_file_to_markdown
 from deerflow.uploads.layout import artifact_url_for_virtual_path, conversion_virtual_path
@@ -128,10 +132,6 @@ def _make_file_sandbox_readable(file_path: os.PathLike[str] | str) -> None:
     os.chmod(file_path, readable_mode, **chmod_kwargs)
 
 
-def _uses_thread_data_mounts(sandbox_provider: SandboxProvider) -> bool:
-    return bool(getattr(sandbox_provider, "uses_thread_data_mounts", False))
-
-
 def _get_uploads_config_value(app_config: AppConfig, key: str, default: object) -> object:
     """Read a value from the uploads config, supporting dict and attribute access."""
     uploads_cfg = getattr(app_config, "uploads", None)
@@ -413,13 +413,18 @@ async def upload_files(
     reserved_coordination_keys: set[str] = set()
     total_size = 0
     sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
-    sync_to_sandbox = not _uses_thread_data_mounts(sandbox_provider)
+    sync_to_sandbox = not await sandbox_provider_uses_thread_data_mounts_async(sandbox_provider)
     sandbox = None
     if sync_to_sandbox:
         sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-        sandbox = sandbox_provider.get(sandbox_id)
-        if sandbox is None:
-            raise HTTPException(status_code=500, detail="Failed to acquire sandbox")
+        sync_to_sandbox = not sandbox_provider_uses_thread_data_mounts(
+            sandbox_provider,
+            refresh=False,
+        )
+        if sync_to_sandbox:
+            sandbox = sandbox_provider.get(sandbox_id)
+            if sandbox is None:
+                raise HTTPException(status_code=500, detail="Failed to acquire sandbox")
     auto_convert_documents = _auto_convert_documents_enabled(config)
     current_filename = "request"
 
diff --git a/backend/docs/API.md b/backend/docs/API.md
index e98a829a5..649eaa628 100644
--- a/backend/docs/API.md
+++ b/backend/docs/API.md
@@ -617,11 +617,11 @@ Content-Type: multipart/form-data
     {
       "filename": "document.pdf",
       "size": 1234567,
-      "path": ".deer-flow/threads/abc123/user-data/uploads/document.pdf",
+      "path": ".deer-flow/users/alice/threads/abc123/user-data/uploads/document.pdf",
       "virtual_path": "/mnt/user-data/uploads/document.pdf",
       "artifact_url": "/api/threads/abc123/artifacts/mnt/user-data/uploads/document.pdf",
       "markdown_file": "document.pdf.md",
-      "markdown_path": ".deer-flow/threads/abc123/user-data/.upload-conversions/document.pdf.md",
+      "markdown_path": ".deer-flow/users/alice/threads/abc123/user-data/.upload-conversions/document.pdf.md",
       "markdown_virtual_path": "/mnt/user-data/.upload-conversions/document.pdf.md",
       "markdown_artifact_url": "/api/threads/abc123/artifacts/mnt/user-data/.upload-conversions/document.pdf.md"
     }
@@ -653,7 +653,7 @@ GET /api/threads/{thread_id}/uploads/list
     {
       "filename": "document.pdf",
       "size": 1234567,
-      "path": ".deer-flow/threads/abc123/user-data/uploads/document.pdf",
+      "path": ".deer-flow/users/alice/threads/abc123/user-data/uploads/document.pdf",
       "virtual_path": "/mnt/user-data/uploads/document.pdf",
       "artifact_url": "/api/threads/abc123/artifacts/mnt/user-data/uploads/document.pdf",
       "extension": ".pdf",
@@ -687,7 +687,7 @@ fail the stricter cross-platform validation applied to new uploads.
 
 ### Thread Cleanup
 
-Remove DeerFlow-managed local thread files under `.deer-flow/threads/{thread_id}` after the LangGraph thread itself has been deleted.
+Remove DeerFlow-managed local thread files under `.deer-flow/users/{user_id}/threads/{thread_id}` after the LangGraph thread itself has been deleted.
 
 ```http
 DELETE /api/threads/{thread_id}
diff --git a/backend/docs/PATH_EXAMPLES.md b/backend/docs/PATH_EXAMPLES.md
index 9386d5848..4bd9e516e 100644
--- a/backend/docs/PATH_EXAMPLES.md
+++ b/backend/docs/PATH_EXAMPLES.md
@@ -7,7 +7,7 @@ DeerFlow 的文件上传系统返回三种不同的路径，每种路径用于
 ### 1. 实际文件系统路径 (path)
 
 ```
-.deer-flow/threads/{thread_id}/user-data/uploads/document.pdf
+.deer-flow/users/{user_id}/threads/{thread_id}/user-data/uploads/document.pdf
 ```
 
 **用途：**
@@ -19,7 +19,7 @@ DeerFlow 的文件上传系统返回三种不同的路径，每种路径用于
 ```python
 # Python 代码中直接访问
 from pathlib import Path
-file_path = Path("backend/.deer-flow/threads/abc123/user-data/uploads/document.pdf")
+file_path = Path("backend/.deer-flow/users/alice/threads/abc123/user-data/uploads/document.pdf")
 content = file_path.read_bytes()
 ```
 
@@ -99,11 +99,11 @@ async function uploadAndProcess(threadId: string, file: File) {
   console.log('文件信息：', fileInfo);
   // {
   //   filename: "report.pdf",
-  //   path: ".deer-flow/threads/abc123/user-data/uploads/report.pdf",
+  //   path: ".deer-flow/users/alice/threads/abc123/user-data/uploads/report.pdf",
   //   virtual_path: "/mnt/user-data/uploads/report.pdf",
   //   artifact_url: "/api/threads/abc123/artifacts/mnt/user-data/uploads/report.pdf",
   //   markdown_file: "report.pdf.md",
-  //   markdown_path: ".deer-flow/threads/abc123/user-data/.upload-conversions/report.pdf.md",
+  //   markdown_path: ".deer-flow/users/alice/threads/abc123/user-data/.upload-conversions/report.pdf.md",
   //   markdown_virtual_path: "/mnt/user-data/.upload-conversions/report.pdf.md",
   //   markdown_artifact_url: "/api/threads/abc123/artifacts/mnt/user-data/.upload-conversions/report.pdf.md"
   // }
@@ -132,26 +132,26 @@ async function uploadAndProcess(threadId: string, file: File) {
 
 | 场景 | 使用的路径类型 | 示例 |
 |------|---------------|------|
-| 服务器后端代码直接访问 | `path` | `.deer-flow/threads/abc123/user-data/uploads/file.pdf` |
+| 服务器后端代码直接访问 | `path` | `.deer-flow/users/alice/threads/abc123/user-data/uploads/file.pdf` |
 | Agent 工具调用 | `virtual_path` | `/mnt/user-data/uploads/file.pdf` |
 | 前端下载/预览 | `artifact_url` | `/api/threads/abc123/artifacts/mnt/user-data/uploads/file.pdf` |
 | Agent 读取生成 Markdown | `markdown_virtual_path` | `/mnt/user-data/.upload-conversions/file.pdf.md` |
 | 前端读取生成 Markdown | `markdown_artifact_url` | `/api/threads/abc123/artifacts/mnt/user-data/.upload-conversions/file.pdf.md` |
-| 备份脚本 | `path` | `.deer-flow/threads/abc123/user-data/uploads/file.pdf` |
-| 日志记录 | `path` | `.deer-flow/threads/abc123/user-data/uploads/file.pdf` |
+| 备份脚本 | `path` | `.deer-flow/users/alice/threads/abc123/user-data/uploads/file.pdf` |
+| 日志记录 | `path` | `.deer-flow/users/alice/threads/abc123/user-data/uploads/file.pdf` |
 
 ## 代码示例集合
 
 ### Python - 后端处理
 
 ```python
-from pathlib import Path
-from deerflow.agents.middlewares.thread_data_middleware import THREAD_DATA_BASE_DIR
+from deerflow.config.paths import get_paths
+from deerflow.uploads.manager import normalize_filename
 
-def process_uploaded_file(thread_id: str, filename: str):
-    # 使用实际路径
-    base_dir = Path.cwd() / THREAD_DATA_BASE_DIR / thread_id / "user-data" / "uploads"
-    file_path = base_dir / filename
+def process_uploaded_file(user_id: str, thread_id: str, filename: str):
+    # 使用与请求所有者相同的用户隔离桶
+    base_dir = get_paths().sandbox_uploads_dir(thread_id, user_id=user_id)
+    file_path = base_dir / normalize_filename(filename)
 
     # 直接读取
     with open(file_path, 'rb') as f:
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
index d5d32cd95..03f67567f 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
@@ -280,6 +280,13 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 return False
         return mounted
 
+    def refresh_thread_data_mount_capabilities(self) -> bool:
+        """Refresh remote mount compatibility when its negotiated value is stale."""
+        backend = getattr(self, "_backend", None)
+        if isinstance(backend, RemoteSandboxBackend):
+            return backend.refresh_capabilities_if_stale()
+        return False
+
     # ── Factory methods ──────────────────────────────────────────────────
 
     def _create_backend(self) -> SandboxBackend:
@@ -763,7 +770,11 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         return (user_id, thread_id)
 
     @staticmethod
-    def _deterministic_sandbox_id(thread_id: str, user_id: str) -> str:
+    def _deterministic_sandbox_id(
+        thread_id: str,
+        user_id: str,
+        mount_contract_version: int = SANDBOX_MOUNT_CONTRACT_VERSION,
+    ) -> str:
         """Generate a deterministic sandbox ID from user/thread scope.
 
         Includes user_id so a previously-created default-bucket sandbox cannot be
@@ -774,7 +785,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         satisfy the new thread acquisition. Reconciliation may still enumerate
         and adopt its old ID for normal orphan cleanup.
         """
-        return hashlib.sha256(f"mount-v{SANDBOX_MOUNT_CONTRACT_VERSION}:{user_id}:{thread_id}".encode()).hexdigest()[:16]
+        return hashlib.sha256(f"mount-v{mount_contract_version}:{user_id}:{thread_id}".encode()).hexdigest()[:16]
 
     def _assert_active_identity_available_locked(
         self,
@@ -810,9 +821,6 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
     def _get_extra_mounts(self, thread_id: str | None, *, user_id: str | None = None) -> list[tuple[str, str, bool]]:
         """Collect all extra mounts for a sandbox (thread-specific + skills)."""
-        backend = getattr(self, "_backend", None)
-        if isinstance(backend, RemoteSandboxBackend):
-            backend.refresh_capabilities_if_stale()
         mounts: list[tuple[str, str, bool]] = []
 
         if thread_id:
@@ -1342,9 +1350,26 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
     def _sandbox_id_for_thread(self, thread_id: str | None, user_id: str | None) -> str:
         """Return deterministic IDs for thread sandboxes and random IDs otherwise."""
-        return self._deterministic_sandbox_id(thread_id, self._effective_acquire_user_id(user_id)) if thread_id else str(uuid.uuid4())[:8]
+        if thread_id is None:
+            return str(uuid.uuid4())[:8]
+        backend = getattr(self, "_backend", None)
+        mount_contract_version = SANDBOX_MOUNT_CONTRACT_VERSION
+        if isinstance(backend, RemoteSandboxBackend):
+            mount_contract_version, _capability_known = backend.mount_contract_snapshot()
+        return self._deterministic_sandbox_id(
+            thread_id,
+            self._effective_acquire_user_id(user_id),
+            mount_contract_version,
+        )
 
-    def _reuse_in_process_sandbox(self, thread_id: str | None, *, user_id: str | None = None, post_lock: bool = False) -> str | None:
+    def _reuse_in_process_sandbox(
+        self,
+        thread_id: str | None,
+        *,
+        user_id: str | None = None,
+        post_lock: bool = False,
+        expected_sandbox_id: str | None = None,
+    ) -> str | None:
         """Reuse an active in-process sandbox for a thread if one is still tracked."""
         if thread_id is None:
             return None
@@ -1356,6 +1381,14 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 return None
 
             existing_id = self._thread_sandboxes[key]
+            if expected_sandbox_id is not None and existing_id != expected_sandbox_id:
+                del self._thread_sandboxes[key]
+                logger.info(
+                    "Cached sandbox %s uses a superseded mount contract; expected %s",
+                    existing_id,
+                    expected_sandbox_id,
+                )
+                return None
             if self._being_torn_down_locally(existing_id):
                 # A reaper thread in this process is stopping this container.
                 # Same answer as a peer's `del:` lease: cold-start instead.
@@ -1382,18 +1415,39 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             user_id=effective_user_id,
         )
 
+        replacement_client = None
+        if validated_info is not None and info is not None and validated_info.sandbox_url != info.sandbox_url:
+            replacement_client = AioSandbox(
+                id=existing_id,
+                base_url=validated_info.sandbox_url,
+            )
+
+        replaced_client = None
+        reuse_is_stale = False
         with self._lock:
             if self._thread_sandboxes.get(key) != existing_id:
-                return None
-            if existing_id not in self._sandboxes:
+                reuse_is_stale = True
+            elif existing_id not in self._sandboxes:
                 self._thread_sandboxes.pop(key, None)
-                return None
-            if validated_info is not None:
-                self._sandbox_infos[existing_id] = validated_info
+                reuse_is_stale = True
+            else:
+                if validated_info is not None:
+                    self._sandbox_infos[existing_id] = validated_info
+                if replacement_client is not None:
+                    replaced_client = self._sandboxes[existing_id]
+                    self._sandboxes[existing_id] = replacement_client
 
-            suffix = " (post-lock check)" if post_lock else ""
-            logger.info(f"Reusing in-process sandbox {existing_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
-            self._last_activity[existing_id] = time.time()
+                suffix = " (post-lock check)" if post_lock else ""
+                logger.info(f"Reusing in-process sandbox {existing_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
+                self._last_activity[existing_id] = time.time()
+
+        if reuse_is_stale:
+            if replacement_client is not None:
+                replacement_client.close()
+            return None
+
+        if replaced_client is not None:
+            replaced_client.close()
 
         # Fail closed: an OwnershipBackendError propagates rather than handing out
         # a sandbox we could not publish ownership for.
@@ -1542,7 +1596,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
     def _recheck_cached_sandbox(self, thread_id: str, sandbox_id: str, *, user_id: str) -> str | None:
         """Re-check in-memory caches after acquiring the cross-process file lock."""
-        return self._reuse_in_process_sandbox(thread_id, user_id=user_id, post_lock=True) or self._reclaim_warm_pool_sandbox(
+        return self._reuse_in_process_sandbox(
+            thread_id,
+            user_id=user_id,
+            post_lock=True,
+            expected_sandbox_id=sandbox_id,
+        ) or self._reclaim_warm_pool_sandbox(
             thread_id,
             sandbox_id,
             user_id=user_id,
@@ -1878,6 +1937,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         Returns:
             The ID of the acquired sandbox environment.
         """
+        self.refresh_thread_data_mount_capabilities()
         effective_user_id = self._effective_acquire_user_id(user_id)
         if thread_id:
             thread_lock = self._get_thread_lock(thread_id, effective_user_id)
@@ -1893,6 +1953,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         event loop and using async-native readiness polling for newly created
         sandboxes.
         """
+        await asyncio.to_thread(self.refresh_thread_data_mount_capabilities)
         effective_user_id = self._effective_acquire_user_id(user_id)
         if thread_id:
             thread_lock = self._get_thread_lock(thread_id, effective_user_id)
@@ -1913,12 +1974,19 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                  is needed — any process can derive the same container name)
         """
         self._ensure_skills_projection(user_id)
-        cached_id = self._reuse_in_process_sandbox(thread_id, user_id=user_id)
+        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
+        expected_cached_id = sandbox_id if thread_id and isinstance(self._backend, RemoteSandboxBackend) else None
+        if expected_cached_id is None:
+            cached_id = self._reuse_in_process_sandbox(thread_id, user_id=user_id)
+        else:
+            cached_id = self._reuse_in_process_sandbox(
+                thread_id,
+                user_id=user_id,
+                expected_sandbox_id=expected_cached_id,
+            )
         if cached_id is not None:
             return cached_id
 
-        # Deterministic ID for thread-specific, random for anonymous
-        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
         if thread_id:
             key = self._thread_key(thread_id, user_id)
             with self._lock:
@@ -1941,12 +2009,24 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
     async def _acquire_internal_async(self, thread_id: str | None, *, user_id: str) -> str:
         """Async counterpart to ``_acquire_internal``."""
         await asyncio.to_thread(self._ensure_skills_projection, user_id)
-        cached_id = await asyncio.to_thread(self._reuse_in_process_sandbox, thread_id, user_id=user_id)
+        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
+        expected_cached_id = sandbox_id if thread_id and isinstance(self._backend, RemoteSandboxBackend) else None
+        if expected_cached_id is None:
+            cached_id = await asyncio.to_thread(
+                self._reuse_in_process_sandbox,
+                thread_id,
+                user_id=user_id,
+            )
+        else:
+            cached_id = await asyncio.to_thread(
+                self._reuse_in_process_sandbox,
+                thread_id,
+                user_id=user_id,
+                expected_sandbox_id=expected_cached_id,
+            )
         if cached_id is not None:
             return cached_id
 
-        # Deterministic ID for thread-specific, random for anonymous
-        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
         if thread_id:
             key = self._thread_key(thread_id, user_id)
             with self._lock:
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
index 09f1815c9..8b2e4a550 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
@@ -171,6 +171,14 @@ class RemoteSandboxBackend(SandboxBackend):
         """Whether the peer definitively answered capability negotiation."""
         return self._mount_contract_capability_known
 
+    def mount_contract_snapshot(self) -> tuple[int, bool]:
+        """Return one capability snapshot for a complete caller operation."""
+        with self._capability_probe_lock:
+            return (
+                self._mount_contract_version,
+                self._mount_contract_capability_known,
+            )
+
     @property
     def requires_create_validation(self) -> bool:
         """Require idempotent POST so the Provisioner checks requested mounts."""
@@ -227,9 +235,13 @@ class RemoteSandboxBackend(SandboxBackend):
             self._mount_contract_version = 0
             self._mount_contract_capability_known = False
             self._capability_probe_failures += 1
+            retry_exponent = min(
+                self._capability_probe_failures - 1,
+                int(_CAPABILITY_RETRY_MAX_SECONDS).bit_length(),
+            )
             retry_seconds = min(
                 _CAPABILITY_RETRY_MAX_SECONDS,
-                float(2 ** (self._capability_probe_failures - 1)),
+                float(2**retry_exponent),
             )
             self._capability_next_probe_at = time.monotonic() + retry_seconds
             logger.warning("Provisioner mount capabilities are unavailable; mounted thread data will fail closed until compatibility can be verified")
@@ -251,11 +263,11 @@ class RemoteSandboxBackend(SandboxBackend):
         Calls ``POST /api/sandboxes`` which creates a dedicated Pod +
         NodePort Service in k3s.
         """
-        if self._mount_contract_version < _UPLOAD_MOUNT_CONTRACT_VERSION:
-            self.refresh_capabilities_if_stale()
+        self.refresh_capabilities_if_stale()
+        mount_contract_version, _capability_known = self.mount_contract_snapshot()
         effective_user_id = user_id or get_effective_user_id()
-        if self._mount_contract_version < _UPLOAD_MOUNT_CONTRACT_VERSION and thread_id is not None and effective_user_id != DEFAULT_USER_ID:
-            raise RuntimeError(f"Provisioner mount contract v{self._mount_contract_version} cannot isolate user {effective_user_id!r}; upgrade the Provisioner before creating authenticated-user sandboxes")
+        if mount_contract_version < _UPLOAD_MOUNT_CONTRACT_VERSION and thread_id is not None and effective_user_id != DEFAULT_USER_ID:
+            raise RuntimeError(f"Provisioner mount contract v{mount_contract_version} cannot isolate user {effective_user_id!r}; upgrade the Provisioner before creating authenticated-user sandboxes")
         return self._provisioner_create(
             thread_id,
             sandbox_id,
@@ -263,6 +275,7 @@ class RemoteSandboxBackend(SandboxBackend):
             user_id=user_id,
             provision_lark_cli_runtime=provision_lark_cli_runtime,
             provision_lark_cli_broker=provision_lark_cli_broker,
+            required_mount_contract_version=mount_contract_version,
         )
 
     def destroy(self, info: SandboxInfo) -> None:
@@ -336,8 +349,11 @@ class RemoteSandboxBackend(SandboxBackend):
         user_id: str | None = None,
         provision_lark_cli_runtime: bool = False,
         provision_lark_cli_broker: bool = False,
+        required_mount_contract_version: int | None = None,
     ) -> SandboxInfo:
         """POST /api/sandboxes → create Pod + Service."""
+        if required_mount_contract_version is None:
+            required_mount_contract_version, _capability_known = self.mount_contract_snapshot()
         effective_user_id = user_id or get_effective_user_id()
         include_legacy_skills = user_should_see_legacy_skills(effective_user_id)
         payload = {
@@ -367,7 +383,7 @@ class RemoteSandboxBackend(SandboxBackend):
             if not isinstance(data, dict):
                 raise RuntimeError("Provisioner mount contract response is not an object")
             effective_thread_id = thread_id or sandbox_id
-            if self._mount_contract_version >= _UPLOAD_MOUNT_CONTRACT_VERSION:
+            if required_mount_contract_version >= _UPLOAD_MOUNT_CONTRACT_VERSION:
                 response_contract = (
                     data.get("sandbox_id"),
                     data.get("user_id"),
diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
index ccc231700..578d6a340 100644
--- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
+++ b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py
@@ -175,3 +175,29 @@ def set_sandbox_provider(provider: SandboxProvider) -> None:
     global _default_sandbox_provider
     with _provider_lock:
         _default_sandbox_provider = provider
+
+
+def sandbox_provider_uses_thread_data_mounts(
+    provider: SandboxProvider,
+    *,
+    refresh: bool = True,
+) -> bool:
+    """Return one provider mount-mode decision, optionally refreshing it first."""
+    if refresh:
+        refresher = getattr(provider, "refresh_thread_data_mount_capabilities", None)
+        if callable(refresher):
+            refresher()
+    return bool(getattr(provider, "uses_thread_data_mounts", False))
+
+
+async def sandbox_provider_uses_thread_data_mounts_async(
+    provider: SandboxProvider,
+    *,
+    refresh: bool = True,
+) -> bool:
+    """Async mount-mode decision that keeps capability probes off the event loop."""
+    return await asyncio.to_thread(
+        sandbox_provider_uses_thread_data_mounts,
+        provider,
+        refresh=refresh,
+    )
diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py
index a504ff818..7ac1c26d5 100644
--- a/backend/packages/harness/deerflow/uploads/manager.py
+++ b/backend/packages/harness/deerflow/uploads/manager.py
@@ -34,6 +34,8 @@ logger = logging.getLogger(__name__)
 UPLOAD_STAGING_PREFIX = ".upload-"
 UPLOAD_STAGING_SUFFIX = ".part"
 UPLOAD_DELETION_TRANSACTION_PREFIX = ".upload-delete-"
+_UPLOAD_DELETION_RESTORE_INTENT = "restore"
+_UPLOAD_DELETION_DISCARD_INTENT = "discard"
 _WINDOWS_FORBIDDEN_FILENAME_CHARS = frozenset('<>:"|?*')
 
 
@@ -467,6 +469,7 @@ def rollback_published_upload(publication: PublishedUpload) -> None:
             publication.path.parent,
             publication.path,
             publication.identity,
+            recover_on_crash=False,
         )
     except (FileNotFoundError, UnsafeUploadPathError):
         # The pathname was replaced after the optimistic identity check. The
@@ -476,6 +479,7 @@ def rollback_published_upload(publication: PublishedUpload) -> None:
         if owned_conversion is not None:
             owned_conversion.unlink(missing_ok=True)
         staged_path.unlink()
+        _finish_deletion_transaction(staged_path)
     except BaseException:
         _restore_staged_deletion(
             staged_path,
@@ -585,15 +589,26 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> in
     return removed
 
 
-def _deletion_transaction_inode(transaction_name: str) -> int | None:
-    """Decode the selected generation inode embedded in a transaction name."""
+def _deletion_transaction_metadata(transaction_name: str) -> tuple[int, bool] | None:
+    """Decode the selected inode and crash intent from a transaction name."""
     if not (transaction_name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and transaction_name.endswith(UPLOAD_STAGING_SUFFIX)):
         return None
     body = transaction_name[len(UPLOAD_DELETION_TRANSACTION_PREFIX) : -len(UPLOAD_STAGING_SUFFIX)]
+    intent, separator, remaining = body.partition("-")
+    if separator and intent in {
+        _UPLOAD_DELETION_RESTORE_INTENT,
+        _UPLOAD_DELETION_DISCARD_INTENT,
+    }:
+        recover_on_crash = intent == _UPLOAD_DELETION_RESTORE_INTENT
+        body = remaining
+    else:
+        # Transactions created before the intent field was introduced always
+        # represented user-requested deletion and therefore recover on crash.
+        recover_on_crash = True
     inode_hex, separator, token = body.partition("-")
     if not separator or not inode_hex or len(token) != 32 or any(character not in "0123456789abcdef" for character in inode_hex + token):
         return None
-    return int(inode_hex, 16)
+    return int(inode_hex, 16), recover_on_crash
 
 
 def write_upload_file_no_symlink(base_dir: Path, filename: str, data: bytes) -> Path:
@@ -751,10 +766,11 @@ def _finish_deletion_transaction(staged_path: Path) -> None:
 
 def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
     """Restore a crash-abandoned tombstone whose basename records its target."""
-    expected_inode = _deletion_transaction_inode(transaction_dir.name)
-    if expected_inode is None:
+    metadata = _deletion_transaction_metadata(transaction_dir.name)
+    if metadata is None:
         logger.warning("Refusing malformed upload deletion transaction name: %s", transaction_dir)
         return False
+    expected_inode, recover_on_crash = metadata
     try:
         transaction_stat = os.lstat(transaction_dir)
     except FileNotFoundError:
@@ -829,6 +845,10 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
             staged_path.unlink()
             _finish_deletion_transaction(staged_path)
             return True
+        if not recover_on_crash:
+            staged_path.unlink()
+            _finish_deletion_transaction(staged_path)
+            return True
         _restore_staged_deletion(
             staged_path,
             uploads_dir / original_name,
@@ -865,11 +885,14 @@ def _stage_primary_deletion(
     base_dir: Path,
     primary_path: Path,
     identity: UploadIdentity,
+    *,
+    recover_on_crash: bool = True,
 ) -> tuple[Path, UploadStageLease]:
     """Atomically move the selected entry into the protected conversion namespace."""
     staging_dir = ensure_conversion_dir(base_dir)
+    intent = _UPLOAD_DELETION_RESTORE_INTENT if recover_on_crash else _UPLOAD_DELETION_DISCARD_INTENT
     while True:
-        transaction_dir = staging_dir / (f"{UPLOAD_DELETION_TRANSACTION_PREFIX}{identity.inode:x}-{secrets.token_hex(16)}{UPLOAD_STAGING_SUFFIX}")
+        transaction_dir = staging_dir / (f"{UPLOAD_DELETION_TRANSACTION_PREFIX}{intent}-{identity.inode:x}-{secrets.token_hex(16)}{UPLOAD_STAGING_SUFFIX}")
         stage_lease = UploadStageLease.acquire(staging_dir, transaction_dir.name)
         try:
             transaction_dir.mkdir(mode=0o700)
diff --git a/backend/packages/harness/deerflow/uploads/sandbox_sync.py b/backend/packages/harness/deerflow/uploads/sandbox_sync.py
index 759fea7e3..b49530ff8 100644
--- a/backend/packages/harness/deerflow/uploads/sandbox_sync.py
+++ b/backend/packages/harness/deerflow/uploads/sandbox_sync.py
@@ -9,6 +9,10 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+from deerflow.sandbox.sandbox_provider import (
+    sandbox_provider_uses_thread_data_mounts,
+    sandbox_provider_uses_thread_data_mounts_async,
+)
 from deerflow.uploads.async_helpers import run_upload_io_cancellation_safe, wait_for_task_completion
 from deerflow.uploads.layout import conversion_virtual_path
 from deerflow.uploads.manager import make_upload_file_sandbox_readable, upload_virtual_path
@@ -66,9 +70,11 @@ def prepare_upload_deletion(
     user_id: str | None,
 ) -> Callable[[str], None] | None:
     """Return a lease-safe remote deletion hook for an explicitly synced sandbox."""
-    if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+    if sandbox_provider_uses_thread_data_mounts(sandbox_provider):
         return None
     sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
+    if sandbox_provider_uses_thread_data_mounts(sandbox_provider, refresh=False):
+        return None
     sandbox = sandbox_provider.get(sandbox_id)
     if sandbox is None:
         raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
@@ -82,9 +88,11 @@ async def prepare_upload_deletion_async(
     user_id: str | None,
 ) -> Callable[[str], None] | None:
     """Async counterpart that keeps remote acquisition off the event loop."""
-    if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+    if await sandbox_provider_uses_thread_data_mounts_async(sandbox_provider):
         return None
     sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=user_id)
+    if sandbox_provider_uses_thread_data_mounts(sandbox_provider, refresh=False):
+        return None
     sandbox = sandbox_provider.get(sandbox_id)
     if sandbox is None:
         raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
@@ -133,11 +141,14 @@ def make_upload_paths_available(
 ) -> 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):
+    if sandbox_provider_uses_thread_data_mounts(sandbox_provider):
         _make_paths_readable(sync_paths)
         return SandboxSyncReceipt(sandbox=None)
 
     sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
+    if sandbox_provider_uses_thread_data_mounts(sandbox_provider, refresh=False):
+        _make_paths_readable(sync_paths)
+        return SandboxSyncReceipt(sandbox=None)
     sandbox = sandbox_provider.get(sandbox_id)
     if sandbox is None:
         raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
@@ -153,11 +164,14 @@ async def make_upload_paths_available_async(
 ) -> 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):
+    if await sandbox_provider_uses_thread_data_mounts_async(sandbox_provider):
         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)
+    if sandbox_provider_uses_thread_data_mounts(sandbox_provider, refresh=False):
+        await run_upload_io_cancellation_safe(_make_paths_readable, sync_paths)
+        return SandboxSyncReceipt(sandbox=None)
     sandbox = sandbox_provider.get(sandbox_id)
     if sandbox is None:
         raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py
index c376f9784..6d1b8a320 100644
--- a/backend/tests/test_aio_sandbox_provider.py
+++ b/backend/tests/test_aio_sandbox_provider.py
@@ -94,6 +94,23 @@ def test_remote_mount_override_uses_current_provisioner_contract():
     assert provider.uses_thread_data_mounts is True
 
 
+def test_remote_mount_mode_refreshes_stale_current_contract(monkeypatch):
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider")
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
+    provider._config = {"thread_data_mounts": True}
+    provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    provider._backend._mount_contract_version = aio_mod.SANDBOX_MOUNT_CONTRACT_VERSION
+    provider._backend._mount_contract_capability_known = True
+    provider._backend._capability_next_probe_at = 0.0
+    response = MagicMock(status_code=404)
+    monkeypatch.setattr(remote_mod.requests, "get", MagicMock(return_value=response))
+
+    assert provider_mod.sandbox_provider_uses_thread_data_mounts(provider) is False
+    remote_mod.requests.get.assert_called_once()
+
+
 # ── ensure_thread_dirs ───────────────────────────────────────────────────────
 
 
@@ -809,6 +826,64 @@ async def test_remote_warm_pool_revalidates_current_contract_before_reclaim_asyn
     assert calls[0]["sandbox_id"] == "sandbox-warm-async"
 
 
+def test_remote_active_cache_replaces_client_when_validated_url_changes(tmp_path, monkeypatch):
+    provider, old_client, aio_mod = _make_provider_with_active_sandbox(tmp_path, "sandbox-active-url")
+    provider._thread_sandboxes = {("alice", "thread-1"): "sandbox-active-url"}
+
+    def create(thread_id, sandbox_id, **_kwargs):
+        return aio_mod.SandboxInfo(
+            sandbox_id=sandbox_id,
+            sandbox_url="http://new-sandbox-host",
+            user_id="alice",
+            thread_id=thread_id,
+            mount_contract_version=2,
+        )
+
+    provider._backend = SimpleNamespace(
+        requires_create_validation=True,
+        is_alive=MagicMock(return_value=True),
+        create=create,
+    )
+    provider._ensure_skills_projection = lambda _user_id: None
+    provider._get_extra_mounts = lambda *_args, **_kwargs: []
+    provider._lark_integration_active = lambda _user_id: False
+    provider._lark_broker_active = lambda _user_id: False
+    new_client = SimpleNamespace(id="sandbox-active-url", base_url="http://new-sandbox-host", close=MagicMock())
+    monkeypatch.setattr(aio_mod, "AioSandbox", lambda **_kwargs: new_client)
+
+    assert provider._acquire_internal("thread-1", user_id="alice") == "sandbox-active-url"
+    assert provider.get("sandbox-active-url") is new_client
+    old_client.close.assert_called_once_with()
+
+
+def test_remote_sandbox_id_changes_with_negotiated_mount_contract(tmp_path):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    provider = _make_provider(tmp_path)
+    provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    provider._backend._mount_contract_version = 0
+    legacy_id = provider._sandbox_id_for_thread("thread-1", "default")
+
+    provider._backend._mount_contract_version = 2
+    current_id = provider._sandbox_id_for_thread("thread-1", "default")
+
+    assert legacy_id != current_id
+
+
+def test_cached_legacy_sandbox_is_not_reused_after_contract_upgrade(tmp_path):
+    provider, _sandbox, _ = _make_provider_with_active_sandbox(tmp_path, "legacy-id")
+    provider._thread_sandboxes = {("default", "thread-1"): "legacy-id"}
+
+    assert (
+        provider._reuse_in_process_sandbox(
+            "thread-1",
+            user_id="default",
+            expected_sandbox_id="current-id",
+        )
+        is None
+    )
+    assert ("default", "thread-1") not in provider._thread_sandboxes
+
+
 @pytest.mark.anyio
 async def test_acquire_async_uses_async_readiness_polling(monkeypatch):
     """AioSandboxProvider async creation must not use sync readiness polling."""
@@ -1588,7 +1663,7 @@ def test_aio_forced_collision_never_overwrites_active_tenant(
     monkeypatch.setattr(
         aio_mod.AioSandboxProvider,
         "_deterministic_sandbox_id",
-        staticmethod(lambda thread_id, user_id: "deadbeefdeadbeef"),
+        staticmethod(lambda thread_id, user_id, mount_contract_version=aio_mod.SANDBOX_MOUNT_CONTRACT_VERSION: "deadbeefdeadbeef"),
     )
 
     sandbox_id = provider.acquire("thread-a", user_id="user-a")
diff --git a/backend/tests/test_artifacts_router.py b/backend/tests/test_artifacts_router.py
index d3485d76c..388fcb5d8 100644
--- a/backend/tests/test_artifacts_router.py
+++ b/backend/tests/test_artifacts_router.py
@@ -225,6 +225,39 @@ def test_update_artifact_syncs_non_mounted_sandbox(tmp_path, monkeypatch) -> Non
 
     assert provider.sandbox.updates == [("/mnt/user-data/outputs/note.txt", b"after")]
     assert provider.released == ["sandbox-1"]
+
+
+def test_update_artifact_rechecks_mount_mode_after_acquire(tmp_path, monkeypatch) -> None:
+    artifact_path = tmp_path / "note.txt"
+    artifact_path.write_text("before", encoding="utf-8")
+
+    class ProviderThatBecomesMounted(_RemoteSandboxProvider):
+        async def acquire_async(self, _thread_id: str, *, user_id: str | None = None) -> str:
+            self.uses_thread_data_mounts = True
+            return "sandbox-1"
+
+        def get(self, sandbox_id: str):
+            raise AssertionError("mounted mode must not explicitly synchronize artifacts")
+
+    provider = ProviderThatBecomesMounted()
+    _patch_artifact_update_dependencies(monkeypatch, artifact_path, provider)
+
+    response = asyncio.run(
+        call_unwrapped(
+            artifacts_router.update_artifact,
+            "thread-1",
+            "mnt/user-data/outputs/note.txt",
+            artifacts_router.ArtifactUpdateRequest(
+                content="after",
+                expected_sha256=_artifact_sha256("before"),
+            ),
+            _make_request(),
+        )
+    )
+
+    assert artifact_path.read_text(encoding="utf-8") == "after"
+    assert response.sha256 == _artifact_sha256("after")
+    assert provider.released == ["sandbox-1"]
     assert artifact_path.read_text(encoding="utf-8") == "after"
 
 
diff --git a/backend/tests/test_remote_sandbox_backend.py b/backend/tests/test_remote_sandbox_backend.py
index bb24a1e4f..d45b51cce 100644
--- a/backend/tests/test_remote_sandbox_backend.py
+++ b/backend/tests/test_remote_sandbox_backend.py
@@ -166,13 +166,23 @@ def test_create_delegates_to_provisioner_create(monkeypatch, expected_user_id):
     backend._mount_contract_version = 2
     expected = SandboxInfo(sandbox_id="abc123", sandbox_url="http://k3s:31001")
 
-    def mock_create(thread_id: str, sandbox_id: str, extra_mounts=None, *, user_id=None, provision_lark_cli_runtime=False, provision_lark_cli_broker=False):
+    def mock_create(
+        thread_id: str,
+        sandbox_id: str,
+        extra_mounts=None,
+        *,
+        user_id=None,
+        provision_lark_cli_runtime=False,
+        provision_lark_cli_broker=False,
+        required_mount_contract_version=None,
+    ):
         assert thread_id == "thread-1"
         assert sandbox_id == "abc123"
         assert extra_mounts == [("/host", "/container", False)]
         assert user_id == expected_user_id
         assert provision_lark_cli_runtime is True
         assert provision_lark_cli_broker is False
+        assert required_mount_contract_version == 2
         return expected
 
     monkeypatch.setattr(backend, "_provisioner_create", mock_create)
@@ -275,6 +285,45 @@ def test_current_contract_create_returns_verified_identity(monkeypatch):
     )
 
 
+def test_create_keeps_v2_response_validation_when_capability_changes_during_post(monkeypatch):
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 2
+    backend._mount_contract_capability_known = True
+    monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda _user_id: False)
+
+    def old_peer_response(*_args, **_kwargs):
+        backend._mount_contract_version = 0
+        backend._mount_contract_capability_known = False
+        return _StubResponse(
+            payload={
+                "sandbox_id": "abc123",
+                "sandbox_url": "http://legacy.local",
+            }
+        )
+
+    monkeypatch.setattr(requests, "post", old_peer_response)
+
+    with pytest.raises(RuntimeError, match="mount contract response"):
+        backend.create("thread-1", "abc123", user_id="alice")
+
+
+def test_capability_retry_backoff_does_not_overflow_after_long_outage(monkeypatch):
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._capability_probe_failures = 1024
+    backend._capability_next_probe_at = 0.0
+    now = [100.0]
+    monkeypatch.setattr(remote_backend_mod.time, "monotonic", lambda: now[0])
+    monkeypatch.setattr(
+        requests,
+        "get",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(requests.ConnectionError("still down")),
+    )
+
+    assert backend.refresh_capabilities_if_stale() is True
+    assert backend._capability_probe_failures == 1025
+    assert backend._capability_next_probe_at == 130.0
+
+
 def test_provisioner_create_returns_sandbox_info(monkeypatch):
     backend = RemoteSandboxBackend("http://provisioner:8002")
     monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda user_id: True)
diff --git a/backend/tests/test_upload_sandbox_sync.py b/backend/tests/test_upload_sandbox_sync.py
new file mode 100644
index 000000000..672414548
--- /dev/null
+++ b/backend/tests/test_upload_sandbox_sync.py
@@ -0,0 +1,82 @@
+import asyncio
+import stat
+
+from deerflow.uploads.sandbox_sync import (
+    make_upload_paths_available,
+    make_upload_paths_available_async,
+    prepare_upload_deletion,
+    prepare_upload_deletion_async,
+)
+
+
+class _ProviderThatBecomesMounted:
+    def __init__(self) -> None:
+        self.uses_thread_data_mounts = False
+        self.get_calls = 0
+
+    def refresh_thread_data_mount_capabilities(self) -> bool:
+        return False
+
+    def acquire(self, _thread_id: str, *, user_id: str | None = None) -> str:
+        self.uses_thread_data_mounts = True
+        return "mounted-sandbox"
+
+    async def acquire_async(self, _thread_id: str, *, user_id: str | None = None) -> str:
+        self.uses_thread_data_mounts = True
+        return "mounted-sandbox"
+
+    def get(self, _sandbox_id: str):
+        self.get_calls += 1
+        raise AssertionError("mounted mode must not explicitly synchronize files")
+
+
+def test_prepare_deletion_rechecks_mount_mode_after_acquire():
+    provider = _ProviderThatBecomesMounted()
+
+    assert prepare_upload_deletion(provider, "thread-1", user_id="alice") is None
+    assert provider.get_calls == 0
+
+
+def test_prepare_deletion_async_rechecks_mount_mode_after_acquire():
+    provider = _ProviderThatBecomesMounted()
+
+    assert asyncio.run(prepare_upload_deletion_async(provider, "thread-1", user_id="alice")) is None
+    assert provider.get_calls == 0
+
+
+def test_make_paths_available_rechecks_mount_mode_after_acquire(tmp_path):
+    provider = _ProviderThatBecomesMounted()
+    path = tmp_path / "notes.txt"
+    path.write_text("notes", encoding="utf-8")
+    path.chmod(0o600)
+
+    receipt = make_upload_paths_available(
+        provider,
+        "thread-1",
+        user_id="alice",
+        paths=[(path, "/mnt/user-data/uploads/notes.txt")],
+    )
+
+    assert receipt.sandbox is None
+    assert stat.S_IMODE(path.stat().st_mode) == 0o644
+    assert provider.get_calls == 0
+
+
+def test_make_paths_available_async_rechecks_mount_mode_after_acquire(tmp_path):
+    provider = _ProviderThatBecomesMounted()
+    path = tmp_path / "notes.txt"
+    path.write_text("notes", encoding="utf-8")
+    path.chmod(0o600)
+
+    receipt = asyncio.run(
+        make_upload_paths_available_async(
+            provider,
+            "thread-1",
+            user_id="alice",
+            paths=[(path, "/mnt/user-data/uploads/notes.txt")],
+        )
+    )
+
+    assert receipt.sandbox is None
+    assert stat.S_IMODE(path.stat().st_mode) == 0o644
+    assert provider.get_calls == 0
diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py
index db898cebd..0e22d45bf 100644
--- a/backend/tests/test_uploads_manager.py
+++ b/backend/tests/test_uploads_manager.py
@@ -523,6 +523,17 @@ class TestUploadPublication:
         assert publication.path.read_bytes() == b"old"
         assert owned_conversion.read_text(encoding="utf-8") == "generated"
 
+    def test_successful_rollback_removes_deletion_transaction_directory(self, tmp_path):
+        publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
+        conversion_dir = conversion_path_for_upload(publication.path).parent
+        try:
+            rollback_published_upload(publication)
+        finally:
+            publication.release()
+
+        assert not publication.path.exists()
+        assert not list(conversion_dir.glob(".upload-delete-*.part"))
+
     def test_staging_unlink_failure_is_not_reported_as_success(self, tmp_path):
         staged = create_upload_staging_file(tmp_path)
         staged.handle.write(b"payload")
@@ -892,6 +903,34 @@ class TestCleanupStaleUploadStagingFiles:
         assert primary.read_bytes() == b"original"
         assert not staged_path.exists()
 
+    def test_restores_legacy_intentless_deletion_transaction(self, tmp_path):
+        import deerflow.uploads.manager as upload_manager_module
+
+        uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        primary = uploads / "report.pdf"
+        primary.write_bytes(b"original")
+        identity = UploadIdentity.from_path(primary)
+        staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
+            uploads,
+            primary,
+            identity,
+        )
+        stage_lease.release()
+        legacy_transaction_dir = staged_path.parent.with_name(
+            staged_path.parent.name.replace(
+                ".upload-delete-restore-",
+                ".upload-delete-",
+                1,
+            )
+        )
+        staged_path.parent.rename(legacy_transaction_dir)
+        legacy_staged_path = legacy_transaction_dir / staged_path.name
+
+        assert cleanup_stale_upload_staging_files(tmp_path) == 1
+        assert primary.read_bytes() == b"original"
+        assert not legacy_staged_path.exists()
+
     def test_refuses_replaced_deletion_tombstone_after_crash(self, tmp_path):
         import deerflow.uploads.manager as upload_manager_module
 
@@ -915,6 +954,26 @@ class TestCleanupStaleUploadStagingFiles:
         assert not primary.exists()
         assert staged_path.read_bytes() == b"replacement"
 
+    def test_crashed_upload_rollback_finishes_deletion_instead_of_restoring(self, tmp_path):
+        import deerflow.uploads.manager as upload_manager_module
+
+        uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        primary = uploads / "failed.pdf"
+        primary.write_bytes(b"never committed")
+        identity = UploadIdentity.from_path(primary)
+        staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
+            uploads,
+            primary,
+            identity,
+            recover_on_crash=False,
+        )
+        stage_lease.release()
+
+        assert cleanup_stale_upload_staging_files(tmp_path) == 1
+        assert not primary.exists()
+        assert not staged_path.exists()
+
 
 # ---------------------------------------------------------------------------
 # delete_file_safe
diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py
index 3c7e89b49..a9a47cd49 100644
--- a/backend/tests/test_uploads_router.py
+++ b/backend/tests/test_uploads_router.py
@@ -189,6 +189,44 @@ def test_sandbox_sync_failure_rolls_back_published_generation(tmp_path):
     assert not (thread_uploads_dir / "notes.txt").exists()
 
 
+def test_upload_rechecks_mount_mode_after_sandbox_acquire(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    class ProviderThatBecomesMounted:
+        uses_thread_data_mounts = False
+
+        @staticmethod
+        def refresh_thread_data_mount_capabilities() -> bool:
+            return False
+
+        async def acquire_async(self, _thread_id, *, user_id=None):
+            self.uses_thread_data_mounts = True
+            return "mounted-sandbox"
+
+        @staticmethod
+        def get(_sandbox_id):
+            raise AssertionError("mounted mode must not explicitly synchronize uploads")
+
+    provider = ProviderThatBecomesMounted()
+    with (
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=provider),
+    ):
+        result = asyncio.run(
+            call_unwrapped(
+                uploads.upload_files,
+                "thread-mounted-after-acquire",
+                request=MagicMock(),
+                files=[UploadFile(filename="notes.txt", file=BytesIO(b"payload"))],
+                config=SimpleNamespace(),
+            )
+        )
+
+    assert result.success is True
+    assert (thread_uploads_dir / "notes.txt").read_bytes() == b"payload"
+
+
 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)