fix: close upload rollback and rollout races

README.md

@@ -989,8 +989,12 @@ can set `sandbox.thread_data_mounts: true` to skip that per-upload sandbox
 acquire and sync. Leave the field unset for automatic detection; setting it
 incorrectly can make uploaded files unavailable inside the sandbox. During a
 mixed-version rollout, a new Gateway probes the Provisioner mount-contract
-version and temporarily falls back to explicit synchronization when the peer
-does not yet advertise the current contract.
+version. A confirmed legacy peer falls back to explicit synchronization for
+the default no-auth user, but authenticated user buckets fail closed because
+legacy Provisioners cannot isolate equal thread IDs across users. Upgrade the
+Provisioner first for authenticated deployments. If the capability probe is
+temporarily unreachable, explicitly requested mounted mode also fails closed
+instead of creating a Pod whose read-only conversion mount cannot be verified.
 
 This is the difference between a chatbot with tool access and an agent with an actual execution environment.
 

backend/AGENTS.md

@@ -700,7 +700,7 @@ that cannot tell sibling branches apart.
 **Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved.
 **Implementations**:
 - `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` file APIs honour the `/mnt/user-data` contract uniformly with AIO. The more-specific `/mnt/user-data/.upload-conversions` mapping rejects writes through those structured file APIs even though the aggregate `/mnt/user-data` mapping is writable. This is not an OS isolation boundary: explicitly enabled Local host bash operates on host paths outside `PathMapping` write enforcement and must remain disabled for untrusted tasks. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Public, custom, legacy, and managed integration skill mappings point at stable enabled-only projection roots rather than raw skill directories.
-- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths — it never touches the ownership store (that would be blocking IO on the event loop); ownership is published on acquire/reclaim and refreshed off the event loop by the dedicated renewal thread (`_renew_owned_leases`). `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner backends=False), while the optional `sandbox.thread_data_mounts` boolean requests mounted mode for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Remote mounted mode additionally requires the Provisioner to advertise the current mount-contract version; a new Gateway talking to an older peer automatically falls back to explicit synchronization. Thread mounts create and expose `.upload-conversions` explicitly as read-only, separate from the writable uploads mount. Non-mounted providers omit that nested read-only mount and instead synchronize only the requested generated file into the writable sandbox copy. Local-container and hostPath-provisioner mounts use the same stable skill projection roots; PVC-backed skills remain governed by the operator-supplied PVC layout until PVC materialization is implemented. Readiness probes and `agent_sandbox` clients classify loopback/private IPs, single-label cluster hosts, and Docker/Podman internal hostnames as direct control-plane destinations and set `trust_env=False`; external FQDNs and public IPs retain environment proxy support.
+- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths — it never touches the ownership store (that would be blocking IO on the event loop); ownership is published on acquire/reclaim and refreshed off the event loop by the dedicated renewal thread (`_renew_owned_leases`). `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner backends=False), while the optional `sandbox.thread_data_mounts` boolean requests mounted mode for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Remote mounted mode additionally requires a reachable Provisioner advertising the current mount-contract version; an unavailable probe fails closed, while a confirmed legacy peer may fall back to explicit synchronization only for the `default` no-auth user. Thread mounts create and expose `.upload-conversions` explicitly as read-only, separate from the writable uploads mount. Non-mounted providers omit that nested read-only mount and instead synchronize only the requested generated file into the writable sandbox copy. Local-container and hostPath-provisioner mounts use the same stable skill projection roots; PVC-backed skills remain governed by the operator-supplied PVC layout until PVC materialization is implemented. Readiness probes and `agent_sandbox` clients classify loopback/private IPs, single-label cluster hosts, and Docker/Podman internal hostnames as direct control-plane destinations and set `trust_env=False`; external FQDNs and public IPs retain environment proxy support.
 - `E2BSandboxProvider` (`packages/harness/deerflow/community/e2b_sandbox/`) provides E2B remote isolation.
   New sandboxes receive a one-shot upload from the enabled-only public, custom,
   legacy, and managed integration projections. Existing E2B VMs keep their
