diff --git a/README.md b/README.md
index 18b1b70b4..f601059c9 100644
--- a/README.md
+++ b/README.md
@@ -963,7 +963,7 @@ Each task gets its own execution environment with a full filesystem view — ski
 
 Uploads from the Web UI, embedded client, and IM channels share one collision-safe storage rule. A completed payload is published only if its candidate name does not exist; concurrent `report.pdf` uploads become `report.pdf`, `report_1.pdf`, `report_2.pdf`, and so on without replacing one another. Collision suffixes remain within the filesystem's 255-byte UTF-8 component limit, including names whose original suffix consumes nearly the whole limit. A busy candidate lease is treated as another collision instead of waiting, so inverse multi-file batches cannot deadlock. The selected name is leased through conversion and sandbox synchronization, so deleting that exact name waits for its active upload lifecycle while unrelated filenames continue concurrently. Internal staging names matching `.upload-*.part` and basenames that cannot be represented losslessly on Windows are rejected for new uploads and reported through `skipped_files`; exact legacy POSIX names already returned by the list endpoint—including literal backslashes and names made only from dots/spaces—remain deletable after upgrade.
 
-Optional document conversions are system-owned assets under `/mnt/user-data/.upload-conversions/`. Normal targets use `<actual-upload-name>.md`; names that would exceed the filesystem component limit use a deterministic UTF-8-safe prefix plus the full SHA-256 digest. The exact generated path is returned through the upload response and omitted from the primary upload listing. Mounted AIO sandboxes expose this namespace through a read-only mount; the remote Provisioner verifies its exact user/thread source, and a mount-contract version in deterministic sandbox IDs prevents reuse of pre-upgrade containers that lack it. Local structured file APIs enforce the same rule through path mappings. Local host bash is outside that mapping boundary and must remain disabled for untrusted tasks. Non-mounted remote providers receive a private synchronized copy that may be writable but cannot mutate the authoritative host conversion or lock state. Direct Markdown uploads supply their own outline/preview, while other formats use only their exact generated asset. Deleting a primary removes only its exact generated asset and never infers that a user-uploaded sibling such as `uploads/report.md` is disposable.
+Optional document conversions are system-owned assets under `/mnt/user-data/.upload-conversions/`. Normal targets use `<actual-upload-name>.md`; names that would exceed the filesystem component limit use a deterministic UTF-8-safe prefix plus the full SHA-256 digest. The exact generated path is returned through the upload response and omitted from the primary upload listing. Mounted AIO sandboxes expose this namespace through a read-only mount; the remote Provisioner verifies its exact user/thread source and the concrete Pod mount signature. A mount-contract version in deterministic sandbox IDs prevents a pre-upgrade container from satisfying a new acquisition for the same thread; older containers may still be enumerated and adopted only for normal orphan cleanup. Local structured file APIs enforce the same rule through path mappings. Local host bash is outside that mapping boundary and must remain disabled for untrusted tasks. Non-mounted remote providers receive a private synchronized copy that may be writable but cannot mutate the authoritative host conversion or lock state. Direct Markdown uploads use one verified exclusive file descriptor for outline and preview reads, while other formats use only their exact generated asset. Deleting a primary removes its exact host and explicitly synchronized sandbox copies plus its generated asset, and never infers that a user-uploaded sibling such as `uploads/report.md` is disposable.
 
 The built-in `grep` tool searches either one text file or all matching text files below a directory, so an agent can search an uploaded document directly without first broadening the request to the entire uploads directory.
 
@@ -987,7 +987,10 @@ sandboxes receive uploaded files through explicit synchronization. Deployments
 where both sides are guaranteed to share the same thread user-data directories
 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.
+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.
 
 This is the difference between a chatbot with tool access and an agent with an actual execution environment.
 
diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index 025bd2597..557d57a61 100644
--- a/backend/AGENTS.md
+++ b/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 takes precedence for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Setting it `true` skips upload-time sandbox acquire/sync; a false positive leaves uploads unavailable to the sandbox. Thread mounts create and expose `.upload-conversions` explicitly as read-only, separate from the writable uploads mount. Non-mounted providers instead receive only the requested generated file as a private synchronized copy; that copy may be writable but does not expose or mutate authoritative host files or lock state. 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 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.
 - `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
@@ -1464,9 +1464,9 @@ Multi-file upload with automatic document conversion:
 - 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.
 - Filenames containing NUL, `<`, `>`, or reserved model-context boundary markers are rejected before staging so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed.
 - Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files. Cancellation during staging creation drains the worker and aborts the returned stage before propagating. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools.
-- Generated Markdown is owned by `user-data/.upload-conversions/<actual-primary-filename>.md` and is omitted from primary upload listings. Deletion removes only the selected primary and that exact generated asset; it never guesses or deletes a legacy/user-owned `uploads/<stem>.md` sibling. Outline extraction reads this owned conversion for non-Markdown uploads and reads a verified regular primary directly when the uploaded file itself is Markdown.
+- 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`; omission preserves backend auto-detection. The remote mount allowlist forwards `/mnt/user-data/.upload-conversions` only when it is the read-only sibling of the same thread's uploads directory; the Provisioner independently verifies the exact user/thread host source and creates a read-only nested hostPath/PVC mount over the writable `/mnt/user-data` parent. `SANDBOX_MOUNT_CONTRACT_VERSION` is included in deterministic AIO sandbox IDs, so a rolling upgrade that adds a required mount cold-starts the new identity instead of reusing an older container that lacks it; old instances remain under normal orphan cleanup.
+- 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.
 - Agent receives uploaded file list via `UploadsMiddleware`
 
 See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details.
diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py
index c3ffe596f..b8ee3ef9f 100644
--- a/backend/app/gateway/routers/uploads.py
+++ b/backend/app/gateway/routers/uploads.py
@@ -35,6 +35,7 @@ from deerflow.uploads.manager import (
     upload_artifact_url,
     upload_virtual_path,
 )
+from deerflow.uploads.sandbox_sync import prepare_upload_deletion_async
 from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS
 from deerflow.utils.file_io import run_file_io
 from deerflow.utils.thread_id import ThreadId
@@ -316,9 +317,18 @@ def _list_uploaded_files_for_thread(thread_id: str, user_id: str) -> dict:
     return result
 
 
-def _delete_uploaded_file_for_thread(thread_id: str, filename: str, user_id: str) -> dict:
+def _delete_uploaded_file_for_thread(
+    thread_id: str,
+    filename: str,
+    user_id: str,
+    delete_remote_copy=None,
+) -> dict:
     uploads_dir = get_uploads_dir(thread_id, user_id=user_id)
