fix: prevent upload lease deadlocks

README.md

@@ -961,7 +961,7 @@ DeerFlow doesn't just *talk* about doing things. It has its own computer.
 
 Each task gets its own execution environment with a full filesystem view — skills, workspace, uploads, outputs. The agent reads, writes, and edits files. It can view images and, when configured safely, execute shell commands.
 
-Uploads from the Web UI, embedded client, and IM channels share one collision-safe storage rule. A completed payload is published only if its candidate name does not exist; concurrent `report.pdf` uploads become `report.pdf`, `report_1.pdf`, `report_2.pdf`, and so on without replacing one another. The selected name is leased through conversion and sandbox synchronization, so deleting that exact name waits for its active upload lifecycle while unrelated filenames continue concurrently. Internal staging names matching `.upload-*.part` are rejected.
+Uploads from the Web UI, embedded client, and IM channels share one collision-safe storage rule. A completed payload is published only if its candidate name does not exist; concurrent `report.pdf` uploads become `report.pdf`, `report_1.pdf`, `report_2.pdf`, and so on without replacing one another. A busy candidate lease is treated as another collision instead of waiting, so inverse multi-file batches cannot deadlock. The selected name is leased through conversion and sandbox synchronization, so deleting that exact name waits for its active upload lifecycle while unrelated filenames continue concurrently. Internal staging names matching `.upload-*.part` and basenames that cannot be represented losslessly on Windows are rejected.
 
 Optional document conversions are system-owned assets under `/mnt/user-data/.upload-conversions/`. Normal targets use `<actual-upload-name>.md`; names that would exceed the filesystem component limit use a deterministic UTF-8-safe prefix plus the full SHA-256 digest. The exact generated path is returned through the upload response and omitted from the primary upload listing. Mounted AIO sandboxes expose this namespace through a read-only mount, and Local structured file APIs enforce the same rule through path mappings. Local host bash is outside that mapping boundary and must remain disabled for untrusted tasks. Non-mounted remote providers receive a private synchronized copy that may be writable but cannot mutate the authoritative host conversion or lock state. Deleting a primary removes only its exact generated asset and never infers that a user-uploaded sibling such as `uploads/report.md` is disposable.
 

backend/AGENTS.md

@@ -1461,11 +1461,11 @@ 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, so case and Unicode-normalization aliases cannot bypass an active generation on case-insensitive filesystems. The original filename remains the published name. Multi-file Gateway and embedded-client requests reserve those portable keys while retaining their publication leases, so a later alias chooses an `_N` candidate instead of waiting on its own batch. Alias-based deletion resolves the primary's actual directory entry by inode before deriving a long-name conversion path. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result.
+- 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. 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 `_N` instead of deadlocking while retaining earlier generations. Alias-based deletion resolves the primary's actual directory entry by inode, requires that entry to remain exclusive, and only then derives a long-name conversion path. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result.
 - Filenames containing NUL, `<`, `>`, or reserved model-context boundary markers are rejected before staging so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed.
 - Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files. Cancellation during staging creation drains the worker and aborts the returned stage before propagating. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools.
 - Generated Markdown is owned by `user-data/.upload-conversions/<actual-primary-filename>.md` and is omitted from primary upload listings. Deletion removes only the selected primary and that exact generated asset; it never guesses or deletes a legacy/user-owned `uploads/<stem>.md` sibling.
-- Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; potentially blocking publication and deletion name-lease acquisition uses a separate pool so waiters cannot starve general-pool release work, and 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. 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.
+- Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; only operations that may block waiting for a name lease use the separate lease-wait pool. Work needed by an existing lease holder and non-blocking publication stays on the general pool, so waiters cannot starve conversion, rollback, or release. Cold sandbox-provider construction is also offloaded. Gateway, embedded-client, and IM ingresses share provider-aware publication: mounted providers make the exact host paths sandbox-readable; non-mounted providers acquire the sandbox and synchronize the primary plus generated conversion to their exact virtual paths. Each ingress records attempted remote paths before the write can commit and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those paths before host rollback and lease release; the command fallback requires a per-call unpredictable exact success trailer. Embedded multi-file calls retain every publication and receipt until the whole response is built, then roll back the complete batch on failure. WeChat download publication uses the cancellation-safe async lease adapter, so cancellation drains and rolls back a publication worker that completes late.
 - Mounted upload paths skip both sandbox acquisition and per-file synchronization. For AIO remote/provisioner deployments this requires an explicit, accurate `sandbox.thread_data_mounts: true`; omission preserves backend auto-detection.
 - Agent receives uploaded file list via `UploadsMiddleware`
 

backend/app/gateway/routers/uploads.py

