fix: bind remote upload deletion generations

backend/AGENTS.md

@@ -1464,7 +1464,7 @@ 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. Deletion rejects an inode that moved outside the requested name's lease. Primary deletion and rollback move the selected primary and its exact generated conversion into one inode-and-intent-tagged transaction directory under the protected conversion namespace before side effects; the primary lives in a dedicated child directory so legal filenames such as `.commit`, `.restore`, and `.conversion` cannot collide with transaction controls. User deletion starts with restore-on-crash intent. Immediately before the first possible remote mutation, a persisted commit marker makes crash recovery discard the authoritative host tombstones; a live, fully compensated remote failure clears it while restoring both host entries. A live rollback writes a restore marker before publishing either hard link, so recovery can finish restoring both entries even when the transaction began with discard intent. A partial remote failure is compensated from the staged authoritative bytes before host restoration; if compensation fails, every remote target is driven back toward the idempotent deletion outcome and host deletion commits instead of publishing a split generation. Failed-upload rollback tombstones otherwise use discard recovery, while legacy primary-only transactions retain their recorded restore/discard behavior. Neither transaction kind is treated as an ordinary disposable upload stage. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result. Embedded-client finalization attempts every lease release and conversion-pool shutdown independently, logging cleanup failures rather than changing an already-committed response or stranding later leases.
 - Filenames containing NUL, `<`, `>`, or reserved model-context boundary markers are rejected before staging so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed.
 - Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files, while `.upload-delete-<restore|discard>-<inode>-<nonce>.part/` transaction directories are resolved according to their persisted crash intent. Legacy intent-less deletion transactions retain restore behavior. Cancellation during staging creation drains the worker and aborts the returned stage before propagating. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools.
-- Generated Markdown is owned by `user-data/.upload-conversions/<actual-primary-filename>.md` and is omitted from primary upload listings. Deletion holds the generation lease while it stages both authoritative host paths, writes a durable remote-deletion journal, persists the remote-mutation phase, removes an explicitly synchronized sandbox primary/conversion, and commits the host transaction. A remote failure is reported and restores the host generation only after every possibly removed remote path has been re-published; failed compensation instead retries idempotent removal of every remote target and commits host deletion. Unconfirmed paths keep the journal and commit marker after the host tombstones are removed; Gateway startup and later embedded deletion preparation reacquire the recorded thread sandbox and retry before ordinary transaction cleanup. Live journals remain protected by their stage lease. 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.
+- 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 stages both authoritative host paths, writes a durable remote-deletion journal, persists the remote-mutation phase, removes an explicitly synchronized sandbox primary/conversion, and commits the host transaction. A remote failure is reported and restores the host generation only after every possibly removed remote path has been re-published; failed compensation instead retries idempotent removal of every remote target and commits host deletion. Unconfirmed paths keep the journal, commit marker, and exact hidden host tombstones; that retained primary persistently reserves its portable filename coordination key, so a new upload advances to `_N` rather than letting an old path-based retry delete the new remote generation. Gateway startup and later embedded deletion preparation retry only against the journal's exact recorded sandbox ID—never a replacement sandbox acquired for the thread—before ordinary transaction cleanup removes the tombstones. Live journals remain protected by their stage lease, and reconciliation takes the same name lease used by publication. Exact legacy POSIX basenames use the existing-file deletion validator during retry. It never guesses or deletes a legacy/user-owned `uploads/<stem>.md` sibling. Outline extraction opens one descriptor, verifies its `fstat` against the current exclusive regular directory entry, and uses that same descriptor for both outline and preview reads.
 - Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; only operations that may block waiting for a name lease use the separate lease-wait pool. Work needed by an existing lease holder and non-blocking publication stays on the general pool, so waiters cannot starve conversion, rollback, or release. Cold sandbox-provider construction is also offloaded. Gateway, embedded-client, and IM ingresses share provider-aware publication: mounted providers make the exact host paths sandbox-readable; non-mounted providers acquire the sandbox and synchronize the primary plus generated conversion to their exact virtual paths. Each ingress records attempted remote paths before the write can commit and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those paths before host rollback and lease release; the command fallback requires a per-call unpredictable exact success trailer. Embedded multi-file calls retain every publication and receipt until the whole response is built, then roll back the complete batch on failure. WeChat download publication uses the cancellation-safe async lease adapter, so cancellation drains and rolls back a publication worker that completes late.
 - Mounted upload paths skip both sandbox acquisition and per-file synchronization. For AIO remote/provisioner deployments this requires an explicit, accurate `sandbox.thread_data_mounts: true` plus a Provisioner advertising a compatible mount contract; every upload/delete/artifact decision waits for any in-flight stale capability probe in a worker thread and uses one atomic snapshot. The same frozen version selects the deterministic sandbox ID, mount set, cache/reuse candidate, create precondition, and response validation; if the peer changes versions during acquisition, the whole attempt restarts with a new ID. The required version is sent in the create JSON, and the Provisioner rejects a mismatch before any Kubernetes read or write, preventing a rolling upgrade from creating an orphan under the prior version's ID. After acquisition, consumers use the immutable contract recorded on that sandbox instance rather than the provider's mutable global capability. Unavailable or legacy results use explicit synchronization and are retried without a Gateway restart. A confirmed legacy peer may create thread sandboxes only for the `default` no-auth user, because its primary user-data mount cannot isolate equal thread IDs across authenticated users; authenticated rolling deployments upgrade the Provisioner first. The Provisioner validates the request before its idempotent fast path, labels Pods with hashed user/thread identity and contract version, stores the exact identity plus a mount-signature annotation, and compares the live Pod specification before reuse. Every remote acquisition path, including active-cache and warm-pool reuse, deliberately replays the idempotent create request instead of trusting GET discovery or a liveness response. Contracts newer than v2 are accepted as forward-compatible, but the response must echo the exact frozen version, user, and thread; legacy responses may omit the version or return exact integer zero, never strings or booleans. The negotiated mount-contract version namespaces deterministic AIO sandbox IDs so a container created under another version cannot satisfy the acquisition; active reuse also replaces its client when validation returns a changed service URL. Reconciliation may still enumerate/adopt the old ID for orphan cleanup; it is not selected for the new identity.
 - Agent receives uploaded file list via `UploadsMiddleware`

