fix(uploads): stop Windows replacement uploads from being skipped by the multi-link guard (#6821)

* fix(uploads): stop Windows replacement uploads from being skipped by the multi-link guard

NTFS never decrements st_nlink when one of a file's names is removed, so
every upload published through os.link(staging, dest) + unlink keeps
reading st_nlink == 2 on Windows. The non-exclusive replacement guards
then reject that destination forever, and DeerFlowClient.upload_files
turns the UnsafeUploadPathError into a log line - a replacement upload
is silently skipped (Closes #6820).

- Add _replacement_destination_is_hardlinked: POSIX-authoritative
  st_nlink > 1, skipped on Windows, used only on the not exclusive
  replacement path. The exclusive (first-claim) path keeps the strict
  count check, as does the POSIX O_NOFOLLOW branch.
- Decide "published alias" in cleanup_stale_upload_staging_files by
  inode identity against siblings (st_dev/st_ino via os.lstat) instead
  of the link count; DirEntry.stat reports zeros for both on Windows,
  matching the existing _staging_entry_stat note.
- Guard the POSIX-only st_nlink assertions in the affected tests and
  mark the three inbound-attachment mode-bit tests POSIX-only, mirroring
  the existing skipif in the same class.

* fix(uploads): restore the exclusive post-open link guard and harden the staging sweep

Address the three review findings on 5353715:

- Restore the strict post-open st_nlink != 1 check for exclusive=True
  on the no-O_NOFOLLOW (Windows fallback) branch, which the previous
  commit dropped there against what the PR description claimed. Only
  the non-exclusive replacement path defers to the Windows-aware guard.
- Never match a degenerate st_ino == 0 in the sibling-inode sweep: on
  FAT/exFAT or some SMB volumes every entry would look like a published
  alias and in-flight uploads would be deleted at any age. Such volumes
  fall back to the min_age guard only.
- Build one {name: stat} map per uploads directory before the entry
  loop so the sweep scans each directory once instead of rescanning it
  for every staging file (O(N) instead of O(N^2) lstat calls).

Tests: 407 passed / 27 skipped across test_uploads_manager.py,
test_uploads_router.py, test_upload_copy_publication.py,
test_upload_windows_staging_aliases.py, test_channel_file_attachments.py
(Windows 11, Python 3.13 venv). The three new regression tests fail on
5353715 and pass with this change.

* style(uploads): apply Ruff formatting

* fix(uploads): report staging directory scan failures

---------

Co-authored-by: sjr666666 <sjr666666@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2435,6 +2435,9 @@ Failed upload commits report the original error even if temporary-file cleanup
 also fails, for example because of a Windows sharing violation.
 Once the file is published, a temporary-file cleanup failure is logged without
 failing the upload; hidden staging files are left for the startup sweep.
+If that sweep cannot scan an uploads directory, it logs a warning and preserves
+its files while continuing cleanup elsewhere. A directory that disappears
+during cleanup is skipped quietly.
 
 Uploads, new skill support files, and new local sandbox paths reject Windows
 reserved device names on every platform, including `COM¹`, `LPT²`, the console

backend/packages/harness/deerflow/uploads/AGENTS.md

@@ -9,6 +9,9 @@ entry count at debug level. Permission and other operational errors propagate.
 Keep both entry and directory race coverage in
 `backend/tests/test_uploads_manager.py`.
 
+Staging cleanup is best-effort: warn with the directory and exception on scan
+errors, keep its files and continue elsewhere. A vanished directory stays quiet.
+
 Embedded first-time uploads use `publish_upload_copy` for suffix retries from the
 normalized original name, for originals and companions. With `snapshot_path`,
 capture the opened source once before retries and retain its original `fstat`

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

@@ -167,6 +167,26 @@ def validate_path_traversal(path: Path, base: Path) -> None:
         raise PathTraversalError("Path traversal detected") from None
 
 
+def _replacement_destination_is_hardlinked(st: os.stat_result) -> bool:
+    """Whether a *replacement* destination shows an untrusted extra hard link.
+
+    Only used on the non-exclusive replacement path, where the module is about to
+    overwrite a file it published itself. On POSIX ``st_nlink`` is authoritative,
+    so ``st_nlink > 1`` means a second name — possibly outside the uploads tree —
+    aliases the destination and overwriting it would mutate that other file.
+
+    Windows cannot answer that question with ``st_nlink``: NTFS leaves the count at
+    2 once one of the two names of a published upload is unlinked, and unlinking a
+    hard link does not decrement it either. The count there is therefore *always*
+    at least 2 for a file this module just published, which made the guard reject
+    every legitimate replacement upload (``copy_upload_file_no_symlink``) and made
+    the alias-reclaiming sweep unable to clear it. Treat the count as unreliable on
+    Windows; the symlink and regular-file guards on the same destination, and the
+    ``exclusive`` (first-claim) path used by every untrusted caller, are unchanged.
+    """
+    return os.name != "nt" and st.st_nlink > 1
+
+
 def validate_upload_destination(base_dir: Path, filename: str, *, exclusive: bool = False) -> Path:
     """Validate a destination; exclusive creation treats regular hardlinks as collisions."""
     safe_name = normalize_filename(filename)
@@ -179,7 +199,7 @@ def validate_upload_destination(base_dir: Path, filename: str, *, exclusive: boo
 
     if st is not None and not stat.S_ISREG(st.st_mode):
         raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}")
-    if not exclusive and st is not None and st.st_nlink > 1:
+    if not exclusive and st is not None and _replacement_destination_is_hardlinked(st):
         raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
 
     validate_path_traversal(dest, base_dir)
@@ -197,21 +217,80 @@ def _staging_entry_stat(entry: os.DirEntry[str]) -> os.stat_result:
     return os.lstat(entry.path)
 
 
+def _uploads_dir_entry_stats(uploads_dir: Path) -> dict[str, os.stat_result]:
+    """One ``lstat`` per entry, keyed by name.
+
+    Built once per *uploads_dir* so the stale-staging sweep stays linear in the
+    number of entries instead of rescanning the directory for every staging file.
+    Entries that cannot be stat'ed are omitted: a file that cannot be stat'ed is
+    kept, never removed on a guess.
+    """
+    try:
+        with os.scandir(uploads_dir) as entries:
+            stats: dict[str, os.stat_result] = {}
+            for entry in entries:
+                # ``_staging_entry_stat`` uses ``os.lstat`` rather than
+                # ``entry.stat``: on Windows a ``DirEntry`` stat reports zero
+                # for ``st_dev``/``st_ino`` and leaves ``st_nlink`` at zero
+                # (the same limitation the ``st_nlink`` note above records),
+                # so it can never match a real inode there.
+                try:
+                    stats[entry.name] = _staging_entry_stat(entry)
+                except FileNotFoundError:
+                    continue  # vanished mid-scan; nothing left to clean
+                except OSError:
+                    logger.warning("Could not stat upload staging file %s; keeping it", entry.path, exc_info=True)
+                    continue
+            return stats
+    except FileNotFoundError:
+        return {}  # directory vanished during cleanup
+    except OSError:
+        logger.warning("Could not scan uploads directory for stale staging files: %s", uploads_dir, exc_info=True)
+        return {}
+
+
+def _staging_name_shares_inode_with_sibling(sibling_stats: dict[str, os.stat_result], name: str, st: os.stat_result) -> bool:
+    """Whether another entry in the uploads dir names the same inode as *name*.
+
+    A commit publishes a staged part with ``os.link``, so its bytes carry two names
+    for an instant and the staged name is an alias once the final name exists. Any
+    sibling sharing ``(st_dev, st_ino)`` marks it as such an alias. Inode
+    identity is the portable signal: unlike ``st_nlink`` it stays meaningful on
+    Windows, where the count is not decremented when a hard link is removed, so the
+    sweep can still reclaim a published alias there instead of mistaking it for an
+    in-flight upload and leaving the destination stuck behind the multi-link check.
+
+    A zero ``st_ino`` on either side never matches: on filesystems that cannot
+    answer inode identity (FAT/exFAT, some SMB/network shares) every name would
+    look like every other name's alias and the sweep would delete in-flight
+    uploads at any age. Such a volume falls back to the ``min_age`` guard only.
+    """
+    if st.st_ino == 0:
+        return False
+    for sibling_name, sibling_st in sibling_stats.items():
+        if sibling_name == name or sibling_st.st_ino == 0:
+            continue
+        if sibling_st.st_dev == st.st_dev and sibling_st.st_ino == st.st_ino:
+            return True
+    return False
+
+
 def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, min_age: timedelta = UPLOAD_STAGING_MIN_AGE) -> int:
     """Remove orphaned Gateway upload staging files left by a hard crash.
 
-    A lone ``.upload-*.part`` (``st_nlink == 1``) is removed only once its mtime
-    is older than *min_age*. The uploads directories may live on a volume
-    shared by several Gateway replicas, so at startup such a file can belong to
-    an upload another replica is still writing; each chunk write refreshes its
-    mtime, which keeps it younger than the guard until it is committed or
-    abandoned.
+    A lone ``.upload-*.part`` is removed only once its mtime is older than
+    *min_age*. The uploads directories may live on a volume shared by several
+    Gateway replicas, so at startup such a file can belong to an upload another
+    replica is still writing; each chunk write refreshes its mtime, which keeps it
+    younger than the guard until it is committed or abandoned. "Lone" is decided by
+    whether a sibling entry names the same inode rather than by ``st_nlink``, which
+    Windows does not decrement when a hard link is removed.
 
-    A staging name with ``st_nlink > 1`` is removed at any age: the commit's
-    ``os.link`` already published those bytes under their final name, so the
-    staged name is only an alias left behind by a crash (or a deferred removal
-    that never ran). Keeping it would make that destination fail the
-    multi-link safety check on the next replacement upload (embedded
+    A staging name that shares an inode with a sibling is removed at any age: the
+    commit's ``os.link`` already published those bytes under their final name, so
+    the staged name is only an alias left behind by a crash (or a deferred removal
+    that never ran). Keeping it would make that destination fail the multi-link
+    safety check on the next replacement upload (embedded
     ``DeerFlowClient.upload_files`` goes through ``copy_upload_file_no_symlink``).
     Removing the alias cannot affect an upload in flight: a staged part gains
     its second link only through its own commit.
@@ -224,31 +303,19 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None, *, mi
     for uploads_dir in _iter_upload_dirs(root):
         if not uploads_dir.is_dir():
             continue
-        try:
-            with os.scandir(uploads_dir) as entries:
-                for entry in entries:
-                    if not is_upload_staging_file(entry.name) or not entry.is_file(follow_symlinks=False):
-                        continue
-                    try:
-                        st = _staging_entry_stat(entry)
-                    except FileNotFoundError:
-                        continue
-                    except OSError:
-                        logger.warning("Could not stat upload staging file %s; keeping it", entry.path, exc_info=True)
-                        continue
-                    if st.st_nlink <= 1 and st.st_mtime >= cutoff:
-                        continue  # a lone, young part may still be in flight on another replica
-                    try:
-                        os.unlink(entry.path)
-                        removed += 1
-                    except FileNotFoundError:
-                        pass
-                    except OSError:
-                        logger.warning("Failed to remove stale upload staging file: %s", entry.path, exc_info=True)
-        except FileNotFoundError:
-            continue
-        except OSError:
-            logger.warning("Failed to scan uploads directory for stale staging files: %s", uploads_dir, exc_info=True)
+        entry_stats = _uploads_dir_entry_stats(uploads_dir)
+        for name, st in entry_stats.items():
+            if not is_upload_staging_file(name) or not stat.S_ISREG(st.st_mode):
+                continue
+            if not _staging_name_shares_inode_with_sibling(entry_stats, name, st) and st.st_mtime >= cutoff:
+                continue  # a lone, young part may still be in flight on another replica
+            try:
+                os.unlink(uploads_dir / name)
+                removed += 1
+            except FileNotFoundError:
+                pass
+            except OSError:
+                logger.warning("Failed to remove stale upload staging file: %s", uploads_dir / name, exc_info=True)
     return removed
 
 
@@ -309,7 +376,7 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo
     # Note: a narrow race window remains between the pre-open lstat and open(); the
     # path-traversal check mitigates escapes from base_dir but cannot prevent an
     # attacker who can atomically replace dest with a symlink after the check.
-    if not exclusive and st is not None and st.st_nlink > 1:
+    if not exclusive and st is not None and _replacement_destination_is_hardlinked(st):
         raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
 
     flags = os.O_WRONLY | os.O_CREAT
@@ -325,7 +392,7 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo
 
     if pre_open_st is not None and not stat.S_ISREG(pre_open_st.st_mode):
         raise UnsafeUploadPathError(f"Upload destination is not a regular file: {safe_name}")
-    if not exclusive and pre_open_st is not None and pre_open_st.st_nlink > 1:
+    if not exclusive and pre_open_st is not None and _replacement_destination_is_hardlinked(pre_open_st):
         raise UnsafeUploadPathError(f"Upload destination has multiple links: {safe_name}")
 
     try:
@@ -337,7 +404,10 @@ def open_upload_file_no_symlink(base_dir: Path, filename: str, *, exclusive: boo
 
     try:
         opened_stat = os.fstat(fd)
-        if not stat.S_ISREG(opened_stat.st_mode) or opened_stat.st_nlink > 1:
+        # The exclusive path keeps the strict post-open single-link check on every
+        # platform (a just-exclusively-created file must have exactly one link);
+        # only the non-exclusive replacement path defers to the Windows-aware guard.
+        if not stat.S_ISREG(opened_stat.st_mode) or (exclusive and opened_stat.st_nlink != 1) or (not exclusive and _replacement_destination_is_hardlinked(opened_stat)):
             raise UnsafeUploadPathError(f"Upload destination is not an exclusive regular file: {safe_name}")
         os.ftruncate(fd, 0)
         fh = os.fdopen(fd, "wb")

backend/tests/test_channel_file_attachments.py

@@ -534,6 +534,7 @@ class TestInboundFileIngestion:
 
 
 class TestInboundFileSandboxPerms:
+    @pytest.mark.skipif(os.name != "posix", reason="POSIX-only: Windows CRT chmod cannot express the 0644 mode this test asserts")
     def test_make_inbound_file_sandbox_readable_sets_group_other_read(self, tmp_path):
         from app.channels.manager import _make_inbound_file_sandbox_readable
 
@@ -548,6 +549,7 @@ class TestInboundFileSandboxPerms:
         assert mode & stat.S_IRGRP
         assert mode & stat.S_IROTH
 
+    @pytest.mark.skipif(os.name != "posix", reason="POSIX-only: Windows CRT chmod cannot express the 0644 mode this test asserts")
     def test_make_inbound_file_sandbox_readable_preserves_owner_bits(self, tmp_path):
         from app.channels.manager import _make_inbound_file_sandbox_readable
 
@@ -563,6 +565,7 @@ class TestInboundFileSandboxPerms:
         assert mode == 0o644
         assert not (mode & stat.S_IWOTH)
 
+    @pytest.mark.skipif(os.name != "posix", reason="POSIX-only: Windows CRT chmod cannot express the 0600 mode this test asserts")
     def test_make_inbound_file_sandbox_readable_skips_symlink(self, tmp_path):
         from support.symlinks import symlink_or_skip
 

backend/tests/test_upload_copy_publication.py

@@ -99,7 +99,10 @@ def test_publishes_with_supported_link_options(upload_paths, monkeypatch, nofoll
     dest = copy_upload_file_no_replace(uploads, "notes.txt", source)
 
     assert dest.read_bytes() == b"uploaded content"
-    assert dest.stat().st_nlink == 1
+    if os.name != "nt":
+        # Windows never decrements the link count when the staged name is removed,
+        # so a freshly published upload reads back as ``st_nlink == 2`` there.
+        assert dest.stat().st_nlink == 1
     assert calls == [{"follow_symlinks": False} if nofollow else {}]
 
 

backend/tests/test_uploads_manager.py

@@ -14,6 +14,7 @@ import pytest
 from deerflow.uploads.manager import (
     PathTraversalError,
     UnsafeUploadPathError,
+    _staging_name_shares_inode_with_sibling,
     apply_upload_sandbox_permits,
     claim_unique_filename,
     cleanup_stale_upload_staging_files,
@@ -244,12 +245,18 @@ class TestWriteUploadFileNoSymlink:
 
         with pytest.raises(FileExistsError):
             write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True)
-        # The default replacement path still rejects hardlinks outright.
-        with pytest.raises(UnsafeUploadPathError, match="multiple links"):
-            write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement")
-
-        assert source.read_bytes() == b"existing upload"
-        assert (tmp_path / "notes.txt").read_bytes() == b"existing upload"
+        # The default replacement path still rejects hardlinks outright, where the
+        # platform maintains a trustworthy link count. Windows keeps a published
+        # upload at ``st_nlink == 2`` forever, so the replacement overwrites the
+        # destination (and, being a hard link, ``source`` with it).
+        if os.name == "nt":
+            assert write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement").read_bytes() == b"replacement"
+            assert source.read_bytes() == b"replacement"
+        else:
+            with pytest.raises(UnsafeUploadPathError, match="multiple links"):
+                write_upload_file_no_symlink(tmp_path, "notes.txt", b"replacement")
+            assert source.read_bytes() == b"existing upload"
+            assert (tmp_path / "notes.txt").read_bytes() == b"existing upload"
 
     @pytest.mark.parametrize("without_nofollow", [False, True])
     def test_exclusive_write_preserves_existing_file(self, tmp_path, monkeypatch, without_nofollow):
@@ -299,6 +306,23 @@ class TestWriteUploadFileNoSymlink:
         assert result == tmp_path / "notes.txt"
         assert (tmp_path / "notes.txt").read_bytes() == b"hello"
 
+    def test_exclusive_write_keeps_post_open_link_guard_without_no_follow(self, tmp_path, monkeypatch):
+        """Without ``O_NOFOLLOW`` (the Windows fallback branch), the exclusive
+        path keeps its strict post-open single-link check: a file that opens
+        with more than one link is rejected on every platform. Only the
+        non-exclusive replacement path defers to the Windows-aware guard."""
+        monkeypatch.delattr(os, "O_NOFOLLOW", raising=False)
+        real_fstat = os.fstat
+
+        def twin_linked_fstat(fd):
+            st = real_fstat(fd)
+            return os.stat_result((st.st_mode, st.st_ino, st.st_dev, 2, st.st_uid, st.st_gid, st.st_size, st.st_atime, st.st_mtime, st.st_ctime))
+
+        monkeypatch.setattr(os, "fstat", twin_linked_fstat)
+
+        with pytest.raises(UnsafeUploadPathError, match="not an exclusive regular file"):
+            write_upload_file_no_symlink(tmp_path, "notes.txt", b"new attachment", exclusive=True)
+
     def test_open_uses_nonblocking_flag_when_available(self, tmp_path):
         if not hasattr(os, "O_NONBLOCK"):
             pytest.skip("O_NONBLOCK not available on this platform")
@@ -697,6 +721,88 @@ class TestCleanupStaleUploadStagingFiles:
         assert young.exists()
         assert not old.exists()
 
+    def test_sibling_inode_match_ignores_degenerate_zero_inodes(self):
+        """On volumes that cannot answer inode identity (FAT/exFAT, some SMB
+        shares) ``st_ino`` reads as 0; a zero-inode sibling must never mark a
+        staging file as a published alias, or the sweep would delete in-flight
+        uploads at any age."""
+
+        def fake_stat(ino, dev=1):
+            return os.stat_result((stat.S_IFREG | 0o644, ino, dev, 1, 0, 0, 4, 0.0, 0.0, 0.0))
+
+        siblings = {
+            "published.txt": fake_stat(42),
+            ".upload-zero.part": fake_stat(0),
+        }
+
+        # A degenerate zero inode matches nothing, not even a sibling reporting
+        # the same degenerate zero.
+        assert not _staging_name_shares_inode_with_sibling(siblings, ".upload-zero.part", fake_stat(0))
+        assert not _staging_name_shares_inode_with_sibling(siblings, ".upload-new.part", fake_stat(0))
+        # A real shared-inode alias is still detected.
+        assert _staging_name_shares_inode_with_sibling(siblings, ".upload-alias.part", fake_stat(42))
+        # A different device never matches, even with the same inode number.
+        assert not _staging_name_shares_inode_with_sibling(siblings, ".upload-alias.part", fake_stat(42, dev=2))
+
+    def test_sweep_scans_each_uploads_dir_once(self, tmp_path, monkeypatch):
+        """The sweep builds its inode map once per uploads directory instead of
+        rescanning the directory for every staging file, so N orphan parts cost
+        one directory scan, not N."""
+        uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads"
+        uploads_dir.mkdir(parents=True)
+        for i in range(5):
+            part = uploads_dir / f".upload-{i}.part"
+            part.write_text("partial")
+            _set_age(part, timedelta(days=2))
+
+        real_scandir = os.scandir
+        sweep_scans: list[str] = []
+
+        def counting_scandir(path, /):
+            if os.path.abspath(os.fspath(path)) == os.path.abspath(os.fspath(uploads_dir)):
+                sweep_scans.append(os.fspath(path))
+            return real_scandir(path)
+
+        monkeypatch.setattr(os, "scandir", counting_scandir)
+
+        removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 5
+        assert len(sweep_scans) == 1
+
+    @pytest.mark.parametrize("error_code", [errno.EACCES, errno.EIO, errno.ENOENT])
+    def test_failed_directory_scan_preserves_files_and_continues_cleanup(self, tmp_path, monkeypatch, caplog, error_code):
+        failed_dir = tmp_path / "threads" / "failed" / "user-data" / "uploads"
+        healthy_dir = tmp_path / "threads" / "healthy" / "user-data" / "uploads"
+        for directory in (failed_dir, healthy_dir):
+            directory.mkdir(parents=True)
+            part = directory / ".upload-old.part"
+            part.write_text("partial", encoding="utf-8")
+            _set_age(part, timedelta(days=2))
+
+        real_scandir = os.scandir
+
+        def fail_one_scan(path):
+            if os.path.abspath(os.fspath(path)) == os.path.abspath(os.fspath(failed_dir)):
+                raise OSError(error_code, "scan failed")
+            return real_scandir(path)
+
+        monkeypatch.setattr(os, "scandir", fail_one_scan)
+        with caplog.at_level(logging.WARNING, logger="deerflow.uploads.manager"):
+            removed = cleanup_stale_upload_staging_files(tmp_path)
+
+        assert removed == 1
+        assert (failed_dir / ".upload-old.part").exists()
+        assert not (healthy_dir / ".upload-old.part").exists()
+        warnings = [record for record in caplog.records if record.name == "deerflow.uploads.manager" and record.levelno >= logging.WARNING]
+        if error_code == errno.ENOENT:
+            assert warnings == []
+        else:
+            assert len(warnings) == 1
+            assert "scan uploads directory for stale staging files" in warnings[0].getMessage()
+            assert str(failed_dir) in warnings[0].getMessage()
+            assert warnings[0].exc_info[1].errno == error_code
+
     def test_keeps_staging_file_when_its_age_cannot_be_read(self, tmp_path, caplog):
         uploads_dir = tmp_path / "threads" / "thread-1" / "user-data" / "uploads"
         uploads_dir.mkdir(parents=True)
@@ -719,7 +825,12 @@ class TestCleanupStaleUploadStagingFiles:
         leaves the final file and its ``.part`` alias sharing one inode. The
         alias is reclaimed on the next startup regardless of age, or the
         destination would fail the multi-link check on its next replacement;
-        a lone young part next to it is still treated as in flight."""
+        a lone young part next to it is still treated as in flight.
+
+        The multi-link check reads ``st_nlink``, which Windows never decrements
+        when a hard link is removed, so the pre-sweep rejection is only asserted
+        where that count is authoritative; the Windows path is exercised by the
+        replacement at the end of the test."""
         uploads_dir = tmp_path / "users" / "owner-1" / "threads" / "thread-1" / "user-data" / "uploads"
         uploads_dir.mkdir(parents=True)
         published = uploads_dir / "notes.txt"
@@ -731,8 +842,9 @@ class TestCleanupStaleUploadStagingFiles:
             pytest.skip(f"hard links unsupported here: {exc}")
         in_flight = uploads_dir / ".upload-inflight.part"
         in_flight.write_bytes(b"partial")
-        with pytest.raises(UnsafeUploadPathError, match="multiple links"):
-            validate_upload_destination(uploads_dir, "notes.txt")
+        if os.name != "nt":
+            with pytest.raises(UnsafeUploadPathError, match="multiple links"):
+                validate_upload_destination(uploads_dir, "notes.txt")
 
         removed = cleanup_stale_upload_staging_files(tmp_path)
 
@@ -740,7 +852,8 @@ class TestCleanupStaleUploadStagingFiles:
         assert not alias.exists()
         assert in_flight.exists()
         assert published.read_bytes() == b"published bytes"
-        assert os.lstat(published).st_nlink == 1
+        if os.name != "nt":
+            assert os.lstat(published).st_nlink == 1
         assert validate_upload_destination(uploads_dir, "notes.txt") == published
         replacement = tmp_path / "replacement.txt"
         replacement.write_bytes(b"replacement bytes")

backend/tests/test_uploads_router.py

@@ -833,9 +833,13 @@ def test_startup_cleanup_reclaims_a_published_staging_alias_immediately(tmp_path
     assert result.success is True
     names = sorted(p.name for p in thread_uploads_dir.iterdir())
     assert names[-1] == "notes.txt" and len(names) == 2 and names[0].startswith(".upload-") and names[0].endswith(".part")
-    assert os.lstat(thread_uploads_dir / "notes.txt").st_nlink == 2
-    with pytest.raises(ValueError, match="multiple links"):
-        validate_upload_destination(thread_uploads_dir, "notes.txt")
+    if os.name != "nt":
+        # ``st_nlink`` is authoritative only where the OS decrements it; Windows
+        # keeps a published-then-unlinked upload at 2, so the multi-link rejection
+        # is asserted on POSIX and the sweep below covers the Windows path.
+        assert os.lstat(thread_uploads_dir / "notes.txt").st_nlink == 2
+        with pytest.raises(ValueError, match="multiple links"):
+            validate_upload_destination(thread_uploads_dir, "notes.txt")
 
     assert cleanup_stale_upload_staging_files(tmp_path) == 1