-    return delete_file_safe(uploads_dir, filename)
+    return delete_file_safe(
+        uploads_dir,
+        filename,
+        delete_remote_copy=delete_remote_copy,
+    )
 
 
 async def _write_upload_file_with_limits(
@@ -416,6 +426,8 @@ async def upload_files(
     try:
         for file in files:
             if not file.filename:
+                logger.warning("Skipping multipart upload with an empty filename")
+                skipped_files.append("")
                 continue
             current_filename = file.filename
 
@@ -561,11 +573,19 @@ async def list_uploaded_files(thread_id: ThreadId, request: Request) -> UploadLi
 async def delete_uploaded_file(thread_id: ThreadId, filename: str, request: Request) -> dict:
     """Delete a file from a thread's uploads directory."""
     try:
+        user_id = get_effective_user_id()
+        sandbox_provider = await asyncio.to_thread(get_sandbox_provider)
+        delete_remote_copy = await prepare_upload_deletion_async(
+            sandbox_provider,
+            thread_id,
+            user_id=user_id,
+        )
         return await _run_upload_lease_io_cancellation_safe(
             _delete_uploaded_file_for_thread,
             thread_id,
             filename,
-            get_effective_user_id(),
+            user_id,
+            delete_remote_copy,
         )
     except FileNotFoundError:
         raise HTTPException(status_code=404, detail=f"File not found: {filename}")
diff --git a/backend/docs/FILE_UPLOAD.md b/backend/docs/FILE_UPLOAD.md
index a2ea3b2d5..3b6012bdd 100644
--- a/backend/docs/FILE_UPLOAD.md
+++ b/backend/docs/FILE_UPLOAD.md
@@ -54,7 +54,7 @@ POST /api/threads/{thread_id}/uploads
 
 所有上传入口都先完整写入同目录暂存文件，再以“不替换已有条目”的原子操作发布。同名碰撞依次命名为 `document.pdf`、`document_1.pdf`、`document_2.pdf`；响应中的 `filename` 和各路径字段始终使用实际发布名。系统内部保留 `.upload-*.part` 作为暂存命名空间；使用该模式的 basename、包含 NUL、`<` 或 `>`、包含保留模型上下文边界标记，或无法在 Windows 上无损表示（如设备名、尾随点/空格或保留字符）的文件名会在创建暂存文件前被拒绝，以保证所有已接受的文件名和 Agent 可见路径都能无损呈现。若请求在 staging 创建尚未返回时被取消，Gateway 会等待创建结束并精确 abort 该临时文件。
 
-实际发布名会在转换、权限调整、沙箱同步和响应构造期间持有同名租约。大小写、Unicode 规范化及 Win32 尾随后缀等可移植文件系统别名共用同一协调键，避免用别名绕过 generation lease。发布遇到正在使用的协调键时不会等待，而会继续选择 `_N` 候选，因此逆序并发批次不会互相持锁；删除仍会等待目标 generation 生命周期完成，并在 hard-link 歧义下拒绝误报成功。跨进程协调使用 `.upload-conversions/.locks/` 下稳定保留的摘要锁文件，该目录属于内部实现，不应由 Agent 或部署脚本修改或清理。最终 lease release 是明确提交点：如果新的取消恰在 release 期间到达，系统会先完成 release 并返回已构造的成功结果，而不会把已提交文件报告成取消。Gateway 拒绝不安全文件名时会把原名加入 `skipped_files` 并返回 `success: false`，不会把“上传 0 个文件”误报为成功。
+实际发布名会在转换、权限调整、沙箱同步和响应构造期间持有同名租约。大小写、Unicode 规范化及 Win32 尾随后缀等可移植文件系统别名共用同一协调键，避免用别名绕过 generation lease。发布遇到正在使用的协调键时不会等待，而会继续选择 `_N` 候选，因此逆序并发批次不会互相持锁；删除仍会等待目标 generation 生命周期完成，并在 hard-link 歧义下拒绝误报成功。跨进程协调使用 `.upload-conversions/.locks/` 下稳定保留的摘要锁文件，该目录属于内部实现，不应由 Agent 或部署脚本修改或清理。最终 lease release 是明确提交点：如果新的取消恰在 release 期间到达，系统会先完成 release 并返回已构造的成功结果，而不会把已提交文件报告成取消。Gateway 拒绝不安全或空的 multipart 文件名时会把原名（空值记为 `""`）加入 `skipped_files` 并返回 `success: false`，不会把“上传 0 个文件”误报为成功。
 
 ### 2. 查询上传限制
 ```
@@ -110,7 +110,7 @@ DELETE /api/threads/{thread_id}/uploads/{filename}
 }
 ```
 
-删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束，然后只额外删除它精确拥有的生成资产。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除；新上传仍执行严格的跨平台文件名校验。
+删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先删除精确同步到沙箱的主文件和转换副本；远端删除失败会返回错误并保留宿主机主文件。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除；新上传仍执行严格的跨平台文件名校验。
 
 ## 支持的文档格式
 
@@ -130,9 +130,9 @@ Deletion:  删除 report.pdf 时只删除 .upload-conversions/report.pdf.md；
            /mnt/user-data/uploads/report.md 永远不会被推断为生成文件或自动删除。
 ```
 
-通常生成名为 `<实际主文件名>.md`。如果这一文件名组件会超过 255 个 UTF-8 字节，系统会使用 UTF-8 安全截断的主文件名前缀、完整 SHA-256 摘要和 `.md`，并在响应中返回精确的 `markdown_*` 路径。上传名称碰撞产生的 `_N` 名称同样始终限制在 255 个 UTF-8 字节内；极端情况下，如果后缀本身占满几乎整个组件，系统会截断完整原名后再追加 `_N`。客户端和 Agent 不应自行拼接生成路径。AIO 挂载模式把 `.upload-conversions` 显式挂载为只读；远端 Provisioner 会再次校验该挂载必须来自同一用户/线程并覆盖在可写父挂载之上。挂载契约版本进入 AIO sandbox ID，因此升级前创建、缺少该目录的容器不会被新版本复用。Local 的结构化文件 API 通过只读路径映射拒绝写入，但可选的 Local 宿主机 bash 不受该映射约束，不应对不受信任任务启用。非挂载远端沙箱得到的是独立同步副本，该副本可能可写，但不会修改宿主机上的权威生成文件或内部锁。
+通常生成名为 `<实际主文件名>.md`。如果这一文件名组件会超过 255 个 UTF-8 字节，系统会使用 UTF-8 安全截断的主文件名前缀、完整 SHA-256 摘要和 `.md`，并在响应中返回精确的 `markdown_*` 路径。上传名称碰撞产生的 `_N` 名称同样始终限制在 255 个 UTF-8 字节内；极端情况下，如果后缀本身占满几乎整个组件，系统会截断完整原名后再追加 `_N`。客户端和 Agent 不应自行拼接生成路径。AIO 挂载模式把 `.upload-conversions` 显式挂载为只读；远端 Provisioner 会再次校验该挂载必须来自同一用户/线程并覆盖在可写父挂载之上。挂载契约版本进入 AIO sandbox ID，因此旧容器不能满足新版本对该线程的获取；它仍可能被枚举并仅用于常规孤儿清理。Local 的结构化文件 API 通过只读路径映射拒绝写入，但可选的 Local 宿主机 bash 不受该映射约束，不应对不受信任任务启用。非挂载远端沙箱不会安装嵌套只读转换挂载，而是得到独立同步副本；该副本可能可写，但不会修改宿主机上的权威生成文件或内部锁。
 
-直接上传的 `.md` 主文件会从自身提取 outline/preview；PDF、Office 等其他格式只读取其精确拥有的 `.upload-conversions` 转换件，不会把用户独立上传的同名 Markdown 误认为转换结果。
+直接上传的 `.md` 主文件会从同一个已验证的排他常规文件描述符提取 outline/preview，路径在验证后被替换成 symlink 或 hardlink 也不会读取替换目标；PDF、Office 等其他格式只读取其精确拥有的 `.upload-conversions` 转换件，不会把用户独立上传的同名 Markdown 误认为转换结果。
 
 默认情况下，自动转换是关闭的，以避免在网关主机上对不受信任的 Office/PDF 上传执行解析。只有在受信任部署中明确接受此风险时，才应将 `uploads.auto_convert_documents` 设置为 `true`。
 
@@ -187,7 +187,8 @@ read_file(path="/mnt/user-data/.upload-conversions/document.pdf.md")
 - Gateway、嵌入式 `DeerFlowClient` 和 IM 通道都会执行同一沙箱可见性步骤：挂载型 provider 调整精确发布路径的读取权限；非挂载 provider 获取沙箱后，把本次主文件及生成转换件精确同步到各自虚拟路径
 - 非挂载同步副本是沙箱私有副本；任一路径失败（包括远端已落盘但传输随后报错）、后续响应构造失败或请求取消时，会对本次尝试的精确远端路径执行幂等撤销，再回滚宿主文件
 - 嵌入式 `DeerFlowClient.upload_files()` 以整批为事务边界：后续文件失败会逆序撤销本次调用中此前成功的所有远端副本和宿主 generation
-- 如果 Gateway 与远端沙箱保证挂载同一份线程 user-data（例如正确对齐的共享 PVC、NFS 或 hostPath），可设置 `sandbox.thread_data_mounts: true`；上传路由会跳过 sandbox acquire 和逐文件同步
+- 如果 Gateway 与远端沙箱保证挂载同一份线程 user-data（例如正确对齐的共享 PVC、NFS 或 hostPath），可设置 `sandbox.thread_data_mounts: true`；只有 Provisioner 能通过 `/api/capabilities` 证明当前挂载契约时，上传路由才会跳过 sandbox acquire 和逐文件同步
+- 新 Gateway 与旧 Provisioner 混合部署时会自动降级为显式同步并省略嵌套只读转换挂载，因此组件可按任一顺序滚动升级
 - 不确定挂载关系时应省略该配置并保留自动检测。错误地设为 `true` 会导致文件只存在于 Gateway 存储、沙箱内不可见
 
 ## 测试示例
diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py
index 0cba5f23b..9408acada 100644
--- a/backend/packages/harness/deerflow/client.py
+++ b/backend/packages/harness/deerflow/client.py
@@ -1671,8 +1671,21 @@ class DeerFlowClient:
             PermissionError: If path traversal is detected.
         """
         validate_thread_id(thread_id)
+        from deerflow.sandbox.sandbox_provider import get_sandbox_provider
+        from deerflow.uploads.sandbox_sync import prepare_upload_deletion
+
         uploads_dir = get_uploads_dir(thread_id)
-        return delete_file_safe(uploads_dir, filename)
+        sandbox_provider = get_sandbox_provider(app_config=self._app_config)
+        delete_remote_copy = prepare_upload_deletion(
+            sandbox_provider,
+            thread_id,
+            user_id=get_effective_user_id(),
+        )
+        return delete_file_safe(
+            uploads_dir,
+            filename,
+            delete_remote_copy=delete_remote_copy,
+        )
 
     # ------------------------------------------------------------------
     # Public API — artifacts
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
index c3434b502..2bd433fea 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py
@@ -261,10 +261,21 @@ 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.
         """
-        override = self._config.get("thread_data_mounts")
-        if override is not None:
-            return override
-        return isinstance(self._backend, LocalContainerBackend)
+        config = getattr(self, "_config", {})
+        backend = getattr(self, "_backend", None)
+        override = config.get("thread_data_mounts")
+        # Mount helpers are also used as pure shape inspectors before provider
+        # 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:
+                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,
+                )
+                return False
+        return mounted
 
     # ── Factory methods ──────────────────────────────────────────────────
 
@@ -281,7 +292,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         if provisioner_url:
             logger.info(f"Using remote sandbox backend with provisioner at {provisioner_url}")
             api_key = self._config.get("provisioner_api_key", "")
-            return RemoteSandboxBackend(provisioner_url=provisioner_url, api_key=api_key)
+            backend = RemoteSandboxBackend(provisioner_url=provisioner_url, api_key=api_key)
+            backend.probe_capabilities()
+            return backend
 
         logger.info("Using local container sandbox backend")
         return LocalContainerBackend(
@@ -754,9 +767,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
         reused for an auth/channel run that should mount a user-scoped bucket.
 
         The mount-contract version is part of the identity. During a rolling
-        upgrade, a container created before a required mount was added is never
-        discovered or reclaimed by the new provider. It remains eligible for
-        normal orphan cleanup while the first new-version acquire cold-starts.
+        upgrade, a container created before a required mount was added cannot
+        satisfy the new thread acquisition. Reconciliation may still enumerate
+        and adopt its old ID for normal orphan cleanup.
         """
         return hashlib.sha256(f"mount-v{SANDBOX_MOUNT_CONTRACT_VERSION}:{user_id}:{thread_id}".encode()).hexdigest()[:16]
 
@@ -798,6 +811,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
 
         if thread_id:
             mounts.extend(self._get_thread_mounts(thread_id, user_id=user_id))
+            if not self.uses_thread_data_mounts:
+                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}")
 
         skills_mounts = self._get_skills_mounts(user_id=user_id)
@@ -1496,6 +1511,13 @@ 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):
+            raise SandboxIdentityCollisionError(
+                info.sandbox_id,
+                (info.user_id or "unknown", info.thread_id or "unknown"),
+                (user_id, thread_id),
+            )
+
         key = self._thread_key(thread_id, user_id)
         with self._lock:
             if self._being_torn_down_locally(info.sandbox_id):
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
index 438f6b8f7..1b29c4f75 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py
@@ -44,6 +44,7 @@ _PROVISIONER_EXTRA_MOUNT_PATHS = {
 
 _UPLOADS_CONTAINER_PATH = "/mnt/user-data/uploads"
 _UPLOAD_CONVERSIONS_CONTAINER_PATH = "/mnt/user-data/.upload-conversions"
+_UPLOAD_MOUNT_CONTRACT_VERSION = 2
 
 _LARK_CLI_RUNTIME_CONTAINER_PATH = "/mnt/integrations/lark-cli/runtime"
 _LARK_CLI_CONFIG_CONTAINER_PATH = "/mnt/integrations/lark-cli/config"
@@ -142,6 +143,7 @@ class RemoteSandboxBackend(SandboxBackend):
         """
         self._provisioner_url = provisioner_url.rstrip("/")
         self._api_key = api_key
+        self._mount_contract_version = 0
 
     @property
     def provisioner_url(self) -> str:
@@ -150,6 +152,27 @@ class RemoteSandboxBackend(SandboxBackend):
     def _auth_headers(self) -> dict[str, str]:
         return {"X-API-Key": self._api_key} if self._api_key else {}
 
+    @property
+    def mount_contract_version(self) -> int:
+        """Provisioner mount contract negotiated during provider startup."""
+        return self._mount_contract_version
+
+    def probe_capabilities(self) -> None:
+        """Negotiate optional Provisioner capabilities without breaking old peers."""
+        try:
+            response = requests.get(
+                f"{self._provisioner_url}/api/capabilities",
+                headers=self._auth_headers(),
+                timeout=5,
+            )
+            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
+        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")
+
     # ── SandboxBackend interface ──────────────────────────────────────────
 
     def create(
@@ -330,9 +353,19 @@ class RemoteSandboxBackend(SandboxBackend):
                 return None
             resp.raise_for_status()
             data = resp.json()
+            if data.get("mount_contract_version") != _UPLOAD_MOUNT_CONTRACT_VERSION:
+                logger.info(
+                    "Provisioner discovery ignored sandbox %s with mount contract %r",
+                    sandbox_id,
+                    data.get("mount_contract_version"),
+                )
+                return None
             return SandboxInfo(
                 sandbox_id=sandbox_id,
                 sandbox_url=data["sandbox_url"],
+                user_id=data.get("user_id"),
+                thread_id=data.get("thread_id"),
+                mount_contract_version=data.get("mount_contract_version"),
             )
         except requests.RequestException as exc:
             logger.debug(f"Provisioner discover failed for {sandbox_id}: {exc}")
diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/sandbox_info.py b/backend/packages/harness/deerflow/community/aio_sandbox/sandbox_info.py
index 8b445dee3..aefa6ee19 100644
--- a/backend/packages/harness/deerflow/community/aio_sandbox/sandbox_info.py
+++ b/backend/packages/harness/deerflow/community/aio_sandbox/sandbox_info.py
@@ -19,6 +19,9 @@ class SandboxInfo:
     sandbox_url: str  # e.g. http://localhost:8080 or http://k3s:30001
     container_name: str | None = None  # Only for local container backend
     container_id: str | None = None  # Only for local container backend
+    user_id: str | None = None  # Provisioner-verified identity for remote discovery
+    thread_id: str | None = None
+    mount_contract_version: int | None = None
     created_at: float = field(default_factory=time.time)
 
     def to_dict(self) -> dict:
@@ -27,6 +30,9 @@ class SandboxInfo:
             "sandbox_url": self.sandbox_url,
             "container_name": self.container_name,
             "container_id": self.container_id,
+            "user_id": self.user_id,
+            "thread_id": self.thread_id,
+            "mount_contract_version": self.mount_contract_version,
             "created_at": self.created_at,
         }
 
@@ -37,5 +43,8 @@ class SandboxInfo:
             sandbox_url=data.get("sandbox_url", data.get("base_url", "")),
             container_name=data.get("container_name"),
             container_id=data.get("container_id"),
+            user_id=data.get("user_id"),
+            thread_id=data.get("thread_id"),
+            mount_contract_version=data.get("mount_contract_version"),
             created_at=data.get("created_at", time.time()),
         )
diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py
index 0303690e3..969f9dc01 100644
--- a/backend/packages/harness/deerflow/uploads/manager.py
+++ b/backend/packages/harness/deerflow/uploads/manager.py
@@ -10,7 +10,7 @@ import os
 import secrets
 import shutil
 import stat
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 from pathlib import Path, PureWindowsPath
 from typing import BinaryIO
@@ -616,12 +616,92 @@ def _find_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> P
     return matching_path
 
 
-def delete_file_safe(base_dir: Path, filename: str) -> dict:
+def _restore_staged_deletion(
+    staged_path: Path,
+    original_path: Path,
+    identity: UploadIdentity,
+) -> None:
+    """Restore a staged primary without replacing a newly-created entry."""
+    try:
+        staged_stat = os.lstat(staged_path)
+    except FileNotFoundError:
+        return
+    if not stat.S_ISREG(staged_stat.st_mode) or (
+        staged_stat.st_dev,
+        staged_stat.st_ino,
+    ) != (identity.device, identity.inode):
+        raise UnsafeUploadPathError("Staged upload deletion changed identity")
+    try:
+        original_stat = os.lstat(original_path)
+    except FileNotFoundError:
+        original_stat = None
+    if original_stat is not None:
+        if stat.S_ISREG(original_stat.st_mode) and (
+            original_stat.st_dev,
+            original_stat.st_ino,
+        ) == (identity.device, identity.inode):
+            staged_path.unlink()
+            return
+        raise UnsafeUploadPathError("Upload name was recreated while deletion was being rolled back")
+    try:
+        os.link(staged_path, original_path, follow_symlinks=False)
+    except FileExistsError as exc:
+        raise UnsafeUploadPathError("Upload name was recreated while deletion was being rolled back") from exc
+    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."""
+    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)
+        try:
+            os.link(primary_path, staged_path, follow_symlinks=False)
+        except FileExistsError:
+            stage_lease.release()
+            continue
+        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)
+                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)
+                except BaseException:
+                    logger.warning(
+                        "Failed to restore staged upload deletion: %s",
+                        primary_path,
+                        exc_info=True,
+                    )
+            stage_lease.release()
+            raise
+
+
+def delete_file_safe(
+    base_dir: Path,
+    filename: str,
+    *,
+    delete_remote_copy: Callable[[str], None] | None = None,
+) -> dict:
     """Delete a primary upload and only its exact owned conversion.
 
     Args:
         base_dir: Directory containing the file.
         filename: Name of file to delete.
+        delete_remote_copy: Optional provider hook invoked while the selected
+            primary is staged and its generation lease is still held.
     Returns:
         Dict with success and message.
 
@@ -655,12 +735,25 @@ def delete_file_safe(base_dir: Path, filename: str) -> dict:
         identity = UploadIdentity(device=file_stat.st_dev, inode=file_stat.st_ino)
         actual_file_path = _find_upload_path_by_identity(base_dir, identity)
         owned_conversion = existing_conversion_path_for_upload(actual_file_path)
-        if owned_conversion is not None:
-            owned_conversion.unlink(missing_ok=True)
         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):
             raise UnsafeUploadPathError("Upload is no longer an exclusive directory entry")
-        actual_file_path.unlink()
+        staged_path, stage_lease = _stage_primary_deletion(
+            base_dir,
+            actual_file_path,
+            identity,
+        )
+        try:
+            if delete_remote_copy is not None:
+                delete_remote_copy(actual_file_path.name)
+            if owned_conversion is not None:
+                owned_conversion.unlink(missing_ok=True)
+            staged_path.unlink()
+        except BaseException:
+            _restore_staged_deletion(staged_path, actual_file_path, identity)
+            raise
+        finally:
+            stage_lease.release()
     finally:
         lease.release()
 
diff --git a/backend/packages/harness/deerflow/uploads/sandbox_sync.py b/backend/packages/harness/deerflow/uploads/sandbox_sync.py
index 164e7b8fb..759fea7e3 100644
--- a/backend/packages/harness/deerflow/uploads/sandbox_sync.py
+++ b/backend/packages/harness/deerflow/uploads/sandbox_sync.py
@@ -4,12 +4,14 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 from deerflow.uploads.async_helpers import run_upload_io_cancellation_safe, wait_for_task_completion
-from deerflow.uploads.manager import make_upload_file_sandbox_readable
+from deerflow.uploads.layout import conversion_virtual_path
+from deerflow.uploads.manager import make_upload_file_sandbox_readable, upload_virtual_path
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +46,51 @@ def _remove_remote_paths(sandbox: Any, virtual_paths: tuple[str, ...]) -> None:
         raise first_error
 
 
+def _deletion_hook_for_sandbox(sandbox: Any) -> Callable[[str], None]:
+    def delete_remote_copy(filename: str) -> None:
+        _remove_remote_paths(
+            sandbox,
+            (
+                conversion_virtual_path(filename),
+                upload_virtual_path(filename),
+            ),
+        )
+
+    return delete_remote_copy
+
+
+def prepare_upload_deletion(
+    sandbox_provider: Any,
+    thread_id: str,
+    *,
+    user_id: str | None,
+) -> Callable[[str], None] | None:
+    """Return a lease-safe remote deletion hook for an explicitly synced sandbox."""
+    if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+        return None
+    sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
+    sandbox = sandbox_provider.get(sandbox_id)
+    if sandbox is None:
+        raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
+    return _deletion_hook_for_sandbox(sandbox)
+
+
+async def prepare_upload_deletion_async(
+    sandbox_provider: Any,
+    thread_id: str,
+    *,
+    user_id: str | None,
+) -> Callable[[str], None] | None:
+    """Async counterpart that keeps remote acquisition off the event loop."""
+    if getattr(sandbox_provider, "uses_thread_data_mounts", False):
+        return None
+    sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=user_id)
+    sandbox = sandbox_provider.get(sandbox_id)
+    if sandbox is None:
+        raise RuntimeError(f"Sandbox {sandbox_id!r} not found after acquire")
+    return _deletion_hook_for_sandbox(sandbox)
+
+
 def rollback_sandbox_sync(receipt: SandboxSyncReceipt) -> None:
     """Remove every exact remote path recorded by *receipt*."""
     if receipt.sandbox is not None and receipt.virtual_paths:
diff --git a/backend/packages/harness/deerflow/utils/file_outline.py b/backend/packages/harness/deerflow/utils/file_outline.py
index 9739f491d..4b22de9ad 100644
--- a/backend/packages/harness/deerflow/utils/file_outline.py
+++ b/backend/packages/harness/deerflow/utils/file_outline.py
@@ -10,7 +10,9 @@ import logging
 import os
 import re
 import stat
+from contextlib import contextmanager
 from pathlib import Path
+from typing import TextIO
 
 from deerflow.uploads.layout import (
     UnsafeConversionPathError,
@@ -67,6 +69,67 @@ def _clean_bold_title(raw: str) -> str:
     return merged
 
 
+def _extract_outline_from_stream(stream: TextIO) -> list[dict]:
+    outline: list[dict] = []
+    for lineno, line in enumerate(stream, 1):
+        stripped = line.strip()
+        if not stripped:
+            continue
+
+        if stripped.startswith("#"):
+            title = _clean_bold_title(stripped.lstrip("#").strip())
+            if title:
+                outline.append({"title": title, "line": lineno})
+        elif m := _BOLD_HEADING_RE.match(stripped):
+            title = m.group(1).strip()
+            if title:
+                outline.append({"title": title, "line": lineno})
+        elif _SPLIT_BOLD_HEADING_RE.match(stripped):
+            title = " ".join(re.findall(r"\*\*([^*]+)\*\*", stripped))
+            if title:
+                outline.append({"title": title, "line": lineno})
+
+        if len(outline) > MAX_OUTLINE_ENTRIES:
+            outline.pop()
+            outline.append({"truncated": True})
+            break
+    return outline
+
+
+@contextmanager
+def _open_verified_markdown(md_path: Path):
+    """Open one exclusive regular file and keep that descriptor for all reads."""
+    descriptor: int | None = None
+    try:
+        descriptor = os.open(md_path, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0))
+        descriptor_stat = os.fstat(descriptor)
+        path_stat = os.lstat(md_path)
+    except (OSError, UnicodeError):
+        if descriptor is not None:
+            os.close(descriptor)
+        yield None
+        return
+
+    verified = (
+        stat.S_ISREG(descriptor_stat.st_mode) and descriptor_stat.st_nlink == 1 and stat.S_ISREG(path_stat.st_mode) and path_stat.st_nlink == 1 and (descriptor_stat.st_dev, descriptor_stat.st_ino) == (path_stat.st_dev, path_stat.st_ino)
+    )
+    if not verified:
+        os.close(descriptor)
+        yield None
+        return
+
+    try:
+        stream = os.fdopen(descriptor, "r", encoding="utf-8")
+    except (OSError, UnicodeError):
+        os.close(descriptor)
+        yield None
+        return
+    try:
+        yield stream
+    finally:
+        stream.close()
+
+
 def extract_outline(md_path: Path) -> list[dict]:
     """Extract document outline (headings) from a Markdown file.
 
@@ -94,41 +157,13 @@ def extract_outline(md_path: Path) -> list[dict]:
         render a "showing first N headings" hint without re-scanning the file.
         Returns an empty list if the file cannot be read or has no headings.
     """
-    outline: list[dict] = []
-    try:
-        with md_path.open(encoding="utf-8") as f:
-            for lineno, line in enumerate(f, 1):
-                stripped = line.strip()
-                if not stripped:
-                    continue
-
-                # Style 1: standard Markdown heading
-                if stripped.startswith("#"):
-                    title = _clean_bold_title(stripped.lstrip("#").strip())
-                    if title:
-                        outline.append({"title": title, "line": lineno})
-
-                # Style 2: single bold block with SEC structural keyword
-                elif m := _BOLD_HEADING_RE.match(stripped):
-                    title = m.group(1).strip()
-                    if title:
-                        outline.append({"title": title, "line": lineno})
-
-                # Style 3: split-bold heading — **<num>** **<title>**
-                # Regex already enforces max 4 blocks and non-numeric second block.
-                elif _SPLIT_BOLD_HEADING_RE.match(stripped):
-                    title = " ".join(re.findall(r"\*\*([^*]+)\*\*", stripped))
-                    if title:
-                        outline.append({"title": title, "line": lineno})
-
-                if len(outline) > MAX_OUTLINE_ENTRIES:
-                    outline.pop()
-                    outline.append({"truncated": True})
-                    break
-    except Exception:
-        return []
-
-    return outline
+    with _open_verified_markdown(md_path) as stream:
+        if stream is None:
+            return []
+        try:
+            return _extract_outline_from_stream(stream)
+        except (OSError, UnicodeError):
+            return []
 
 
 def extract_outline_for_file(file_path: Path) -> tuple[list[dict], list[str]]:
@@ -146,12 +181,6 @@ def extract_outline_for_file(file_path: Path) -> tuple[list[dict], list[str]]:
           Empty when outline is non-empty (no fallback needed).
     """
     if file_path.suffix.lower() == ".md":
-        try:
-            primary_stat = os.lstat(file_path)
-        except OSError:
-            return [], []
-        if not stat.S_ISREG(primary_stat.st_mode) or primary_stat.st_nlink != 1:
-            return [], []
         md_path = file_path
     else:
         try:
@@ -162,21 +191,24 @@ def extract_outline_for_file(file_path: Path) -> tuple[list[dict], list[str]]:
     if md_path is None:
         return [], []
 
-    outline = extract_outline(md_path)
-    if outline:
-        logger.debug("Extracted %d outline entries from %s", len(outline), file_path.name)
-        return outline, []
+    with _open_verified_markdown(md_path) as stream:
+        if stream is None:
+            return [], []
+        try:
+            outline = _extract_outline_from_stream(stream)
+            if outline:
+                logger.debug("Extracted %d outline entries from %s", len(outline), file_path.name)
+                return outline, []
 
-    # outline is empty — read the first few non-empty lines as a content preview
-    preview: list[str] = []
-    try:
-        with md_path.open(encoding="utf-8") as f:
-            for line in f:
+            stream.seek(0)
+            preview: list[str] = []
+            for line in stream:
                 stripped = line.strip()
                 if stripped:
                     preview.append(stripped)
                 if len(preview) >= _OUTLINE_PREVIEW_LINES:
                     break
-    except Exception:
-        logger.debug("Failed to read preview lines from %s", md_path, exc_info=True)
-    return [], preview
+            return [], preview
+        except (OSError, UnicodeError):
+            logger.debug("Failed to read outline/preview lines from %s", md_path, exc_info=True)
+            return [], []
diff --git a/backend/tests/blocking_io/test_uploads_router.py b/backend/tests/blocking_io/test_uploads_router.py
index 8ac16f20d..ab089cb15 100644
--- a/backend/tests/blocking_io/test_uploads_router.py
+++ b/backend/tests/blocking_io/test_uploads_router.py
@@ -149,6 +149,7 @@ async def test_list_uploaded_files_does_not_block_event_loop(tmp_path: Path, mon
 
 async def test_delete_uploaded_file_does_not_block_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     _reset_paths(tmp_path, monkeypatch)
+    monkeypatch.setattr(uploads, "get_sandbox_provider", lambda: _MountedProvider())
     uploads_dir = await _thread_uploads_dir("t-delete")
     target = uploads_dir / "notes.txt"
     await asyncio.to_thread(target.write_bytes, b"delete me")
diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py
index fb8ebd468..9ebc5ab5c 100644
--- a/backend/tests/test_aio_sandbox_provider.py
+++ b/backend/tests/test_aio_sandbox_provider.py
@@ -62,6 +62,37 @@ def test_thread_data_mounts_override_precedes_backend_detection(backend_is_local
     assert provider.uses_thread_data_mounts is expected
 
 
+def test_thread_data_mounts_defaults_to_mounted_for_uninitialized_mount_helper():
+    """Pure mount-shape inspection historically works without running provider init."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
+
+    assert provider.uses_thread_data_mounts is True
+
+
+def test_remote_mount_override_downgrades_without_current_provisioner_contract():
+    """A new Gateway must remain usable while the Provisioner is still old."""
+    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")
+    provider._backend._mount_contract_version = 0
+
+    assert provider.uses_thread_data_mounts is False
+
+
+def test_remote_mount_override_uses_current_provisioner_contract():
+    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")
+    provider._backend._mount_contract_version = aio_mod.SANDBOX_MOUNT_CONTRACT_VERSION
+
+    assert provider.uses_thread_data_mounts is True
+
+
 # ── ensure_thread_dirs ───────────────────────────────────────────────────────
 
 
@@ -262,6 +293,7 @@ def test_get_extra_mounts_provisioner_payload_has_unique_container_paths(tmp_pat
     monkeypatch.setattr(remote_backend, "user_should_see_legacy_skills", lambda *_args, **_kwargs: False)
 
     provider = _make_provider(tmp_path)
+    provider._backend = object.__new__(aio_mod.LocalContainerBackend)
     mounts = provider._get_extra_mounts("thread-1", user_id="alice")
     container_paths = [container for _host, container, _read_only in mounts]
 
@@ -298,6 +330,117 @@ def test_get_extra_mounts_provisioner_payload_has_unique_container_paths(tmp_pat
     }
 
 
+def test_nonmounted_remote_provider_omits_read_only_conversion_mount(tmp_path, monkeypatch):
+    """Explicit-sync mode must not write into a nested read-only mount."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    monkeypatch.setattr(aio_mod, "get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_skills_mounts", staticmethod(lambda **_kwargs: []))
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_user_skill_mounts", staticmethod(lambda **_kwargs: []))
+    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_get_lark_cli_runtime_mounts", staticmethod(lambda **_kwargs: []))
+    provider = _make_provider(tmp_path)
+    provider._config["thread_data_mounts"] = False
+    provider._backend = object()
+
+    mounts = provider._get_extra_mounts("thread-1", user_id="alice")
+
+    assert "/mnt/user-data/uploads" in {container for _host, container, _read_only in mounts}
+    assert "/mnt/user-data/.upload-conversions" not in {container for _host, container, _read_only in mounts}
+
+
+def test_remote_backend_negotiates_mount_contract_capability(monkeypatch):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+    requested: dict[str, object] = {}
+
+    class _Response:
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {"mount_contract_version": 2}
+
+    def _get(url, *, headers, timeout):
+        requested.update(url=url, headers=headers, timeout=timeout)
+        return _Response()
+
+    monkeypatch.setattr(remote_mod.requests, "get", _get)
+    backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002", api_key="secret")
+
+    backend.probe_capabilities()
+
+    assert backend.mount_contract_version == 2
+    assert requested == {
+        "url": "http://provisioner:8002/api/capabilities",
+        "headers": {"X-API-Key": "secret"},
+        "timeout": 5,
+    }
+
+
+def test_remote_backend_treats_missing_mount_capability_as_legacy(monkeypatch):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+
+    class _Response:
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {"lark_cli_init_image": False}
+
+    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
+
+
+def test_remote_discovery_carries_verified_identity(monkeypatch):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+
+    class _Response:
+        status_code = 200
+
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {
+                "sandbox_url": "http://sandbox.local",
+                "user_id": "alice",
+                "thread_id": "thread-1",
+                "mount_contract_version": 2,
+            }
+
+    monkeypatch.setattr(remote_mod.requests, "get", lambda *_args, **_kwargs: _Response())
+    backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+
+    info = backend.discover("sandbox-1")
+
+    assert info is not None
+    assert (info.user_id, info.thread_id, info.mount_contract_version) == (
+        "alice",
+        "thread-1",
+        2,
+    )
+
+
+def test_remote_discovery_ignores_legacy_unverified_response(monkeypatch):
+    remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
+
+    class _Response:
+        status_code = 200
+
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {"sandbox_url": "http://legacy.local"}
+
+    monkeypatch.setattr(remote_mod.requests, "get", lambda *_args, **_kwargs: _Response())
+    backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
+
+    assert backend.discover("sandbox-old") is None
+
+
 @pytest.mark.parametrize(
     "conversion_mount",
     [
@@ -1101,6 +1244,31 @@ def test_aio_mount_contract_upgrade_does_not_reuse_pre_upgrade_container_id():
     assert current_id == hashlib.sha256(f"mount-v{aio_mod.SANDBOX_MOUNT_CONTRACT_VERSION}:{user_id}:{thread_id}".encode()).hexdigest()[:16]
 
 
+def test_reconcile_may_adopt_pre_upgrade_id_only_for_orphan_cleanup(tmp_path):
+    """Versioning blocks new acquisition reuse, not orphan enumeration/reaping."""
+    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
+    provider = _make_provider(tmp_path)
+    provider._lock = aio_mod.threading.Lock()
+    provider._warm_pool = {}
+    provider._warm_pool_identity = {}
+    provider._sandboxes = {}
+    provider._sandbox_infos = {}
+    provider._unowned_since = {}
+    old_id = hashlib.sha256(b"alice:thread-1").hexdigest()[:16]
+    old_info = aio_mod.SandboxInfo(
+        sandbox_id=old_id,
+        sandbox_url="http://old-sandbox",
+    )
+    provider._backend = SimpleNamespace(list_running=MagicMock(return_value=[old_info]))
+
+    provider._reconcile_orphans()
+
+    current_id = provider._deterministic_sandbox_id("thread-1", "alice")
+    assert old_id in provider._warm_pool
+    assert current_id != old_id
+    assert current_id not in provider._warm_pool
+
+
 def test_aio_forced_collision_never_overwrites_active_tenant(
     tmp_path,
     monkeypatch,
diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py
index f925939f0..946eb2773 100644
--- a/backend/tests/test_client.py
+++ b/backend/tests/test_client.py
@@ -2609,7 +2609,12 @@ class TestUploads:
             conversion.parent.mkdir()
             conversion.write_text("generated")
 
-            with patch("deerflow.client.get_uploads_dir", return_value=uploads_dir), patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir):
+            provider = MagicMock(uses_thread_data_mounts=True)
+            with (
+                patch("deerflow.client.get_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=provider),
+            ):
                 result = client.delete_upload("thread-1", "delete-me.txt")
 
             assert result["success"] is True
@@ -2620,7 +2625,11 @@ class TestUploads:
 
     def test_delete_upload_not_found(self, client):
         with tempfile.TemporaryDirectory() as tmp:
-            with patch("deerflow.client.get_uploads_dir", return_value=Path(tmp)):
+            provider = MagicMock(uses_thread_data_mounts=True)
+            with (
+                patch("deerflow.client.get_uploads_dir", return_value=Path(tmp)),
+                patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=provider),
+            ):
                 with pytest.raises(FileNotFoundError):
                     client.delete_upload("thread-1", "nope.txt")
 
@@ -2632,7 +2641,11 @@ class TestUploads:
         legacy = uploads_dir / filename
         legacy.write_bytes(b"legacy")
 
-        with patch("deerflow.client.get_uploads_dir", return_value=uploads_dir):
+        provider = MagicMock(uses_thread_data_mounts=True)
+        with (
+            patch("deerflow.client.get_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=provider),
+        ):
             result = client.delete_upload("thread-1", legacy.name)
 
         assert result["success"] is True
@@ -2641,10 +2654,42 @@ class TestUploads:
     def test_delete_upload_path_traversal(self, client):
         with tempfile.TemporaryDirectory() as tmp:
             uploads_dir = Path(tmp)
-            with patch("deerflow.client.get_uploads_dir", return_value=uploads_dir), patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir):
+            provider = MagicMock(uses_thread_data_mounts=True)
+            with (
+                patch("deerflow.client.get_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+                patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=provider),
+            ):
                 with pytest.raises(PathTraversalError):
                     client.delete_upload("thread-1", "../../etc/passwd")
 
+    def test_delete_upload_removes_explicitly_synced_remote_paths(self, client, tmp_path):
+        uploads_dir = tmp_path / "uploads"
+        uploads_dir.mkdir()
+        primary = uploads_dir / "report.pdf"
+        primary.write_bytes(b"pdf")
+        conversion = conversion_path_for_upload(primary)
+        conversion.parent.mkdir()
+        conversion.write_text("generated", encoding="utf-8")
+        provider = MagicMock(uses_thread_data_mounts=False)
+        provider.acquire.return_value = "remote-1"
+        sandbox = MagicMock()
+        provider.get.return_value = sandbox
+
+        with (
+            patch("deerflow.client.get_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.sandbox.sandbox_provider.get_sandbox_provider", return_value=provider),
+        ):
+            result = client.delete_upload("thread-1", "report.pdf")
+
+        assert result["success"] is True
+        assert sandbox.remove_file.call_args_list == [
+            (("/mnt/user-data/uploads/report.pdf",),),
+            (("/mnt/user-data/.upload-conversions/report.pdf.md",),),
+        ]
+        assert not primary.exists()
+        assert not conversion.exists()
+
 
 # ---------------------------------------------------------------------------
 # Artifacts
diff --git a/backend/tests/test_list_uploaded_files_tool.py b/backend/tests/test_list_uploaded_files_tool.py
index b72e9a59d..6c8b5aa53 100644
--- a/backend/tests/test_list_uploaded_files_tool.py
+++ b/backend/tests/test_list_uploaded_files_tool.py
@@ -223,6 +223,47 @@ class TestListUploadedFiles:
         assert extract_outline_for_file(symlink) == ([], [])
         assert extract_outline_for_file(hardlink) == ([], [])
 
+    @pytest.mark.parametrize("replacement", ["symlink", "hardlink"])
+    def test_direct_markdown_outline_reads_one_verified_descriptor(
+        self,
+        tmp_path,
+        monkeypatch,
+        replacement,
+    ):
+        """Replacing the path after validation must not expose replacement bytes."""
+        import deerflow.utils.file_outline as outline_module
+
+        uploads_dir = _uploads_dir(tmp_path)
+        primary = uploads_dir / "notes.md"
+        primary.write_text("# Safe heading\n", encoding="utf-8")
+        outside = tmp_path / "outside.md"
+        outside.write_text("# SECRET VIA REPLACEMENT\n", encoding="utf-8")
+        real_lstat = outline_module.os.lstat
+        replaced = False
+
+        def lstat_then_replace(path):
+            nonlocal replaced
+            result = real_lstat(path)
+            if Path(path) == primary and not replaced:
+                replaced = True
+                primary.unlink()
+                if replacement == "symlink":
+                    try:
+                        primary.symlink_to(outside)
+                    except OSError as exc:
+                        if getattr(exc, "winerror", None) == 1314:
+                            pytest.skip("Windows symlink privilege is not available")
+                        raise
+                else:
+                    os.link(outside, primary)
+            return result
+
+        monkeypatch.setattr(outline_module.os, "lstat", lstat_then_replace)
+
+        outline, preview = extract_outline_for_file(primary)
+
+        assert all("SECRET" not in str(item) for item in [*outline, *preview])
+
     def test_long_filename_result_includes_exact_generated_markdown_path(self, tmp_path):
         uploads_dir = _uploads_dir(tmp_path)
         filename = f"{'a' * 250}.pdf"
diff --git a/backend/tests/test_provisioner_mount_contract.py b/backend/tests/test_provisioner_mount_contract.py
index 6ed0e346f..420ce99b0 100644
--- a/backend/tests/test_provisioner_mount_contract.py
+++ b/backend/tests/test_provisioner_mount_contract.py
@@ -18,6 +18,7 @@ def _literal_assignment(path: Path, name: str):
 
 def test_gateway_and_provisioner_extra_mount_contracts_match() -> None:
     gateway_path = REPO_ROOT / "backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py"
+    provider_path = REPO_ROOT / "backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py"
     provisioner_path = REPO_ROOT / "docker/provisioner/app.py"
 
     gateway_paths = _literal_assignment(gateway_path, "_PROVISIONER_EXTRA_MOUNT_PATHS")
@@ -27,3 +28,11 @@ def test_gateway_and_provisioner_extra_mount_contracts_match() -> None:
     assert "/mnt/user-data/.upload-conversions" in gateway_paths
     assert "/mnt/integrations/lark-cli/runtime" in gateway_paths
     assert _literal_assignment(provisioner_path, "MAX_EXTRA_MOUNTS") == 9
+    assert _literal_assignment(gateway_path, "_UPLOAD_MOUNT_CONTRACT_VERSION") == _literal_assignment(
+        provider_path,
+        "SANDBOX_MOUNT_CONTRACT_VERSION",
+    )
+    assert _literal_assignment(provisioner_path, "MOUNT_CONTRACT_VERSION") == _literal_assignment(
+        provider_path,
+        "SANDBOX_MOUNT_CONTRACT_VERSION",
+    )
diff --git a/backend/tests/test_provisioner_pvc_volumes.py b/backend/tests/test_provisioner_pvc_volumes.py
index 12b2e806b..89481b033 100644
--- a/backend/tests/test_provisioner_pvc_volumes.py
+++ b/backend/tests/test_provisioner_pvc_volumes.py
@@ -62,15 +62,28 @@ class TestBuildVolumes:
             "user-data",
         ]
 
-    def test_hostpath_userdata_includes_thread_id(self, provisioner_module):
-        """hostPath user-data path should include thread_id."""
+    @pytest.mark.parametrize("user_id", ["alice", "bob"])
+    def test_hostpath_userdata_uses_exact_user_thread_root(self, provisioner_module, user_id):
+        """Parent and nested upload mounts must resolve from one tenant root."""
         provisioner_module.USERDATA_PVC_NAME = ""
-        volumes = provisioner_module._build_volumes("my-thread-42")
-        userdata_vol = volumes[-1]
-        path = userdata_vol.host_path.path
-        assert "my-thread-42" in path
-        assert path.endswith("user-data")
+        provisioner_module.DEER_FLOW_HOST_BASE_DIR = "/state"
+        conversion = provisioner_module.ExtraMount(
+            host_path=f"/state/users/{user_id}/threads/my-thread-42/user-data/.upload-conversions",
+            container_path="/mnt/user-data/.upload-conversions",
+            read_only=True,
+        )
+
+        volumes = provisioner_module._build_volumes(
+            "my-thread-42",
+            user_id=user_id,
+            extra_mounts=[conversion],
+        )
+
+        userdata_vol = next(volume for volume in volumes if volume.name == "user-data")
+        conversion_vol = next(volume for volume in volumes if volume.name == "extra-0")
+        assert userdata_vol.host_path.path == f"/state/users/{user_id}/threads/my-thread-42/user-data"
         assert userdata_vol.host_path.type == "DirectoryOrCreate"
+        assert conversion_vol.host_path.path == (f"/state/users/{user_id}/threads/my-thread-42/user-data/.upload-conversions")
 
     # ── PVC mode (single-volume fallback) ──────────────────────────────
 
diff --git a/backend/tests/test_provisioner_request_threading.py b/backend/tests/test_provisioner_request_threading.py
index c0e74ea6f..1df2fdc65 100644
--- a/backend/tests/test_provisioner_request_threading.py
+++ b/backend/tests/test_provisioner_request_threading.py
@@ -58,6 +58,7 @@ class _RecordingCoreV1:
         self.service_read_counts: dict[str, int] = {}
         self.created_pods: list[str] = []
         self.created_pod_specs: dict[str, object] = {}
+        self.existing_pod_specs: dict[str, object] = {}
         self.created_services: list[str] = []
 
     def _record_k8s_call(self) -> None:
@@ -86,6 +87,11 @@ class _RecordingCoreV1:
 
     def read_namespaced_pod(self, _name: str, _namespace: str):
         self._record_k8s_call()
+        sandbox_id = _name.removeprefix("sandbox-")
+        pod = self.existing_pod_specs.get(sandbox_id) or self.created_pod_specs.get(sandbox_id)
+        if pod is not None:
+            pod.status = SimpleNamespace(phase="Running")
+            return pod
         return SimpleNamespace(status=SimpleNamespace(phase="Running"))
 
     def create_namespaced_pod(self, _namespace: str, pod) -> None:
@@ -172,6 +178,11 @@ async def test_sandbox_business_routes_run_k8s_client_off_event_loop_thread(
     )
     monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
     monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret")
+    fake_core_v1.existing_pod_specs["sandbox-existing"] = provisioner_module._build_pod(
+        "sandbox-existing",
+        "thread-1",
+        user_id="user-1",
+    )
 
     with _detect_provisioner_blocking_io(provisioner_module):
         transport = httpx.ASGITransport(app=provisioner_module.app)
@@ -189,6 +200,129 @@ async def test_sandbox_business_routes_run_k8s_client_off_event_loop_thread(
         assert fake_core_v1.created_services == [expected_created_sandbox]
 
 
+def test_existing_sandbox_rejects_cross_tenant_reuse(
+    monkeypatch: pytest.MonkeyPatch,
+    provisioner_module,
+) -> None:
+    fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
+    fake_core_v1.existing_pod_specs["sandbox-existing"] = provisioner_module._build_pod(
+        "sandbox-existing",
+        "thread-1",
+        user_id="alice",
+    )
+    monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
+
+    with pytest.raises(provisioner_module.HTTPException) as exc_info:
+        provisioner_module.create_sandbox(
+            provisioner_module.CreateSandboxRequest(
+                sandbox_id="sandbox-existing",
+                thread_id="thread-1",
+                user_id="bob",
+            )
+        )
+
+    assert exc_info.value.status_code == 409
+    assert "another user" in exc_info.value.detail
+
+
+def test_existing_sandbox_validates_read_only_conversion_before_fast_path(
+    monkeypatch: pytest.MonkeyPatch,
+    provisioner_module,
+) -> None:
+    fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
+    fake_core_v1.existing_pod_specs["sandbox-existing"] = provisioner_module._build_pod(
+        "sandbox-existing",
+        "thread-1",
+        user_id="alice",
+    )
+    monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
+
+    with pytest.raises(provisioner_module.HTTPException) as exc_info:
+        provisioner_module.create_sandbox(
+            provisioner_module.CreateSandboxRequest(
+                sandbox_id="sandbox-existing",
+                thread_id="thread-1",
+                user_id="alice",
+                extra_mounts=[
+                    provisioner_module.ExtraMount(
+                        host_path=("/.deer-flow/users/alice/threads/thread-1/user-data/.upload-conversions"),
+                        container_path="/mnt/user-data/.upload-conversions",
+                        read_only=False,
+                    )
+                ],
+            )
+        )
+
+    assert exc_info.value.status_code == 400
+    assert "read-only" in exc_info.value.detail
+
+
+def test_existing_sandbox_rejects_tampered_mount_spec(
+    monkeypatch: pytest.MonkeyPatch,
+    provisioner_module,
+) -> None:
+    fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
+    existing = provisioner_module._build_pod(
+        "sandbox-existing",
+        "thread-1",
+        user_id="alice",
+    )
+    userdata = next(volume for volume in existing.spec.volumes if volume.name == "user-data")
+    userdata.host_path.path = "/state/users/mallory/threads/thread-1/user-data"
+    fake_core_v1.existing_pod_specs["sandbox-existing"] = existing
+    monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
+
+    with pytest.raises(provisioner_module.HTTPException) as exc_info:
+        provisioner_module.create_sandbox(
+            provisioner_module.CreateSandboxRequest(
+                sandbox_id="sandbox-existing",
+                thread_id="thread-1",
+                user_id="alice",
+            )
+        )
+
+    assert exc_info.value.status_code == 409
+    assert "incompatible or unverifiable" in exc_info.value.detail
+
+
+def test_existing_sandbox_contract_ignores_kubernetes_service_account_injection(
+    monkeypatch: pytest.MonkeyPatch,
+    provisioner_module,
+) -> None:
+    """Admission-added service-account mounts are outside the DeerFlow contract."""
+    fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
+    existing = provisioner_module._build_pod(
+        "sandbox-existing",
+        "thread-1",
+        user_id="alice",
+    )
+    existing.spec.volumes.append(
+        provisioner_module.k8s_client.V1Volume(
+            name="kube-api-access-abcde",
+            projected=provisioner_module.k8s_client.V1ProjectedVolumeSource(sources=[]),
+        )
+    )
+    existing.spec.containers[0].volume_mounts.append(
+        provisioner_module.k8s_client.V1VolumeMount(
+            name="kube-api-access-abcde",
+            mount_path="/var/run/secrets/kubernetes.io/serviceaccount",
+            read_only=True,
+        )
+    )
+    fake_core_v1.existing_pod_specs["sandbox-existing"] = existing
+    monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
+
+    response = provisioner_module.create_sandbox(
+        provisioner_module.CreateSandboxRequest(
+            sandbox_id="sandbox-existing",
+            thread_id="thread-1",
+            user_id="alice",
+        )
+    )
+
+    assert response.status == "Running"
+
+
 @pytest.mark.parametrize(
     ("include_legacy_skills", "expected_mount_names"),
     [
diff --git a/backend/tests/test_remote_sandbox_backend.py b/backend/tests/test_remote_sandbox_backend.py
index 1c3bb0eb5..4c19cb283 100644
--- a/backend/tests/test_remote_sandbox_backend.py
+++ b/backend/tests/test_remote_sandbox_backend.py
@@ -473,7 +473,15 @@ def test_provisioner_discover_returns_info_on_success(monkeypatch):
     backend = RemoteSandboxBackend("http://provisioner:8002")
 
     def mock_get(url: str, timeout: int, headers=None):
-        return _StubResponse(payload={"sandbox_id": "abc123", "sandbox_url": "http://k3s:31001"})
+        return _StubResponse(
+            payload={
+                "sandbox_id": "abc123",
+                "sandbox_url": "http://k3s:31001",
+                "user_id": "alice",
+                "thread_id": "thread-1",
+                "mount_contract_version": 2,
+            }
+        )
 
     monkeypatch.setattr(requests, "get", mock_get)
 
@@ -481,6 +489,11 @@ def test_provisioner_discover_returns_info_on_success(monkeypatch):
     assert info is not None
     assert info.sandbox_id == "abc123"
     assert info.sandbox_url == "http://k3s:31001"
+    assert (info.user_id, info.thread_id, info.mount_contract_version) == (
+        "alice",
+        "thread-1",
+        2,
+    )
 
 
 def test_provisioner_discover_returns_none_on_request_exception(monkeypatch):
diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py
index 61ea9f2e7..c7258a16f 100644
--- a/backend/tests/test_uploads_manager.py
+++ b/backend/tests/test_uploads_manager.py
@@ -908,6 +908,36 @@ class TestDeleteFileSafe:
         assert primary.read_bytes() == b"payload"
         assert alias.read_bytes() == b"payload"
 
+    def test_delete_hardlink_after_identity_scan_preserves_conversion(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"payload")
+        conversion = conversion_path_for_upload(primary)
+        conversion.parent.mkdir()
+        conversion.write_text("generated", encoding="utf-8")
+        alias = uploads / "alias.pdf"
+        real_find = upload_manager_module._find_upload_path_by_identity
+
+        def add_alias_after_scan(base_dir, identity):
+            found = real_find(base_dir, identity)
+            os.link(found, alias)
+            return found
+
+        with patch.object(
+            upload_manager_module,
+            "_find_upload_path_by_identity",
+            side_effect=add_alias_after_scan,
+        ):
+            with pytest.raises(UnsafeUploadPathError, match="exclusive"):
+                delete_file_safe(uploads, primary.name)
+
+        assert primary.read_bytes() == b"payload"
+        assert alias.read_bytes() == b"payload"
+        assert conversion.read_text(encoding="utf-8") == "generated"
+
     def test_delete_removes_owned_conversion_but_preserves_legacy_sibling(self, tmp_path):
         uploads = tmp_path / "user-data" / "uploads"
         uploads.mkdir(parents=True)
diff --git a/backend/tests/test_uploads_router.py b/backend/tests/test_uploads_router.py
index b0bac68ca..3c7e89b49 100644
--- a/backend/tests/test_uploads_router.py
+++ b/backend/tests/test_uploads_router.py
@@ -433,7 +433,10 @@ async def test_waiting_delete_cannot_starve_general_io_lease_release(tmp_path, m
     single_worker = ThreadPoolExecutor(max_workers=1)
     monkeypatch.setattr(file_io_module, "_FILE_IO_EXECUTOR", single_worker)
 
-    with patch.object(uploads, "get_uploads_dir", return_value=tmp_path):
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=tmp_path),
+        patch.object(uploads, "get_sandbox_provider", return_value=_mounted_provider()),
+    ):
         deletion = asyncio.create_task(call_unwrapped(uploads.delete_uploaded_file, "thread-delete", "notes.txt", request=MagicMock()))
         await asyncio.sleep(0.05)
         release = asyncio.create_task(uploads._run_file_io_commit(publication.release))
@@ -1293,7 +1296,10 @@ def test_delete_uploaded_file_removes_owned_conversion_and_preserves_user_markdo
     conversion.parent.mkdir()
     conversion.write_text("converted", encoding="utf-8")
 
-    with patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir):
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=_mounted_provider()),
+    ):
         result = asyncio.run(call_unwrapped(uploads.delete_uploaded_file, "thread-aio", "report.pdf", request=MagicMock()))
 
     assert result == {"success": True, "message": "Deleted report.pdf"}
@@ -1310,7 +1316,10 @@ def test_delete_uploaded_file_accepts_listed_legacy_posix_filename(tmp_path, fil
     legacy = thread_uploads_dir / filename
     legacy.write_bytes(b"legacy")
 
-    with patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir):
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=_mounted_provider()),
+    ):
         result = asyncio.run(
             call_unwrapped(
                 uploads.delete_uploaded_file,
@@ -1324,6 +1333,102 @@ def test_delete_uploaded_file_accepts_listed_legacy_posix_filename(tmp_path, fil
     assert not legacy.exists()
 
 
+def test_delete_uploaded_file_removes_explicitly_synced_remote_paths(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    primary = thread_uploads_dir / "report.pdf"
+    primary.write_bytes(b"pdf")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_text("generated", encoding="utf-8")
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = False
+    provider.acquire_async = AsyncMock(return_value="remote-1")
+    sandbox = MagicMock()
+    provider.get.return_value = sandbox
+
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=provider),
+    ):
+        result = asyncio.run(
+            call_unwrapped(
+                uploads.delete_uploaded_file,
+                "thread-remote",
+                "report.pdf",
+                request=MagicMock(),
+            )
+        )
+
+    assert result["success"] is True
+    assert sandbox.remove_file.call_args_list == [
+        (("/mnt/user-data/uploads/report.pdf",),),
+        (("/mnt/user-data/.upload-conversions/report.pdf.md",),),
+    ]
+    assert not primary.exists()
+    assert not conversion.exists()
+
+
+def test_delete_uploaded_file_preserves_host_when_remote_delete_fails(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    primary = thread_uploads_dir / "notes.txt"
+    primary.write_bytes(b"notes")
+    provider = MagicMock()
+    provider.uses_thread_data_mounts = False
+    provider.acquire_async = AsyncMock(return_value="remote-1")
+    sandbox = MagicMock()
+
+    def fail_remote_delete(_virtual_path):
+        assert not primary.exists(), "primary must be staged before remote deletion"
+        raise OSError("remote unavailable")
+
+    sandbox.remove_file.side_effect = fail_remote_delete
+    provider.get.return_value = sandbox
+
+    with (
+        patch.object(uploads, "get_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=provider),
+    ):
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(
+                call_unwrapped(
+                    uploads.delete_uploaded_file,
+                    "thread-remote",
+                    "notes.txt",
+                    request=MagicMock(),
+                )
+            )
+
+    assert exc_info.value.status_code == 500
+    assert primary.read_bytes() == b"notes"
+
+
+@pytest.mark.parametrize("filename", ["", None])
+def test_upload_files_reports_empty_multipart_filename_as_skipped(tmp_path, filename):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+
+    with (
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=_mounted_provider()),
+    ):
+        result = asyncio.run(
+            call_unwrapped(
+                uploads.upload_files,
+                "thread-local",
+                request=MagicMock(),
+                files=[UploadFile(filename=filename, file=BytesIO(b"data"))],
+                config=SimpleNamespace(),
+            )
+        )
+
+    assert result.success is False
+    assert result.files == []
+    assert result.skipped_files == [""]
+    assert result.message == "Successfully uploaded 0 file(s); skipped 1 unsafe file(s)"
+
+
 def test_auto_convert_documents_enabled_defaults_to_false_on_config_errors():
     class BrokenConfig:
         def __getattribute__(self, name):
diff --git a/config.example.yaml b/config.example.yaml
index 56af387d1..17ae9b39e 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -1277,6 +1277,8 @@ sandbox:
 #   # guarantees both sides use the same thread user-data directories, such as
 #   # 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.
 #   # thread_data_mounts: true
 #
 #   # Optional: Additional mount directories from host to container
diff --git a/docker/provisioner/README.md b/docker/provisioner/README.md
index 9d73f2f11..285026cc2 100644
--- a/docker/provisioner/README.md
+++ b/docker/provisioner/README.md
@@ -88,18 +88,27 @@ 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.
+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.
 
 **Response**:
 ```json
 {
   "sandbox_id": "abc-123",
   "sandbox_url": "http://host.docker.internal:32123",
-  "status": "Pending"
+  "status": "Pending",
+  "user_id": "user-789",
+  "thread_id": "thread-456",
+  "mount_contract_version": 2
 }
 ```
 
-**Idempotent**: Calling with the same `sandbox_id` returns the existing sandbox info.
+**Conditionally idempotent**: Calling with the same `sandbox_id` returns the
+existing sandbox only when its tenant labels, contract version, stored mount
+signature, and live Pod mount specification match the request. Mismatches return
+`409` instead of reusing an unverifiable or cross-tenant Pod.
 
 ### `GET /api/sandboxes/{sandbox_id}`
 Get status and URL of a specific sandbox.
@@ -153,8 +162,8 @@ The provisioner is configured via environment variables (set in [docker-compose-
 | `SANDBOX_IMAGE` | `enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest` | AIO-compatible container image for sandbox Pods |
 | `LARK_CLI_INIT_IMAGE` | empty (feature off) | Optional lark-cli init image (Pattern A). When set, sandbox Pods requesting the lark-cli runtime get an init container + shared `emptyDir` that provisions `lark-cli`, instead of a hostPath/PVC runtime mount. See [`docker/lark-cli-init`](../lark-cli-init/README.md) |
 | `LARK_CLI_BROKER_IMAGE` | empty (feature off) | Optional lark-cli broker image (Pattern B, issue #4338). When set, sandbox Pods requesting the broker get a shim init container + a `lark-cli-broker` sidecar that holds the credentials; the plaintext `config`/`data` are mounted into the **sidecar only**, never the sandbox. Supersedes `LARK_CLI_INIT_IMAGE` when both are set. See [`docker/lark-cli-broker`](../lark-cli-broker/README.md) |
-| `THREADS_HOST_PATH` | - | **Host machine** path to threads data directory (must be absolute) |
-| `DEER_FLOW_HOST_BASE_DIR` | `/.deer-flow` | **Host machine** DeerFlow data root containing global and per-user `skills_view` projections |
+| `THREADS_HOST_PATH` | - | Legacy fallback used only to infer the DeerFlow root when `DEER_FLOW_HOST_BASE_DIR` is empty |
+| `DEER_FLOW_HOST_BASE_DIR` | `/.deer-flow` | **Host machine** DeerFlow data root. HostPath user-data resolves to `users/{user_id}/threads/{thread_id}/user-data`, matching Gateway storage and the PVC subpath contract |
 | `SKILLS_PVC_NAME` | empty (use hostPath) | PVC name for skills volume; when set, sandbox Pods use PVC instead of hostPath |
 | `SKILLS_PVC_SUBPATH_TEMPLATE` | empty | Optional `subPath` template for `SKILLS_PVC_NAME`. Supports `{user_id}` and `{thread_id}`. When empty, the skills PVC root is mounted unchanged |
 | `USERDATA_PVC_NAME` | empty (use hostPath) | PVC name for user-data volume; when set, uses PVC with `subPath: deer-flow/users/{user_id}/threads/{thread_id}/user-data` |
@@ -250,10 +259,17 @@ kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}'
      - Read Namespaces (to create `deer-flow` if missing)
 
 4. **Host Paths**:
-   - `DEER_FLOW_HOST_BASE_DIR` and `THREADS_HOST_PATH` must be **absolute paths on the host machine**
+   - `DEER_FLOW_HOST_BASE_DIR` must be an **absolute path on the host machine**
    - These paths are mounted into sandbox Pods via K8s HostPath volumes
    - The paths must exist and be readable by the K8s node
 
+   Upgrading from a Provisioner that mounted
+   `THREADS_HOST_PATH/{thread_id}/user-data` changes the HostPath source to
+   `DEER_FLOW_HOST_BASE_DIR/users/{user_id}/threads/{thread_id}/user-data`.
+   Before enabling mounted mode, copy any sandbox-written workspace/output data
+   needed from the legacy tree into the matching user tree, preserve the legacy
+   tree for rollback, and verify both paths before deleting the old copy.
+
 ### Docker Compose Setup
 
 The provisioner runs as part of the docker-compose-dev stack:
diff --git a/docker/provisioner/app.py b/docker/provisioner/app.py
index 672e4b17b..095456df9 100644
--- a/docker/provisioner/app.py
+++ b/docker/provisioner/app.py
@@ -29,6 +29,8 @@ Architecture (docker-compose-dev):
 
 from __future__ import annotations
 
+import hashlib
+import json
 import logging
 import os
 import posixpath
@@ -76,7 +78,9 @@ LARK_CLI_BROKER_IMAGE = os.environ.get("LARK_CLI_BROKER_IMAGE", "")
 # Optional comma-separated lark-cli subcommand denylist forwarded to the broker
 # sidecar (issue #4338 hardening). Empty ⇒ no subcommand is blocked. See the
 # broker README's "subcommand denylist" section.
-LARK_CLI_BROKER_DENY_SUBCOMMANDS = os.environ.get("DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS", "")
+LARK_CLI_BROKER_DENY_SUBCOMMANDS = os.environ.get(
+    "DEERFLOW_LARK_BROKER_DENY_SUBCOMMANDS", ""
+)
 LARK_CLI_CONFIG_CONTAINER_PATH = "/mnt/integrations/lark-cli/config"
 LARK_CLI_DATA_CONTAINER_PATH = "/mnt/integrations/lark-cli/data"
 # Where the broker sidecar reads the per-user credentials (sidecar-only paths).
@@ -95,15 +99,22 @@ SANDBOX_SERVICE_TYPE = os.environ.get("SANDBOX_SERVICE_TYPE", "NodePort")
 try:
     SANDBOX_CONTAINER_PORT = int(SANDBOX_CONTAINER_PORT_RAW)
 except ValueError as exc:
-    raise RuntimeError(f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT_RAW!r}; expected an integer TCP port") from exc
+    raise RuntimeError(
+        f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT_RAW!r}; expected an integer TCP port"
+    ) from exc
 if not (1 <= SANDBOX_CONTAINER_PORT <= 65535):
-    raise RuntimeError(f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT}; expected a value in [1, 65535]")
+    raise RuntimeError(
+        f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT}; expected a value in [1, 65535]"
+    )
 if SANDBOX_SERVICE_TYPE not in {"NodePort", "ClusterIP"}:
-    raise RuntimeError(f"Invalid SANDBOX_SERVICE_TYPE={SANDBOX_SERVICE_TYPE!r}; expected 'NodePort' or 'ClusterIP'")
+    raise RuntimeError(
+        f"Invalid SANDBOX_SERVICE_TYPE={SANDBOX_SERVICE_TYPE!r}; expected 'NodePort' or 'ClusterIP'"
+    )
 SAFE_THREAD_ID_PATTERN = r"^[A-Za-z0-9_-]{1,64}$"
 SAFE_USER_ID_PATTERN = r"^[A-Za-z0-9_\-]+$"
 DEFAULT_USER_ID = "default"
 MAX_EXTRA_MOUNTS = 9
+MOUNT_CONTRACT_VERSION = 2
 ALLOWED_EXTRA_MOUNT_PATHS = {
     "/mnt/user-data/.upload-conversions",
     "/mnt/acp-workspace",
@@ -115,6 +126,12 @@ ALLOWED_EXTRA_MOUNT_PATHS = {
 }
 UPLOAD_CONVERSIONS_CONTAINER_PATH = "/mnt/user-data/.upload-conversions"
 UPLOAD_CONVERSIONS_DIRNAME = ".upload-conversions"
+MOUNT_CONTRACT_LABEL = "deerflow.dev/mount-contract"
+USER_ID_HASH_LABEL = "deerflow.dev/user-id-hash"
+THREAD_ID_HASH_LABEL = "deerflow.dev/thread-id-hash"
+USER_ID_ANNOTATION = "deerflow.dev/user-id"
+THREAD_ID_ANNOTATION = "deerflow.dev/thread-id"
+MOUNT_SIGNATURE_ANNOTATION = "deerflow.dev/mount-signature"
 
 # Path to the kubeconfig *inside* the provisioner container.
 # Typically the host's ~/.kube/config is mounted here.
@@ -164,7 +181,9 @@ def _is_path_under_base(path: str, base: str) -> bool:
     if not base:
         return False
     try:
-        return os.path.commonpath([os.path.normpath(path), os.path.normpath(base)]) == os.path.normpath(base)
+        return os.path.commonpath(
+            [os.path.normpath(path), os.path.normpath(base)]
+        ) == os.path.normpath(base)
     except ValueError:
         return False
 
@@ -172,9 +191,14 @@ def _is_path_under_base(path: str, base: str) -> bool:
 def _normalize_extra_mount_container_path(container_path: str) -> str:
     normalized = posixpath.normpath(container_path)
     if not normalized.startswith("/"):
-        raise HTTPException(status_code=400, detail=f"Extra mount path must be absolute: {container_path}")
+        raise HTTPException(
+            status_code=400,
+            detail=f"Extra mount path must be absolute: {container_path}",
+        )
     if normalized not in ALLOWED_EXTRA_MOUNT_PATHS:
-        raise HTTPException(status_code=400, detail=f"Unsupported extra mount path: {container_path}")
+        raise HTTPException(
+            status_code=400, detail=f"Unsupported extra mount path: {container_path}"
+        )
     return normalized
 
 
@@ -188,7 +212,9 @@ def _validated_extra_mounts(
     if not extra_mounts:
         return []
     if len(extra_mounts) > MAX_EXTRA_MOUNTS:
-        raise HTTPException(status_code=400, detail=f"Too many extra mounts; max is {MAX_EXTRA_MOUNTS}")
+        raise HTTPException(
+            status_code=400, detail=f"Too many extra mounts; max is {MAX_EXTRA_MOUNTS}"
+        )
 
     host_base_dir = _host_base_dir_for_extra_mounts()
     seen_container_paths: set[str] = set()
@@ -196,16 +222,26 @@ def _validated_extra_mounts(
     for mount in extra_mounts:
         host_path = os.path.normpath(mount.host_path)
         if not os.path.isabs(host_path):
-            raise HTTPException(status_code=400, detail=f"Extra mount host path must be absolute: {mount.host_path}")
+            raise HTTPException(
+                status_code=400,
+                detail=f"Extra mount host path must be absolute: {mount.host_path}",
+            )
         if not _is_path_under_base(host_path, host_base_dir):
-            raise HTTPException(status_code=400, detail=f"Extra mount host path is outside DeerFlow state: {mount.host_path}")
+            raise HTTPException(
+                status_code=400,
+                detail=f"Extra mount host path is outside DeerFlow state: {mount.host_path}",
+            )
 
         container_path = _normalize_extra_mount_container_path(mount.container_path)
         if container_path == UPLOAD_CONVERSIONS_CONTAINER_PATH:
             if not mount.read_only:
-                raise HTTPException(status_code=400, detail="Upload conversion mount must be read-only")
+                raise HTTPException(
+                    status_code=400, detail="Upload conversion mount must be read-only"
+                )
             if thread_id is None:
-                raise HTTPException(status_code=400, detail="Upload conversion mount requires a thread")
+                raise HTTPException(
+                    status_code=400, detail="Upload conversion mount requires a thread"
+                )
             expected_host_path = os.path.normpath(
                 join_host_path(
                     host_base_dir,
@@ -223,7 +259,9 @@ def _validated_extra_mounts(
                     detail="Upload conversion mount must match the requested user and thread",
                 )
         if container_path in seen_container_paths:
-            raise HTTPException(status_code=400, detail=f"Duplicate extra mount path: {container_path}")
+            raise HTTPException(
+                status_code=400, detail=f"Duplicate extra mount path: {container_path}"
+            )
         seen_container_paths.add(container_path)
 
         validated.append(
@@ -278,10 +316,16 @@ def _runtime_provided_extra_mounts(
         dropped = {LARK_CLI_RUNTIME_CONTAINER_PATH}
     if not extra_mounts or not dropped:
         return list(extra_mounts or [])
-    return [mount for mount in extra_mounts if posixpath.normpath(mount.container_path) not in dropped]
+    return [
+        mount
+        for mount in extra_mounts
+        if posixpath.normpath(mount.container_path) not in dropped
+    ]
 
 
-def _lark_broker_credential_mounts(extra_mounts: list["ExtraMount"] | None) -> dict[str, "ExtraMount"]:
+def _lark_broker_credential_mounts(
+    extra_mounts: list["ExtraMount"] | None,
+) -> dict[str, "ExtraMount"]:
     """Extract the config/data credential mounts the broker sidecar needs.
 
     Keyed by container path so the caller can wire each into the sidecar's fixed
@@ -304,12 +348,21 @@ def _lark_broker_credential_mounts(extra_mounts: list["ExtraMount"] | None) -> d
 def _extra_mount_pvc_sub_path(host_path: str) -> str:
     host_base_dir = _host_base_dir_for_extra_mounts()
     if not _is_path_under_base(host_path, host_base_dir):
-        raise HTTPException(status_code=400, detail=f"Extra mount host path is outside DeerFlow state: {host_path}")
+        raise HTTPException(
+            status_code=400,
+            detail=f"Extra mount host path is outside DeerFlow state: {host_path}",
+        )
 
     rel_path = os.path.relpath(os.path.normpath(host_path), host_base_dir)
-    rel_parts = [part for part in rel_path.replace(os.sep, "/").split("/") if part and part != "."]
+    rel_parts = [
+        part
+        for part in rel_path.replace(os.sep, "/").split("/")
+        if part and part != "."
+    ]
     if not rel_parts or any(part == ".." for part in rel_parts):
-        raise HTTPException(status_code=400, detail=f"Invalid extra mount host path: {host_path}")
+        raise HTTPException(
+            status_code=400, detail=f"Invalid extra mount host path: {host_path}"
+        )
     return posixpath.join("deer-flow", *rel_parts)
 
 
@@ -326,18 +379,26 @@ def _init_k8s_client() -> k8s_client.CoreV1Api:
     """
     if os.path.exists(KUBECONFIG_PATH):
         if os.path.isdir(KUBECONFIG_PATH):
-            raise RuntimeError(f"KUBECONFIG_PATH points to a directory, expected a file: {KUBECONFIG_PATH}")
+            raise RuntimeError(
+                f"KUBECONFIG_PATH points to a directory, expected a file: {KUBECONFIG_PATH}"
+            )
         try:
             k8s_config.load_kube_config(config_file=KUBECONFIG_PATH)
             logger.info(f"Loaded kubeconfig from {KUBECONFIG_PATH}")
         except Exception as exc:
-            raise RuntimeError(f"Failed to load kubeconfig from {KUBECONFIG_PATH}: {exc}") from exc
+            raise RuntimeError(
+                f"Failed to load kubeconfig from {KUBECONFIG_PATH}: {exc}"
+            ) from exc
     else:
-        logger.warning(f"Kubeconfig not found at {KUBECONFIG_PATH}; trying in-cluster config")
+        logger.warning(
+            f"Kubeconfig not found at {KUBECONFIG_PATH}; trying in-cluster config"
+        )
         try:
             k8s_config.load_incluster_config()
         except Exception as exc:
-            raise RuntimeError(f"Failed to initialize Kubernetes client. No kubeconfig at {KUBECONFIG_PATH}, and in-cluster config is unavailable: {exc}") from exc
+            raise RuntimeError(
+                f"Failed to initialize Kubernetes client. No kubeconfig at {KUBECONFIG_PATH}, and in-cluster config is unavailable: {exc}"
+            ) from exc
 
     # When connecting from inside Docker to the host's K8s API, the
     # kubeconfig may reference ``localhost`` or ``127.0.0.1``.  We
@@ -363,11 +424,17 @@ def _wait_for_kubeconfig(timeout: int = 30) -> None:
                 logger.info(f"Found kubeconfig file at {KUBECONFIG_PATH}")
                 return
             if os.path.isdir(KUBECONFIG_PATH):
-                raise RuntimeError(f"Kubeconfig path is a directory. Please mount a kubeconfig file at {KUBECONFIG_PATH}.")
-            raise RuntimeError(f"Kubeconfig path exists but is not a regular file: {KUBECONFIG_PATH}")
+                raise RuntimeError(
+                    f"Kubeconfig path is a directory. Please mount a kubeconfig file at {KUBECONFIG_PATH}."
+                )
+            raise RuntimeError(
+                f"Kubeconfig path exists but is not a regular file: {KUBECONFIG_PATH}"
+            )
         logger.info(f"Waiting for kubeconfig at {KUBECONFIG_PATH} …")
         time.sleep(2)
-    logger.warning(f"Kubeconfig not found at {KUBECONFIG_PATH} after {timeout}s; will attempt in-cluster Kubernetes config")
+    logger.warning(
+        f"Kubeconfig not found at {KUBECONFIG_PATH} after {timeout}s; will attempt in-cluster Kubernetes config"
+    )
 
 
 def _ensure_namespace() -> None:
@@ -412,8 +479,12 @@ app = FastAPI(title="DeerFlow Sandbox Provisioner", lifespan=lifespan)
 async def verify_api_key(request: Request, call_next):
     if request.url.path.startswith("/api/"):
         key = request.headers.get("X-API-Key", "")
-        if not PROVISIONER_API_KEY or not secrets.compare_digest(key, PROVISIONER_API_KEY):
-            logger.warning("provisioner auth rejected: %s %s", request.method, request.url.path)
+        if not PROVISIONER_API_KEY or not secrets.compare_digest(
+            key, PROVISIONER_API_KEY
+        ):
+            logger.warning(
+                "provisioner auth rejected: %s %s", request.method, request.url.path
+            )
             return Response(status_code=401, content="Unauthorized")
     return await call_next(request)
 
@@ -449,6 +520,9 @@ class SandboxResponse(BaseModel):
     sandbox_id: str
     sandbox_url: str
     status: str
+    user_id: str | None = None
+    thread_id: str | None = None
+    mount_contract_version: int | None = None
 
 
 # ── K8s resource helpers ─────────────────────────────────────────────────
@@ -471,6 +545,131 @@ def _sandbox_url(sandbox_id: str, node_port: int | None = None) -> str:
     return f"http://{NODE_HOST}:{node_port}"
 
 
+def _identity_hash(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
+
+
+def _mount_signature(pod: k8s_client.V1Pod) -> str:
+    """Hash the concrete volumes and container mounts that define isolation."""
+    contract_container_names = {
+        "sandbox",
+        "lark-cli-shim-init",
+        "lark-cli-init",
+        "lark-cli-broker",
+    }
+    contract_volume_names: set[str] = set()
+    container_payload = []
+    for container in [*(pod.spec.containers or []), *(pod.spec.init_containers or [])]:
+        if container.name not in contract_container_names:
+            continue
+        mounts = sorted(
+            ((mount.name or ""), mount.to_dict())
+            for mount in (container.volume_mounts or [])
+            if mount.mount_path == "/mnt"
+            or (mount.mount_path or "").startswith("/mnt/")
+            or (mount.mount_path or "").startswith("/var/lark/")
+        )
+        contract_volume_names.update(name for name, _payload in mounts)
+        container_payload.append((container.name or "", mounts))
+    volume_payload = sorted(
+        ((volume.name or ""), volume.to_dict())
+        for volume in (pod.spec.volumes or [])
+        if volume.name in contract_volume_names
+    )
+    payload = json.dumps(
+        {"volumes": volume_payload, "containers": sorted(container_payload)},
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _set_mount_contract_metadata(
+    pod: k8s_client.V1Pod,
+    *,
+    user_id: str,
+    thread_id: str,
+) -> k8s_client.V1Pod:
+    labels = dict(pod.metadata.labels or {})
+    labels.update(
+        {
+            MOUNT_CONTRACT_LABEL: str(MOUNT_CONTRACT_VERSION),
+            USER_ID_HASH_LABEL: _identity_hash(user_id),
+            THREAD_ID_HASH_LABEL: _identity_hash(thread_id),
+        }
+    )
+    pod.metadata.labels = labels
+    annotations = dict(pod.metadata.annotations or {})
+    annotations.update(
+        {
+            USER_ID_ANNOTATION: user_id,
+            THREAD_ID_ANNOTATION: thread_id,
+            MOUNT_SIGNATURE_ANNOTATION: _mount_signature(pod),
+        }
+    )
+    pod.metadata.annotations = annotations
+    return pod
+
+
+def _validate_existing_sandbox_contract(
+    sandbox_id: str,
+    *,
+    expected_pod: k8s_client.V1Pod | None = None,
+    user_id: str | None = None,
+    thread_id: str | None = None,
+) -> tuple[str, str]:
+    """Fail closed unless an existing Pod proves its tenant and mount contract."""
+    try:
+        pod = core_v1.read_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE)
+    except ApiException as exc:
+        raise HTTPException(
+            status_code=409,
+            detail=f"Existing sandbox '{sandbox_id}' has no verifiable Pod contract",
+        ) from exc
+
+    labels = pod.metadata.labels or {}
+    annotations = pod.metadata.annotations or {}
+    actual_user_id = annotations.get(USER_ID_ANNOTATION)
+    actual_thread_id = annotations.get(THREAD_ID_ANNOTATION)
+    expected_signature = annotations.get(MOUNT_SIGNATURE_ANNOTATION)
+    if (
+        labels.get(MOUNT_CONTRACT_LABEL) != str(MOUNT_CONTRACT_VERSION)
+        or not actual_user_id
+        or not actual_thread_id
+        or labels.get(USER_ID_HASH_LABEL) != _identity_hash(actual_user_id)
+        or labels.get(THREAD_ID_HASH_LABEL) != _identity_hash(actual_thread_id)
+        or not expected_signature
+        or _mount_signature(pod) != expected_signature
+    ):
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Existing sandbox '{sandbox_id}' uses an incompatible or unverifiable mount contract; "
+                "destroy it before retrying"
+            ),
+        )
+
+    if user_id is not None and actual_user_id != user_id:
+        raise HTTPException(
+            status_code=409,
+            detail=f"Existing sandbox '{sandbox_id}' belongs to another user",
+        )
+    if thread_id is not None and actual_thread_id != thread_id:
+        raise HTTPException(
+            status_code=409,
+            detail=f"Existing sandbox '{sandbox_id}' belongs to another thread",
+        )
+    if (
+        expected_pod is not None
+        and _mount_signature(expected_pod) != expected_signature
+    ):
+        raise HTTPException(
+            status_code=409,
+            detail=f"Existing sandbox '{sandbox_id}' does not match the requested mount contract",
+        )
+    return actual_user_id, actual_thread_id
+
+
 def _build_extra_volumes(
     extra_mounts: list[ExtraMount] | None = None,
     *,
@@ -549,7 +748,9 @@ def _build_volumes(
     if SKILLS_PVC_NAME:
         # PVC mode: three-way subPath not yet supported; fall back to
         # single-volume mount for backward compatibility.
-        logger.warning("SKILLS_PVC_NAME is set — three-way skills layout is not supported in PVC mode yet; falling back to single /mnt/skills mount")
+        logger.warning(
+            "SKILLS_PVC_NAME is set — three-way skills layout is not supported in PVC mode yet; falling back to single /mnt/skills mount"
+        )
         volumes.append(
             k8s_client.V1Volume(
                 name="skills",
@@ -615,7 +816,14 @@ def _build_volumes(
         userdata_vol = k8s_client.V1Volume(
             name="user-data",
             host_path=k8s_client.V1HostPathVolumeSource(
-                path=join_host_path(THREADS_HOST_PATH, thread_id, "user-data"),
+                path=join_host_path(
+                    _host_base_dir_for_extra_mounts(),
+                    "users",
+                    user_id,
+                    "threads",
+                    thread_id,
+                    "user-data",
+                ),
                 type="DirectoryOrCreate",
             ),
         )
@@ -634,7 +842,9 @@ def _build_volumes(
     )
     # The runtime emptyDir is shared by the init container (writer) and the
     # sandbox container (reader) in both Pattern A and Pattern B (shim).
-    if _lark_cli_runtime_enabled(provision_lark_cli_runtime) or _lark_cli_broker_enabled(provision_lark_cli_broker):
+    if _lark_cli_runtime_enabled(
+        provision_lark_cli_runtime
+    ) or _lark_cli_broker_enabled(provision_lark_cli_broker):
         volumes.append(
             k8s_client.V1Volume(
                 name=LARK_CLI_RUNTIME_VOLUME_NAME,
@@ -666,7 +876,9 @@ def _build_volumes(
                         name=volume_name,
                         host_path=k8s_client.V1HostPathVolumeSource(
                             path=mount.host_path,
-                            type="Directory" if mount.read_only else "DirectoryOrCreate",
+                            type="Directory"
+                            if mount.read_only
+                            else "DirectoryOrCreate",
                         ),
                     )
                 )
@@ -731,7 +943,9 @@ def _build_volume_mounts(
         read_only=False,
     )
     if USERDATA_PVC_NAME:
-        userdata_mount.sub_path = f"deer-flow/users/{user_id}/threads/{thread_id}/user-data"
+        userdata_mount.sub_path = (
+            f"deer-flow/users/{user_id}/threads/{thread_id}/user-data"
+        )
     mounts.append(userdata_mount)
     mounts.extend(
         _build_extra_volume_mounts(
@@ -745,7 +959,9 @@ def _build_volume_mounts(
         )
     )
     # Sandbox reads the runtime dir (real binary in Pattern A, shim in Pattern B).
-    if _lark_cli_runtime_enabled(provision_lark_cli_runtime) or _lark_cli_broker_enabled(provision_lark_cli_broker):
+    if _lark_cli_runtime_enabled(
+        provision_lark_cli_runtime
+    ) or _lark_cli_broker_enabled(provision_lark_cli_broker):
         mounts.append(
             k8s_client.V1VolumeMount(
                 name=LARK_CLI_RUNTIME_VOLUME_NAME,
@@ -772,7 +988,9 @@ def _build_lark_cli_init_containers(
         mount_path=LARK_CLI_RUNTIME_CONTAINER_PATH,
         read_only=False,
     )
-    secure = k8s_client.V1SecurityContext(privileged=False, allow_privilege_escalation=False)
+    secure = k8s_client.V1SecurityContext(
+        privileged=False, allow_privilege_escalation=False
+    )
     if _lark_cli_broker_enabled(provision_lark_cli_broker):
         return [
             k8s_client.V1Container(
@@ -780,7 +998,12 @@ def _build_lark_cli_init_containers(
                 image=LARK_CLI_BROKER_IMAGE,
                 image_pull_policy="IfNotPresent",
                 args=["install-shim", LARK_CLI_RUNTIME_CONTAINER_PATH],
-                env=[k8s_client.V1EnvVar(name="LARK_CLI_RUNTIME_DEST", value=LARK_CLI_RUNTIME_CONTAINER_PATH)],
+                env=[
+                    k8s_client.V1EnvVar(
+                        name="LARK_CLI_RUNTIME_DEST",
+                        value=LARK_CLI_RUNTIME_CONTAINER_PATH,
+                    )
+                ],
                 volume_mounts=[runtime_mount],
                 security_context=secure,
             )
@@ -819,8 +1042,16 @@ def _build_lark_cli_broker_sidecars(
     credential_mounts = _lark_broker_credential_mounts(extra_mounts)
     volume_mounts: list[k8s_client.V1VolumeMount] = []
     for container_path, volume_name, sidecar_path in (
-        (LARK_CLI_CONFIG_CONTAINER_PATH, LARK_BROKER_CONFIG_VOLUME_NAME, LARK_BROKER_SIDECAR_CONFIG_PATH),
-        (LARK_CLI_DATA_CONTAINER_PATH, LARK_BROKER_DATA_VOLUME_NAME, LARK_BROKER_SIDECAR_DATA_PATH),
+        (
+            LARK_CLI_CONFIG_CONTAINER_PATH,
+            LARK_BROKER_CONFIG_VOLUME_NAME,
+            LARK_BROKER_SIDECAR_CONFIG_PATH,
+        ),
+        (
+            LARK_CLI_DATA_CONTAINER_PATH,
+            LARK_BROKER_DATA_VOLUME_NAME,
+            LARK_BROKER_SIDECAR_DATA_PATH,
+        ),
     ):
         mount = credential_mounts.get(container_path)
         if mount is None:
@@ -834,8 +1065,12 @@ def _build_lark_cli_broker_sidecars(
             sidecar_mount.sub_path = _extra_mount_pvc_sub_path(mount.host_path)
         volume_mounts.append(sidecar_mount)
     broker_env = [
-        k8s_client.V1EnvVar(name="LARKSUITE_CLI_CONFIG_DIR", value=LARK_BROKER_SIDECAR_CONFIG_PATH),
-        k8s_client.V1EnvVar(name="LARKSUITE_CLI_DATA_DIR", value=LARK_BROKER_SIDECAR_DATA_PATH),
+        k8s_client.V1EnvVar(
+            name="LARKSUITE_CLI_CONFIG_DIR", value=LARK_BROKER_SIDECAR_CONFIG_PATH
+        ),
+        k8s_client.V1EnvVar(
+            name="LARKSUITE_CLI_DATA_DIR", value=LARK_BROKER_SIDECAR_DATA_PATH
+        ),
     ]
     # Forward the optional subcommand denylist so the broker refuses secret-dump
     # subcommands (issue #4338 hardening); omitted when unset ⇒ nothing blocked.
@@ -874,9 +1109,12 @@ def _build_pod(
 ) -> k8s_client.V1Pod:
     """Construct a Pod manifest for a single sandbox."""
     init_containers = (
-        _build_lark_cli_init_containers(provision_lark_cli_runtime, provision_lark_cli_broker) or None
+        _build_lark_cli_init_containers(
+            provision_lark_cli_runtime, provision_lark_cli_broker
+        )
+        or None
     )
-    return k8s_client.V1Pod(
+    pod = k8s_client.V1Pod(
         metadata=k8s_client.V1ObjectMeta(
             name=_pod_name(sandbox_id),
             namespace=K8S_NAMESPACE,
@@ -894,7 +1132,11 @@ def _build_pod(
                     image=SANDBOX_IMAGE,
                     image_pull_policy="IfNotPresent",
                     env=(
-                        [k8s_client.V1EnvVar(name="DEERFLOW_LARK_BROKER_URL", value=LARK_BROKER_URL)]
+                        [
+                            k8s_client.V1EnvVar(
+                                name="DEERFLOW_LARK_BROKER_URL", value=LARK_BROKER_URL
+                            )
+                        ]
                         if _lark_cli_broker_enabled(provision_lark_cli_broker)
                         else None
                     ),
@@ -950,7 +1192,9 @@ def _build_pod(
                         allow_privilege_escalation=True,
                     ),
                 ),
-                *_build_lark_cli_broker_sidecars(provision_lark_cli_broker, extra_mounts),
+                *_build_lark_cli_broker_sidecars(
+                    provision_lark_cli_broker, extra_mounts
+                ),
             ],
             init_containers=init_containers,
             volumes=_build_volumes(
@@ -964,9 +1208,15 @@ def _build_pod(
             restart_policy="Always",
         ),
     )
+    return _set_mount_contract_metadata(pod, user_id=user_id, thread_id=thread_id)
 
 
-def _build_service(sandbox_id: str) -> k8s_client.V1Service:
+def _build_service(
+    sandbox_id: str,
+    *,
+    user_id: str = DEFAULT_USER_ID,
+    thread_id: str | None = None,
+) -> k8s_client.V1Service:
     """Construct a Service manifest for the configured access mode."""
     return k8s_client.V1Service(
         metadata=k8s_client.V1ObjectMeta(
@@ -977,6 +1227,9 @@ def _build_service(sandbox_id: str) -> k8s_client.V1Service:
                 "sandbox-id": sandbox_id,
                 "app.kubernetes.io/name": "deer-flow",
                 "app.kubernetes.io/component": "sandbox",
+                MOUNT_CONTRACT_LABEL: str(MOUNT_CONTRACT_VERSION),
+                USER_ID_HASH_LABEL: _identity_hash(user_id),
+                THREAD_ID_HASH_LABEL: _identity_hash(thread_id or sandbox_id),
             },
         ),
         spec=k8s_client.V1ServiceSpec(
@@ -1007,7 +1260,9 @@ def _url_from_service(svc, sandbox_id: str) -> str | None:
     return None
 
 
-def _sandbox_access_url(sandbox_id: str, *, tolerate_read_errors: bool = False) -> str | None:
+def _sandbox_access_url(
+    sandbox_id: str, *, tolerate_read_errors: bool = False
+) -> str | None:
     """Read the sandbox Service and return its backend-facing URL when ready."""
     try:
         svc = core_v1.read_namespaced_service(_svc_name(sandbox_id), K8S_NAMESPACE)
@@ -1057,6 +1312,7 @@ async def capabilities():
     return {
         "lark_cli_init_image": bool(LARK_CLI_INIT_IMAGE),
         "lark_cli_broker_image": bool(LARK_CLI_BROKER_IMAGE),
+        "mount_contract_version": MOUNT_CONTRACT_VERSION,
     }
 
 
@@ -1073,6 +1329,17 @@ def create_sandbox(req: CreateSandboxRequest):
     include_legacy_skills = req.include_legacy_skills
     provision_lark_cli_runtime = req.provision_lark_cli_runtime
     provision_lark_cli_broker = req.provision_lark_cli_broker
+    # Build and validate the requested contract before the idempotent fast path;
+    # an existing Service must never bypass tenant or read-only mount checks.
+    expected_pod = _build_pod(
+        sandbox_id,
+        thread_id,
+        user_id=user_id,
+        include_legacy_skills=include_legacy_skills,
+        extra_mounts=req.extra_mounts,
+        provision_lark_cli_runtime=provision_lark_cli_runtime,
+        provision_lark_cli_broker=provision_lark_cli_broker,
+    )
 
     logger.info(
         "Received request to create sandbox '%s' for thread '%s' user '%s' include_legacy_skills=%s provision_lark_cli_runtime=%s provision_lark_cli_broker=%s",
@@ -1087,34 +1354,46 @@ def create_sandbox(req: CreateSandboxRequest):
     # ── Fast path: sandbox already exists ────────────────────────────
     existing_url = _sandbox_access_url(sandbox_id, tolerate_read_errors=True)
     if existing_url:
+        _validate_existing_sandbox_contract(
+            sandbox_id,
+            expected_pod=expected_pod,
+            user_id=user_id,
+            thread_id=thread_id,
+        )
         return SandboxResponse(
             sandbox_id=sandbox_id,
             sandbox_url=existing_url,
             status=_get_pod_phase(sandbox_id),
+            user_id=user_id,
+            thread_id=thread_id,
+            mount_contract_version=MOUNT_CONTRACT_VERSION,
         )
 
     # ── Create Pod ───────────────────────────────────────────────────
     try:
         core_v1.create_namespaced_pod(
             K8S_NAMESPACE,
-            _build_pod(
-                sandbox_id,
-                thread_id,
-                user_id=user_id,
-                include_legacy_skills=include_legacy_skills,
-                extra_mounts=req.extra_mounts,
-                provision_lark_cli_runtime=provision_lark_cli_runtime,
-                provision_lark_cli_broker=provision_lark_cli_broker,
-            ),
+            expected_pod,
         )
         logger.info(f"Created Pod {_pod_name(sandbox_id)}")
     except ApiException as exc:
         if exc.status != 409:  # 409 = AlreadyExists
-            raise HTTPException(status_code=500, detail=f"Pod creation failed: {exc.reason}")
+            raise HTTPException(
+                status_code=500, detail=f"Pod creation failed: {exc.reason}"
+            )
+        _validate_existing_sandbox_contract(
+            sandbox_id,
+            expected_pod=expected_pod,
+            user_id=user_id,
+            thread_id=thread_id,
+        )
 
     # ── Create Service ───────────────────────────────────────────────
     try:
-        core_v1.create_namespaced_service(K8S_NAMESPACE, _build_service(sandbox_id))
+        core_v1.create_namespaced_service(
+            K8S_NAMESPACE,
+            _build_service(sandbox_id, user_id=user_id, thread_id=thread_id),
+        )
         logger.info(f"Created Service {_svc_name(sandbox_id)}")
     except ApiException as exc:
         if exc.status != 409:
@@ -1123,7 +1402,9 @@ def create_sandbox(req: CreateSandboxRequest):
                 core_v1.delete_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE)
             except ApiException:
                 pass
-            raise HTTPException(status_code=500, detail=f"Service creation failed: {exc.reason}")
+            raise HTTPException(
+                status_code=500, detail=f"Service creation failed: {exc.reason}"
+            )
 
     # ── Wait until the Service has a usable access URL ───────────────
     sandbox_url: str | None = None
@@ -1134,12 +1415,17 @@ def create_sandbox(req: CreateSandboxRequest):
         time.sleep(0.5)
 
     if not sandbox_url:
-        raise HTTPException(status_code=500, detail="Service access URL was not available in time")
+        raise HTTPException(
+            status_code=500, detail="Service access URL was not available in time"
+        )
 
     return SandboxResponse(
         sandbox_id=sandbox_id,
         sandbox_url=sandbox_url,
         status=_get_pod_phase(sandbox_id),
+        user_id=user_id,
+        thread_id=thread_id,
+        mount_contract_version=MOUNT_CONTRACT_VERSION,
     )
 
 
@@ -1165,7 +1451,9 @@ def destroy_sandbox(sandbox_id: str):
             errors.append(f"pod: {exc.reason}")
 
     if errors:
-        raise HTTPException(status_code=500, detail=f"Partial cleanup: {', '.join(errors)}")
+        raise HTTPException(
+            status_code=500, detail=f"Partial cleanup: {', '.join(errors)}"
+        )
 
     return {"ok": True, "sandbox_id": sandbox_id}
 
@@ -1176,11 +1464,15 @@ def get_sandbox(sandbox_id: str):
     sandbox_url = _sandbox_access_url(sandbox_id)
     if not sandbox_url:
         raise HTTPException(status_code=404, detail=f"Sandbox '{sandbox_id}' not found")
+    user_id, thread_id = _validate_existing_sandbox_contract(sandbox_id)
 
     return SandboxResponse(
         sandbox_id=sandbox_id,
         sandbox_url=sandbox_url,
         status=_get_pod_phase(sandbox_id),
+        user_id=user_id,
+        thread_id=thread_id,
+        mount_contract_version=MOUNT_CONTRACT_VERSION,
     )
 
 
@@ -1193,7 +1485,9 @@ def list_sandboxes():
             label_selector="app=deer-flow-sandbox",
         )
     except ApiException as exc:
-        raise HTTPException(status_code=500, detail=f"Failed to list services: {exc.reason}")
+        raise HTTPException(
+            status_code=500, detail=f"Failed to list services: {exc.reason}"
+        )
 
     sandboxes: list[SandboxResponse] = []
     for svc in services.items: