fix: close sandbox contract and deletion recovery races

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. Remote reuse additionally replays the current idempotent Provisioner create contract (user/thread identity, mounts, and optional runtime/broker flags) before handing out either an active-cache or warm-pool entry. 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 Provisioner advertising the current mount-contract version; unavailable and confirmed-legacy results use explicit synchronization and are re-probed with bounded backoff/cache expiry, while a non-current peer may create thread sandboxes 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.
+- `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. Remote reuse additionally replays the current idempotent Provisioner create contract (user/thread identity, mounts, and optional runtime/broker flags) before handing out either an active-cache or warm-pool entry. 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 Provisioner advertising mount contract v2 or newer; unavailable and confirmed-legacy results use explicit synchronization and are re-probed with bounded backoff/cache expiry, while a legacy peer may create thread sandboxes 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. 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.
+- 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. Once a user deletion stages its primary, its persisted crash intent is commit-on-recovery: startup completes removal of that exact tombstone and its unambiguous host conversion rather than resurrecting a primary after remote or companion deletion may already have occurred; a conversion belonging to a replacement primary is preserved. An exception returned to the live caller still rolls the staged primary back before it is surfaced. Failed-upload rollback tombstones use the same discard recovery, while legacy restore-intent transactions retain basename/recovery-name restoration. Neither transaction kind 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-<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; 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.
+- 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 a compatible mount contract; every upload/delete/artifact decision waits for any in-flight stale capability probe in a worker thread and uses one atomic snapshot. The same frozen version selects the deterministic sandbox ID, mount set, cache/reuse candidate, create request, and response validation; if the peer changes versions during acquisition, the whole attempt restarts with a new ID. After acquisition, consumers use the immutable contract recorded on that sandbox instance rather than the provider's mutable global capability. Unavailable or legacy 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. Contracts newer than v2 are accepted as forward-compatible, but the response must echo the exact frozen version, user, and thread. The negotiated mount-contract version namespaces deterministic AIO sandbox IDs so a container created under another version cannot satisfy the 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.

backend/app/gateway/routers/artifacts.py

@@ -24,7 +24,7 @@ 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,
-    sandbox_provider_uses_thread_data_mounts,
+    sandbox_provider_sandbox_uses_thread_data_mounts,
     sandbox_provider_uses_thread_data_mounts_async,
 )
 from deerflow.utils.thread_id import ThreadId
@@ -469,9 +469,9 @@ async def update_artifact(
             sandbox_provider = get_sandbox_provider()
             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)
-                if not sandbox_provider_uses_thread_data_mounts(
+                if not sandbox_provider_sandbox_uses_thread_data_mounts(
                     sandbox_provider,
-                    refresh=False,
+                    sandbox_id,
                 ):
                     sandbox = sandbox_provider.get(sandbox_id)
                     if sandbox is None:

backend/app/gateway/routers/uploads.py

@@ -16,7 +16,7 @@ from deerflow.config.paths import get_paths
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.sandbox_provider import (
     get_sandbox_provider,
-    sandbox_provider_uses_thread_data_mounts,
+    sandbox_provider_sandbox_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
@@ -417,9 +417,9 @@ async def upload_files(
     sandbox = None
     if sync_to_sandbox:
         sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
-        sync_to_sandbox = not sandbox_provider_uses_thread_data_mounts(
+        sync_to_sandbox = not sandbox_provider_sandbox_uses_thread_data_mounts(
             sandbox_provider,
-            refresh=False,
+            sandbox_id,
         )
         if sync_to_sandbox:
             sandbox = sandbox_provider.get(sandbox_id)

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

@@ -58,7 +58,11 @@ from .ownership import (
     make_sandbox_ownership_store,
     resolve_ownership_config,
 )
-from .remote_backend import RemoteSandboxBackend
+from .remote_backend import (
+    MountContractChangedError,
+    MountContractSnapshot,
+    RemoteSandboxBackend,
+)
 from .sandbox_info import SandboxInfo
 
 SANDBOX_MOUNT_CONTRACT_VERSION = 2
@@ -261,6 +265,15 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         Remote backends may require explicit file sync. Operators can override
         this detection when gateway and remote sandboxes share the same storage.
         """
+        backend = getattr(self, "_backend", None)
+        snapshot = backend.mount_contract_snapshot() if isinstance(backend, RemoteSandboxBackend) else None
+        return self._uses_thread_data_mounts_for_contract(snapshot)
+
+    def _uses_thread_data_mounts_for_contract(
+        self,
+        snapshot: MountContractSnapshot | None,
+    ) -> bool:
+        """Evaluate mounted mode against one immutable remote contract."""
         config = getattr(self, "_config", {})
         backend = getattr(self, "_backend", None)
         override = config.get("thread_data_mounts")
@@ -268,18 +281,53 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         # initialization. Preserve their historical local/mounted default.
         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:
+            effective_snapshot = snapshot or MountContractSnapshot(
+                version=backend.mount_contract_version,
+                capability_known=backend.mount_contract_capability_known,
+            )
+            if effective_snapshot.version < SANDBOX_MOUNT_CONTRACT_VERSION:
+                if not effective_snapshot.capability_known:
                     logger.warning("Provisioner mount compatibility could not be verified; using explicit synchronization until a retry confirms the current contract")
                     return False
                 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,
-                    backend.mount_contract_version,
+                    effective_snapshot.version,
                 )
                 return False
         return mounted
 
+    def thread_data_mounts_mode(self, *, refresh: bool = True) -> bool:
+        """Return mounted mode from one atomic, optionally refreshed snapshot."""
+        backend = getattr(self, "_backend", None)
+        if not isinstance(backend, RemoteSandboxBackend):
+            return self._uses_thread_data_mounts_for_contract(None)
+        snapshot = backend.refresh_capabilities_and_snapshot() if refresh else backend.mount_contract_snapshot()
+        return self._uses_thread_data_mounts_for_contract(snapshot)
+
+    def sandbox_uses_thread_data_mounts(self, sandbox_id: str) -> bool:
+        """Return the mount mode of a specific acquired sandbox instance."""
+        backend = getattr(self, "_backend", None)
+        if not isinstance(backend, RemoteSandboxBackend):
+            return self._uses_thread_data_mounts_for_contract(None)
+        with self._lock:
+            info = self._sandbox_infos.get(sandbox_id)
+        if info is None or info.mount_contract_version is None:
+            return False
+        return self._uses_thread_data_mounts_for_contract(
+            MountContractSnapshot(
+                version=info.mount_contract_version,
+                capability_known=True,
+            )
+        )
+
+    def _acquisition_contract_snapshot(self) -> MountContractSnapshot | None:
+        """Freeze the remote contract used throughout one acquisition attempt."""
+        backend = getattr(self, "_backend", None)
+        if isinstance(backend, RemoteSandboxBackend):
+            return backend.refresh_capabilities_and_snapshot()
+        return None
+
     def refresh_thread_data_mount_capabilities(self) -> bool:
         """Refresh remote mount compatibility when its negotiated value is stale."""
         backend = getattr(self, "_backend", None)
@@ -819,13 +867,20 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
     # ── Mount helpers ────────────────────────────────────────────────────
 
-    def _get_extra_mounts(self, thread_id: str | None, *, user_id: str | None = None) -> list[tuple[str, str, bool]]:
+    def _get_extra_mounts(
+        self,
+        thread_id: str | None,
+        *,
+        user_id: str | None = None,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> list[tuple[str, str, bool]]:
         """Collect all extra mounts for a sandbox (thread-specific + skills)."""
         mounts: list[tuple[str, str, bool]] = []
 
         if thread_id:
             mounts.extend(self._get_thread_mounts(thread_id, user_id=user_id))
-            if not self.uses_thread_data_mounts:
+            mounted = self._uses_thread_data_mounts_for_contract(contract_snapshot) if contract_snapshot is not None else self.uses_thread_data_mounts
+            if not mounted:
                 mounts = [mount for mount in mounts if mount[1] != f"{VIRTUAL_PATH_PREFIX}/{UPLOAD_CONVERSIONS_DIRNAME}"]
             logger.info(f"Adding thread mounts for thread {thread_id}: {mounts}")
 
@@ -1348,14 +1403,21 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 self._thread_locks[key] = threading.Lock()
             return self._thread_locks[key]
 
-    def _sandbox_id_for_thread(self, thread_id: str | None, user_id: str | None) -> str:
+    def _sandbox_id_for_thread(
+        self,
+        thread_id: str | None,
+        user_id: str | None,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Return deterministic IDs for thread sandboxes and random IDs otherwise."""
         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()
+        if contract_snapshot is not None:
+            mount_contract_version = contract_snapshot.version
+        elif isinstance(backend, RemoteSandboxBackend):
+            mount_contract_version = backend.mount_contract_snapshot().version
         return self._deterministic_sandbox_id(
             thread_id,
             self._effective_acquire_user_id(user_id),
@@ -1369,6 +1431,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         user_id: str | None = None,
         post_lock: bool = False,
         expected_sandbox_id: str | None = None,
+        contract_snapshot: MountContractSnapshot | None = None,
     ) -> str | None:
         """Reuse an active in-process sandbox for a thread if one is still tracked."""
         if thread_id is None:
@@ -1413,6 +1476,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             thread_id,
             existing_id,
             user_id=effective_user_id,
+            contract_snapshot=contract_snapshot,
         )
 
         replacement_client = None
@@ -1489,6 +1553,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         sandbox_id: str,
         *,
         user_id: str,
+        contract_snapshot: MountContractSnapshot | None = None,
     ) -> SandboxInfo | None:
         """Replay remote creation so the Provisioner validates current mounts.
 
@@ -1498,14 +1563,23 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         """
         if getattr(self._backend, "requires_create_validation", False) is not True:
             return None
-        extra_mounts = self._get_extra_mounts(thread_id, user_id=user_id)
+        extra_mounts = self._get_extra_mounts(
+            thread_id,
+            user_id=user_id,
+            contract_snapshot=contract_snapshot,
+        )
+        create_kwargs = {
+            "user_id": user_id,
+            "provision_lark_cli_runtime": self._lark_integration_active(user_id),
+            "provision_lark_cli_broker": self._lark_broker_active(user_id),
+        }
+        if isinstance(self._backend, RemoteSandboxBackend) and contract_snapshot is not None:
+            create_kwargs["required_mount_contract_version"] = contract_snapshot.version
         info = self._backend.create(
             thread_id,
             sandbox_id,
             extra_mounts=extra_mounts or None,
-            user_id=user_id,
-            provision_lark_cli_runtime=self._lark_integration_active(user_id),
-            provision_lark_cli_broker=self._lark_broker_active(user_id),
+            **create_kwargs,
         )
         if info.sandbox_id != sandbox_id:
             raise RuntimeError(f"Provisioner validated unexpected sandbox {info.sandbox_id!r}; expected {sandbox_id!r}")
@@ -1518,6 +1592,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         *,
         user_id: str | None = None,
         post_lock: bool = False,
+        contract_snapshot: MountContractSnapshot | None = None,
     ) -> str | None:
         """Promote a warm-pool sandbox back to active tracking if available."""
         if thread_id is None:
@@ -1551,6 +1626,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             thread_id,
             sandbox_id,
             user_id=effective_user_id,
+            contract_snapshot=contract_snapshot,
         )
 
         # Publish ownership before the warm → active transition: a raise here must
@@ -1594,18 +1670,27 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         logger.info(f"Reclaimed warm-pool sandbox {sandbox_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
         return sandbox_id
 
-    def _recheck_cached_sandbox(self, thread_id: str, sandbox_id: str, *, user_id: str) -> str | None:
+    def _recheck_cached_sandbox(
+        self,
+        thread_id: str,
+        sandbox_id: str,
+        *,
+        user_id: str,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> 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,
             expected_sandbox_id=sandbox_id,
+            contract_snapshot=contract_snapshot,
         ) or self._reclaim_warm_pool_sandbox(
             thread_id,
             sandbox_id,
             user_id=user_id,
             post_lock=True,
+            contract_snapshot=contract_snapshot,
         )
 
     def _register_discovered_sandbox(self, thread_id: str, info: SandboxInfo, *, user_id: str) -> str:
@@ -1620,7 +1705,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 prevent. The window is a peer's in-flight container stop, so the
                 thread's next turn discovers nothing and cold-starts cleanly.
         """
-        if isinstance(self._backend, RemoteSandboxBackend) and (info.mount_contract_version != SANDBOX_MOUNT_CONTRACT_VERSION or info.user_id != user_id or info.thread_id != thread_id):
+        if isinstance(self._backend, RemoteSandboxBackend) and (info.mount_contract_version is None or info.mount_contract_version < SANDBOX_MOUNT_CONTRACT_VERSION or info.user_id != user_id or info.thread_id != thread_id):
             raise SandboxIdentityCollisionError(
                 info.sandbox_id,
                 (info.user_id or "unknown", info.thread_id or "unknown"),
@@ -1937,14 +2022,24 @@ 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)
-            with thread_lock:
-                return self._acquire_internal(thread_id, user_id=effective_user_id)
-        else:
-            return self._acquire_internal(thread_id, user_id=effective_user_id)
+        thread_lock = self._get_thread_lock(thread_id, effective_user_id) if thread_id else contextlib.nullcontext()
+        with thread_lock:
+            for attempt in range(3):
+                contract_snapshot = self._acquisition_contract_snapshot()
+                try:
+                    if contract_snapshot is None:
+                        return self._acquire_internal(thread_id, user_id=effective_user_id)
+                    return self._acquire_internal(
+                        thread_id,
+                        user_id=effective_user_id,
+                        contract_snapshot=contract_snapshot,
+                    )
+                except MountContractChangedError:
+                    if attempt == 2:
+                        raise
+                    logger.info("Provisioner mount contract changed during acquire; restarting with a fresh sandbox ID")
+        raise RuntimeError("Sandbox acquisition exhausted mount-contract retries")
 
     async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
         """Acquire a sandbox environment without blocking the event loop.
@@ -1953,19 +2048,42 @@ 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)
             await _acquire_thread_lock_async(thread_lock)
             try:
-                return await self._acquire_internal_async(thread_id, user_id=effective_user_id)
+                return await self._acquire_with_contract_async(thread_id, user_id=effective_user_id)
             finally:
                 thread_lock.release()
 
-        return await self._acquire_internal_async(thread_id, user_id=effective_user_id)
+        return await self._acquire_with_contract_async(thread_id, user_id=effective_user_id)
 
-    def _acquire_internal(self, thread_id: str | None, *, user_id: str) -> str:
+    async def _acquire_with_contract_async(self, thread_id: str | None, *, user_id: str) -> str:
+        """Retry an async acquisition when the Provisioner changes contracts."""
+        for attempt in range(3):
+            contract_snapshot = await asyncio.to_thread(self._acquisition_contract_snapshot)
+            try:
+                if contract_snapshot is None:
+                    return await self._acquire_internal_async(thread_id, user_id=user_id)
+                return await self._acquire_internal_async(
+                    thread_id,
+                    user_id=user_id,
+                    contract_snapshot=contract_snapshot,
+                )
+            except MountContractChangedError:
+                if attempt == 2:
+                    raise
+                logger.info("Provisioner mount contract changed during async acquire; restarting with a fresh sandbox ID")
+        raise RuntimeError("Sandbox acquisition exhausted mount-contract retries")
+
+    def _acquire_internal(
+        self,
+        thread_id: str | None,
+        *,
+        user_id: str,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Internal sandbox acquisition with two-layer consistency.
 
         Layer 1: In-process cache (fastest, covers same-process repeated access)
@@ -1974,7 +2092,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                  is needed — any process can derive the same container name)
         """
         self._ensure_skills_projection(user_id)
-        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
+        if contract_snapshot is None and isinstance(self._backend, RemoteSandboxBackend):
+            contract_snapshot = self._acquisition_contract_snapshot()
+        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id) if contract_snapshot is None else self._sandbox_id_for_thread(thread_id, user_id, contract_snapshot)
         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)
@@ -1983,6 +2103,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 thread_id,
                 user_id=user_id,
                 expected_sandbox_id=expected_cached_id,
+                contract_snapshot=contract_snapshot,
             )
         if cached_id is not None:
             return cached_id
@@ -1993,7 +2114,19 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 self._assert_active_identity_available_locked(sandbox_id, key)
 
         # ── Layer 1.5: Warm pool (container still running, no cold-start) ──
-        reclaimed_id = self._reclaim_warm_pool_sandbox(thread_id, sandbox_id, user_id=user_id)
+        if contract_snapshot is None:
+            reclaimed_id = self._reclaim_warm_pool_sandbox(
+                thread_id,
+                sandbox_id,
+                user_id=user_id,
+            )
+        else:
+            reclaimed_id = self._reclaim_warm_pool_sandbox(
+                thread_id,
+                sandbox_id,
+                user_id=user_id,
+                contract_snapshot=contract_snapshot,
+            )
         if reclaimed_id is not None:
             return reclaimed_id
 
@@ -2002,14 +2135,40 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         # for the same thread_id serialize here: the second process will discover
         # the container started by the first instead of hitting a name-conflict.
         if thread_id:
-            return self._discover_or_create_with_lock(thread_id, sandbox_id, user_id=user_id)
+            if contract_snapshot is None:
+                return self._discover_or_create_with_lock(
+                    thread_id,
+                    sandbox_id,
+                    user_id=user_id,
+                )
+            return self._discover_or_create_with_lock(
+                thread_id,
+                sandbox_id,
+                user_id=user_id,
+                contract_snapshot=contract_snapshot,
+            )
 
-        return self._create_sandbox(thread_id, sandbox_id, user_id=user_id)
+        if contract_snapshot is None:
+            return self._create_sandbox(thread_id, sandbox_id, user_id=user_id)
+        return self._create_sandbox(
+            thread_id,
+            sandbox_id,
+            user_id=user_id,
+            contract_snapshot=contract_snapshot,
+        )
 
-    async def _acquire_internal_async(self, thread_id: str | None, *, user_id: str) -> str:
+    async def _acquire_internal_async(
+        self,
+        thread_id: str | None,
+        *,
+        user_id: str,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Async counterpart to ``_acquire_internal``."""
         await asyncio.to_thread(self._ensure_skills_projection, user_id)
-        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id)
+        if contract_snapshot is None and isinstance(self._backend, RemoteSandboxBackend):
+            contract_snapshot = await asyncio.to_thread(self._acquisition_contract_snapshot)
+        sandbox_id = self._sandbox_id_for_thread(thread_id, user_id) if contract_snapshot is None else self._sandbox_id_for_thread(thread_id, user_id, contract_snapshot)
         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(
@@ -2023,6 +2182,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 thread_id,
                 user_id=user_id,
                 expected_sandbox_id=expected_cached_id,
+                contract_snapshot=contract_snapshot,
             )
         if cached_id is not None:
             return cached_id
@@ -2033,17 +2193,56 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 self._assert_active_identity_available_locked(sandbox_id, key)
 
         # ── Layer 1.5: Warm pool (container still running, no cold-start) ──
-        reclaimed_id = await asyncio.to_thread(self._reclaim_warm_pool_sandbox, thread_id, sandbox_id, user_id=user_id)
+        if contract_snapshot is None:
+            reclaimed_id = await asyncio.to_thread(
+                self._reclaim_warm_pool_sandbox,
+                thread_id,
+                sandbox_id,
+                user_id=user_id,
+            )
+        else:
+            reclaimed_id = await asyncio.to_thread(
+                self._reclaim_warm_pool_sandbox,
+                thread_id,
+                sandbox_id,
+                user_id=user_id,
+                contract_snapshot=contract_snapshot,
+            )
         if reclaimed_id is not None:
             return reclaimed_id
 
         # ── Layer 2: Backend discovery + create (protected by cross-process lock) ──
         if thread_id:
-            return await self._discover_or_create_with_lock_async(thread_id, sandbox_id, user_id=user_id)
+            if contract_snapshot is None:
+                return await self._discover_or_create_with_lock_async(
+                    thread_id,
+                    sandbox_id,
+                    user_id=user_id,
+                )
+            return await self._discover_or_create_with_lock_async(
+                thread_id,
+                sandbox_id,
+                user_id=user_id,
+                contract_snapshot=contract_snapshot,
+            )
 
-        return await self._create_sandbox_async(thread_id, sandbox_id, user_id=user_id)
+        if contract_snapshot is None:
+            return await self._create_sandbox_async(thread_id, sandbox_id, user_id=user_id)
+        return await self._create_sandbox_async(
+            thread_id,
+            sandbox_id,
+            user_id=user_id,
+            contract_snapshot=contract_snapshot,
+        )
 
-    def _discover_or_create_with_lock(self, thread_id: str, sandbox_id: str, *, user_id: str | None = None) -> str:
+    def _discover_or_create_with_lock(
+        self,
+        thread_id: str,
+        sandbox_id: str,
+        *,
+        user_id: str | None = None,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Discover an existing sandbox or create a new one under a cross-process file lock.
 
         The file lock serializes concurrent sandbox creation for the same thread_id
@@ -2061,7 +2260,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 locked = True
                 # Re-check in-process caches under the file lock in case another
                 # thread in this process won the race while we were waiting.
-                cached_id = self._recheck_cached_sandbox(thread_id, sandbox_id, user_id=effective_user_id)
+                cached_id = self._recheck_cached_sandbox(
+                    thread_id,
+                    sandbox_id,
+                    user_id=effective_user_id,
+                    contract_snapshot=contract_snapshot,
+                )
                 if cached_id is not None:
                     return cached_id
 
@@ -2079,12 +2283,30 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 if discovered is not None:
                     return self._register_discovered_sandbox(thread_id, discovered, user_id=effective_user_id)
 
-                return self._create_sandbox(thread_id, sandbox_id, user_id=effective_user_id)
+                if contract_snapshot is None:
+                    return self._create_sandbox(
+                        thread_id,
+                        sandbox_id,
+                        user_id=effective_user_id,
+                    )
+                return self._create_sandbox(
+                    thread_id,
+                    sandbox_id,
+                    user_id=effective_user_id,
+                    contract_snapshot=contract_snapshot,
+                )
             finally:
                 if locked:
                     _unlock_file(lock_file)
 
-    async def _discover_or_create_with_lock_async(self, thread_id: str, sandbox_id: str, *, user_id: str | None = None) -> str:
+    async def _discover_or_create_with_lock_async(
+        self,
+        thread_id: str,
+        sandbox_id: str,
+        *,
+        user_id: str | None = None,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Async counterpart to ``_discover_or_create_with_lock``."""
         paths = get_paths()
         effective_user_id = self._effective_acquire_user_id(user_id)
@@ -2098,7 +2320,13 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             locked = True
             # Re-check in-process caches under the file lock in case another
             # thread in this process won the race while we were waiting.
-            cached_id = await asyncio.to_thread(self._recheck_cached_sandbox, thread_id, sandbox_id, user_id=effective_user_id)
+            cached_id = await asyncio.to_thread(
+                self._recheck_cached_sandbox,
+                thread_id,
+                sandbox_id,
+                user_id=effective_user_id,
+                contract_snapshot=contract_snapshot,
+            )
             if cached_id is not None:
                 return cached_id
 
@@ -2123,7 +2351,18 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
                 # every other step in this coroutine is offloaded.
                 return await asyncio.to_thread(self._register_discovered_sandbox, thread_id, discovered, user_id=effective_user_id)
 
-            return await self._create_sandbox_async(thread_id, sandbox_id, user_id=effective_user_id)
+            if contract_snapshot is None:
+                return await self._create_sandbox_async(
+                    thread_id,
+                    sandbox_id,
+                    user_id=effective_user_id,
+                )
+            return await self._create_sandbox_async(
+                thread_id,
+                sandbox_id,
+                user_id=effective_user_id,
+                contract_snapshot=contract_snapshot,
+            )
         finally:
             if locked:
                 await asyncio.to_thread(_unlock_file, lock_file)
@@ -2184,7 +2423,14 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         finally:
             self._finish_local_teardown(sandbox_id)
 
-    def _create_sandbox(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
+    def _create_sandbox(
+        self,
+        thread_id: str | None,
+        sandbox_id: str,
+        *,
+        user_id: str | None = None,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Create a new sandbox via the backend.
 
         Args:
@@ -2198,7 +2444,15 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             RuntimeError: If sandbox creation or readiness check fails.
         """
         effective_user_id = self._effective_acquire_user_id(user_id)
-        extra_mounts = self._get_extra_mounts(thread_id, user_id=effective_user_id)
+        extra_mounts = (
+            self._get_extra_mounts(thread_id, user_id=effective_user_id)
+            if contract_snapshot is None
+            else self._get_extra_mounts(
+                thread_id,
+                user_id=effective_user_id,
+                contract_snapshot=contract_snapshot,
+            )
+        )
         provision_lark_cli_runtime = self._lark_integration_active(effective_user_id)
         provision_lark_cli_broker = self._lark_broker_active(effective_user_id)
 
@@ -2209,13 +2463,18 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             evicted = self._evict_oldest_warm()
             self._log_replicas_soft_cap(replicas, sandbox_id, evicted)
 
+        create_kwargs = {
+            "user_id": effective_user_id,
+            "provision_lark_cli_runtime": provision_lark_cli_runtime,
+            "provision_lark_cli_broker": provision_lark_cli_broker,
+        }
+        if isinstance(self._backend, RemoteSandboxBackend) and contract_snapshot is not None:
+            create_kwargs["required_mount_contract_version"] = contract_snapshot.version
         info = self._backend.create(
             thread_id,
             sandbox_id,
             extra_mounts=extra_mounts or None,
-            user_id=effective_user_id,
-            provision_lark_cli_runtime=provision_lark_cli_runtime,
-            provision_lark_cli_broker=provision_lark_cli_broker,
+            **create_kwargs,
         )
 
         # Wait for sandbox to be ready
@@ -2229,10 +2488,29 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
         return self._register_created_sandbox(thread_id, sandbox_id, info, user_id=effective_user_id)
 
-    async def _create_sandbox_async(self, thread_id: str | None, sandbox_id: str, *, user_id: str | None = None) -> str:
+    async def _create_sandbox_async(
+        self,
+        thread_id: str | None,
+        sandbox_id: str,
+        *,
+        user_id: str | None = None,
+        contract_snapshot: MountContractSnapshot | None = None,
+    ) -> str:
         """Async counterpart to ``_create_sandbox``."""
         effective_user_id = self._effective_acquire_user_id(user_id)
-        extra_mounts = await asyncio.to_thread(self._get_extra_mounts, thread_id, user_id=effective_user_id)
+        if contract_snapshot is None:
+            extra_mounts = await asyncio.to_thread(
+                self._get_extra_mounts,
+                thread_id,
+                user_id=effective_user_id,
+            )
+        else:
+            extra_mounts = await asyncio.to_thread(
+                self._get_extra_mounts,
+                thread_id,
+                user_id=effective_user_id,
+                contract_snapshot=contract_snapshot,
+            )
         provision_lark_cli_runtime = await asyncio.to_thread(self._lark_integration_active, effective_user_id)
         provision_lark_cli_broker = await asyncio.to_thread(self._lark_broker_active, effective_user_id)
 
@@ -2243,14 +2521,19 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
             evicted = await asyncio.to_thread(self._evict_oldest_warm)
             self._log_replicas_soft_cap(replicas, sandbox_id, evicted)
 
+        create_kwargs = {
+            "user_id": effective_user_id,
+            "provision_lark_cli_runtime": provision_lark_cli_runtime,
+            "provision_lark_cli_broker": provision_lark_cli_broker,
+        }
+        if isinstance(self._backend, RemoteSandboxBackend) and contract_snapshot is not None:
+            create_kwargs["required_mount_contract_version"] = contract_snapshot.version
         info = await asyncio.to_thread(
             self._backend.create,
             thread_id,
             sandbox_id,
             extra_mounts=extra_mounts or None,
-            user_id=effective_user_id,
-            provision_lark_cli_runtime=provision_lark_cli_runtime,
-            provision_lark_cli_broker=provision_lark_cli_broker,
+            **create_kwargs,
         )
 
         # Wait for sandbox to be ready without blocking the event loop.

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

@@ -22,6 +22,7 @@ import os
 import re
 import threading
 import time
+from dataclasses import dataclass
 from pathlib import Path, PureWindowsPath
 
 import requests
@@ -56,6 +57,23 @@ _LARK_CLI_CONFIG_CONTAINER_PATH = "/mnt/integrations/lark-cli/config"
 _LARK_CLI_DATA_CONTAINER_PATH = "/mnt/integrations/lark-cli/data"
 
 
+@dataclass(frozen=True, slots=True)
+class MountContractSnapshot:
+    """One immutable Provisioner capability decision used by an acquisition."""
+
+    version: int
+    capability_known: bool
+
+
+class MountContractChangedError(RuntimeError):
+    """The Provisioner contract changed while an acquisition was in flight."""
+
+    def __init__(self, expected: int, actual: int) -> None:
+        super().__init__(f"Provisioner mount contract changed during sandbox acquisition: expected v{expected}, now v{actual}")
+        self.expected = expected
+        self.actual = actual
+
+
 def _provisioner_extra_mounts_payload(
     extra_mounts: list[tuple[str, str, bool]] | None,
     *,
@@ -171,14 +189,37 @@ class RemoteSandboxBackend(SandboxBackend):
         """Whether the peer definitively answered capability negotiation."""
         return self._mount_contract_capability_known
 
-    def mount_contract_snapshot(self) -> tuple[int, bool]:
+    def mount_contract_snapshot(self) -> MountContractSnapshot:
         """Return one capability snapshot for a complete caller operation."""
         with self._capability_probe_lock:
-            return (
-                self._mount_contract_version,
-                self._mount_contract_capability_known,
+            return MountContractSnapshot(
+                version=self._mount_contract_version,
+                capability_known=self._mount_contract_capability_known,
             )
 
+    def refresh_capabilities_and_snapshot(self) -> MountContractSnapshot:
+        """Wait for an in-flight probe and return a fresh atomic snapshot.
+
+        Unlike ``refresh_capabilities_if_stale()``, this operation deliberately
+        waits for the probe lock.  It is used only by synchronous callers or in
+        a worker thread so an expired v2 result can never be reused as mounted
+        while another thread is proving that the peer is legacy/unavailable.
+        """
+        with self._capability_probe_lock:
+            if time.monotonic() >= self._capability_next_probe_at:
+                self._probe_capabilities_locked()
+            return MountContractSnapshot(
+                version=self._mount_contract_version,
+                capability_known=self._mount_contract_capability_known,
+            )
+
+    def _invalidate_capability_snapshot(self) -> None:
+        """Fail closed and force the next acquisition attempt to re-probe."""
+        with self._capability_probe_lock:
+            self._mount_contract_version = 0
+            self._mount_contract_capability_known = False
+            self._capability_next_probe_at = 0.0
+
     @property
     def requires_create_validation(self) -> bool:
         """Require idempotent POST so the Provisioner checks requested mounts."""
@@ -257,14 +298,20 @@ 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:
         """Create a sandbox Pod + Service via the provisioner.
 
         Calls ``POST /api/sandboxes`` which creates a dedicated Pod +
         NodePort Service in k3s.
         """
-        self.refresh_capabilities_if_stale()
-        mount_contract_version, _capability_known = self.mount_contract_snapshot()
+        snapshot = self.refresh_capabilities_and_snapshot()
+        if required_mount_contract_version is not None and snapshot.version != required_mount_contract_version:
+            raise MountContractChangedError(
+                required_mount_contract_version,
+                snapshot.version,
+            )
+        mount_contract_version = snapshot.version
         effective_user_id = user_id or get_effective_user_id()
         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")
@@ -353,7 +400,7 @@ class RemoteSandboxBackend(SandboxBackend):
     ) -> SandboxInfo:
         """POST /api/sandboxes → create Pod + Service."""
         if required_mount_contract_version is None:
-            required_mount_contract_version, _capability_known = self.mount_contract_snapshot()
+            required_mount_contract_version = self.mount_contract_snapshot().version
         effective_user_id = user_id or get_effective_user_id()
         include_legacy_skills = user_should_see_legacy_skills(effective_user_id)
         payload = {
@@ -383,21 +430,34 @@ 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
+            response_version = data.get("mount_contract_version")
             if required_mount_contract_version >= _UPLOAD_MOUNT_CONTRACT_VERSION:
+                if type(response_version) is not int or response_version != required_mount_contract_version:
+                    self._invalidate_capability_snapshot()
+                    raise MountContractChangedError(
+                        required_mount_contract_version,
+                        response_version if type(response_version) is int else 0,
+                    )
                 response_contract = (
                     data.get("sandbox_id"),
                     data.get("user_id"),
                     data.get("thread_id"),
-                    data.get("mount_contract_version"),
+                    response_version,
                 )
                 expected_contract = (
                     sandbox_id,
                     effective_user_id,
                     effective_thread_id,
-                    _UPLOAD_MOUNT_CONTRACT_VERSION,
+                    required_mount_contract_version,
                 )
                 if response_contract != expected_contract:
                     raise RuntimeError(f"Provisioner mount contract response does not match the requested sandbox identity: expected={expected_contract!r}, received={response_contract!r}")
+            elif (type(response_version) is int and response_version != required_mount_contract_version) or (type(response_version) is not int and required_mount_contract_version != 0):
+                self._invalidate_capability_snapshot()
+                raise MountContractChangedError(
+                    required_mount_contract_version,
+                    response_version if type(response_version) is int else 0,
+                )
             logger.info(f"Provisioner created sandbox {sandbox_id}: sandbox_url={data['sandbox_url']}")
             return SandboxInfo(
                 sandbox_id=sandbox_id,

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

@@ -1,4 +1,5 @@
 import asyncio
+import inspect
 import threading
 from abc import ABC, abstractmethod
 
@@ -183,13 +184,34 @@ def sandbox_provider_uses_thread_data_mounts(
     refresh: bool = True,
 ) -> bool:
     """Return one provider mount-mode decision, optionally refreshing it first."""
+    # ``MagicMock`` and some proxy providers fabricate arbitrary attributes from
+    # ``__getattr__``.  Only opt into the extended contract when the attribute
+    # really exists on the instance or its type; otherwise the legacy boolean is
+    # the source of truth.
+    static_mode_resolver = inspect.getattr_static(provider, "thread_data_mounts_mode", None)
+    mode_resolver = getattr(provider, "thread_data_mounts_mode", None) if static_mode_resolver is not None else None
+    if callable(mode_resolver):
+        return bool(mode_resolver(refresh=refresh))
     if refresh:
-        refresher = getattr(provider, "refresh_thread_data_mount_capabilities", None)
+        static_refresher = inspect.getattr_static(provider, "refresh_thread_data_mount_capabilities", None)
+        refresher = getattr(provider, "refresh_thread_data_mount_capabilities", None) if static_refresher is not None else None
         if callable(refresher):
             refresher()
     return bool(getattr(provider, "uses_thread_data_mounts", False))
 
 
+def sandbox_provider_sandbox_uses_thread_data_mounts(
+    provider: SandboxProvider,
+    sandbox_id: str,
+) -> bool:
+    """Return the immutable mount mode recorded for one acquired sandbox."""
+    static_mode_resolver = inspect.getattr_static(provider, "sandbox_uses_thread_data_mounts", None)
+    mode_resolver = getattr(provider, "sandbox_uses_thread_data_mounts", None) if static_mode_resolver is not None else None
+    if callable(mode_resolver):
+        return bool(mode_resolver(sandbox_id))
+    return sandbox_provider_uses_thread_data_mounts(provider, refresh=False)
+
+
 async def sandbox_provider_uses_thread_data_mounts_async(
     provider: SandboxProvider,
     *,

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

@@ -819,6 +819,55 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
             device=staged_stat.st_dev,
             inode=staged_stat.st_ino,
         )
+        if not recover_on_crash:
+            visible_matches: list[Path] = []
+            if staged_stat.st_nlink == 2:
+                # A discard transaction never restores a visible peer.  If a
+                # previous attempt crashed after relinking the tombstone (or a
+                # rollback raced with recovery), remove the one verified peer
+                # first; a crash after this unlink leaves an idempotent nlink=1
+                # tombstone for the next startup pass.
+                with os.scandir(uploads_dir) as upload_entries:
+                    for upload_entry in upload_entries:
+                        try:
+                            upload_stat = upload_entry.stat(follow_symlinks=False)
+                        except FileNotFoundError:
+                            continue
+                        if stat.S_ISREG(upload_stat.st_mode) and (
+                            upload_stat.st_dev,
+                            upload_stat.st_ino,
+                        ) == (identity.device, identity.inode):
+                            visible_matches.append(Path(upload_entry.path))
+                if len(visible_matches) != 1:
+                    logger.warning(
+                        "Refusing ambiguous upload discard recovery with %s visible aliases: %s",
+                        len(visible_matches),
+                        staged_path,
+                    )
+                    return False
+            original_path = uploads_dir / original_name
+            try:
+                current_original_stat = os.lstat(original_path)
+            except FileNotFoundError:
+                current_original_stat = None
+            delete_conversion = current_original_stat is None or (stat.S_ISREG(current_original_stat.st_mode) and (current_original_stat.st_dev, current_original_stat.st_ino) == (identity.device, identity.inode))
+            if delete_conversion:
+                owned_conversion = existing_conversion_path_for_upload(original_path)
+                if owned_conversion is not None:
+                    owned_conversion.unlink(missing_ok=True)
+            else:
+                # The upload name was reused after the crash.  Its conversion
+                # path is generation-ambiguous, so preserve it rather than
+                # deleting a possible replacement generation's companion.
+                logger.warning(
+                    "Preserving replacement upload conversion during discard recovery: %s",
+                    original_path,
+                )
+            if visible_matches:
+                visible_matches[0].unlink()
+            staged_path.unlink()
+            _finish_deletion_transaction(staged_path)
+            return True
         if staged_stat.st_nlink == 2:
             # A previous recovery may have crashed after publishing the visible
             # hard link but before removing the tombstone.  Accept only one
@@ -845,10 +894,6 @@ 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,
@@ -1008,6 +1053,7 @@ def delete_file_safe(
             base_dir,
             actual_file_path,
             identity,
+            recover_on_crash=False,
         )
         try:
             if delete_remote_copy is not None:

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

@@ -10,6 +10,7 @@ from pathlib import Path
 from typing import Any
 
 from deerflow.sandbox.sandbox_provider import (
+    sandbox_provider_sandbox_uses_thread_data_mounts,
     sandbox_provider_uses_thread_data_mounts,
     sandbox_provider_uses_thread_data_mounts_async,
 )
@@ -73,7 +74,7 @@ def prepare_upload_deletion(
     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):
+    if sandbox_provider_sandbox_uses_thread_data_mounts(sandbox_provider, sandbox_id):
         return None
     sandbox = sandbox_provider.get(sandbox_id)
     if sandbox is None:
@@ -91,7 +92,7 @@ async def prepare_upload_deletion_async(
     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):
+    if sandbox_provider_sandbox_uses_thread_data_mounts(sandbox_provider, sandbox_id):
         return None
     sandbox = sandbox_provider.get(sandbox_id)
     if sandbox is None:
@@ -146,7 +147,7 @@ def make_upload_paths_available(
         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):
+    if sandbox_provider_sandbox_uses_thread_data_mounts(sandbox_provider, sandbox_id):
         _make_paths_readable(sync_paths)
         return SandboxSyncReceipt(sandbox=None)
     sandbox = sandbox_provider.get(sandbox_id)
@@ -169,7 +170,7 @@ async def make_upload_paths_available_async(
         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):
+    if sandbox_provider_sandbox_uses_thread_data_mounts(sandbox_provider, sandbox_id):
         await run_upload_io_cancellation_safe(_make_paths_readable, sync_paths)
         return SandboxSyncReceipt(sandbox=None)
     sandbox = sandbox_provider.get(sandbox_id)

backend/tests/test_aio_sandbox_provider.py

@@ -6,6 +6,7 @@ import hashlib
 import importlib
 import stat
 import threading
+import time
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -884,6 +885,128 @@ def test_cached_legacy_sandbox_is_not_reused_after_contract_upgrade(tmp_path):
     assert ("default", "thread-1") not in provider._thread_sandboxes
 
 
+@pytest.mark.anyio
+async def test_acquire_internal_async_does_not_block_on_capability_snapshot(tmp_path):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    provider = _make_provider(tmp_path)
+    backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 2
+    backend._mount_contract_capability_known = True
+    backend._capability_next_probe_at = time.monotonic() + 30
+    provider._backend = backend
+    provider._ensure_skills_projection = lambda _user_id: None
+    provider._reuse_in_process_sandbox = lambda *_args, **_kwargs: "cached-v2"
+    backend._capability_probe_lock.acquire()
+    release_probe = threading.Timer(0.15, backend._capability_probe_lock.release)
+    release_probe.start()
+    loop = asyncio.get_running_loop()
+    started = loop.time()
+
+    task = asyncio.create_task(provider._acquire_internal_async("thread-1", user_id="default"))
+    await asyncio.sleep(0.02)
+    ticker_elapsed = loop.time() - started
+    result = await task
+    release_probe.join(timeout=1)
+
+    assert ticker_elapsed < 0.08
+    assert result == "cached-v2"
+
+
+def test_acquire_restarts_with_a_new_id_when_mount_contract_changes(tmp_path):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    provider = _make_provider(tmp_path)
+    provider._thread_locks = {}
+    provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    snapshots = iter(
+        [
+            remote_mod.MountContractSnapshot(version=2, capability_known=True),
+            remote_mod.MountContractSnapshot(version=0, capability_known=True),
+        ]
+    )
+    provider._acquisition_contract_snapshot = lambda: next(snapshots)
+    attempted_ids: list[str] = []
+
+    def acquire_internal(thread_id, *, user_id, contract_snapshot):
+        sandbox_id = provider._sandbox_id_for_thread(thread_id, user_id, contract_snapshot)
+        attempted_ids.append(sandbox_id)
+        if contract_snapshot.version == 2:
+            raise remote_mod.MountContractChangedError(2, 0)
+        return sandbox_id
+
+    provider._acquire_internal = acquire_internal
+
+    result = provider.acquire("thread-1", user_id="default")
+
+    assert attempted_ids == [
+        provider._deterministic_sandbox_id("thread-1", "default", 2),
+        provider._deterministic_sandbox_id("thread-1", "default", 0),
+    ]
+    assert result == attempted_ids[-1]
+
+
+@pytest.mark.anyio
+async def test_acquire_async_restarts_with_a_new_id_when_mount_contract_changes(tmp_path):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    provider = _make_provider(tmp_path)
+    provider._thread_locks = {}
+    provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    snapshots = iter(
+        [
+            remote_mod.MountContractSnapshot(version=2, capability_known=True),
+            remote_mod.MountContractSnapshot(version=3, capability_known=True),
+        ]
+    )
+    provider._acquisition_contract_snapshot = lambda: next(snapshots)
+    attempted_ids: list[str] = []
+
+    async def acquire_internal(thread_id, *, user_id, contract_snapshot):
+        sandbox_id = provider._sandbox_id_for_thread(thread_id, user_id, contract_snapshot)
+        attempted_ids.append(sandbox_id)
+        if contract_snapshot.version == 2:
+            raise remote_mod.MountContractChangedError(2, 3)
+        return sandbox_id
+
+    provider._acquire_internal_async = acquire_internal
+
+    result = await provider.acquire_async("thread-1", user_id="default")
+
+    assert attempted_ids == [
+        provider._deterministic_sandbox_id("thread-1", "default", 2),
+        provider._deterministic_sandbox_id("thread-1", "default", 3),
+    ]
+    assert result == attempted_ids[-1]
+
+
+@pytest.mark.parametrize(
+    ("global_version", "sandbox_version", "expected"),
+    [
+        (2, 0, False),
+        (0, 2, True),
+    ],
+)
+def test_sandbox_mount_mode_uses_instance_contract_not_global_capability(
+    tmp_path,
+    global_version,
+    sandbox_version,
+    expected,
+):
+    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 = _make_provider(tmp_path)
+    provider._config["thread_data_mounts"] = True
+    provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+    provider._backend._mount_contract_version = global_version
+    provider._sandbox_infos = {}
+    provider._sandbox_infos["sandbox-1"] = aio_mod.SandboxInfo(
+        sandbox_id="sandbox-1",
+        sandbox_url="http://sandbox.local",
+        mount_contract_version=sandbox_version,
+    )
+
+    assert provider_mod.sandbox_provider_sandbox_uses_thread_data_mounts(provider, "sandbox-1") is expected
+
+
 @pytest.mark.anyio
 async def test_acquire_async_uses_async_readiness_polling(monkeypatch):
     """AioSandboxProvider async creation must not use sync readiness polling."""

backend/tests/test_remote_sandbox_backend.py

@@ -1,5 +1,10 @@
 from __future__ import annotations
 
+import importlib
+import threading
+import time
+from unittest.mock import MagicMock
+
 import pytest
 import requests
 
@@ -217,7 +222,6 @@ def test_create_rejects_nondefault_user_on_unverified_legacy_provisioner(monkeyp
 @pytest.mark.parametrize(
     "response_payload",
     [
-        {"sandbox_id": "abc123", "sandbox_url": "http://k3s:31001"},
         {
             "sandbox_id": "abc123",
             "sandbox_url": "http://k3s:31001",
@@ -232,16 +236,9 @@ def test_create_rejects_nondefault_user_on_unverified_legacy_provisioner(monkeyp
             "thread_id": "other-thread",
             "mount_contract_version": 2,
         },
-        {
-            "sandbox_id": "abc123",
-            "sandbox_url": "http://k3s:31001",
-            "user_id": "alice",
-            "thread_id": "thread-1",
-            "mount_contract_version": 1,
-        },
     ],
 )
-def test_current_contract_create_rejects_unverified_or_mismatched_response(monkeypatch, response_payload):
+def test_current_contract_create_rejects_mismatched_identity(monkeypatch, response_payload):
     backend = RemoteSandboxBackend("http://provisioner:8002")
     backend._mount_contract_version = 2
     backend._mount_contract_capability_known = True
@@ -256,6 +253,37 @@ def test_current_contract_create_rejects_unverified_or_mismatched_response(monke
         backend.create("thread-1", "abc123", user_id="alice")
 
 
+@pytest.mark.parametrize(
+    "response_payload",
+    [
+        {"sandbox_id": "abc123", "sandbox_url": "http://k3s:31001"},
+        {
+            "sandbox_id": "abc123",
+            "sandbox_url": "http://k3s:31001",
+            "user_id": "alice",
+            "thread_id": "thread-1",
+            "mount_contract_version": 1,
+        },
+    ],
+)
+def test_current_contract_create_restarts_after_response_contract_change(monkeypatch, response_payload):
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 2
+    backend._mount_contract_capability_known = True
+    backend._capability_next_probe_at = time.monotonic() + 30
+    monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda _user_id: False)
+    monkeypatch.setattr(
+        requests,
+        "post",
+        lambda *_args, **_kwargs: _StubResponse(payload=response_payload),
+    )
+
+    with pytest.raises(remote_backend_mod.MountContractChangedError):
+        backend.create("thread-1", "abc123", user_id="alice")
+
+    assert backend._capability_next_probe_at == 0.0
+
+
 def test_current_contract_create_returns_verified_identity(monkeypatch):
     backend = RemoteSandboxBackend("http://provisioner:8002")
     backend._mount_contract_version = 2
@@ -303,9 +331,35 @@ def test_create_keeps_v2_response_validation_when_capability_changes_during_post
 
     monkeypatch.setattr(requests, "post", old_peer_response)
 
-    with pytest.raises(RuntimeError, match="mount contract response"):
+    with pytest.raises(remote_backend_mod.MountContractChangedError):
         backend.create("thread-1", "abc123", user_id="alice")
 
+    assert backend._capability_next_probe_at == 0.0
+
+
+def test_legacy_create_restarts_when_response_advertises_another_version(monkeypatch):
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 0
+    backend._mount_contract_capability_known = True
+    backend._capability_next_probe_at = time.monotonic() + 30
+    monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda _user_id: False)
+    monkeypatch.setattr(
+        requests,
+        "post",
+        lambda *_args, **_kwargs: _StubResponse(
+            payload={
+                "sandbox_id": "abc123",
+                "sandbox_url": "http://legacy.local",
+                "mount_contract_version": 1,
+            }
+        ),
+    )
+
+    with pytest.raises(remote_backend_mod.MountContractChangedError):
+        backend.create("thread-1", "abc123", user_id="default")
+
+    assert backend._capability_next_probe_at == 0.0
+
 
 def test_capability_retry_backoff_does_not_overflow_after_long_outage(monkeypatch):
     backend = RemoteSandboxBackend("http://provisioner:8002")
@@ -324,6 +378,85 @@ def test_capability_retry_backoff_does_not_overflow_after_long_outage(monkeypatc
     assert backend._capability_next_probe_at == 130.0
 
 
+def test_mount_mode_waits_for_an_inflight_stale_probe(monkeypatch):
+    provider_mod = importlib.import_module("deerflow.sandbox.sandbox_provider")
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 2
+    backend._mount_contract_capability_known = True
+    backend._capability_next_probe_at = 0.0
+    provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
+    provider._config = {"thread_data_mounts": True}
+    provider._backend = backend
+    result: list[bool] = []
+    completed = threading.Event()
+
+    backend._capability_probe_lock.acquire()
+    try:
+        worker = threading.Thread(
+            target=lambda: (
+                result.append(provider_mod.sandbox_provider_uses_thread_data_mounts(provider)),
+                completed.set(),
+            ),
+            daemon=True,
+        )
+        worker.start()
+        assert not completed.wait(0.05)
+        backend._mount_contract_version = 0
+        backend._mount_contract_capability_known = True
+        backend._capability_next_probe_at = time.monotonic() + 30
+    finally:
+        backend._capability_probe_lock.release()
+
+    assert completed.wait(1)
+    worker.join(timeout=1)
+    assert result == [False]
+
+
+def test_create_restarts_when_required_mount_contract_changed(monkeypatch):
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 0
+    backend._mount_contract_capability_known = True
+    backend._capability_next_probe_at = time.monotonic() + 30
+    post = MagicMock()
+    monkeypatch.setattr(requests, "post", post)
+
+    with pytest.raises(remote_backend_mod.MountContractChangedError):
+        backend.create(
+            "thread-1",
+            "sandbox-v2",
+            user_id="default",
+            required_mount_contract_version=2,
+        )
+
+    post.assert_not_called()
+
+
+def test_create_accepts_and_validates_future_compatible_contract(monkeypatch):
+    backend = RemoteSandboxBackend("http://provisioner:8002")
+    backend._mount_contract_version = 3
+    backend._mount_contract_capability_known = True
+    backend._capability_next_probe_at = time.monotonic() + 30
+    monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda _user_id: False)
+    monkeypatch.setattr(
+        requests,
+        "post",
+        lambda *_args, **_kwargs: _StubResponse(
+            payload={
+                "sandbox_id": "sandbox-v3",
+                "sandbox_url": "http://v3.local",
+                "user_id": "alice",
+                "thread_id": "thread-1",
+                "mount_contract_version": 3,
+            }
+        ),
+    )
+
+    info = backend.create("thread-1", "sandbox-v3", user_id="alice")
+
+    assert info.mount_contract_version == 3
+
+
 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

@@ -974,6 +974,76 @@ class TestCleanupStaleUploadStagingFiles:
         assert not primary.exists()
         assert not staged_path.exists()
 
+    def test_crashed_discard_removes_a_republished_visible_link(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,
+        )
+        os.link(staged_path, primary)
+        stage_lease.release()
+
+        assert cleanup_stale_upload_staging_files(tmp_path) == 1
+        assert not primary.exists()
+        assert not staged_path.exists()
+
+    def test_crashed_committed_delete_removes_primary_and_conversion(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"delete me")
+        conversion = conversion_path_for_upload(primary)
+        conversion.parent.mkdir(parents=True, exist_ok=True)
+        conversion.write_text("generated", encoding="utf-8")
+        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 conversion.exists()
+        assert not staged_path.exists()
+
+    def test_crashed_committed_delete_preserves_replacement_conversion(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"old generation")
+        old_identity = UploadIdentity.from_path(primary)
+        staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
+            uploads,
+            primary,
+            old_identity,
+            recover_on_crash=False,
+        )
+        primary.write_bytes(b"replacement generation")
+        conversion = conversion_path_for_upload(primary)
+        conversion.parent.mkdir(exist_ok=True)
+        conversion.write_text("replacement conversion", encoding="utf-8")
+        stage_lease.release()
+
+        assert cleanup_stale_upload_staging_files(tmp_path) == 1
+        assert primary.read_bytes() == b"replacement generation"
+        assert conversion.read_text(encoding="utf-8") == "replacement conversion"
+        assert not staged_path.exists()
+
 
 # ---------------------------------------------------------------------------
 # delete_file_safe