@@ -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. Alias-based deletion resolves the primary's actual directory entry by inode, requires that entry to remain exclusive, and only then derives a long-name conversion path. 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 the protected conversion namespace before revalidating its identity, so a pathname replacement between check and removal is restored rather than unlinked. 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. 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; otherwise the Gateway uses explicit synchronization and does not send the nested read-only conversion mount. 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. `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 reachable Provisioner advertising the current mount contract; an unavailable capability probe fails closed instead of being treated as confirmed legacy. A confirmed legacy peer may use explicit synchronization 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. Remote acquisition deliberately replays the idempotent create request instead of trusting GET discovery, so an existing Pod must match the complete mounts requested by the current run. `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.
 - Agent receives uploaded file list via `UploadsMiddleware`
 
 See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details.

backend/docs/FILE_UPLOAD.md

@@ -188,7 +188,8 @@ read_file(path="/mnt/user-data/.upload-conversions/document.pdf.md")
 - 非挂载同步副本是沙箱私有副本；任一路径失败（包括远端已落盘但传输随后报错）、后续响应构造失败或请求取消时，会对本次尝试的精确远端路径执行幂等撤销，再回滚宿主文件
 - 嵌入式 `DeerFlowClient.upload_files()` 以整批为事务边界：后续文件失败会逆序撤销本次调用中此前成功的所有远端副本和宿主 generation
 - 如果 Gateway 与远端沙箱保证挂载同一份线程 user-data（例如正确对齐的共享 PVC、NFS 或 hostPath），可设置 `sandbox.thread_data_mounts: true`；只有 Provisioner 能通过 `/api/capabilities` 证明当前挂载契约时，上传路由才会跳过 sandbox acquire 和逐文件同步
-- 新 Gateway 与旧 Provisioner 混合部署时会自动降级为显式同步并省略嵌套只读转换挂载，因此组件可按任一顺序滚动升级
+- 新 Gateway 与已确认的旧 Provisioner 混合部署时，只有 `default` 无认证用户可降级为显式同步；旧 Provisioner 的主挂载不包含 `user_id`，认证用户必须先升级 Provisioner，否则创建沙箱会 fail closed
+- `/api/capabilities` 暂时不可达时，显式请求的挂载模式 fail closed，不会把缺少嵌套只读转换挂载的 Pod 标记为当前契约；远端获取会通过幂等创建重新校验本次请求的完整 Pod 挂载签名，而不是只信任 discovery 响应
 - 不确定挂载关系时应省略该配置并保留自动检测。错误地设为 `true` 会导致文件只存在于 Gateway 存储、沙箱内不可见
 
 ## 测试示例

backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py

@@ -269,6 +269,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         mounted = override if override is not None else backend is None or isinstance(backend, LocalContainerBackend)
         if mounted and isinstance(backend, RemoteSandboxBackend):
             if backend.mount_contract_version < SANDBOX_MOUNT_CONTRACT_VERSION:
