diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 67780e13e5..57e96279f0 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -1777,6 +1777,7 @@ class DeerFlowClient: """ validate_thread_id(thread_id) from deerflow.uploads.companions import invalidate_overwritten_upload, register_companion + from deerflow.utils.atomic_write import cleanup_failed_atomic_write from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown # Validate all files upfront to avoid partial uploads. @@ -1824,7 +1825,7 @@ class DeerFlowClient: conversion_source = None if src_path.suffix.lower() in CONVERTIBLE_EXTENSIONS: if conversion_sources is None: - conversion_sources = tempfile.TemporaryDirectory(prefix="deerflow-upload-source-") + conversion_sources = tempfile.TemporaryDirectory(prefix="deerflow-upload-source-", ignore_cleanup_errors=True) conversion_source = Path(conversion_sources.name) / dest_name if dest_name in existing_names: if conversion_source is not None: @@ -1836,7 +1837,9 @@ class DeerFlowClient: dest_name = dest.name except UnsafeUploadPathError: if conversion_source is not None: - conversion_source.unlink(missing_ok=True) + # Skipping one unsafe file must survive a failed discard, + # or a locked temp turns the skip into a batch failure. + cleanup_failed_atomic_write(conversion_source) logger.warning("Skipping upload with unsafe destination: %s", dest_name) skipped_files.append(dest_name) continue @@ -1905,12 +1908,16 @@ class DeerFlowClient: seen_names.discard(unique_md_name) if conversion_source is not None: - conversion_source.unlink(missing_ok=True) + # The upload already succeeded; a capture we cannot discard + # must not replace that result or drop the later files. + cleanup_failed_atomic_write(conversion_source) uploaded_files.append(info) finally: if conversion_pool is not None: conversion_pool.shutdown(wait=True) if conversion_sources is not None: + # ignore_cleanup_errors keeps a locked capture from turning a + # decided batch outcome into a cleanup error. conversion_sources.cleanup() message = f"Successfully uploaded {len(uploaded_files)} file(s)" diff --git a/backend/tests/test_client_upload_unsafe_skip.py b/backend/tests/test_client_upload_unsafe_skip.py new file mode 100644 index 0000000000..305fc11d20 --- /dev/null +++ b/backend/tests/test_client_upload_unsafe_skip.py @@ -0,0 +1,109 @@ +"""An unsafe upload must stay a skip even when its capture cannot be discarded. + +``upload_files`` rejects a destination that resolves outside the uploads +directory and moves on to the next file. On Windows the staged conversion +capture can still be locked at that point, so an unguarded discard turned the +skip into a ``PermissionError`` for the whole batch. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from deerflow.client import DeerFlowClient +from deerflow.uploads.manager import UnsafeUploadPathError + + +@pytest.fixture +def upload_case(monkeypatch, tmp_path): + import deerflow.client as client_module + + uploads = tmp_path / "thread/user-data/uploads" + uploads.mkdir(parents=True) + monkeypatch.setattr(client_module, "ensure_uploads_dir", lambda *_args, **_kwargs: uploads) + return DeerFlowClient.__new__(DeerFlowClient), uploads, client_module + + +def test_unsafe_upload_stays_a_skip_when_discarding_its_capture_fails(upload_case, tmp_path, monkeypatch): + client, uploads, client_module = upload_case + + unsafe = tmp_path / "unsafe.pdf" + unsafe.write_bytes(b"%PDF unsafe") + keep = tmp_path / "keep.txt" + keep.write_bytes(b"keep me") + + real_publish = client_module.publish_upload_copy + + def publish(uploads_dir, dest_name, src_path, **_kwargs): + if Path(src_path).name == unsafe.name: + raise UnsafeUploadPathError("destination escapes the uploads directory") + return real_publish(uploads_dir, dest_name, src_path, **_kwargs) + + monkeypatch.setattr(client_module, "publish_upload_copy", publish) + + real_unlink = Path.unlink + + def unlink(self, *args, **kwargs): + if "deerflow-upload-source-" in str(self): + raise PermissionError("capture is still open") + return real_unlink(self, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", unlink) + + result = client.upload_files("skip-thread", [unsafe, keep]) + + assert result["skipped_files"] == [unsafe.name] + assert [item["filename"] for item in result["files"]] == [keep.name] + assert (uploads / keep.name).read_bytes() == b"keep me" + + +def test_successful_upload_survives_a_capture_it_cannot_discard(upload_case, tmp_path, monkeypatch): + """A locked capture must not replace a completed upload or drop later files.""" + client, uploads, _client_module = upload_case + + convertible = tmp_path / "report.pdf" + convertible.write_bytes(b"%PDF report") + keep = tmp_path / "keep.txt" + keep.write_bytes(b"keep me") + + real_unlink = Path.unlink + + def unlink(self, *args, **kwargs): + if "deerflow-upload-source-" in str(self): + raise PermissionError("capture is still open") + return real_unlink(self, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", unlink) + + result = client.upload_files("successful-thread", [convertible, keep]) + + assert result["skipped_files"] == [] + assert [item["filename"] for item in result["files"]] == [convertible.name, keep.name] + assert (uploads / keep.name).read_bytes() == b"keep me" + + +def test_capture_directory_cleanup_failure_does_not_fail_the_batch(upload_case, tmp_path, monkeypatch): + """The ``finally`` rmtree must not replace an outcome the batch already decided.""" + import tempfile + + client, _uploads, _client_module = upload_case + + convertible = tmp_path / "report.pdf" + convertible.write_bytes(b"%PDF report") + + real_rmtree = tempfile.TemporaryDirectory._rmtree.__func__ + + def rmtree(cls, name, ignore_errors=False, repeated=False): + # A locked capture: exactly what CPython's onexc re-raises when the + # directory's cleanup is not told to ignore errors. + if "deerflow-upload-source-" in str(name) and not ignore_errors: + raise PermissionError("capture directory is still open") + return real_rmtree(cls, name, ignore_errors=ignore_errors, repeated=repeated) + + monkeypatch.setattr(tempfile.TemporaryDirectory, "_rmtree", classmethod(rmtree)) + + result = client.upload_files("cleanup-thread", [convertible]) + + assert [item["filename"] for item in result["files"]] == [convertible.name]