fix: persist remote upload deletion recovery

backend/AGENTS.md

@@ -1461,10 +1461,10 @@ Multi-file upload with automatic document conversion:
 - Reuses one conversion worker per request when called from an active event loop
 - Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()`
 - Every ingress stages a complete payload and atomically publishes it without replacing an existing entry. Collisions across requests, processes, HTTP, embedded, and IM adapters use `name.ext`, `name_1.ext`, `name_2.ext`; storage that cannot provide atomic no-replace publication fails explicitly.
-- Exact-name generation leases use a portable NFC-plus-casefold coordination key with Win32 trailing-dot/space folding, so filesystem aliases cannot bypass an active generation; filenames that Windows cannot represent losslessly are rejected before staging, while exact legacy POSIX basenames remain deletable after upgrade. Legacy deletion has a POSIX-only lease mode for literal backslashes and components made entirely from dots/spaces; it does not weaken new-upload validation. The original filename remains the published name. Publication tries each candidate lease without blocking and treats a busy canonical key as a collision, so same-batch and inverse concurrent batches advance to a UTF-8-bounded `_N` candidate instead of deadlocking while retaining earlier generations; pathological long suffixes fall back to truncating the complete basename. Deletion rejects an inode that moved outside the requested name's lease. Primary deletion and rollback 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, so a same-name replacement can never inherit the prior generation's Markdown. User deletion starts with restore-on-crash intent; after explicit remote deletion succeeds, a persisted commit marker switches recovery to discard before authoritative host tombstones are removed. 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, 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.
+- 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, 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 remote compensation commits host deletion. 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 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.
 - 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/app/gateway/app.py

@@ -40,8 +40,10 @@ from app.gateway.routers import (
 from app.gateway.trace_middleware import TraceMiddleware, resolve_trace_enabled
 from deerflow.config import app_config as deerflow_app_config
 from deerflow.logging_config import DEFAULT_LOG_DATE_FORMAT, DEFAULT_LOG_FORMAT, configure_logging
+from deerflow.sandbox.sandbox_provider import get_sandbox_provider
 from deerflow.tracing.monocle import setup_monocle_tracing_if_enabled
 from deerflow.uploads.manager import cleanup_stale_upload_staging_files
+from deerflow.uploads.sandbox_sync import reconcile_pending_remote_deletions
 
 AppConfig = deerflow_app_config.AppConfig
 get_app_config = deerflow_app_config.get_app_config
@@ -274,6 +276,16 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     except Exception:
         logger.warning("tiktoken warm-up skipped", exc_info=True)
 
+    try:
+        reconciled_remote_deletions = await asyncio.to_thread(
+            reconcile_pending_remote_deletions,
+            sandbox_provider_factory=lambda: get_sandbox_provider(app_config=startup_config),
+        )
+        if reconciled_remote_deletions:
+            logger.info("Reconciled %d pending remote upload deletion(s)", reconciled_remote_deletions)
+    except Exception:
+        logger.warning("Pending remote upload deletion reconciliation skipped", exc_info=True)
+
     try:
         removed_upload_staging_files = await asyncio.to_thread(cleanup_stale_upload_staging_files)
         if removed_upload_staging_files:

backend/docs/FILE_UPLOAD.md

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

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

@@ -37,9 +37,11 @@ UPLOAD_STAGING_SUFFIX = ".part"
 UPLOAD_DELETION_TRANSACTION_PREFIX = ".upload-delete-"
 _UPLOAD_DELETION_RESTORE_INTENT = "restore"
 _UPLOAD_DELETION_DISCARD_INTENT = "discard"
+_UPLOAD_DELETION_PRIMARY_DIRNAME = "primary"
 _UPLOAD_DELETION_CONVERSION_TOMBSTONE = ".conversion"
 _UPLOAD_DELETION_COMMIT_MARKER = ".commit"
 _UPLOAD_DELETION_RESTORE_MARKER = ".restore"
+_UPLOAD_DELETION_REMOTE_JOURNAL = ".remote-delete.json"
 _WINDOWS_FORBIDDEN_FILENAME_CHARS = frozenset('<>:"|?*')
 
 
@@ -47,6 +49,10 @@ class RemoteDeletionCommitRequiredError(RuntimeError):
     """Remote deletion could not be compensated, so host deletion must commit."""
 
 
+class RemoteDeletionCompensatedError(RuntimeError):
+    """Remote deletion failed but its side effects were fully compensated."""
+
+
 @dataclass(slots=True)
 class StagedUpload:
     """A complete-or-in-progress upload stored under a hidden temporary name."""
@@ -735,15 +741,29 @@ def _restore_staged_deletion(
 
 
 def _staged_conversion_path(staged_path: Path) -> Path:
-    return staged_path.parent / _UPLOAD_DELETION_CONVERSION_TOMBSTONE
+    return _deletion_transaction_dir_for_staged_path(staged_path) / _UPLOAD_DELETION_CONVERSION_TOMBSTONE
+
+
+def _deletion_transaction_dir_for_staged_path(staged_path: Path) -> Path:
+    """Return the transaction root for both current and legacy layouts."""
+    parent = staged_path.parent
+    if parent.name == _UPLOAD_DELETION_PRIMARY_DIRNAME:
+        candidate = parent.parent
+        if candidate.name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and candidate.name.endswith(UPLOAD_STAGING_SUFFIX):
+            return candidate
+    return parent
 
 
 def _staged_deletion_commit_marker(staged_path: Path) -> Path:
-    return staged_path.parent / _UPLOAD_DELETION_COMMIT_MARKER
+    return _deletion_transaction_dir_for_staged_path(staged_path) / _UPLOAD_DELETION_COMMIT_MARKER
 
 
 def _staged_deletion_restore_marker(staged_path: Path) -> Path:
-    return staged_path.parent / _UPLOAD_DELETION_RESTORE_MARKER
+    return _deletion_transaction_dir_for_staged_path(staged_path) / _UPLOAD_DELETION_RESTORE_MARKER
+
+
+def _staged_deletion_remote_journal(staged_path: Path) -> Path:
+    return _deletion_transaction_dir_for_staged_path(staged_path) / _UPLOAD_DELETION_REMOTE_JOURNAL
 
 
 def _create_deletion_phase_marker(marker: Path, *, error_message: str) -> None:
@@ -759,10 +779,15 @@ def _create_deletion_phase_marker(marker: Path, *, error_message: str) -> None:
         os.fsync(descriptor)
     finally:
         os.close(descriptor)
+    directory_descriptor = os.open(marker.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    try:
+        os.fsync(directory_descriptor)
+    finally:
+        os.close(directory_descriptor)
 
 
 def _mark_staged_deletion_committed(staged_path: Path) -> None:
-    """Persist that remote side effects completed and host recovery must discard."""
+    """Persist that remote side effects may begin and recovery must discard."""
     _create_deletion_phase_marker(
         _staged_deletion_commit_marker(staged_path),
         error_message="Unsafe upload deletion commit marker",
@@ -778,11 +803,23 @@ def _mark_staged_deletion_restore(staged_path: Path) -> None:
 
 
 def _clear_staged_deletion_commit(staged_path: Path) -> None:
-    _staged_deletion_commit_marker(staged_path).unlink(missing_ok=True)
+    _unlink_deletion_control_durably(_staged_deletion_commit_marker(staged_path))
 
 
 def _clear_staged_deletion_restore(staged_path: Path) -> None:
-    _staged_deletion_restore_marker(staged_path).unlink(missing_ok=True)
+    _unlink_deletion_control_durably(_staged_deletion_restore_marker(staged_path))
+
+
+def _unlink_deletion_control_durably(path: Path) -> None:
+    try:
+        path.unlink()
+    except FileNotFoundError:
+        return
+    directory_descriptor = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    try:
+        os.fsync(directory_descriptor)
+    finally:
+        os.close(directory_descriptor)
 
 
 def _restore_staged_conversion(staged_path: Path, restored_primary_path: Path) -> None:
@@ -821,8 +858,9 @@ def _discard_staged_deletion(staged_path: Path) -> None:
     """Commit deletion of the exact primary and conversion tombstones."""
     _staged_conversion_path(staged_path).unlink(missing_ok=True)
     staged_path.unlink(missing_ok=True)
-    _clear_staged_deletion_commit(staged_path)
     _clear_staged_deletion_restore(staged_path)
+    if not _staged_deletion_remote_journal(staged_path).exists():
+        _clear_staged_deletion_commit(staged_path)
     _finish_deletion_transaction(staged_path)
 
 
@@ -857,9 +895,25 @@ def _link_staged_entry_as_recovery(staged_path: Path, original_path: Path) -> Pa
 
 def _finish_deletion_transaction(staged_path: Path) -> None:
     """Remove the now-empty transaction directory, when this is the new layout."""
-    transaction_dir = staged_path.parent
+    transaction_dir = _deletion_transaction_dir_for_staged_path(staged_path)
     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():
+        # 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.
+        return
+    try:
+        primary_dir.rmdir()
+    except FileNotFoundError:
+        pass
+    except OSError:
+        logger.warning(
+            "Failed to remove completed upload deletion primary directory: %s",
+            primary_dir,
+            exc_info=True,
+        )
     try:
         transaction_dir.rmdir()
     except FileNotFoundError:
@@ -893,7 +947,9 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
     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_entries = [
+    primary_dir_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_PRIMARY_DIRNAME and entry.is_dir(follow_symlinks=False)]
+    remote_journal_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_REMOTE_JOURNAL and primary_dir_entries]
+    legacy_primary_entries = [
         entry
         for entry in entries
         if entry.name
@@ -902,14 +958,45 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
             _UPLOAD_DELETION_COMMIT_MARKER,
             _UPLOAD_DELETION_RESTORE_MARKER,
         }
+        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 or len(commit_entries) > 1 or len(restore_entries) > 1 or len(primary_entries) > 1:
+    if (
+        len(conversion_entries) > 1
+        or len(commit_entries) > 1
+        or len(restore_entries) > 1
+        or len(remote_journal_entries) > 1
+        or len(primary_dir_entries) > 1
+        or len(legacy_primary_entries) > 1
+        or (primary_dir_entries and legacy_primary_entries)
+    ):
         logger.warning(
             "Refusing malformed upload deletion transaction with %s entries: %s",
             len(entries),
             transaction_dir,
         )
         return False
+    if remote_journal_entries:
+        remote_journal_stat = remote_journal_entries[0].stat(follow_symlinks=False)
+        if not stat.S_ISREG(remote_journal_stat.st_mode) or remote_journal_stat.st_nlink != 1:
+            logger.warning("Refusing malformed remote upload deletion journal: %s", transaction_dir)
+            return False
+    primary_dir: Path | None = None
+    if primary_dir_entries:
+        primary_dir_entry = primary_dir_entries[0]
+        if not primary_dir_entry.is_dir(follow_symlinks=False):
+            logger.warning("Refusing malformed upload deletion primary directory: %s", transaction_dir)
+            return False
+        primary_dir = Path(primary_dir_entry.path)
+        nested_primary_entries = list(os.scandir(primary_dir))
+        if len(nested_primary_entries) > 1:
+            logger.warning("Refusing upload deletion transaction with multiple primary entries: %s", transaction_dir)
+            return False
+        primary_entries = nested_primary_entries
+    else:
+        # Compatibility with transactions written before primaries were
+        # isolated from fixed control names in a dedicated directory.
+        primary_entries = legacy_primary_entries
     if conversion_entries:
         conversion_stat = conversion_entries[0].stat(follow_symlinks=False)
         if not stat.S_ISREG(conversion_stat.st_mode) or conversion_stat.st_nlink not in {1, 2}:
@@ -921,6 +1008,15 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
             logger.warning("Refusing malformed upload deletion commit marker: %s", transaction_dir)
             return False
         recover_on_crash = False
+    elif remote_journal_entries:
+        if not recover_on_crash:
+            logger.warning("Refusing discard transaction with an uncommitted remote journal: %s", transaction_dir)
+            return False
+        # The process crashed after preparing the durable remote operation but
+        # before persisting permission to start it. No remote side effect could
+        # have begun, so discard the prepared journal and restore the host.
+        _unlink_deletion_control_durably(Path(remote_journal_entries[0].path))
+        remote_journal_entries = []
     if restore_entries:
         restore_stat = restore_entries[0].stat(follow_symlinks=False)
         if not stat.S_ISREG(restore_stat.st_mode) or restore_stat.st_nlink != 1:
@@ -937,8 +1033,15 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
         )
         return False
     if not primary_entries:
-        if restore_entries and not conversion_entries:
-            Path(restore_entries[0].path).unlink()
+        if remote_journal_entries:
+            # sandbox_sync reconciles this durable remote cleanup before the
+            # transaction and its commit marker can be removed.
+            return False
+        if recover_on_crash and not commit_entries and not conversion_entries:
+            if restore_entries:
+                Path(restore_entries[0].path).unlink()
+            if primary_dir is not None:
+                primary_dir.rmdir()
             transaction_dir.rmdir()
             return True
         if not commit_entries:
@@ -947,6 +1050,8 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
         if conversion_entries:
             Path(conversion_entries[0].path).unlink()
         Path(commit_entries[0].path).unlink()
+        if primary_dir is not None:
+            primary_dir.rmdir()
         transaction_dir.rmdir()
         return True
     entry = primary_entries[0]
@@ -1101,11 +1206,19 @@ def _stage_primary_deletion(
         except FileExistsError:
             stage_lease.release()
             continue
-        staged_path = transaction_dir / primary_path.name
+        primary_dir = transaction_dir / _UPLOAD_DELETION_PRIMARY_DIRNAME
+        try:
+            primary_dir.mkdir(mode=0o700)
+        except BaseException:
+            transaction_dir.rmdir()
+            stage_lease.release()
+            raise
+        staged_path = primary_dir / primary_path.name
 
         try:
             os.rename(primary_path, staged_path)
         except BaseException:
+            primary_dir.rmdir()
             transaction_dir.rmdir()
             stage_lease.release()
             raise
@@ -1224,16 +1337,44 @@ def delete_file_safe(
         try:
             if delete_remote_copy is not None:
                 staged_conversion = _staged_conversion_path(staged_path)
+                prepare_remote_delete = getattr(delete_remote_copy, "prepare", None)
+                if callable(prepare_remote_delete):
+                    prepare_remote_delete(
+                        companion_name,
+                        staged_path,
+                        staged_conversion if staged_conversion.exists() else None,
+                    )
+                # Once a remote mutation may begin, crash recovery must never
+                # resurrect a host generation whose sandbox copies may already
+                # be gone. A live, fully compensated failure clears this marker
+                # in _restore_staged_deletion below.
+                try:
+                    _mark_staged_deletion_committed(staged_path)
+                except BaseException:
+                    abort_remote_delete = getattr(delete_remote_copy, "abort_prepared", None)
+                    if callable(abort_remote_delete):
+                        abort_remote_delete()
+                    raise
                 delete_remote_copy(
                     companion_name,
                     staged_path,
                     staged_conversion if staged_conversion.exists() else None,
                 )
                 remote_delete_committed = True
-                _mark_staged_deletion_committed(staged_path)
             else:
                 _mark_staged_deletion_committed(staged_path)
             _discard_staged_deletion(staged_path)
+        except RemoteDeletionCompensatedError:
+            # Clear the irreversible phase before removing the durable remote
+            # journal. A crash between these steps is recovered as a prepared,
+            # never-started transaction and restores the authoritative host
+            # generation.
+            _clear_staged_deletion_commit(staged_path)
+            abort_remote_delete = getattr(delete_remote_copy, "abort_prepared", None)
+            if callable(abort_remote_delete):
+                abort_remote_delete()
+            _restore_staged_deletion(staged_path, actual_file_path, identity)
+            raise
         except RemoteDeletionCommitRequiredError:
             # At least one remote mutation could not be rolled back.  Restoring
             # the host generation would publish a permanently split view, so

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

@@ -3,12 +3,16 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
+import os
+import stat
 from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+from deerflow.config.paths import get_paths
 from deerflow.sandbox.sandbox_provider import (
     sandbox_provider_sandbox_uses_thread_data_mounts,
     sandbox_provider_uses_thread_data_mounts,
@@ -16,13 +20,22 @@ 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.manager import (
+    _UPLOAD_DELETION_COMMIT_MARKER,
+    _UPLOAD_DELETION_PRIMARY_DIRNAME,
     RemoteDeletionCommitRequiredError,
+    RemoteDeletionCompensatedError,
+    _staged_deletion_remote_journal,
+    cleanup_stale_upload_staging_files,
     make_upload_file_sandbox_readable,
+    normalize_filename,
     upload_virtual_path,
 )
 
 logger = logging.getLogger(__name__)
+_REMOTE_DELETE_CONVERGENCE_ATTEMPTS = 3
+_REMOTE_DELETE_JOURNAL_MAX_BYTES = 4096
 
 
 @dataclass(frozen=True, slots=True)
@@ -55,14 +68,66 @@ def _remove_remote_paths(sandbox: Any, virtual_paths: tuple[str, ...]) -> None:
         raise first_error
 
 
-def _deletion_hook_for_sandbox(
-    sandbox: Any,
-) -> Callable[[str, Path, Path | None], None]:
-    def delete_remote_copy(
+def _unlink_journal_durably(journal_path: Path) -> None:
+    journal_path.unlink(missing_ok=True)
+    directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    try:
+        os.fsync(directory_descriptor)
+    finally:
+        os.close(directory_descriptor)
+
+
+def _read_remote_delete_journal(journal_path: Path) -> dict[str, Any]:
+    path_stat = os.lstat(journal_path)
+    if not stat.S_ISREG(path_stat.st_mode) or path_stat.st_nlink != 1:
+        raise ValueError("journal is not an exclusive regular file")
+    descriptor = os.open(journal_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+    try:
+        descriptor_stat = os.fstat(descriptor)
+        if not stat.S_ISREG(descriptor_stat.st_mode) or descriptor_stat.st_nlink != 1 or (descriptor_stat.st_dev, descriptor_stat.st_ino) != (path_stat.st_dev, path_stat.st_ino):
+            raise ValueError("journal changed identity while opening")
+        chunks: list[bytes] = []
+        remaining = _REMOTE_DELETE_JOURNAL_MAX_BYTES + 1
+        while remaining:
+            chunk = os.read(descriptor, remaining)
+            if not chunk:
+                break
+            chunks.append(chunk)
+            remaining -= len(chunk)
+        payload = b"".join(chunks)
+    finally:
+        os.close(descriptor)
+    if len(payload) > _REMOTE_DELETE_JOURNAL_MAX_BYTES:
+        raise ValueError("journal exceeds maximum size")
+    data = json.loads(payload.decode("utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError("journal payload is not an object")
+    return data
+
+
+@dataclass(slots=True)
+class _SandboxDeletionHook:
+    sandbox: Any
+    sandbox_id: str | None = None
+    thread_id: str | None = None
+    user_id: str | None = None
+    _journal_path: Path | None = None
+    _prepared_targets: tuple[tuple[str, bytes | None], ...] | None = None
+
+    def _remove_journal(self) -> None:
+        journal_path = self._journal_path
+        if journal_path is None:
+            return
+        _unlink_journal_durably(journal_path)
+        self._journal_path = None
+
+    def prepare(
+        self,
         filename: str,
         primary_path: Path,
         conversion_path: Path | None,
     ) -> None:
+        """Persist enough context to retry an interrupted remote deletion."""
         targets = (
             (upload_virtual_path(filename), primary_path.read_bytes()),
             (
@@ -70,10 +135,90 @@ def _deletion_hook_for_sandbox(
                 conversion_path.read_bytes() if conversion_path is not None else None,
             ),
         )
+        journal_path = _staged_deletion_remote_journal(primary_path)
+        payload = json.dumps(
+            {
+                "version": 1,
+                "sandbox_id": self.sandbox_id,
+                "thread_id": self.thread_id,
+                "user_id": self.user_id,
+                "filename": filename,
+                "virtual_paths": [path for path, _bytes in targets],
+            },
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode("utf-8")
+        descriptor = os.open(journal_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        try:
+            view = memoryview(payload)
+            while view:
+                written = os.write(descriptor, view)
+                if written <= 0:
+                    raise OSError("Failed to write remote upload deletion journal")
+                view = view[written:]
+            os.fsync(descriptor)
+        except BaseException:
+            os.close(descriptor)
+            journal_path.unlink(missing_ok=True)
+            raise
+        else:
+            os.close(descriptor)
+        try:
+            directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+            try:
+                os.fsync(directory_descriptor)
+            finally:
+                os.close(directory_descriptor)
+        except BaseException:
+            journal_path.unlink(missing_ok=True)
+            raise
+        self._journal_path = journal_path
+        self._prepared_targets = targets
+
+    def abort_prepared(self) -> None:
+        """Forget a journal when no remote side effect was allowed to start."""
+        self._remove_journal()
+        self._prepared_targets = None
+
+    def _converge_to_deleted(self, virtual_paths: tuple[str, ...]) -> tuple[str, ...]:
+        """Retry idempotent removal and return paths still not confirmed absent."""
+        pending = list(dict.fromkeys(virtual_paths))
+        for _attempt in range(_REMOTE_DELETE_CONVERGENCE_ATTEMPTS):
+            still_pending: list[str] = []
+            for virtual_path in pending:
+                try:
+                    self.sandbox.remove_file(virtual_path)
+                except BaseException:
+                    still_pending.append(virtual_path)
+                    logger.warning(
+                        "Failed to converge sandbox upload deletion: %s",
+                        virtual_path,
+                        exc_info=True,
+                    )
+            pending = still_pending
+            if not pending:
+                break
+        return tuple(pending)
+
+    def __call__(
+        self,
+        filename: str,
+        primary_path: Path,
+        conversion_path: Path | None,
+    ) -> None:
+        targets = self._prepared_targets
+        if targets is None:
+            targets = (
+                (upload_virtual_path(filename), primary_path.read_bytes()),
+                (
+                    conversion_virtual_path(filename),
+                    conversion_path.read_bytes() if conversion_path is not None else None,
+                ),
+            )
         removed: list[tuple[str, bytes | None]] = []
         for virtual_path, authoritative_bytes in targets:
             try:
-                sandbox.remove_file(virtual_path)
+                self.sandbox.remove_file(virtual_path)
             except BaseException as delete_error:
                 # A remote failure is ambiguous: the server may have removed
                 # the file before the client observed the error.  Re-publish
@@ -84,7 +229,7 @@ def _deletion_hook_for_sandbox(
                     if rollback_bytes is None:
                         continue
                     try:
-                        sandbox.update_file(rollback_path, rollback_bytes)
+                        self.sandbox.update_file(rollback_path, rollback_bytes)
                     except BaseException:
                         compensation_failed = True
                         logger.warning(
@@ -93,11 +238,132 @@ def _deletion_hook_for_sandbox(
                             exc_info=True,
                         )
                 if compensation_failed:
-                    raise RemoteDeletionCommitRequiredError("Remote upload deletion could not be compensated") from delete_error
-                raise
+                    # Compensation is no longer an all-or-nothing rollback.
+                    # Drive every target, including paths successfully written
+                    # back above, to the deletion outcome before committing the
+                    # authoritative host deletion. Repeated remove_file calls
+                    # are idempotent and narrow transient transport failures.
+                    pending = self._converge_to_deleted(tuple(path for path, _bytes in targets))
+                    if not pending:
+                        self._remove_journal()
+                    detail = f"; unconfirmed remote paths: {pending!r}" if pending else ""
+                    raise RemoteDeletionCommitRequiredError(f"Remote upload deletion could not be compensated{detail}") from delete_error
+                raise RemoteDeletionCompensatedError(str(delete_error)) from delete_error
             removed.append((virtual_path, authoritative_bytes))
+        self._remove_journal()
 
-    return delete_remote_copy
+
+def _deletion_hook_for_sandbox(
+    sandbox: Any,
+    *,
+    sandbox_id: str | None = None,
+    thread_id: str | None = None,
+    user_id: str | None = None,
+) -> _SandboxDeletionHook:
+    return _SandboxDeletionHook(
+        sandbox=sandbox,
+        sandbox_id=sandbox_id,
+        thread_id=thread_id,
+        user_id=user_id,
+    )
+
+
+def _pending_remote_deletion_journals(base_dir: Path) -> list[Path]:
+    patterns = (
+        "threads/*/user-data/.upload-conversions/.upload-delete-*.part/.remote-delete.json",
+        "users/*/threads/*/user-data/.upload-conversions/.upload-delete-*.part/.remote-delete.json",
+    )
+    journals: list[Path] = []
+    for pattern in patterns:
+        for path in base_dir.glob(pattern):
+            try:
+                primary_dir_stat = os.lstat(path.parent / _UPLOAD_DELETION_PRIMARY_DIRNAME)
+                commit_stat = os.lstat(path.parent / _UPLOAD_DELETION_COMMIT_MARKER)
+            except FileNotFoundError:
+                continue
+            if stat.S_ISDIR(primary_dir_stat.st_mode) and not stat.S_ISLNK(primary_dir_stat.st_mode) and stat.S_ISREG(commit_stat.st_mode) and commit_stat.st_nlink == 1:
+                journals.append(path)
+    return sorted(journals)
+
+
+def reconcile_pending_remote_deletions(
+    *,
+    sandbox_provider_factory: Callable[[], Any],
+    base_dir: Path | str | None = None,
+) -> int:
+    """Retry durable remote upload deletions before host transaction cleanup."""
+    root = Path(base_dir) if base_dir is not None else get_paths().base_dir
+    journals = _pending_remote_deletion_journals(root)
+    if not journals:
+        return 0
+    sandbox_provider: Any | None = None
+    reconciled = 0
+    for journal_path in journals:
+        transaction_dir = journal_path.parent
+        stage_lease = UploadStageLease.try_acquire(transaction_dir.parent, transaction_dir.name)
+        if stage_lease is None:
+            continue
+        try:
+            try:
+                data = _read_remote_delete_journal(journal_path)
+                if data.get("version") != 1:
+                    raise ValueError("unsupported journal version")
+                filename = data.get("filename")
+                thread_id = data.get("thread_id")
+                user_id = data.get("user_id")
+                sandbox_id = data.get("sandbox_id")
+                virtual_paths = data.get("virtual_paths")
+                expected_paths = [
+                    upload_virtual_path(filename) if isinstance(filename, str) else None,
+                    conversion_virtual_path(filename) if isinstance(filename, str) else None,
+                ]
+                if (
+                    not isinstance(filename, str)
+                    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 virtual_paths != expected_paths
+                ):
+                    raise ValueError("invalid remote deletion journal fields")
+                if normalize_filename(filename) != filename:
+                    raise ValueError("invalid remote deletion journal 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,
+                    )
+                    continue
+                _unlink_journal_durably(journal_path)
+                reconciled += 1
+            except BaseException:
+                logger.warning(
+                    "Failed to reconcile remote upload deletion journal: %s",
+                    journal_path,
+                    exc_info=True,
+                )
+        finally:
+            stage_lease.release()
+    return reconciled
 
 
 def prepare_upload_deletion(
@@ -107,6 +373,8 @@ def prepare_upload_deletion(
     user_id: str | None,
 ) -> Callable[[str, Path, Path | None], None] | None:
     """Return a lease-safe remote deletion hook for an explicitly synced sandbox."""
+    if reconcile_pending_remote_deletions(sandbox_provider_factory=lambda: sandbox_provider):
+        cleanup_stale_upload_staging_files()
     if sandbox_provider_uses_thread_data_mounts(sandbox_provider):
         return None
     sandbox_id = sandbox_provider.acquire(thread_id, user_id=user_id)
@@ -115,7 +383,12 @@ def prepare_upload_deletion(
     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)
+    return _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id=sandbox_id,
+        thread_id=thread_id,
+        user_id=user_id,
+    )
 
 
 async def prepare_upload_deletion_async(
@@ -125,6 +398,12 @@ async def prepare_upload_deletion_async(
     user_id: str | None,
 ) -> Callable[[str, Path, Path | None], None] | None:
     """Async counterpart that keeps remote acquisition off the event loop."""
+    reconciled = await asyncio.to_thread(
+        reconcile_pending_remote_deletions,
+        sandbox_provider_factory=lambda: sandbox_provider,
+    )
+    if reconciled:
+        await asyncio.to_thread(cleanup_stale_upload_staging_files)
     if await sandbox_provider_uses_thread_data_mounts_async(sandbox_provider):
         return None
     sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=user_id)
@@ -133,7 +412,12 @@ async def prepare_upload_deletion_async(
     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)
+    return _deletion_hook_for_sandbox(
+        sandbox,
+        sandbox_id=sandbox_id,
+        thread_id=thread_id,
+        user_id=user_id,
+    )
 
 
 def rollback_sandbox_sync(receipt: SandboxSyncReceipt) -> None:

backend/tests/test_upload_sandbox_sync.py

@@ -1,14 +1,181 @@
 import asyncio
 import stat
 
+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.sandbox_sync import (
+    _deletion_hook_for_sandbox,
     make_upload_paths_available,
     make_upload_paths_available_async,
     prepare_upload_deletion,
     prepare_upload_deletion_async,
+    reconcile_pending_remote_deletions,
 )
 
 
+def test_failed_compensation_deletes_paths_that_were_already_republished(tmp_path):
+    primary_virtual = "/mnt/user-data/uploads/report.pdf"
+    conversion_virtual = "/mnt/user-data/.upload-conversions/report.pdf.md"
+
+    class PartialFailureSandbox:
+        def __init__(self) -> None:
+            self.files = {
+                primary_virtual: b"pdf",
+                conversion_virtual: b"generated",
+            }
+            self.remove_calls: list[str] = []
+            self.conversion_failure_injected = False
+
+        def remove_file(self, path: str) -> None:
+            self.remove_calls.append(path)
+            self.files.pop(path, None)
+            if path == conversion_virtual and not self.conversion_failure_injected:
+                self.conversion_failure_injected = True
+                raise OSError("conversion response lost")
+
+        def update_file(self, path: str, content: bytes) -> None:
+            if path == primary_virtual:
+                raise OSError("primary compensation unavailable")
+            self.files[path] = content
+
+    primary = tmp_path / "report.pdf"
+    primary.write_bytes(b"pdf")
+    conversion = tmp_path / "report.pdf.md"
+    conversion.write_bytes(b"generated")
+    sandbox = PartialFailureSandbox()
+
+    with pytest.raises(RemoteDeletionCommitRequiredError):
+        _deletion_hook_for_sandbox(sandbox)("report.pdf", primary, conversion)
+
+    assert sandbox.files == {}
+    assert sandbox.remove_calls == [
+        primary_virtual,
+        conversion_virtual,
+        primary_virtual,
+        conversion_virtual,
+    ]
+
+
+def test_unconfirmed_remote_delete_is_persisted_and_retried_after_restart(tmp_path):
+    primary_virtual = "/mnt/user-data/uploads/report.pdf"
+    conversion_virtual = "/mnt/user-data/.upload-conversions/report.pdf.md"
+
+    class RestartableSandbox:
+        def __init__(self) -> None:
+            self.files = {
+                primary_virtual: b"pdf",
+                conversion_virtual: b"generated",
+            }
+            self.unavailable = True
+
+        def remove_file(self, path: str) -> None:
+            if path == primary_virtual:
+                self.files.pop(path, None)
+                return
+            if self.unavailable:
+                self.files.pop(path, None)
+                raise OSError("conversion deletion unconfirmed")
+            self.files.pop(path, None)
+
+        def update_file(self, _path: str, _content: bytes) -> None:
+            raise OSError("compensation unavailable")
+
+    class Provider:
+        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")
+    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)
+
+    assert not primary.exists()
+    assert not conversion.exists()
+    journals = list(conversion.parent.glob(".upload-delete-*.part/.remote-delete.json"))
+    assert len(journals) == 1
+    assert cleanup_stale_upload_staging_files(tmp_path) == 0
+
+    sandbox.unavailable = False
+    provider = Provider(sandbox)
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: provider,
+            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"))
+
+
+def test_prepared_remote_journal_without_commit_marker_restores_host(tmp_path):
+    import deerflow.uploads.manager as upload_manager_module
+
+    uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    primary = uploads / "report.pdf"
+    primary.write_bytes(b"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,
+    )
+    hook = _deletion_hook_for_sandbox(
+        object(),
+        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))
+
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: (_ for _ in ()).throw(AssertionError("live journal must be skipped")),
+            base_dir=tmp_path,
+        )
+        == 0
+    )
+    stage_lease.release()
+    assert (
+        reconcile_pending_remote_deletions(
+            sandbox_provider_factory=lambda: (_ for _ in ()).throw(AssertionError("uncommitted journal must be skipped")),
+            base_dir=tmp_path,
+        )
+        == 0
+    )
+
+    assert cleanup_stale_upload_staging_files(tmp_path) == 1
+    assert primary.read_bytes() == b"pdf"
+    assert conversion.read_bytes() == b"generated"
+    assert not list(conversion.parent.glob(".upload-delete-*.part"))
+
+
 class _ProviderThatBecomesMounted:
     def __init__(self) -> None:
         self.uses_thread_data_mounts = False

backend/tests/test_uploads_manager.py

@@ -911,12 +911,13 @@ class TestCleanupStaleUploadStagingFiles:
         assert primary.read_bytes() == b"original"
         assert not staged_path.exists()
 
-    def test_restores_legacy_intentless_deletion_transaction(self, tmp_path):
+    @pytest.mark.parametrize("filename", ["report.pdf", "primary", ".remote-delete.json"])
+    def test_restores_legacy_intentless_deletion_transaction(self, tmp_path, filename):
         import deerflow.uploads.manager as upload_manager_module
 
         uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
         uploads.mkdir(parents=True)
-        primary = uploads / "report.pdf"
+        primary = uploads / filename
         primary.write_bytes(b"original")
         identity = UploadIdentity.from_path(primary)
         staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
@@ -925,14 +926,21 @@ class TestCleanupStaleUploadStagingFiles:
             identity,
         )
         stage_lease.release()
-        legacy_transaction_dir = staged_path.parent.with_name(
-            staged_path.parent.name.replace(
+        transaction_dir = staged_path.parent.parent
+        legacy_transaction_dir = transaction_dir.with_name(
+            transaction_dir.name.replace(
                 ".upload-delete-restore-",
                 ".upload-delete-",
                 1,
             )
         )
-        staged_path.parent.rename(legacy_transaction_dir)
+        current_primary_dir = staged_path.parent
+        temporary_primary_dir = transaction_dir / ".current-primary"
+        current_primary_dir.rename(temporary_primary_dir)
+        legacy_staged_path = transaction_dir / staged_path.name
+        (temporary_primary_dir / staged_path.name).rename(legacy_staged_path)
+        temporary_primary_dir.rmdir()
+        transaction_dir.rename(legacy_transaction_dir)
         legacy_staged_path = legacy_transaction_dir / staged_path.name
 
         assert cleanup_stale_upload_staging_files(tmp_path) == 1
@@ -1078,6 +1086,32 @@ class TestCleanupStaleUploadStagingFiles:
         assert conversion.read_text(encoding="utf-8") == "conversion"
         assert not staged_path.exists()
 
+    @pytest.mark.parametrize("filename", [".commit", ".restore", ".conversion", ".remote-delete.json", "primary"])
+    def test_crash_recovery_restores_control_named_primary_and_conversion(self, tmp_path, filename):
+        import deerflow.uploads.manager as upload_manager_module
+
+        uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        primary = uploads / filename
+        primary.write_bytes(b"primary")
+        conversion = conversion_path_for_upload(primary)
+        conversion.parent.mkdir()
+        conversion.write_text("conversion", encoding="utf-8")
+        identity = UploadIdentity.from_path(primary)
+        staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
+            uploads,
+            primary,
+            identity,
+            recover_on_crash=True,
+            conversion_path=conversion,
+        )
+        stage_lease.release()
+
+        assert cleanup_stale_upload_staging_files(tmp_path) == 1
+        assert primary.read_bytes() == b"primary"
+        assert conversion.read_text(encoding="utf-8") == "conversion"
+        assert not staged_path.exists()
+
     def test_crash_after_remote_delete_commits_primary_and_conversion(self, tmp_path):
         import deerflow.uploads.manager as upload_manager_module
 
@@ -1296,7 +1330,52 @@ class TestDeleteFileSafe:
         assert primary.read_bytes() == b"PDF"
         assert owned_conversion.read_text(encoding="utf-8") == "generated"
 
-    def test_remote_success_does_not_restore_host_when_commit_marker_fails(self, tmp_path):
+    @pytest.mark.parametrize("filename", [".commit", ".restore", ".conversion", ".remote-delete.json", "primary"])
+    def test_failed_remote_delete_restores_control_named_primary_and_conversion(self, tmp_path, filename):
+        uploads = tmp_path / "user-data" / "uploads"
+        uploads.mkdir(parents=True)
+        primary = uploads / filename
+        primary.write_bytes(b"primary")
+        owned_conversion = conversion_path_for_upload(primary)
+        owned_conversion.parent.mkdir()
+        owned_conversion.write_text("generated", encoding="utf-8")
+
+        with pytest.raises(OSError, match="remote unavailable"):
+            delete_file_safe(
+                uploads,
+                filename,
+                delete_remote_copy=lambda _name, _primary, _conversion: (_ for _ in ()).throw(OSError("remote unavailable")),
+            )
+
+        assert primary.read_bytes() == b"primary"
+        assert owned_conversion.read_text(encoding="utf-8") == "generated"
+        assert not list(owned_conversion.parent.glob(".upload-delete-*.part"))
+
+    def test_remote_delete_phase_is_persisted_before_the_hook_runs(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"primary")
+        observed_marker = False
+
+        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")
+
+        with pytest.raises(OSError, match="remote unavailable"):
+            delete_file_safe(
+                uploads,
+                primary.name,
+                delete_remote_copy=observe_phase_then_fail,
+            )
+
+        assert observed_marker
+        assert primary.read_bytes() == b"primary"
+
+    def test_commit_marker_failure_prevents_remote_side_effect_and_restores_host(self, tmp_path):
         import deerflow.uploads.manager as upload_manager_module
 
         uploads = tmp_path / "user-data" / "uploads"
@@ -1320,9 +1399,9 @@ class TestDeleteFileSafe:
                     delete_remote_copy=lambda name, _primary, _conversion: remote_names.append(name),
                 )
 
-        assert remote_names == ["report.pdf"]
-        assert not primary.exists()
-        assert not owned_conversion.exists()
+        assert remote_names == []
+        assert primary.read_bytes() == b"PDF"
+        assert owned_conversion.read_text(encoding="utf-8") == "generated"
         assert not list(owned_conversion.parent.glob(".upload-delete-*.part"))
 
     def test_failed_delete_preserves_old_generation_when_name_is_recreated(self, tmp_path):