+                if not backend.mount_contract_capability_known:
+                    raise RuntimeError("Provisioner mount compatibility could not be verified; retry after the Provisioner capability endpoint is reachable")
                 logger.warning(
                     "Configured remote thread-data mounts require Provisioner mount contract v%s; peer advertises v%s, so uploads will use explicit synchronization during the rolling upgrade",
                     SANDBOX_MOUNT_CONTRACT_VERSION,
@@ -1936,7 +1938,16 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                     return cached_id
 
                 # Backend discovery: another process may have created the container.
-                discovered = self._backend.discover(sandbox_id)
+                discovered = (
+                    None
+                    if getattr(
+                        self._backend,
+                        "requires_create_validation",
+                        False,
+                    )
+                    is True
+                    else self._backend.discover(sandbox_id)
+                )
                 if discovered is not None:
                     return self._register_discovered_sandbox(thread_id, discovered, user_id=effective_user_id)
 
@@ -1965,7 +1976,19 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
             # Backend discovery is sync because local discovery may inspect
             # Docker and perform a health check; keep it off the event loop.
-            discovered = await asyncio.to_thread(self._backend.discover, sandbox_id)
+            discovered = (
+                None
+                if getattr(
+                    self._backend,
+                    "requires_create_validation",
+                    False,
+                )
+                is True
+                else await asyncio.to_thread(
+                    self._backend.discover,
+                    sandbox_id,
+                )
+            )
             if discovered is not None:
                 # Registration publishes ownership, which is blocking store IO
                 # (filesystem or network depending on the backend) — same reason

backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py

@@ -24,7 +24,7 @@ from pathlib import Path, PureWindowsPath
 
 import requests
 
-from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.runtime.user_context import DEFAULT_USER_ID, get_effective_user_id
 from deerflow.skills.storage import user_should_see_legacy_skills
 
 from .backend import SandboxBackend
@@ -144,6 +144,7 @@ class RemoteSandboxBackend(SandboxBackend):
         self._provisioner_url = provisioner_url.rstrip("/")
         self._api_key = api_key
         self._mount_contract_version = 0
+        self._mount_contract_capability_known = False
 
     @property
     def provisioner_url(self) -> str:
@@ -157,6 +158,16 @@ class RemoteSandboxBackend(SandboxBackend):
         """Provisioner mount contract negotiated during provider startup."""
         return self._mount_contract_version
 
+    @property
+    def mount_contract_capability_known(self) -> bool:
+        """Whether the peer definitively answered capability negotiation."""
+        return self._mount_contract_capability_known
+
+    @property
+    def requires_create_validation(self) -> bool:
+        """Require idempotent POST so the Provisioner checks requested mounts."""
+        return True
+
     def probe_capabilities(self) -> None:
         """Negotiate optional Provisioner capabilities without breaking old peers."""
         try:
@@ -165,13 +176,19 @@ class RemoteSandboxBackend(SandboxBackend):
                 headers=self._auth_headers(),
                 timeout=5,
             )
+            if getattr(response, "status_code", None) == 404:
+                self._mount_contract_version = 0
+                self._mount_contract_capability_known = True
+                return
             response.raise_for_status()
             payload = response.json()
             version = payload.get("mount_contract_version", 0) if isinstance(payload, dict) else 0
             self._mount_contract_version = version if isinstance(version, int) and version >= 0 else 0
+            self._mount_contract_capability_known = True
         except (requests.RequestException, ValueError, TypeError):
             self._mount_contract_version = 0
-            logger.warning("Provisioner mount capabilities are unavailable; using explicit upload synchronization for rolling-upgrade compatibility")
+            self._mount_contract_capability_known = False
+            logger.warning("Provisioner mount capabilities are unavailable; mounted thread data will fail closed until compatibility can be verified")
 
     # ── SandboxBackend interface ──────────────────────────────────────────
 
@@ -190,6 +207,9 @@ class RemoteSandboxBackend(SandboxBackend):
         Calls ``POST /api/sandboxes`` which creates a dedicated Pod +
         NodePort Service in k3s.
         """
+        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")
         return self._provisioner_create(
             thread_id,
             sandbox_id,

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

@@ -22,6 +22,7 @@ from deerflow.uploads.layout import (
     UPLOAD_CONVERSIONS_DIRNAME,
     _truncate_utf8,
     artifact_url_for_virtual_path,
+    ensure_conversion_dir,
     existing_conversion_path_for_upload,
     upload_virtual_path,
 )
@@ -460,9 +461,29 @@ def rollback_published_upload(publication: PublishedUpload) -> None:
     if not publication.identity.matches(publication.path):
         return
     owned_conversion = existing_conversion_path_for_upload(publication.path)
-    if owned_conversion is not None:
-        owned_conversion.unlink(missing_ok=True)
-    publication.path.unlink()
+    try:
+        staged_path, stage_lease = _stage_primary_deletion(
+            publication.path.parent,
+            publication.path,
+            publication.identity,
+        )
+    except (FileNotFoundError, UnsafeUploadPathError):
+        # The pathname was replaced after the optimistic identity check. The
+        # staging helper restores that replacement without deleting it.
+        return
+    try:
+        if owned_conversion is not None:
+            owned_conversion.unlink(missing_ok=True)
+        staged_path.unlink()
+    except BaseException:
+        _restore_staged_deletion(
+            staged_path,
+            publication.path,
+            publication.identity,
+        )
+        raise
+    finally:
+        stage_lease.release()
 
 
 def replace_system_owned_staged_file(staged: StagedUpload, filename: str) -> Path:
@@ -650,35 +671,76 @@ def _restore_staged_deletion(
     staged_path.unlink()
 
 
+def _restore_unexpected_staged_entry(staged_path: Path, original_path: Path) -> None:
+    """Restore an entry moved during an identity race without unlinking it."""
+    try:
+        os.link(staged_path, original_path, follow_symlinks=False)
+    except FileExistsError:
+        # A non-cooperating writer recreated the original name while the entry
+        # was staged. Preserve the moved payload under a collision-resistant
+        # visible recovery name instead of deleting either generation.
+        while True:
+            recovery_path = original_path.with_name(f"{original_path.stem}_recovered_{secrets.token_hex(8)}{original_path.suffix}")
+            try:
+                os.link(staged_path, recovery_path, follow_symlinks=False)
+                break
+            except FileExistsError:
+                continue
+        logger.warning(
+            "Upload entry changed during deletion; preserved the raced entry as %s",
+            recovery_path,
+        )
+    staged_path.unlink()
+
+
 def _stage_primary_deletion(
     base_dir: Path,
     primary_path: Path,
     identity: UploadIdentity,
 ) -> tuple[Path, UploadStageLease]:
-    """Move the selected directory entry behind a leased no-replace hard link."""
+    """Atomically move the selected entry into the protected conversion namespace."""
+    staging_dir = ensure_conversion_dir(base_dir)
     while True:
-        staged_path = base_dir / (f"{UPLOAD_STAGING_PREFIX}delete-{secrets.token_hex(16)}{UPLOAD_STAGING_SUFFIX}")
-        stage_lease = UploadStageLease.acquire(base_dir, staged_path.name)
+        staged_path = staging_dir / (f"{UPLOAD_STAGING_PREFIX}delete-{secrets.token_hex(16)}{UPLOAD_STAGING_SUFFIX}")
+        stage_lease = UploadStageLease.acquire(staging_dir, staged_path.name)
         try:
-            os.link(primary_path, staged_path, follow_symlinks=False)
-        except FileExistsError:
+            os.lstat(staged_path)
+        except FileNotFoundError:
+            pass
+        else:
             stage_lease.release()
             continue
+
+        try:
+            os.rename(primary_path, staged_path)
         except BaseException:
             stage_lease.release()
             raise
 
         try:
-            primary_path.unlink()
             staged_stat = os.lstat(staged_path)
             if not stat.S_ISREG(staged_stat.st_mode) or staged_stat.st_nlink != 1 or (staged_stat.st_dev, staged_stat.st_ino) != (identity.device, identity.inode):
-                _restore_staged_deletion(staged_path, primary_path, identity)
+                _restore_unexpected_staged_entry(staged_path, primary_path)
                 raise UnsafeUploadPathError("Upload is no longer an exclusive directory entry")
             return staged_path, stage_lease
         except BaseException:
             if staged_path.exists():
                 try:
-                    _restore_staged_deletion(staged_path, primary_path, identity)
+                    staged_stat = os.lstat(staged_path)
+                    if (staged_stat.st_dev, staged_stat.st_ino) == (
+                        identity.device,
+                        identity.inode,
+                    ):
+                        _restore_staged_deletion(
+                            staged_path,
+                            primary_path,
+                            identity,
+                        )
+                    else:
+                        _restore_unexpected_staged_entry(
+                            staged_path,
+                            primary_path,
+                        )
                 except BaseException:
                     logger.warning(
                         "Failed to restore staged upload deletion: %s",
@@ -734,6 +796,8 @@ def delete_file_safe(
 
         identity = UploadIdentity(device=file_stat.st_dev, inode=file_stat.st_ino)
         actual_file_path = _find_upload_path_by_identity(base_dir, identity)
+        if portable_name_coordination_key(actual_file_path.name) != portable_name_coordination_key(safe_name):
+            raise UnsafeUploadPathError("Upload name changed outside the requested generation lease")
         owned_conversion = existing_conversion_path_for_upload(actual_file_path)
         final_stat = os.lstat(actual_file_path)
         if not stat.S_ISREG(final_stat.st_mode) or final_stat.st_nlink != 1 or (final_stat.st_dev, final_stat.st_ino) != (identity.device, identity.inode):

backend/tests/test_aio_sandbox_provider.py

@@ -78,6 +78,7 @@ def test_remote_mount_override_downgrades_without_current_provisioner_contract()
     provider._config = {"thread_data_mounts": True}
     provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
     provider._backend._mount_contract_version = 0
+    provider._backend._mount_contract_capability_known = True
 
     assert provider.uses_thread_data_mounts is False
 
@@ -368,6 +369,7 @@ def test_remote_backend_negotiates_mount_contract_capability(monkeypatch):
     backend.probe_capabilities()
 
     assert backend.mount_contract_version == 2
+    assert backend.mount_contract_capability_known is True
     assert requested == {
         "url": "http://provisioner:8002/api/capabilities",
         "headers": {"X-API-Key": "secret"},
@@ -391,6 +393,51 @@ def test_remote_backend_treats_missing_mount_capability_as_legacy(monkeypatch):
     backend.probe_capabilities()
 
     assert backend.mount_contract_version == 0
+    assert backend.mount_contract_capability_known is True
+
+
+def test_remote_backend_treats_missing_capability_endpoint_as_known_legacy(monkeypatch):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+
+    class _Response:
+        status_code = 404
+
+        def raise_for_status(self):
+            raise AssertionError("404 legacy response should not be raised")
+
+    monkeypatch.setattr(remote_mod.requests, "get", lambda *_args, **_kwargs: _Response())
+    backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+
+    backend.probe_capabilities()
+
+    assert backend.mount_contract_version == 0
+    assert backend.mount_contract_capability_known is True
+
+
+def test_remote_backend_distinguishes_unavailable_capability_probe(monkeypatch):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    monkeypatch.setattr(
+        remote_mod.requests,
+        "get",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(remote_mod.requests.ConnectionError("not ready")),
+    )
+    backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+
+    backend.probe_capabilities()
+
+    assert backend.mount_contract_version == 0
+    assert backend.mount_contract_capability_known is False
+
+
+def test_remote_mount_override_fails_closed_when_capability_probe_is_unavailable():
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_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")
+
+    with pytest.raises(RuntimeError, match="could not be verified"):
+        _ = provider.uses_thread_data_mounts
 
 
 def test_remote_discovery_carries_verified_identity(monkeypatch):
@@ -515,6 +562,58 @@ def test_discover_or_create_only_unlocks_when_lock_succeeds(tmp_path, monkeypatc
     assert unlock_calls == []
 
 
+def test_remote_contract_backend_revalidates_existing_pod_via_create(tmp_path, monkeypatch):
+    """A GET cannot prove the live Pod matches the mounts requested this run."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    provider._backend = SimpleNamespace(
+        requires_create_validation=True,
+        discover=MagicMock(side_effect=AssertionError("unvalidated discovery must be skipped")),
+    )
+    monkeypatch.setattr(provider, "_recheck_cached_sandbox", lambda *_args, **_kwargs: None)
+
+    with patch.object(provider, "_create_sandbox", return_value="sandbox-v2") as create:
+        result = provider._discover_or_create_with_lock(
+            "thread-1",
+            "sandbox-v2",
+            user_id="alice",
+        )
+
+    assert result == "sandbox-v2"
+    provider._backend.discover.assert_not_called()
+    create.assert_called_once_with("thread-1", "sandbox-v2", user_id="alice")
+
+
+@pytest.mark.anyio
+async def test_remote_contract_backend_revalidates_existing_pod_via_create_async(tmp_path, monkeypatch):
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    provider._backend = SimpleNamespace(
+        requires_create_validation=True,
+        discover=MagicMock(side_effect=AssertionError("unvalidated discovery must be skipped")),
+    )
+    monkeypatch.setattr(provider, "_recheck_cached_sandbox", lambda *_args, **_kwargs: None)
+    create_calls: list[tuple[str, str, str | None]] = []
+
+    async def create(thread_id, sandbox_id, *, user_id=None):
+        create_calls.append((thread_id, sandbox_id, user_id))
+        return sandbox_id
+
+    monkeypatch.setattr(provider, "_create_sandbox_async", create)
+
+    result = await provider._discover_or_create_with_lock_async(
+        "thread-1",
+        "sandbox-v2",
+        user_id="alice",
+    )
+
+    assert result == "sandbox-v2"
+    provider._backend.discover.assert_not_called()
+    assert create_calls == [("thread-1", "sandbox-v2", "alice")]
+
+
 @pytest.mark.anyio
 async def test_acquire_async_uses_async_readiness_polling(monkeypatch):
     """AioSandboxProvider async creation must not use sync readiness polling."""
