fix(uploads): skip files that vanish mid-scan in list_files_in_dir (#6533)

* fix(uploads): skip files that vanish mid-scan in list_files_in_dir

The listing iterates scandir entries and stats each survivor. A
concurrent upload delete or the staging sweep can remove a file in
between, and the resulting FileNotFoundError escapes as a 500 on the
listing endpoint. is_file() does not cover this: on POSIX it answers
from the scandir d_type without touching the filesystem again.

Catch OSError around the stat and skip the entry on ENOENT/ENOTDIR/
ELOOP, the same tolerance the chmod path just above already applies to
replacement races. Operational errors such as EACCES still propagate.

Regression tests pin both directions: a vanished entry is skipped while
the rest of the listing survives, and a permission error still raises.

Fixes #6097

* fix(uploads): log vanished entries and fix comment per review

Address willem-bd's review on #6533: the skip now emits a debug line so
"my file disappeared from the list" reports are diagnosable without
changing the response contract, and the race comment names the actual
concurrent removers (the DELETE endpoint, possibly from another
replica, and sandbox processes) instead of the staging sweep, which
cannot remove a listed non-staging file.

The directory-level race between the is_dir() guard and os.scandir()
stays out of scope; it is covered by the still-open #6096 and its PR.

Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com>

* fix(uploads): tolerate directory races during listing

* docs(uploads): scope listing guidance to avoid inherited size limits

* fix(uploads): log directory races during listing

---------

Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2214,6 +2214,11 @@ remains downloadable but cannot be attached directly. Download it, rename it,
 and upload it to the thread. This change does not recover or migrate older
 thread uploads that already match the staging pattern.
 
+Upload listings remain available during concurrent cleanup: removed files are
+omitted, and if the upload directory disappears or is replaced by a file, the
+listing returns any entries already collected. Permission and other operational
+errors still surface.
+
 This is the difference between a chatbot with tool access and an agent with an actual execution environment.
 
 ```

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

@@ -0,0 +1,10 @@
+# Uploads
+
+`manager.py::list_files_in_dir` returns a best-effort snapshot under concurrent
+cleanup. Entry stat races (`ENOENT`, `ENOTDIR`, `ELOOP`) are skipped with debug
+logging. Directory removal or replacement by a non-directory during scanning
+returns entries already collected and logs the directory path and retained
+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`.

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

@@ -462,6 +462,10 @@ def apply_upload_sandbox_permits(file_path: os.PathLike[str] | str, extra_mode_b
 def list_files_in_dir(directory: Path) -> dict:
     """List files (not directories) in *directory*.
 
+    Listing is best-effort under concurrent deletion: vanished entries are
+    omitted, and a directory that disappears or becomes a non-directory
+    returns the entries collected so far. Operational errors still propagate.
+
     Args:
         directory: Directory to scan.
 
@@ -474,22 +478,38 @@ def list_files_in_dir(directory: Path) -> dict:
         return {"files": [], "count": 0}
 
     files = []
-    with os.scandir(directory) as entries:
-        for entry in sorted(entries, key=lambda e: e.name):
-            if is_upload_staging_file(entry.name):
-                continue
-            if not entry.is_file(follow_symlinks=False):
-                continue
-            st = entry.stat(follow_symlinks=False)
-            files.append(
-                {
-                    "filename": entry.name,
-                    "size": st.st_size,
-                    "path": entry.path,
-                    "extension": Path(entry.name).suffix,
-                    "modified": st.st_mtime,
-                }
-            )
+    try:
+        with os.scandir(directory) as entries:
+            for entry in sorted(entries, key=lambda e: e.name):
+                if is_upload_staging_file(entry.name):
+                    continue
+                if not entry.is_file(follow_symlinks=False):
+                    continue
+                try:
+                    st = entry.stat(follow_symlinks=False)
+                except OSError as exc:
+                    # The entry can vanish between the scandir sweep and this stat
+                    # (the DELETE endpoint via delete_file_safe, possibly from
+                    # another replica, or a sandbox process removing its own file).
+                    # Same policy as the chmod path above: skip expected races,
+                    # surface operational errors like EACCES.
+                    if exc.errno in {errno.ENOENT, errno.ENOTDIR, errno.ELOOP}:
+                        logger.debug("Skipped upload entry that vanished mid-scan: %s", entry.path)
+                        continue
+                    raise
+                files.append(
+                    {
+                        "filename": entry.name,
+                        "size": st.st_size,
+                        "path": entry.path,
+                        "extension": Path(entry.name).suffix,
+                        "modified": st.st_mtime,
+                    }
+                )
+    except (FileNotFoundError, NotADirectoryError):
+        # Thread deletion or a sandbox process may remove or replace the
+        # directory after the is_dir() check. Keep the collected snapshot.
+        logger.debug("Uploads directory vanished mid-scan, keeping partial snapshot of %d entries: %s", len(files), directory)
     return {"files": files, "count": len(files)}
 
 

backend/tests/test_agent_guidance_check.py

@@ -43,6 +43,7 @@ EXPECTED_GUIDANCE_PATHS = {
     "backend/packages/harness/deerflow/tracing/AGENTS.md",
     "backend/packages/harness/deerflow/tui/AGENTS.md",
     "backend/packages/harness/deerflow/typesafe/AGENTS.md",
+    "backend/packages/harness/deerflow/uploads/AGENTS.md",
     "backend/packages/harness/deerflow/utils/AGENTS.md",
     "frontend/src/AGENTS.md",
     "scripts/AGENTS.md",

backend/tests/test_uploads_manager.py

@@ -1,6 +1,7 @@
 """Tests for deerflow.uploads.manager — shared upload management logic."""
 
 import errno
+import logging
 import os
 import shutil
 import stat
@@ -378,6 +379,41 @@ class TestCopyUploadFileNoSymlink:
 # ---------------------------------------------------------------------------
 
 
+class _VanishingDirEntry:
+    """DirEntry stand-in whose stat() raises, simulating a file removed mid-scan."""
+
+    def __init__(self, real, error):
+        self._real = real
+        self._error = error
+
+    @property
+    def name(self):
+        return self._real.name
+
+    @property
+    def path(self):
+        return self._real.path
+
+    def is_file(self, follow_symlinks=False):
+        return True
+
+    def stat(self, follow_symlinks=False):
+        raise self._error
+
+
+class _StaticScandir:
+    """Context manager mimicking os.scandir over a fixed entry list."""
+
+    def __init__(self, entries):
+        self._entries = entries
+
+    def __enter__(self):
+        return iter(self._entries)
+
+    def __exit__(self, *exc_info):
+        return False
+
+
 class TestListFilesInDir:
     def test_empty_dir(self, tmp_path):
         result = list_files_in_dir(tmp_path)
@@ -387,6 +423,42 @@ class TestListFilesInDir:
         result = list_files_in_dir(tmp_path / "nope")
         assert result == {"files": [], "count": 0}
 
+    @pytest.mark.parametrize("replacement", ["removed", "file"])
+    def test_directory_replaced_after_is_dir_check(self, tmp_path, monkeypatch, caplog, replacement):
+        directory = tmp_path / "uploads"
+        directory.mkdir()
+        real_scandir = os.scandir
+
+        def replace_before_scandir(path):
+            assert path == directory
+            directory.rmdir()
+            if replacement == "file":
+                directory.write_text("replacement", encoding="utf-8")
+            return real_scandir(path)
+
+        monkeypatch.setattr(os, "scandir", replace_before_scandir)
+
+        with caplog.at_level(logging.DEBUG, logger="deerflow.uploads.manager"):
+            assert list_files_in_dir(directory) == {"files": [], "count": 0}
+
+        assert (
+            "deerflow.uploads.manager",
+            logging.DEBUG,
+            f"Uploads directory vanished mid-scan, keeping partial snapshot of 0 entries: {directory}",
+        ) in caplog.record_tuples
+
+    @pytest.mark.parametrize("error_code", [errno.EACCES, errno.EIO])
+    def test_scandir_operational_error_still_propagates(self, tmp_path, monkeypatch, error_code):
+        def fail_scandir(path):
+            raise OSError(error_code, os.strerror(error_code), str(path))
+
+        monkeypatch.setattr(os, "scandir", fail_scandir)
+
+        with pytest.raises(OSError) as excinfo:
+            list_files_in_dir(tmp_path)
+
+        assert excinfo.value.errno == error_code
+
     def test_multiple_files_sorted(self, tmp_path):
         (tmp_path / "b.txt").write_text("b")
         (tmp_path / "a.txt").write_text("a")
@@ -416,6 +488,35 @@ class TestListFilesInDir:
         assert result["count"] == 4
         assert [f["filename"] for f in result["files"]] == [".env", ".upload-note.txt", "draft.part", "visible.txt"]
 
+    @pytest.mark.parametrize("error_code", [errno.ENOENT, errno.ENOTDIR, errno.ELOOP])
+    def test_skips_entries_vanishing_mid_scan(self, tmp_path, monkeypatch, caplog, error_code):
+        (tmp_path / "kept.txt").write_text("kept")
+        (tmp_path / "gone.txt").write_text("gone")
+        with os.scandir(tmp_path) as it:
+            real_entries = {e.name: e for e in it}
+        vanished = _VanishingDirEntry(
+            real_entries["gone.txt"],
+            OSError(error_code, os.strerror(error_code), str(tmp_path / "gone.txt")),
+        )
+        monkeypatch.setattr(os, "scandir", lambda path: _StaticScandir([real_entries["kept.txt"], vanished]))
+
+        with caplog.at_level(logging.DEBUG, logger="deerflow.uploads.manager"):
+            result = list_files_in_dir(tmp_path)
+
+        assert result["count"] == 1
+        assert result["files"][0]["filename"] == "kept.txt"
+        assert any("vanished mid-scan" in record.message and str(tmp_path / "gone.txt") in record.message for record in caplog.records)
+
+    def test_stat_permission_error_still_propagates(self, tmp_path, monkeypatch):
+        (tmp_path / "locked.txt").write_text("locked")
+        with os.scandir(tmp_path) as it:
+            real_entry = next(iter(it))
+        locked = _VanishingDirEntry(real_entry, PermissionError(errno.EACCES, "Permission denied", str(tmp_path / "locked.txt")))
+        monkeypatch.setattr(os, "scandir", lambda path: _StaticScandir([locked]))
+
+        with pytest.raises(PermissionError):
+            list_files_in_dir(tmp_path)
+
 
 # ---------------------------------------------------------------------------
 # cleanup_stale_upload_staging_files