backend/docs/FILE_UPLOAD.md

@@ -110,7 +110,7 @@ DELETE /api/threads/{thread_id}/uploads/{filename}
 }
 ```
 
-删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先暂存宿主机主文件与精确转换副本、持久化远端删除日志，再删除沙箱中的对应副本。远端失败且所有可能的副作用都成功补偿时，请求会返回错误并恢复宿主机文件；如果补偿无法完整确认，系统会重试将全部远端目标收敛到删除状态并提交宿主机删除，避免重新暴露一个远端已经缺失的宿主机 generation，此时请求仍返回错误。持续未确认的远端路径会保留在受保护的日志中，并在 Gateway 重启或后续嵌入式删除准备时重新获取对应线程沙箱继续清理。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除；新上传仍执行严格的跨平台文件名校验。
+删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先暂存宿主机主文件与精确转换副本、持久化远端删除日志，再删除沙箱中的对应副本。远端失败且所有可能的副作用都成功补偿时，请求会返回错误并恢复宿主机文件；如果补偿无法完整确认，系统会重试将全部远端目标收敛到删除状态并提交宿主机删除，避免重新暴露一个远端已经缺失的宿主机 generation，此时请求仍返回错误。持续未确认时，受保护的日志、提交标记和精确宿主机 tombstone 会一起保留；隐藏的主文件 tombstone 持续占用该文件名的可移植协调键，因此新的同名上传会得到 `_N` 名称，不会被旧日志的按路径重试误删。Gateway 重启或后续嵌入式删除准备只会重新连接日志中记录的精确 sandbox ID，不会把清理重定向到该线程后来获取的新 sandbox；远端确认删除后才清理 tombstone。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除，远端日志重试也沿用同一精确旧名称；新上传仍执行严格的跨平台文件名校验。
 
 ## 支持的文档格式
 

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

@@ -297,6 +297,60 @@ def _filename_candidates(name: str) -> Iterator[str]:
         counter += 1
 
 
+def _pending_remote_deletion_reserves_name(base_dir: Path, coordination_key: str) -> bool:
+    """Return whether a durable remote deletion still owns this filename.
+
+    A committed journal deletes fixed sandbox paths.  Until that journal is
+    reconciled, publishing another generation under the same portable name
+    would let the old operation delete the new generation's remote copy.
+    The retained primary tombstone is therefore a persistent name reservation.
+    Callers hold the candidate's name lease while scanning, so reconciliation
+    cannot remove the journal between this check and publication.
+    """
+    conversion_dir = base_dir.parent / UPLOAD_CONVERSIONS_DIRNAME
+    try:
+        conversion_dir_stat = os.lstat(conversion_dir)
+    except FileNotFoundError:
+        return False
+    if stat.S_ISLNK(conversion_dir_stat.st_mode) or not stat.S_ISDIR(conversion_dir_stat.st_mode):
+        raise UnsafeUploadPathError("Unsafe upload conversion directory")
+
+    with os.scandir(conversion_dir) as transactions:
+        for transaction in transactions:
+            if not (transaction.name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and transaction.name.endswith(UPLOAD_STAGING_SUFFIX) and transaction.is_dir(follow_symlinks=False)):
+                continue
+            transaction_dir = Path(transaction.path)
+            journal_path = transaction_dir / _UPLOAD_DELETION_REMOTE_JOURNAL
+            commit_path = transaction_dir / _UPLOAD_DELETION_COMMIT_MARKER
+            primary_dir = transaction_dir / _UPLOAD_DELETION_PRIMARY_DIRNAME
+            try:
+                journal_stat = os.lstat(journal_path)
+                commit_stat = os.lstat(commit_path)
+                primary_dir_stat = os.lstat(primary_dir)
+            except FileNotFoundError:
+                continue
+            if (
+                not stat.S_ISREG(journal_stat.st_mode)
+                or journal_stat.st_nlink != 1
+                or not stat.S_ISREG(commit_stat.st_mode)
+                or commit_stat.st_nlink != 1
+                or stat.S_ISLNK(primary_dir_stat.st_mode)
+                or not stat.S_ISDIR(primary_dir_stat.st_mode)
+            ):
+                raise UnsafeUploadPathError("Unsafe pending remote upload deletion transaction")
+            with os.scandir(primary_dir) as primary_entries:
+                entries = list(primary_entries)
+            if len(entries) != 1:
+                raise UnsafeUploadPathError("Malformed pending remote upload deletion transaction")
+            entry = entries[0]
+            entry_stat = entry.stat(follow_symlinks=False)
+            if not stat.S_ISREG(entry_stat.st_mode) or entry_stat.st_nlink != 1:
+                raise UnsafeUploadPathError("Unsafe pending remote upload deletion tombstone")
+            if portable_name_coordination_key(entry.name) == coordination_key:
+                return True
+    return False
+
+
 def _unlink_matching_upload(path: Path, identity: UploadIdentity) -> None:
     """Remove *path* only while it still names *identity*."""
     if identity.matches(path):
@@ -348,6 +402,9 @@ def publish_staged_upload_leased(
             continue
         linked = False
         try:
+            if _pending_remote_deletion_reserves_name(staged.base_dir, coordination_key):
+                lease.release()
+                continue
             try:
                 os.link(staged.path, candidate, follow_symlinks=False)
                 linked = True
@@ -766,6 +823,14 @@ def _staged_deletion_remote_journal(staged_path: Path) -> Path:
     return _deletion_transaction_dir_for_staged_path(staged_path) / _UPLOAD_DELETION_REMOTE_JOURNAL
 
 
+def _staged_deletion_has_remote_journal(staged_path: Path) -> bool:
+    try:
+        os.lstat(_staged_deletion_remote_journal(staged_path))
+    except FileNotFoundError:
+        return False
+    return True
+
+
 def _create_deletion_phase_marker(marker: Path, *, error_message: str) -> None:
     """Create and validate one durable deletion phase marker."""
     try:
@@ -796,6 +861,11 @@ def _mark_staged_deletion_committed(staged_path: Path) -> None:
 
 def _mark_staged_deletion_restore(staged_path: Path) -> None:
     """Persist that a live rollback began and recovery must finish restoring."""
+    if _staged_deletion_restore_marker(staged_path) == staged_path:
+        # Legacy direct-layout transactions could stage a user primary named
+        # exactly like this control. Their restore intent is already encoded in
+        # the transaction name, and nlink=2 recovery is idempotent.
+        return
     _create_deletion_phase_marker(
         _staged_deletion_restore_marker(staged_path),
         error_message="Unsafe upload deletion restore marker",
@@ -803,10 +873,14 @@ def _mark_staged_deletion_restore(staged_path: Path) -> None:
 
 
 def _clear_staged_deletion_commit(staged_path: Path) -> None:
+    if _staged_deletion_commit_marker(staged_path) == staged_path:
+        return
     _unlink_deletion_control_durably(_staged_deletion_commit_marker(staged_path))
 
 
 def _clear_staged_deletion_restore(staged_path: Path) -> None:
+    if _staged_deletion_restore_marker(staged_path) == staged_path:
+        return
     _unlink_deletion_control_durably(_staged_deletion_restore_marker(staged_path))
 
 
@@ -825,6 +899,11 @@ def _unlink_deletion_control_durably(path: Path) -> None:
 def _restore_staged_conversion(staged_path: Path, restored_primary_path: Path) -> None:
     """Restore the exact conversion moved into a deletion transaction."""
     staged_conversion = _staged_conversion_path(staged_path)
+    if staged_conversion == staged_path:
+        # In the legacy direct layout, a primary named ``.conversion`` occupied
+        # the same path as the control tombstone. The expected-inode decoder has
+        # already established that this entry is the primary, not a companion.
+        return
     try:
         conversion_stat = os.lstat(staged_conversion)
     except FileNotFoundError:
@@ -859,11 +938,22 @@ def _discard_staged_deletion(staged_path: Path) -> None:
     _staged_conversion_path(staged_path).unlink(missing_ok=True)
     staged_path.unlink(missing_ok=True)
     _clear_staged_deletion_restore(staged_path)
-    if not _staged_deletion_remote_journal(staged_path).exists():
+    if not _staged_deletion_has_remote_journal(staged_path):
         _clear_staged_deletion_commit(staged_path)
     _finish_deletion_transaction(staged_path)
 
 
+def _finalize_committed_staged_deletion(staged_path: Path) -> None:
+    """Discard host tombstones only after durable remote cleanup converged.
+
+    A remaining journal must retain the exact primary tombstone so publication
+    can reserve its portable filename until reconciliation finishes.
+    """
+    if _staged_deletion_has_remote_journal(staged_path):
+        return
+    _discard_staged_deletion(staged_path)
+
+
 def _recovery_path_for(original_path: Path) -> Path:
     """Return a collision-resistant visible recovery name within NAME_MAX."""
     marker = f"_recovered_{secrets.token_hex(8)}"
@@ -899,7 +989,11 @@ def _finish_deletion_transaction(staged_path: Path) -> None:
     if not (transaction_dir.name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and transaction_dir.name.endswith(UPLOAD_STAGING_SUFFIX) and transaction_dir.parent.name == UPLOAD_CONVERSIONS_DIRNAME):
         return
     primary_dir = transaction_dir / _UPLOAD_DELETION_PRIMARY_DIRNAME
-    if (transaction_dir / _UPLOAD_DELETION_REMOTE_JOURNAL).exists():
+    try:
+        os.lstat(transaction_dir / _UPLOAD_DELETION_REMOTE_JOURNAL)
+    except FileNotFoundError:
+        pass
+    else:
         # Keep the empty primary container as the on-disk layout discriminator;
         # older direct-layout transactions may legitimately have a primary
         # named exactly like the new journal control.
@@ -944,22 +1038,33 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
     if not entries:
         transaction_dir.rmdir()
         return True
-    conversion_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_CONVERSION_TOMBSTONE]
-    commit_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_COMMIT_MARKER]
-    restore_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_RESTORE_MARKER]
     primary_dir_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_PRIMARY_DIRNAME and entry.is_dir(follow_symlinks=False)]
+    legacy_primary_entries: list[os.DirEntry[str]] = []
+    if not primary_dir_entries:
+        for entry in entries:
+            try:
+                entry_stat = entry.stat(follow_symlinks=False)
+            except FileNotFoundError:
+                continue
+            if stat.S_ISREG(entry_stat.st_mode) and entry_stat.st_ino == expected_inode:
+                legacy_primary_entries.append(entry)
+        if len(legacy_primary_entries) > 1:
+            logger.warning("Refusing legacy upload deletion transaction with multiple matching primaries: %s", transaction_dir)
+            return False
+    legacy_primary_paths = {entry.path for entry in legacy_primary_entries}
+    conversion_entries = [entry for entry in entries if entry.path not in legacy_primary_paths and entry.name == _UPLOAD_DELETION_CONVERSION_TOMBSTONE]
+    commit_entries = [entry for entry in entries if entry.path not in legacy_primary_paths and entry.name == _UPLOAD_DELETION_COMMIT_MARKER]
+    restore_entries = [entry for entry in entries if entry.path not in legacy_primary_paths and entry.name == _UPLOAD_DELETION_RESTORE_MARKER]
     remote_journal_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_REMOTE_JOURNAL and primary_dir_entries]
-    legacy_primary_entries = [
+    legacy_unknown_entries = [
         entry
         for entry in entries
-        if entry.name
-        not in {
-            _UPLOAD_DELETION_CONVERSION_TOMBSTONE,
-            _UPLOAD_DELETION_COMMIT_MARKER,
-            _UPLOAD_DELETION_RESTORE_MARKER,
-        }
+        if not primary_dir_entries
+        and entry.path not in legacy_primary_paths
+        and entry not in conversion_entries
+        and entry not in commit_entries
+        and entry not in restore_entries
         and not (entry.name == _UPLOAD_DELETION_REMOTE_JOURNAL and remote_journal_entries)
-        and not (entry.name == _UPLOAD_DELETION_PRIMARY_DIRNAME and entry.is_dir(follow_symlinks=False))
     ]
     if (
         len(conversion_entries) > 1
@@ -968,6 +1073,7 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
         or len(remote_journal_entries) > 1
         or len(primary_dir_entries) > 1
         or len(legacy_primary_entries) > 1
+        or legacy_unknown_entries
         or (primary_dir_entries and legacy_primary_entries)
     ):
         logger.warning(
@@ -1026,6 +1132,12 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
             logger.warning("Refusing upload deletion transaction with conflicting phase markers: %s", transaction_dir)
             return False
         recover_on_crash = True
+    if remote_journal_entries:
+        # The exact tombstones are both the authoritative recovery bytes and a
+        # persistent reservation for the remote path generation.  Only the
+        # remote reconciler may clear the journal; a later cleanup pass then
+        # commits these host tombstones.
+        return False
     if not recover_on_crash and conversion_entries and conversion_stat.st_nlink != 1:
         logger.warning(
             "Refusing discard transaction with a linked conversion tombstone: %s",
@@ -1333,7 +1445,7 @@ def delete_file_safe(
             recover_on_crash=True,
             conversion_path=owned_conversion,
         )
-        remote_delete_committed = False
+        remote_delete_may_have_started = False
         try:
             if delete_remote_copy is not None:
                 staged_conversion = _staged_conversion_path(staged_path)
@@ -1355,12 +1467,12 @@ def delete_file_safe(
                     if callable(abort_remote_delete):
                         abort_remote_delete()
                     raise
+                remote_delete_may_have_started = True
                 delete_remote_copy(
                     companion_name,
                     staged_path,
                     staged_conversion if staged_conversion.exists() else None,
                 )
-                remote_delete_committed = True
             else:
                 _mark_staged_deletion_committed(staged_path)
             _discard_staged_deletion(staged_path)
@@ -1381,7 +1493,7 @@ def delete_file_safe(
             # keep the persisted discard intent and finish locally when able.
             try:
                 _mark_staged_deletion_committed(staged_path)
-                _discard_staged_deletion(staged_path)
+                _finalize_committed_staged_deletion(staged_path)
             except BaseException:
                 logger.warning(
                     "Failed to finish a deletion after remote compensation failed: %s",
@@ -1390,14 +1502,13 @@ def delete_file_safe(
                 )
             raise
         except BaseException:
-            if remote_delete_committed:
-                # The remote view already committed.  Leave the discard
-                # transaction for startup recovery if cleanup still fails,
-                # instead of resurrecting a host-only generation.  This also
-                # handles a commit-marker write error while the process is
-                # still alive.
+            if remote_delete_may_have_started:
+                # The durable commit marker allowed remote side effects.  Any
+                # non-compensated failure after that point must converge to the
+                # deletion outcome; restoring the host could create a permanent
+                # split even when only journal finalization failed.
                 try:
-                    _discard_staged_deletion(staged_path)
+                    _finalize_committed_staged_deletion(staged_path)
                 except BaseException:
                     logger.warning(
                         "Failed to finish a deletion after the remote copy was removed: %s",

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

@@ -20,16 +20,17 @@ from deerflow.sandbox.sandbox_provider import (
 )
 from deerflow.uploads.async_helpers import run_upload_io_cancellation_safe, wait_for_task_completion
 from deerflow.uploads.layout import conversion_virtual_path
-from deerflow.uploads.lease import UploadStageLease
+from deerflow.uploads.lease import UploadNameLease, UploadStageLease
 from deerflow.uploads.manager import (
     _UPLOAD_DELETION_COMMIT_MARKER,
     _UPLOAD_DELETION_PRIMARY_DIRNAME,
     RemoteDeletionCommitRequiredError,
     RemoteDeletionCompensatedError,
+    _deletion_transaction_metadata,
+    _normalize_existing_filename,
     _staged_deletion_remote_journal,
     cleanup_stale_upload_staging_files,
     make_upload_file_sandbox_readable,
-    normalize_filename,
     upload_virtual_path,
 )
 
@@ -286,6 +287,28 @@ def _pending_remote_deletion_journals(base_dir: Path) -> list[Path]:
     return sorted(journals)
 
 
+def _validate_remote_delete_tombstone(journal_path: Path, filename: str) -> Path:
+    """Validate that a journal still names its retained host generation."""
+    transaction_dir = journal_path.parent
+    metadata = _deletion_transaction_metadata(transaction_dir.name)
+    if metadata is None:
+        raise ValueError("invalid remote deletion transaction name")
+    expected_inode, _recover_on_crash = metadata
+    primary_dir = transaction_dir / _UPLOAD_DELETION_PRIMARY_DIRNAME
+    primary_dir_stat = os.lstat(primary_dir)
+    if stat.S_ISLNK(primary_dir_stat.st_mode) or not stat.S_ISDIR(primary_dir_stat.st_mode):
+        raise ValueError("invalid remote deletion primary directory")
+    with os.scandir(primary_dir) as entries:
+        primary_entries = list(entries)
+    if len(primary_entries) != 1:
+        raise ValueError("remote deletion transaction must retain one primary")
+    primary_entry = primary_entries[0]
+    primary_stat = primary_entry.stat(follow_symlinks=False)
+    if primary_entry.name != filename or not stat.S_ISREG(primary_stat.st_mode) or primary_stat.st_nlink != 1 or primary_stat.st_ino != expected_inode:
+        raise ValueError("remote deletion journal does not match its primary generation")
+    return transaction_dir.parent.parent / "uploads"
+
+
 def reconcile_pending_remote_deletions(
     *,
     sandbox_provider_factory: Callable[[], Any],
@@ -322,39 +345,51 @@ def reconcile_pending_remote_deletions(
                     or not isinstance(thread_id, str)
                     or not thread_id
                     or (user_id is not None and not isinstance(user_id, str))
-                    or (sandbox_id is not None and not isinstance(sandbox_id, str))
+                    or not isinstance(sandbox_id, str)
+                    or not sandbox_id
                     or virtual_paths != expected_paths
                 ):
                     raise ValueError("invalid remote deletion journal fields")
-                if normalize_filename(filename) != filename:
+                if _normalize_existing_filename(filename) != filename:
                     raise ValueError("invalid remote deletion journal filename")
+                uploads_dir = _validate_remote_delete_tombstone(journal_path, filename)
 
                 if sandbox_provider is None:
                     sandbox_provider = sandbox_provider_factory()
-                sandbox = sandbox_provider.get(sandbox_id) if sandbox_id else None
-                effective_sandbox_id = sandbox_id
-                if sandbox is None:
-                    effective_sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
-                    if sandbox_provider_sandbox_uses_thread_data_mounts(sandbox_provider, effective_sandbox_id):
-                        # A newly acquired mounted sandbox reads the already
-                        # committed authoritative host state; no explicit remote
-                        # cache remains to reconcile under this contract.
-                        _unlink_journal_durably(journal_path)
-                        reconciled += 1
-                        continue
-                    sandbox = sandbox_provider.get(effective_sandbox_id)
-                if sandbox is None:
-                    raise RuntimeError(f"Sandbox {effective_sandbox_id!r} not found during upload deletion reconciliation")
-                pending = _SandboxDeletionHook(sandbox)._converge_to_deleted(tuple(virtual_paths))
-                if pending:
-                    logger.warning(
-                        "Remote upload deletion remains pending for %s: %s",
-                        journal_path,
-                        pending,
-                    )
+                name_lease = UploadNameLease.try_acquire(
+                    uploads_dir,
+                    filename,
+                    allow_legacy_posix_filename=True,
+                )
+                if name_lease is None:
                     continue
-                _unlink_journal_durably(journal_path)
-                reconciled += 1
+                try:
+                    sandbox = sandbox_provider.get(sandbox_id)
+                    if sandbox is None:
+                        acquired_sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
+                        if acquired_sandbox_id != sandbox_id:
+                            logger.warning(
+                                "Pending remote upload deletion targets sandbox %r, but thread %r acquired %r; leaving journal pending",
+                                sandbox_id,
+                                thread_id,
+                                acquired_sandbox_id,
+                            )
+                            continue
+                        sandbox = sandbox_provider.get(sandbox_id)
+                    if sandbox is None:
+                        raise RuntimeError(f"Sandbox {sandbox_id!r} not found during upload deletion reconciliation")
+                    pending = _SandboxDeletionHook(sandbox)._converge_to_deleted(tuple(virtual_paths))
+                    if pending:
+                        logger.warning(
+                            "Remote upload deletion remains pending for %s: %s",
+                            journal_path,
+                            pending,
+                        )
+                        continue
+                    _unlink_journal_durably(journal_path)
+                    reconciled += 1
+                finally:
+                    name_lease.release()
             except BaseException:
                 logger.warning(
                     "Failed to reconcile remote upload deletion journal: %s",

backend/tests/test_gateway_lifespan_shutdown.py

@@ -88,7 +88,13 @@ async def _run_lifespan_with_upload_staging_cleanup():
     startup_config = SimpleNamespace(log_level="INFO", memory=SimpleNamespace(token_counting="char", enabled=False, shutdown_flush_timeout_seconds=30.0))
     fake_service = MagicMock()
     fake_service.get_status = MagicMock(return_value={})
-    cleanup_upload_staging_files = MagicMock(return_value=2)
+    startup_events: list[str] = []
+    reconcile_remote_deletions = MagicMock(
+        side_effect=lambda **_kwargs: startup_events.append("reconcile") or 1,
+    )
+    cleanup_upload_staging_files = MagicMock(
+        side_effect=lambda: startup_events.append("cleanup") or 2,
+    )
     close_oidc_service = AsyncMock()
     stop_channel_service = AsyncMock()
 
@@ -100,6 +106,10 @@ async def _run_lifespan_with_upload_staging_cleanup():
         patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)),
         patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime),
         patch("deerflow.skills.projection.ensure_public_skill_projection"),
+        patch(
+            "app.gateway.app.reconcile_pending_remote_deletions",
+            reconcile_remote_deletions,
+        ),
         patch("app.gateway.app.cleanup_stale_upload_staging_files", cleanup_upload_staging_files),
         patch("app.gateway.app.auth.close_oidc_service", close_oidc_service),
         patch("app.channels.service.start_channel_service", side_effect=fake_start),
@@ -108,13 +118,27 @@ async def _run_lifespan_with_upload_staging_cleanup():
         async with lifespan(app):
             pass
 
-    return cleanup_upload_staging_files, close_oidc_service, stop_channel_service
+    return (
+        reconcile_remote_deletions,
+        cleanup_upload_staging_files,
+        startup_events,
+        close_oidc_service,
+        stop_channel_service,
+    )
 
 
 def test_lifespan_sweeps_upload_staging_files_on_startup():
-    cleanup_upload_staging_files, close_oidc_service, stop_channel_service = asyncio.run(_run_lifespan_with_upload_staging_cleanup())
+    (
+        reconcile_remote_deletions,
+        cleanup_upload_staging_files,
+        startup_events,
+        close_oidc_service,
+        stop_channel_service,
+    ) = asyncio.run(_run_lifespan_with_upload_staging_cleanup())
 
+    reconcile_remote_deletions.assert_called_once()
     cleanup_upload_staging_files.assert_called_once_with()
+    assert startup_events == ["reconcile", "cleanup"]
     close_oidc_service.assert_awaited_once()
     stop_channel_service.assert_awaited_once()
 

backend/tests/test_upload_sandbox_sync.py

@@ -1,10 +1,18 @@
 import asyncio
+import json
+import os
 import stat
+from unittest.mock import patch
 
 import pytest
 
-from deerflow.uploads.layout import conversion_path_for_upload
-from deerflow.uploads.manager import RemoteDeletionCommitRequiredError, cleanup_stale_upload_staging_files, delete_file_safe
+from deerflow.uploads.layout import conversion_path_for_upload, conversion_virtual_path, upload_virtual_path
+from deerflow.uploads.manager import (
+    RemoteDeletionCommitRequiredError,
+    cleanup_stale_upload_staging_files,
+    delete_file_safe,
+    publish_upload_bytes,
+)
 from deerflow.uploads.sandbox_sync import (
     _deletion_hook_for_sandbox,
     make_upload_paths_available,
@@ -128,6 +136,335 @@ def test_unconfirmed_remote_delete_is_persisted_and_retried_after_restart(tmp_pa
     assert not list(conversion.parent.glob(".upload-delete-*.part"))
 
 
+@pytest.mark.parametrize("unlink_first", [False, True], ids=["unlink", "directory-fsync"])
+def test_remote_delete_finalize_failure_never_restores_deleted_host_generation(tmp_path, unlink_first):
+    import deerflow.uploads.sandbox_sync as sandbox_sync_module
+
+    primary_virtual = upload_virtual_path("report.pdf")
+    conversion_virtual = conversion_virtual_path("report.pdf")
+
+    class Sandbox:
+        def __init__(self) -> None:
+            self.files = {
+                primary_virtual: b"pdf",
+                conversion_virtual: b"generated",
+            }
+
+        def remove_file(self, path: str) -> None:
+            self.files.pop(path, None)
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / "report.pdf"
+    primary.write_bytes(b"pdf")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_bytes(b"generated")
+    sandbox = Sandbox()
+    deletion_hook = _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id="sandbox-1",
+        thread_id="thread-1",
+        user_id="alice",
+    )
+    real_unlink = sandbox_sync_module._unlink_journal_durably
+
+    def fail_finalize(journal_path):
+        if unlink_first:
+            real_unlink(journal_path)
+        raise OSError("remote deletion journal finalize failed")
+
+    with patch.object(sandbox_sync_module, "_unlink_journal_durably", side_effect=fail_finalize):
+        with pytest.raises(OSError, match="journal finalize failed"):
+            delete_file_safe(uploads, primary.name, delete_remote_copy=deletion_hook)
+
+    assert sandbox.files == {}
+    assert not primary.exists()
+    assert not conversion.exists()
+    transactions = list(conversion.parent.glob(".upload-delete-*.part"))
+    if unlink_first:
+        assert transactions == []
+    else:
+        assert len(transactions) == 1
+        assert (transactions[0] / ".remote-delete.json").is_file()
+        assert (transactions[0] / "primary" / primary.name).read_bytes() == b"pdf"
+        assert cleanup_stale_upload_staging_files(tmp_path) == 0
+
+
+def test_pending_remote_deletion_reserves_name_until_exact_remote_generation_is_gone(tmp_path):
+    old_primary_virtual = upload_virtual_path("report.pdf")
+    old_conversion_virtual = conversion_virtual_path("report.pdf")
+
+    class RestartableSandbox:
+        def __init__(self) -> None:
+            self.files = {
+                old_primary_virtual: b"old pdf",
+                old_conversion_virtual: b"old generated",
+            }
+            self.unavailable = True
+
+        def remove_file(self, path: str) -> None:
+            self.files.pop(path, None)
+            if self.unavailable and path == old_conversion_virtual:
+                raise OSError("old conversion deletion unconfirmed")
+
+        def update_file(self, _path: str, _content: bytes) -> None:
+            raise OSError("compensation unavailable")
+
+    class Provider:
+        uses_thread_data_mounts = False
+
+        def __init__(self, sandbox) -> None:
+            self.sandbox = sandbox
+
+        def get(self, sandbox_id: str):
+            assert sandbox_id == "sandbox-1"
+            return self.sandbox
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / "report.pdf"
+    primary.write_bytes(b"old pdf")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_bytes(b"old generated")
+    sandbox = RestartableSandbox()
+    deletion_hook = _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id="sandbox-1",
+        thread_id="thread-1",
+        user_id="alice",
+    )
+
+    with pytest.raises(RemoteDeletionCommitRequiredError):
+        delete_file_safe(uploads, primary.name, delete_remote_copy=deletion_hook)
+
+    replacement = publish_upload_bytes(uploads, "report.pdf", b"new pdf")
+    assert replacement.name == "report_1.pdf"
+    new_primary_virtual = upload_virtual_path(replacement.name)
+    new_conversion_virtual = conversion_virtual_path(replacement.name)
+    sandbox.files[new_primary_virtual] = b"new pdf"
+    sandbox.files[new_conversion_virtual] = b"new generated"
+
+    sandbox.unavailable = False
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: Provider(sandbox),
+            base_dir=tmp_path,
+        )
+        == 1
+    )
+    assert sandbox.files == {
+        new_primary_virtual: b"new pdf",
+        new_conversion_virtual: b"new generated",
+    }
+    assert cleanup_stale_upload_staging_files(tmp_path) == 1
+    assert not list(conversion.parent.glob(".upload-delete-*.part"))
+
+
+def test_reconciliation_never_redirects_old_journal_to_new_sandbox(tmp_path):
+    primary_virtual = upload_virtual_path("report.pdf")
+    conversion_virtual = conversion_virtual_path("report.pdf")
+
+    class OldSandbox:
+        def remove_file(self, path: str) -> None:
+            if path == conversion_virtual:
+                raise OSError("old sandbox unavailable")
+
+        def update_file(self, _path: str, _content: bytes) -> None:
+            raise OSError("old sandbox compensation unavailable")
+
+    class NewSandbox:
+        def __init__(self) -> None:
+            self.files = {
+                primary_virtual: b"new primary",
+                conversion_virtual: b"new conversion",
+            }
+
+        def remove_file(self, path: str) -> None:
+            self.files.pop(path, None)
+
+    class RollingProvider:
+        uses_thread_data_mounts = False
+
+        def __init__(self, new_sandbox) -> None:
+            self.new_sandbox = new_sandbox
+
+        def get(self, sandbox_id: str):
+            if sandbox_id == "sandbox-old":
+                return None
+            assert sandbox_id == "sandbox-new"
+            return self.new_sandbox
+
+        def acquire(self, thread_id: str, *, user_id: str | None = None) -> str:
+            assert thread_id == "thread-1"
+            assert user_id == "alice"
+            return "sandbox-new"
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / "report.pdf"
+    primary.write_bytes(b"old primary")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_bytes(b"old conversion")
+    deletion_hook = _deletion_hook_for_sandbox(
+        OldSandbox(),
+        sandbox_id="sandbox-old",
+        thread_id="thread-1",
+        user_id="alice",
+    )
+    with pytest.raises(RemoteDeletionCommitRequiredError):
+        delete_file_safe(uploads, primary.name, delete_remote_copy=deletion_hook)
+
+    new_sandbox = NewSandbox()
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: RollingProvider(new_sandbox),
+            base_dir=tmp_path,
+        )
+        == 0
+    )
+    assert new_sandbox.files == {
+        primary_virtual: b"new primary",
+        conversion_virtual: b"new conversion",
+    }
+    assert len(list(conversion.parent.glob(".upload-delete-*.part/.remote-delete.json"))) == 1
+    assert cleanup_stale_upload_staging_files(tmp_path) == 0
+
+
+def test_reconciliation_rejects_journal_that_does_not_match_retained_generation(tmp_path):
+    report_virtual = upload_virtual_path("report.pdf")
+    report_conversion_virtual = conversion_virtual_path("report.pdf")
+    other_virtual = upload_virtual_path("other.pdf")
+    other_conversion_virtual = conversion_virtual_path("other.pdf")
+
+    class Sandbox:
+        def __init__(self) -> None:
+            self.files = {
+                report_virtual: b"report",
+                report_conversion_virtual: b"report conversion",
+                other_virtual: b"other",
+                other_conversion_virtual: b"other conversion",
+            }
+            self.unavailable = True
+
+        def remove_file(self, path: str) -> None:
+            if self.unavailable and path == report_conversion_virtual:
+                raise OSError("report conversion deletion unconfirmed")
+            self.files.pop(path, None)
+
+        def update_file(self, _path: str, _content: bytes) -> None:
+            raise OSError("compensation unavailable")
+
+    class Provider:
+        uses_thread_data_mounts = False
+
+        def __init__(self, sandbox) -> None:
+            self.sandbox = sandbox
+
+        def get(self, sandbox_id: str):
+            assert sandbox_id == "sandbox-1"
+            return self.sandbox
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / "report.pdf"
+    primary.write_bytes(b"report")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_bytes(b"report conversion")
+    sandbox = Sandbox()
+    deletion_hook = _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id="sandbox-1",
+        thread_id="thread-1",
+        user_id="alice",
+    )
+    with pytest.raises(RemoteDeletionCommitRequiredError):
+        delete_file_safe(uploads, primary.name, delete_remote_copy=deletion_hook)
+
+    journal_path = next(conversion.parent.glob(".upload-delete-*.part/.remote-delete.json"))
+    journal = json.loads(journal_path.read_text(encoding="utf-8"))
+    journal["filename"] = "other.pdf"
+    journal["virtual_paths"] = [other_virtual, other_conversion_virtual]
+    journal_path.write_text(json.dumps(journal), encoding="utf-8")
+    sandbox.unavailable = False
+
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: Provider(sandbox),
+            base_dir=tmp_path,
+        )
+        == 0
+    )
+    assert sandbox.files[other_virtual] == b"other"
+    assert sandbox.files[other_conversion_virtual] == b"other conversion"
+    assert journal_path.is_file()
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX legacy filenames are not representable on Windows")
+@pytest.mark.parametrize("filename", ["CON", "report?.pdf", "trailing ", r"report\draft.pdf"])
+def test_reconciliation_accepts_exact_legacy_posix_filename(tmp_path, filename):
+    primary_virtual = upload_virtual_path(filename)
+    conversion_virtual = conversion_virtual_path(filename)
+
+    class RestartableSandbox:
+        def __init__(self) -> None:
+            self.files = {
+                primary_virtual: b"legacy primary",
+                conversion_virtual: b"legacy conversion",
+            }
+            self.unavailable = True
+
+        def remove_file(self, path: str) -> None:
+            self.files.pop(path, None)
+            if self.unavailable and path == conversion_virtual:
+                raise OSError("legacy conversion deletion unconfirmed")
+
+        def update_file(self, _path: str, _content: bytes) -> None:
+            raise OSError("legacy compensation unavailable")
+
+    class Provider:
+        uses_thread_data_mounts = False
+
+        def __init__(self, sandbox) -> None:
+            self.sandbox = sandbox
+
+        def get(self, sandbox_id: str):
+            assert sandbox_id == "sandbox-1"
+            return self.sandbox
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / filename
+    primary.write_bytes(b"legacy primary")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_bytes(b"legacy conversion")
+    sandbox = RestartableSandbox()
+    deletion_hook = _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id="sandbox-1",
+        thread_id="thread-1",
+        user_id="alice",
+    )
+
+    with pytest.raises(RemoteDeletionCommitRequiredError):
+        delete_file_safe(uploads, filename, delete_remote_copy=deletion_hook)
+
+    sandbox.unavailable = False
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: Provider(sandbox),
+            base_dir=tmp_path,
+        )
+        == 1
+    )
+    assert sandbox.files == {}
+    assert cleanup_stale_upload_staging_files(tmp_path) == 1
+
+
 def test_prepared_remote_journal_without_commit_marker_restores_host(tmp_path):
     import deerflow.uploads.manager as upload_manager_module
 
@@ -176,6 +513,76 @@ def test_prepared_remote_journal_without_commit_marker_restores_host(tmp_path):
     assert not list(conversion.parent.glob(".upload-delete-*.part"))
 
 
+def test_reconciliation_finishes_crash_after_first_remote_delete(tmp_path):
+    import deerflow.uploads.manager as upload_manager_module
+
+    primary_virtual = upload_virtual_path("report.pdf")
+    conversion_virtual = conversion_virtual_path("report.pdf")
+
+    class Sandbox:
+        def __init__(self) -> None:
+            self.files = {
+                primary_virtual: b"pdf",
+                conversion_virtual: b"generated",
+            }
+
+        def remove_file(self, path: str) -> None:
+            self.files.pop(path, None)
+
+    class Provider:
+        uses_thread_data_mounts = False
+
+        def __init__(self, sandbox) -> None:
+            self.sandbox = sandbox
+
+        def get(self, sandbox_id: str):
+            assert sandbox_id == "sandbox-1"
+            return self.sandbox
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / "report.pdf"
+    primary.write_bytes(b"pdf")
+    conversion = conversion_path_for_upload(primary)
+    conversion.parent.mkdir()
+    conversion.write_bytes(b"generated")
+    identity = upload_manager_module.UploadIdentity.from_path(primary)
+    staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
+        uploads,
+        primary,
+        identity,
+        recover_on_crash=True,
+        conversion_path=conversion,
+    )
+    sandbox = Sandbox()
+    hook = _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id="sandbox-1",
+        thread_id="thread-1",
+        user_id="alice",
+    )
+    hook.prepare(
+        primary.name,
+        staged_path,
+        upload_manager_module._staged_conversion_path(staged_path),
+    )
+    upload_manager_module._mark_staged_deletion_committed(staged_path)
+    sandbox.remove_file(primary_virtual)
+    stage_lease.release()
+
+    assert cleanup_stale_upload_staging_files(tmp_path) == 0
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: Provider(sandbox),
+            base_dir=tmp_path,
+        )
+        == 1
+    )
+    assert sandbox.files == {}
+    assert cleanup_stale_upload_staging_files(tmp_path) == 1
+    assert not list(conversion.parent.glob(".upload-delete-*.part"))
+
+
 class _ProviderThatBecomesMounted:
     def __init__(self) -> None:
         self.uses_thread_data_mounts = False
@@ -204,6 +611,30 @@ def test_prepare_deletion_rechecks_mount_mode_after_acquire():
     assert provider.get_calls == 0
 
 
+def test_prepare_deletion_reconciles_and_cleans_before_mount_decision():
+    import deerflow.uploads.sandbox_sync as sandbox_sync_module
+
+    provider = _ProviderThatBecomesMounted()
+    provider.uses_thread_data_mounts = True
+    events: list[str] = []
+
+    with (
+        patch.object(
+            sandbox_sync_module,
+            "reconcile_pending_remote_deletions",
+            side_effect=lambda **_kwargs: events.append("reconcile") or 1,
+        ),
+        patch.object(
+            sandbox_sync_module,
+            "cleanup_stale_upload_staging_files",
+            side_effect=lambda: events.append("cleanup") or 1,
+        ),
+    ):
+        assert prepare_upload_deletion(provider, "thread-1", user_id="alice") is None
+
+    assert events == ["reconcile", "cleanup"]
+
+
 def test_prepare_deletion_async_rechecks_mount_mode_after_acquire():
     provider = _ProviderThatBecomesMounted()
 
@@ -211,6 +642,30 @@ def test_prepare_deletion_async_rechecks_mount_mode_after_acquire():
     assert provider.get_calls == 0
 
 
+def test_prepare_deletion_async_reconciles_and_cleans_before_mount_decision():
+    import deerflow.uploads.sandbox_sync as sandbox_sync_module
+
+    provider = _ProviderThatBecomesMounted()
+    provider.uses_thread_data_mounts = True
+    events: list[str] = []
+
+    with (
+        patch.object(
+            sandbox_sync_module,
+            "reconcile_pending_remote_deletions",
+            side_effect=lambda **_kwargs: events.append("reconcile") or 1,
+        ),
+        patch.object(
+            sandbox_sync_module,
+            "cleanup_stale_upload_staging_files",
+            side_effect=lambda: events.append("cleanup") or 1,
+        ),
+    ):
+        assert asyncio.run(prepare_upload_deletion_async(provider, "thread-1", user_id="alice")) is None
+
+    assert events == ["reconcile", "cleanup"]
+
+
 def test_make_paths_available_rechecks_mount_mode_after_acquire(tmp_path):
     provider = _ProviderThatBecomesMounted()
     path = tmp_path / "notes.txt"

backend/tests/test_uploads_manager.py

@@ -23,6 +23,7 @@ from deerflow.uploads.lease import UploadIdentity, UploadNameLease
 from deerflow.uploads.manager import (
     AtomicUploadPublishError,
     PathTraversalError,
+    RemoteDeletionCompensatedError,
     UnsafeUploadPathError,
     abort_staged_upload,
     claim_unique_filename,
@@ -911,7 +912,10 @@ class TestCleanupStaleUploadStagingFiles:
         assert primary.read_bytes() == b"original"
         assert not staged_path.exists()
 
-    @pytest.mark.parametrize("filename", ["report.pdf", "primary", ".remote-delete.json"])
+    @pytest.mark.parametrize(
+        "filename",
+        ["report.pdf", "primary", ".remote-delete.json", ".commit", ".restore", ".conversion"],
+    )
     def test_restores_legacy_intentless_deletion_transaction(self, tmp_path, filename):
         import deerflow.uploads.manager as upload_manager_module
 
@@ -1340,11 +1344,11 @@ class TestDeleteFileSafe:
         owned_conversion.parent.mkdir()
         owned_conversion.write_text("generated", encoding="utf-8")
 
-        with pytest.raises(OSError, match="remote unavailable"):
+        with pytest.raises(RemoteDeletionCompensatedError, match="remote unavailable"):
             delete_file_safe(
                 uploads,
                 filename,
-                delete_remote_copy=lambda _name, _primary, _conversion: (_ for _ in ()).throw(OSError("remote unavailable")),
+                delete_remote_copy=lambda _name, _primary, _conversion: (_ for _ in ()).throw(RemoteDeletionCompensatedError("remote unavailable")),
             )
 
         assert primary.read_bytes() == b"primary"
@@ -1363,9 +1367,9 @@ class TestDeleteFileSafe:
         def observe_phase_then_fail(_name, staged_primary, _conversion):
             nonlocal observed_marker
             observed_marker = upload_manager_module._staged_deletion_commit_marker(staged_primary).is_file()
-            raise OSError("remote unavailable")
+            raise RemoteDeletionCompensatedError("remote unavailable")
 
-        with pytest.raises(OSError, match="remote unavailable"):
+        with pytest.raises(RemoteDeletionCompensatedError, match="remote unavailable"):
             delete_file_safe(
                 uploads,
                 primary.name,
@@ -1412,9 +1416,9 @@ class TestDeleteFileSafe:
 
         def recreate_then_fail(_filename, _primary_path, _conversion_path):
             primary.write_bytes(b"new generation")
-            raise OSError("remote delete failed")
+            raise RemoteDeletionCompensatedError("remote delete failed")
 
-        with pytest.raises((OSError, UnsafeUploadPathError)):
+        with pytest.raises((RemoteDeletionCompensatedError, UnsafeUploadPathError)):
             delete_file_safe(
                 uploads,
                 primary.name,