@@ -721,6 +820,7 @@ def test_remote_backend_create_forwards_effective_user_id(monkeypatch):
     """Provisioner mode must receive user_id so PVC subPath matches user isolation."""
     remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
     backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 2
     token = set_current_user(SimpleNamespace(id="user-7"))
     posted: dict = {}
 
@@ -758,6 +858,7 @@ def test_remote_backend_create_prefers_explicit_user_id(monkeypatch):
     """Provisioner mode must not fall back to the ambient default for channel runs."""
     remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
     backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 2
     posted: dict = {}
 
     class _Response:

backend/tests/test_remote_sandbox_backend.py

@@ -163,6 +163,7 @@ def test_user_should_see_legacy_skills_follows_storage_visibility_rule(monkeypat
 @pytest.mark.parametrize("expected_user_id", [None, "owner-1"])
 def test_create_delegates_to_provisioner_create(monkeypatch, expected_user_id):
     backend = RemoteSandboxBackend("http://provisioner:8002")
+    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):
@@ -186,6 +187,23 @@ def test_create_delegates_to_provisioner_create(monkeypatch, expected_user_id):
     assert result == expected
 
 
+def test_create_rejects_nondefault_user_on_unverified_legacy_provisioner(monkeypatch):
+    backend = RemoteSandboxBackend("http://legacy-provisioner:8002")
+    called = False
+
+    def unexpected_create(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        raise AssertionError("unsafe legacy create should not be attempted")
+
+    monkeypatch.setattr(backend, "_provisioner_create", unexpected_create)
+
+    with pytest.raises(RuntimeError, match="cannot isolate user 'alice'"):
+        backend.create("shared-thread", "sandbox-1", user_id="alice")
+
+    assert called is False
+
+
 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)