@@ -250,7 +250,7 @@ async def _publish_staged_upload_cancellation_safe(
     reserved_coordination_keys: set[str] | None = None,
 ) -> PublishedUpload:
     publish_task = asyncio.create_task(
-        run_upload_lease_io(
+        run_file_io(
             publish_staged_upload_leased,
             staged,
             filename,

backend/docs/API.md

@@ -636,7 +636,7 @@ Content-Type: multipart/form-data
 - Excel (`.xls`, `.xlsx`)
 - Word (`.doc`, `.docx`)
 
-All upload entry points publish complete payloads without replacing an existing name. Concurrent collisions are returned as `document.pdf`, `document_1.pdf`, `document_2.pdf`, and so on. A published filename remains leased through conversion, permission adjustment, sandbox synchronization, and response construction; portable case and Unicode-normalization aliases share the same coordination key so deletion cannot bypass that lifecycle on case-insensitive filesystems. Mounted providers make the exact published paths sandbox-readable; non-mounted providers receive exact private copies for Gateway, embedded-client, and IM-channel ingresses. If a non-mounted sandbox update later fails or the request is cancelled, DeerFlow removes every exact remote path attempted by that request before rolling back its host generations. Gateway cancellation also drains and aborts an in-flight staging creation. Final lease release is the commit point: cancellation newly arriving during release is delayed and the already-built successful response is returned. Basenames matching the internal `.upload-*.part` staging pattern, containing NUL, `<`, or `>`, or containing the reserved model-context boundary markers are rejected before staging so every accepted model-visible filename and path can be rendered losslessly. Embedded multi-file calls are request-atomic: a later failure rolls back every earlier host and remote generation in that call.
+All upload entry points publish complete payloads without replacing an existing name. Concurrent collisions are returned as `document.pdf`, `document_1.pdf`, `document_2.pdf`, and so on. A published filename remains leased through conversion, permission adjustment, sandbox synchronization, and response construction; portable case, Unicode-normalization, and Win32 trailing-dot/space aliases share the same coordination key. Publication never waits on a busy candidate lease and advances to `_N`, preventing inverse multi-file batches from deadlocking; deletion waits for the target generation and rejects ambiguous hard-linked identities. Mounted providers make the exact published paths sandbox-readable; non-mounted providers receive exact private copies for Gateway, embedded-client, and IM-channel ingresses. If a non-mounted sandbox update later fails or the request is cancelled, DeerFlow removes every exact remote path attempted by that request before rolling back its host generations. Gateway cancellation also drains and aborts an in-flight staging creation. Final lease release is the commit point: cancellation newly arriving during release is delayed and the already-built successful response is returned. Basenames matching the internal `.upload-*.part` staging pattern, containing NUL, `<`, or `>`, containing reserved model-context boundary markers, or invalid/reserved on Windows are rejected before staging so every accepted model-visible filename and path can be rendered losslessly. Embedded multi-file calls are request-atomic: a later failure rolls back every earlier host and remote generation in that call.
 
 Generated Markdown is stored outside the primary namespace and is not returned by the list endpoint. Normal conversion names are `<actual-primary-filename>.md`; if that component would exceed 255 UTF-8 bytes, the response contains a deterministic UTF-8-safe prefix plus the full SHA-256 digest and `.md`. Clients must consume the returned `markdown_*` fields rather than derive the path. Mounted AIO sandboxes use a read-only conversion mount, while Local structured file APIs reject writes through a read-only path mapping; Local host bash is outside that boundary. Non-mounted providers receive a private synchronized copy rather than the authoritative host namespace. Deleting `document.pdf` also deletes only its exact generated conversion; an independent `uploads/document.md` is preserved.
 

backend/docs/FILE_UPLOAD.md

@@ -52,9 +52,9 @@ POST /api/threads/{thread_id}/uploads
 - `virtual_path`: Agent 在沙箱中使用的虚拟路径
 - `artifact_url`: 前端通过 HTTP 访问文件的 URL
 
-所有上传入口都先完整写入同目录暂存文件，再以“不替换已有条目”的原子操作发布。同名碰撞依次命名为 `document.pdf`、`document_1.pdf`、`document_2.pdf`；响应中的 `filename` 和各路径字段始终使用实际发布名。系统内部保留 `.upload-*.part` 作为暂存命名空间；使用该模式的 basename、包含 NUL、`<` 或 `>`、或包含保留模型上下文边界标记的文件名会在创建暂存文件前被拒绝，以保证所有已接受的文件名和 Agent 可见路径都能无损呈现。若请求在 staging 创建尚未返回时被取消，Gateway 会等待创建结束并精确 abort 该临时文件。
+所有上传入口都先完整写入同目录暂存文件，再以“不替换已有条目”的原子操作发布。同名碰撞依次命名为 `document.pdf`、`document_1.pdf`、`document_2.pdf`；响应中的 `filename` 和各路径字段始终使用实际发布名。系统内部保留 `.upload-*.part` 作为暂存命名空间；使用该模式的 basename、包含 NUL、`<` 或 `>`、包含保留模型上下文边界标记，或无法在 Windows 上无损表示（如设备名、尾随点/空格或保留字符）的文件名会在创建暂存文件前被拒绝，以保证所有已接受的文件名和 Agent 可见路径都能无损呈现。若请求在 staging 创建尚未返回时被取消，Gateway 会等待创建结束并精确 abort 该临时文件。
 
-实际发布名会在转换、权限调整、沙箱同步和响应构造期间持有同名租约。大小写及 Unicode 规范化等可移植文件系统别名共用同一协调键，避免在 APFS/Windows 上用别名绕过 generation lease。删除该名称会等待当前生命周期完成；其他文件名仍可并发处理。跨进程协调使用 `.upload-conversions/.locks/` 下稳定保留的摘要锁文件，该目录属于内部实现，不应由 Agent 或部署脚本修改或清理。最终 lease release 是明确提交点：如果新的取消恰在 release 期间到达，系统会先完成 release 并返回已构造的成功结果，而不会把已提交文件报告成取消。
+实际发布名会在转换、权限调整、沙箱同步和响应构造期间持有同名租约。大小写、Unicode 规范化及 Win32 尾随后缀等可移植文件系统别名共用同一协调键，避免用别名绕过 generation lease。发布遇到正在使用的协调键时不会等待，而会继续选择 `_N` 候选，因此逆序并发批次不会互相持锁；删除仍会等待目标 generation 生命周期完成，并在 hard-link 歧义下拒绝误报成功。跨进程协调使用 `.upload-conversions/.locks/` 下稳定保留的摘要锁文件，该目录属于内部实现，不应由 Agent 或部署脚本修改或清理。最终 lease release 是明确提交点：如果新的取消恰在 release 期间到达，系统会先完成 release 并返回已构造的成功结果，而不会把已提交文件报告成取消。
 
 ### 2. 查询上传限制
 ```

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

@@ -1,4 +1,5 @@
 import re
+import secrets
 import shlex
 from abc import ABC, abstractmethod
 
@@ -186,7 +187,8 @@ class Sandbox(ABC):
         """Remove one exact sandbox file using the provider's virtual-path mapping."""
         resolver = getattr(self, "_resolve_path", None)
         resolved = resolver(path) if callable(resolver) else path
-        marker = "__DEERFLOW_REMOVE_FILE_OK__"
-        output = self.execute_command(f"rm -f -- {shlex.quote(resolved)} && printf '%s' {marker}")
-        if marker not in str(output):
+        marker = f"__DEERFLOW_REMOVE_FILE_OK_{secrets.token_hex(16)}__"
+        output = self.execute_command(f"rm -f -- {shlex.quote(resolved)} && printf '%s\\n' {marker}")
+        output_lines = str(output).splitlines()
+        if not output_lines or output_lines[-1] != marker:
             raise OSError(f"Sandbox did not confirm removal of {path}")

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

@@ -15,6 +15,7 @@ from deerflow.uploads.manager import (
     publish_upload_bytes_leased,
     rollback_published_upload,
 )
+from deerflow.utils.file_io import run_file_io
 
 logger = logging.getLogger(__name__)
 
@@ -75,7 +76,7 @@ async def publish_upload_bytes_leased_async(
 ) -> PublishedUpload:
     """Publish bytes off-thread without leaking a lease when cancelled."""
     publish_task = asyncio.create_task(
-        run_upload_lease_io(publish_upload_bytes_leased, base_dir, preferred_filename, data),
+        run_file_io(publish_upload_bytes_leased, base_dir, preferred_filename, data),
         name=f"publish-upload:{preferred_filename}",
     )
     try:

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

@@ -23,6 +23,7 @@ from deerflow.uploads.manager import (
     replace_system_owned_staged_file,
 )
 from deerflow.utils.file_conversion import convert_file_to_markdown
+from deerflow.utils.file_io import run_file_io
 
 logger = logging.getLogger(__name__)
 
@@ -118,8 +119,9 @@ async def _prepare_conversion_cancellation_safe(
     upload_path: Path,
     publication: PublishedUpload | None,
 ) -> _PreparedConversion:
+    prepare_io = run_upload_lease_io if publication is None else run_file_io
     prepare_task = asyncio.create_task(
-        run_upload_lease_io(_prepare_conversion, upload_path, publication),
+        prepare_io(_prepare_conversion, upload_path, publication),
         name=f"prepare-upload-conversion:{upload_path.name}",
     )
     cancelled = await wait_for_task_completion(prepare_task)

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

@@ -38,7 +38,7 @@ _THREAD_LOCKS: dict[tuple[int, int, str], _ThreadLockEntry] = {}
 
 def portable_name_coordination_key(filename: str) -> str:
     """Collapse portable filesystem case and Unicode aliases for lease locking."""
-    return unicodedata.normalize("NFC", filename).casefold()
+    return unicodedata.normalize("NFC", filename).casefold().rstrip(" .")
 
 
 def _acquire_thread_lock(uploads_dir: Path, filename: str) -> tuple[tuple[int, int, str], _ThreadLockEntry]:
@@ -63,6 +63,29 @@ def _acquire_thread_lock(uploads_dir: Path, filename: str) -> tuple[tuple[int, i
     return key, entry
 
 
+def _try_acquire_thread_lock(
+    uploads_dir: Path,
+    filename: str,
+) -> tuple[tuple[int, int, str], _ThreadLockEntry] | None:
+    directory_stat = os.lstat(uploads_dir)
+    if stat.S_ISLNK(directory_stat.st_mode) or not stat.S_ISDIR(directory_stat.st_mode):
+        raise UnsafeUploadPathError("Unsafe upload lease directory")
+    key = (directory_stat.st_dev, directory_stat.st_ino, filename)
+    with _THREAD_LOCKS_GUARD:
+        entry = _THREAD_LOCKS.get(key)
+        if entry is None:
+            entry = _ThreadLockEntry()
+            _THREAD_LOCKS[key] = entry
+        entry.references += 1
+    if entry.lock.acquire(blocking=False):
+        return key, entry
+    with _THREAD_LOCKS_GUARD:
+        entry.references -= 1
+        if entry.references == 0 and _THREAD_LOCKS.get(key) is entry:
+            del _THREAD_LOCKS[key]
+    return None
+
+
 def _release_thread_lock(key: tuple[int, int, str], entry: _ThreadLockEntry) -> None:
     entry.lock.release()
     with _THREAD_LOCKS_GUARD:
@@ -267,14 +290,7 @@ class UploadNameLease:
     @classmethod
     def acquire(cls, uploads_dir: Path, filename: str) -> "UploadNameLease":
         """Acquire the stable name lease, blocking until it is available."""
-        if not filename or Path(filename).name != filename or "\\" in filename:
-            raise UnsafeUploadPathError(f"Unsafe upload lease filename: {filename!r}")
-        if len(filename.encode("utf-8")) > 255:
-            raise UnsafeUploadPathError("Upload lease filename is too long")
-
-        uploads_dir = Path(uploads_dir)
-        coordination_key = portable_name_coordination_key(filename)
-        digest = hashlib.sha256(coordination_key.encode("utf-8")).hexdigest()
+        uploads_dir, coordination_key, digest = cls._validate_request(uploads_dir, filename)
         thread_lock_key, thread_lock_entry = _acquire_thread_lock(uploads_dir, coordination_key)
         lock_file: BinaryIO | None = None
         try:
@@ -295,6 +311,53 @@ class UploadNameLease:
             _release_thread_lock(thread_lock_key, thread_lock_entry)
             raise
 
+    @classmethod
+    def try_acquire(cls, uploads_dir: Path, filename: str) -> "UploadNameLease | None":
+        """Acquire a name lease without waiting, or return ``None`` when busy."""
+        uploads_dir, coordination_key, digest = cls._validate_request(uploads_dir, filename)
+        thread_lock = _try_acquire_thread_lock(uploads_dir, coordination_key)
+        if thread_lock is None:
+            return None
+        thread_lock_key, thread_lock_entry = thread_lock
+        thread_lock_owned = True
+        lock_file: BinaryIO | None = None
+        try:
+            lock_path = ensure_upload_lock_dir(uploads_dir) / f"{digest}.lock"
+            lock_file = _open_lock_file(lock_path)
+            if not _try_lock_file(lock_file):
+                lock_file.close()
+                lock_file = None
+                _release_thread_lock(thread_lock_key, thread_lock_entry)
+                thread_lock_owned = False
+                return None
+            return cls(
+                uploads_dir=uploads_dir,
+                filename=filename,
+                lock_path=lock_path,
+                _lock_file=lock_file,
+                _thread_lock_key=thread_lock_key,
+                _thread_lock_entry=thread_lock_entry,
+            )
+        except BaseException:
+            if lock_file is not None and not lock_file.closed:
+                lock_file.close()
+            if thread_lock_owned:
+                _release_thread_lock(thread_lock_key, thread_lock_entry)
+            raise
+
+    @staticmethod
+    def _validate_request(uploads_dir: Path, filename: str) -> tuple[Path, str, str]:
+        if not filename or Path(filename).name != filename or "\\" in filename:
+            raise UnsafeUploadPathError(f"Unsafe upload lease filename: {filename!r}")
+        if len(filename.encode("utf-8")) > 255:
+            raise UnsafeUploadPathError("Upload lease filename is too long")
+        uploads_dir = Path(uploads_dir)
+        coordination_key = portable_name_coordination_key(filename)
+        if not coordination_key:
+            raise UnsafeUploadPathError(f"Unsafe upload lease filename: {filename!r}")
+        digest = hashlib.sha256(coordination_key.encode("utf-8")).hexdigest()
+        return uploads_dir, coordination_key, digest
+
     @property
     def is_active(self) -> bool:
         """Return whether this object still owns the name lease."""

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

@@ -12,7 +12,7 @@ import shutil
 import stat
 from collections.abc import Iterator
 from dataclasses import dataclass
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
 from typing import BinaryIO
 
 from deerflow.config.paths import get_paths
@@ -32,6 +32,7 @@ logger = logging.getLogger(__name__)
 
 UPLOAD_STAGING_PREFIX = ".upload-"
 UPLOAD_STAGING_SUFFIX = ".part"
+_WINDOWS_FORBIDDEN_FILENAME_CHARS = frozenset('<>:"|?*')
 
 
 @dataclass(slots=True)
@@ -101,6 +102,8 @@ def normalize_filename(filename: str) -> str:
         raise ValueError(f"Filename contains backslash: {filename!r}")
     if "<" in safe or ">" in safe or "--- BEGIN USER INPUT ---" in safe or "--- END USER INPUT ---" in safe:
         raise ValueError(f"Filename contains reserved model-context token: {filename!r}")
+    if safe.endswith((" ", ".")) or PureWindowsPath(safe).is_reserved() or any(character in _WINDOWS_FORBIDDEN_FILENAME_CHARS or ord(character) < 32 for character in safe):
+        raise ValueError(f"Filename is reserved or invalid on Windows: {filename!r}")
     if len(safe.encode("utf-8")) > 255:
         raise ValueError(f"Filename too long: {len(safe)} chars")
     if is_upload_staging_file(safe):
@@ -286,7 +289,9 @@ def publish_staged_upload_leased(
             # its lease. This also lets one multi-file request retain the first
             # generation's lease while choosing a suffix for a duplicate name.
             continue
-        lease = UploadNameLease.acquire(staged.base_dir, candidate_name)
+        lease = UploadNameLease.try_acquire(staged.base_dir, candidate_name)
+        if lease is None:
+            continue
         linked = False
         try:
             try:
@@ -553,6 +558,7 @@ def list_files_in_dir(directory: Path) -> dict:
 
 def _find_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> Path:
     """Return the directory entry that actually names *identity*."""
+    matching_path: Path | None = None
     with os.scandir(base_dir) as entries:
         for entry in entries:
             try:
@@ -563,8 +569,15 @@ def _find_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> P
                 identity.device,
                 identity.inode,
             ):
-                return Path(entry.path)
-    raise UnsafeUploadPathError("Upload directory entry changed during deletion")
+                if entry_stat.st_nlink != 1 or matching_path is not None:
+                    raise UnsafeUploadPathError("Upload is no longer an exclusive directory entry")
+                matching_path = Path(entry.path)
+    if matching_path is None:
+        raise UnsafeUploadPathError("Upload directory entry changed during deletion")
+    matching_stat = os.lstat(matching_path)
+    if not stat.S_ISREG(matching_stat.st_mode) or matching_stat.st_nlink != 1 or (matching_stat.st_dev, matching_stat.st_ino) != (identity.device, identity.inode):
+        raise UnsafeUploadPathError("Upload is no longer an exclusive directory entry")
+    return matching_path
 
 
 def delete_file_safe(base_dir: Path, filename: str) -> dict:
@@ -604,8 +617,9 @@ def delete_file_safe(base_dir: Path, filename: str) -> dict:
         owned_conversion = existing_conversion_path_for_upload(actual_file_path)
         if owned_conversion is not None:
             owned_conversion.unlink(missing_ok=True)
-        if not identity.matches(actual_file_path):
-            raise UnsafeUploadPathError("Upload changed during deletion")
+        final_stat = os.lstat(actual_file_path)
+        if not stat.S_ISREG(final_stat.st_mode) or final_stat.st_nlink != 1 or (final_stat.st_dev, final_stat.st_ino) != (identity.device, identity.inode):
+            raise UnsafeUploadPathError("Upload is no longer an exclusive directory entry")
         actual_file_path.unlink()
     finally:
         lease.release()

backend/tests/test_client.py

@@ -2313,6 +2313,47 @@ class TestUploads:
         assert (uploads_dir / "Report.txt").read_bytes() == b"first"
         assert (uploads_dir / "report_1.txt").read_bytes() == b"second"
 
+    def test_inverse_portable_alias_batches_do_not_deadlock(self, client, tmp_path):
+        import deerflow.client as client_module
+
+        uploads_dir = tmp_path / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        batch_one = tmp_path / "batch-one"
+        batch_two = tmp_path / "batch-two"
+        batch_one.mkdir()
+        batch_two.mkdir()
+        files_one = [batch_one / "A.txt", batch_one / "B.txt"]
+        files_two = [batch_two / "b.txt", batch_two / "a.txt"]
+        for path in files_one + files_two:
+            path.write_bytes(path.name.encode())
+
+        first_publications = threading.Barrier(2)
+        batch_counts = {batch_one: 0, batch_two: 0}
+        counts_lock = threading.Lock()
+        real_publish = client_module.publish_upload_copy_leased
+
+        def pause_after_each_batch_publishes_first(base_dir, preferred_filename, source_path, **kwargs):
+            publication = real_publish(base_dir, preferred_filename, source_path, **kwargs)
+            with counts_lock:
+                batch_counts[source_path.parent] += 1
+                is_first = batch_counts[source_path.parent] == 1
+            if is_first:
+                first_publications.wait(timeout=5)
+            return publication
+
+        with (
+            patch("deerflow.client.ensure_uploads_dir", return_value=uploads_dir),
+            patch("deerflow.client.publish_upload_copy_leased", side_effect=pause_after_each_batch_publishes_first),
+            concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool,
+        ):
+            future_one = pool.submit(client.upload_files, "thread-aliases", files_one)
+            future_two = pool.submit(client.upload_files, "thread-aliases", files_two)
+            result_one = future_one.result(timeout=5)
+            result_two = future_two.result(timeout=5)
+
+        assert [file["filename"] for file in result_one["files"]] == ["A.txt", "B_1.txt"]
+        assert [file["filename"] for file in result_two["files"]] == ["b.txt", "a_1.txt"]
+
     def test_concurrent_client_uploads_preserve_all_payloads(self, client, tmp_path):
         uploads_dir = tmp_path / "user-data" / "uploads"
         uploads_dir.mkdir(parents=True)

backend/tests/test_upload_conversion.py

@@ -3,6 +3,7 @@
 import asyncio
 import hashlib
 import threading
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
@@ -15,9 +16,40 @@ from deerflow.uploads.layout import (
     conversion_virtual_path,
     existing_conversion_path_for_upload,
 )
+from deerflow.uploads.lease import UploadNameLease
 from deerflow.uploads.manager import delete_file_safe, publish_upload_bytes, publish_upload_bytes_leased
 
 
+@pytest.mark.asyncio
+async def test_active_publication_conversion_is_not_starved_by_lease_waiter(tmp_path, monkeypatch):
+    import deerflow.uploads.async_helpers as async_helpers
+
+    uploads = tmp_path / "user-data" / "uploads"
+    uploads.mkdir(parents=True)
+    publication = publish_upload_bytes_leased(uploads, "report.pdf", b"PDF")
+    single_worker = ThreadPoolExecutor(max_workers=1)
+    monkeypatch.setattr(async_helpers, "_UPLOAD_LEASE_EXECUTOR", single_worker)
+    waiter = asyncio.create_task(async_helpers.run_upload_lease_io(UploadNameLease.acquire, uploads, "report.pdf"))
+    await asyncio.sleep(0.05)
+
+    async def fake_convert(_source, output_path=None):
+        output_path.write_text("converted", encoding="utf-8")
+        return output_path
+
+    try:
+        with patch("deerflow.uploads.conversion.convert_file_to_markdown", side_effect=fake_convert):
+            converted = await asyncio.wait_for(
+                convert_uploaded_file_to_markdown(publication.path, publication=publication),
+                timeout=2,
+            )
+        assert converted is not None
+    finally:
+        publication.release()
+        waiting_lease = await asyncio.wait_for(waiter, timeout=2)
+        waiting_lease.release()
+        single_worker.shutdown(wait=True)
+
+
 @pytest.mark.asyncio
 async def test_cancellation_waits_for_converter_before_cleanup_and_lease_release(tmp_path):
     uploads = tmp_path / "user-data" / "uploads"

backend/tests/test_uploads_manager.py

@@ -58,6 +58,16 @@ def _delete_upload_in_process(
         finished.set()
 
 
+def _try_upload_lease_in_process(uploads_dir: str, filename: str, outcomes: Any) -> None:
+    try:
+        lease = UploadNameLease.try_acquire(Path(uploads_dir), filename)
+        outcomes.put(lease is not None)
+        if lease is not None:
+            lease.release()
+    except BaseException as exc:  # pragma: no cover - surfaced in the parent
+        outcomes.put(repr(exc))
+
+
 def _hold_staged_upload_in_process(
     uploads_dir: str,
     started: Any,
@@ -127,6 +137,21 @@ class TestNormalizeFilename:
         with pytest.raises(ValueError, match="NUL"):
             normalize_filename("bad\0name.pdf")
 
+    @pytest.mark.parametrize(
+        "filename",
+        [
+            "report.pdf.",
+            "report.pdf ",
+            "CON",
+            "nul.txt",
+            "report:stream.pdf",
+            "report?.pdf",
+        ],
+    )
+    def test_rejects_windows_reserved_filenames(self, filename):
+        with pytest.raises(ValueError, match="Windows"):
+            normalize_filename(filename)
+
 
 @pytest.mark.parametrize(
     ("first_name", "alias_name"),
@@ -154,6 +179,36 @@ def test_portable_filesystem_aliases_share_one_generation_lease(tmp_path, first_
         alias.release()
 
 
+def test_nonblocking_lease_treats_busy_windows_alias_as_collision(tmp_path):
+    first = UploadNameLease.acquire(tmp_path, "report.pdf")
+    try:
+        assert UploadNameLease.try_acquire(tmp_path, "REPORT.PDF. ") is None
+    finally:
+        first.release()
+
+
+def test_nonblocking_lease_observes_busy_portable_alias_across_processes(tmp_path):
+    first = UploadNameLease.acquire(tmp_path, "Report.pdf")
+    context = multiprocessing.get_context("spawn")
+    outcomes = context.Queue()
+    worker = context.Process(
+        target=_try_upload_lease_in_process,
+        args=(str(tmp_path), "report.pdf", outcomes),
+    )
+    worker.start()
+    try:
+        outcome = outcomes.get(timeout=5)
+        worker.join(5)
+    finally:
+        if worker.is_alive():
+            worker.terminate()
+            worker.join(5)
+        first.release()
+
+    assert worker.exitcode == 0
+    assert outcome is False
+
+
 @pytest.mark.parametrize(
     ("first_name", "alias_name", "expected_alias_name"),
     [
@@ -811,6 +866,25 @@ class TestDeleteFileSafe:
         assert planted.is_symlink()
         assert outside.read_text(encoding="utf-8") == "protected"
 
+    def test_delete_rejects_hardlink_race_without_unlinking_any_alias(self, tmp_path):
+        import deerflow.uploads.manager as upload_manager_module
+
+        primary = tmp_path / "zzz.txt"
+        primary.write_bytes(b"payload")
+        alias = tmp_path / "aaa.txt"
+        real_find = upload_manager_module._find_upload_path_by_identity
+
+        def add_alias_before_scan(base_dir, identity):
+            os.link(primary, alias)
+            return real_find(base_dir, identity)
+
+        with patch.object(upload_manager_module, "_find_upload_path_by_identity", side_effect=add_alias_before_scan):
+            with pytest.raises(UnsafeUploadPathError, match="exclusive"):
+                delete_file_safe(tmp_path, primary.name)
+
+        assert primary.read_bytes() == b"payload"
+        assert alias.read_bytes() == b"payload"
+
     def test_delete_removes_owned_conversion_but_preserves_legacy_sibling(self, tmp_path):
         uploads = tmp_path / "user-data" / "uploads"
         uploads.mkdir(parents=True)

backend/tests/test_uploads_router.py

@@ -80,11 +80,23 @@ def test_sandbox_remove_file_uses_provider_virtual_path_resolution():
         @staticmethod
         def execute_command(command: str) -> str:
             commands.append(command)
-            return "__DEERFLOW_REMOVE_FILE_OK__"
+            return "__DEERFLOW_REMOVE_FILE_OK_abc123__"
 
-    Sandbox.remove_file(FakeSandbox(), "/mnt/user-data/uploads/report final.pdf")
+    with patch("deerflow.sandbox.sandbox.secrets.token_hex", return_value="abc123"):
+        Sandbox.remove_file(FakeSandbox(), "/mnt/user-data/uploads/report final.pdf")
 
-    assert commands == ["rm -f -- '/home/sandbox/uploads/report final.pdf' && printf '%s' __DEERFLOW_REMOVE_FILE_OK__"]
+    assert commands == ["rm -f -- '/home/sandbox/uploads/report final.pdf' && printf '%s\\n' __DEERFLOW_REMOVE_FILE_OK_abc123__"]
+
+
+def test_sandbox_remove_file_rejects_error_output_containing_legacy_marker():
+    class FakeSandbox:
+        @staticmethod
+        def execute_command(_command: str) -> str:
+            return "rm: cannot remove '__DEERFLOW_REMOVE_FILE_OK__': Permission denied"
+
+    with patch("deerflow.sandbox.sandbox.secrets.token_hex", return_value="abc123"):
+        with pytest.raises(OSError, match="did not confirm"):
+            Sandbox.remove_file(FakeSandbox(), "/mnt/user-data/uploads/__DEERFLOW_REMOVE_FILE_OK__")
 
 
 def test_cleanup_uses_publication_identity_not_reused_path(tmp_path):
@@ -394,30 +406,23 @@ async def test_cancellation_after_remote_sync_still_removes_the_completed_copy(t
 
 
 @pytest.mark.asyncio
-async def test_waiting_publication_cannot_starve_lease_release(tmp_path, monkeypatch):
-    import deerflow.utils.file_io as file_io_module
-
+async def test_busy_publication_uses_suffix_without_waiting_for_lease(tmp_path):
     first = UploadNameLease.acquire(tmp_path, "report.pdf")
     staged = create_upload_staging_file(tmp_path)
     staged.handle.write(b"second")
-    single_worker = ThreadPoolExecutor(max_workers=1)
-    monkeypatch.setattr(file_io_module, "_FILE_IO_EXECUTOR", single_worker)
-
     publication_task = asyncio.create_task(uploads._publish_staged_upload_cancellation_safe(staged, "report.pdf"))
-    await asyncio.sleep(0.05)
-    release_task = asyncio.create_task(uploads._run_file_io_cancellation_safe(first.release))
     second = None
     try:
-        await asyncio.wait_for(asyncio.shield(release_task), timeout=0.2)
+        second = await asyncio.wait_for(publication_task, timeout=2)
+        assert first.is_active
     finally:
         if first.is_active:
             first.release()
-        await release_task
-        second = await asyncio.wait_for(publication_task, timeout=2)
-        second.release()
-        single_worker.shutdown(wait=True)
+        if second is not None:
+            second.release()
 
-    assert second.path.name == "report.pdf"
+    assert second is not None
+    assert second.path.name == "report_1.pdf"
 
 
 @pytest.mark.asyncio
@@ -1037,6 +1042,56 @@ def test_upload_files_renames_portable_aliases_within_one_batch(tmp_path):
     assert (thread_uploads_dir / "report_1.txt").read_bytes() == b"second"
 
 
+@pytest.mark.asyncio
+async def test_inverse_portable_alias_gateway_batches_do_not_deadlock(tmp_path):
+    thread_uploads_dir = tmp_path / "uploads"
+    thread_uploads_dir.mkdir(parents=True)
+    provider = _mounted_provider()
+    first_publications = asyncio.Barrier(2)
+    real_publish = uploads._publish_staged_upload_cancellation_safe
+
+    async def pause_after_each_batch_publishes_first(staged, filename, reserved_coordination_keys=None):
+        publication = await real_publish(staged, filename, reserved_coordination_keys)
+        if filename in {"A.txt", "b.txt"}:
+            await first_publications.wait()
+        return publication
+
+    async def run_batch(files):
+        return await call_unwrapped(
+            uploads.upload_files,
+            "thread-aliases",
+            request=MagicMock(),
+            files=files,
+            config=SimpleNamespace(),
+        )
+
+    with (
+        patch.object(uploads, "ensure_uploads_dir", return_value=thread_uploads_dir),
+        patch.object(uploads, "get_sandbox_provider", return_value=provider),
+        patch.object(uploads, "_publish_staged_upload_cancellation_safe", side_effect=pause_after_each_batch_publishes_first),
+    ):
+        result_one, result_two = await asyncio.wait_for(
+            asyncio.gather(
+                run_batch(
+                    [
+                        UploadFile(filename="A.txt", file=BytesIO(b"A")),
+                        UploadFile(filename="B.txt", file=BytesIO(b"B")),
+                    ]
+                ),
+                run_batch(
+                    [
+                        UploadFile(filename="b.txt", file=BytesIO(b"b")),
+                        UploadFile(filename="a.txt", file=BytesIO(b"a")),
+                    ]
+                ),
+            ),
+            timeout=5,
+        )
+
+    assert [file.filename for file in result_one.files] == ["A.txt", "B_1.txt"]
+    assert [file.filename for file in result_two.files] == ["b.txt", "a_1.txt"]
+
+
 def test_upload_files_rejects_dotdot_and_dot_filenames(tmp_path):
     thread_uploads_dir = tmp_path / "uploads"
     thread_uploads_dir.mkdir(parents=True)