backend/tests/test_uploads_manager.py

@@ -19,7 +19,7 @@ from deerflow.uploads.layout import (
     conversion_path_for_upload,
     conversion_virtual_path,
 )
-from deerflow.uploads.lease import UploadNameLease
+from deerflow.uploads.lease import UploadIdentity, UploadNameLease
 from deerflow.uploads.manager import (
     AtomicUploadPublishError,
     PathTraversalError,
@@ -479,6 +479,28 @@ class TestUploadPublication:
 
         assert (tmp_path / "report.pdf").read_bytes() == b"new"
 
+    def test_rollback_does_not_unlink_replacement_after_identity_check(self, tmp_path):
+        publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
+        real_matches = UploadIdentity.matches
+        replaced = False
+
+        def replace_after_match(identity, path):
+            nonlocal replaced
+            matches = real_matches(identity, path)
+            if matches and path == publication.path and not replaced:
+                replaced = True
+                path.unlink()
+                path.write_bytes(b"new")
+            return matches
+
+        try:
+            with patch.object(UploadIdentity, "matches", autospec=True, side_effect=replace_after_match):
+                rollback_published_upload(publication)
+        finally:
+            publication.release()
+
+        assert publication.path.read_bytes() == b"new"
+
     def test_rollback_preserves_primary_when_conversion_removal_fails(self, tmp_path):
         publication = publish_upload_bytes_leased(tmp_path, "report.pdf", b"old")
         owned_conversion = conversion_path_for_upload(publication.path)
@@ -938,6 +960,49 @@ class TestDeleteFileSafe:
         assert alias.read_bytes() == b"payload"
         assert conversion.read_text(encoding="utf-8") == "generated"
 
+    def test_delete_rejects_identity_renamed_outside_requested_name_lease(self, tmp_path):
+        import deerflow.uploads.manager as upload_manager_module
+
+        uploads = tmp_path / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        primary = uploads / "report.pdf"
+        primary.write_bytes(b"old")
+        conversion = conversion_path_for_upload(primary)
+        conversion.parent.mkdir()
+        conversion.write_text("old conversion", encoding="utf-8")
+        renamed = uploads / "other.pdf"
+        renamed_conversion = conversion_path_for_upload(renamed)
+        real_find = upload_manager_module._find_upload_path_by_identity
+        remote_names: list[str] = []
+
+        def rename_after_scan(base_dir, identity):
+            found = real_find(base_dir, identity)
+            found.rename(renamed)
+            conversion.rename(renamed_conversion)
+            return renamed
+
+        def publish_replacement(actual_name):
+            remote_names.append(actual_name)
+            publish_upload_bytes(uploads, actual_name, b"new")
+            renamed_conversion.unlink()
+            renamed_conversion.write_text("new conversion", encoding="utf-8")
+
+        with patch.object(
+            upload_manager_module,
+            "_find_upload_path_by_identity",
+            side_effect=rename_after_scan,
+        ):
+            with pytest.raises(UnsafeUploadPathError, match="name changed"):
+                delete_file_safe(
+                    uploads,
+                    primary.name,
+                    delete_remote_copy=publish_replacement,
+                )
+
+        assert remote_names == []
+        assert renamed.read_bytes() == b"old"
+        assert renamed_conversion.read_text(encoding="utf-8") == "old conversion"
+
     def test_delete_removes_owned_conversion_but_preserves_legacy_sibling(self, tmp_path):
         uploads = tmp_path / "user-data" / "uploads"
         uploads.mkdir(parents=True)

config.example.yaml

@@ -1278,7 +1278,9 @@ sandbox:
 #   # a correctly aligned shared PVC, NFS volume, hostPath, or bind mount.
 #   # true skips per-upload sandbox acquire/sync; false forces explicit sync.
 #   # In provisioner mode, true is honored only when /api/capabilities advertises
-#   # the current mount contract; older peers automatically fall back to sync.
+#   # the current mount contract; an unreachable probe fails closed. Confirmed
+#   # legacy peers may fall back to sync only for the default no-auth user;
+#   # authenticated deployments must upgrade the Provisioner first.
 #   # thread_data_mounts: true
 #
 #   # Optional: Additional mount directories from host to container

docker/provisioner/README.md

@@ -88,10 +88,14 @@ When the Gateway mounts that same storage at its DeerFlow home and the PVC
 subpaths align, set `sandbox.thread_data_mounts: true` in the Gateway's
 `config.yaml` to skip redundant upload-time sandbox acquire/sync. Leave the
 field unset when using unrelated storage or when the mount relationship is
-uncertain. The Gateway probes `/api/capabilities`; if the Provisioner does not
-advertise the current `mount_contract_version`, it temporarily uses explicit
-synchronization and omits the new nested conversion mount so either component
-can be upgraded first.
+uncertain. The Gateway probes `/api/capabilities`. A confirmed legacy
+Provisioner can use explicit synchronization only for the `default` no-auth
+user; its primary user-data mount does not isolate equal thread IDs belonging
+to different users, so authenticated deployments must upgrade the Provisioner
+before the Gateway. If the capability endpoint is temporarily unreachable,
+explicitly requested mounted mode fails closed. Remote acquisition uses the
+idempotent create endpoint even for an existing Pod so the Provisioner compares
+the live Pod against the complete mount signature requested by the current run.
 
 **Response**:
 ```json