Fix: move files per document during scheduled workflow runs

check_scheduled_workflows runs the workflow once per matched document inside the task-wide file move scope, so every rename it triggered was deferred to the end of the task. A long run killed by the Celery time limit would then discard all of the renames.

The task now flushes the pending file moves after each document's workflow run and websocket notification, the same way bulk_update_documents does, so renames land document by document while the scope stays open.

src/documents/tasks.py

@@ -574,6 +574,8 @@ def check_scheduled_workflows() -> None:
 
     Once a document satisfies this condition, and recurring/non-recurring constraints are met, the workflow is run.
     """
+    from documents.file_moves import flush_pending_file_moves
+
     scheduled_workflows = get_workflows_for_trigger(
         WorkflowTrigger.WorkflowTriggerType.SCHEDULED,
     )
@@ -701,6 +703,9 @@ def check_scheduled_workflows() -> None:
                             sender=None,
                             document=document,
                         )
+                        # Move this document's files now rather than at the end
+                        # of the task, so a long run makes steady progress.
+                        flush_pending_file_moves()
 
 
 def update_document_parent_tags(tag: Tag, new_parent: Tag) -> None:

src/documents/tests/test_file_moves.py

@@ -25,7 +25,11 @@ from documents.file_moves import move_files_for_document
 from documents.file_moves import request_file_move
 from documents.models import CustomField
 from documents.models import Document
+from documents.models import Workflow
+from documents.models import WorkflowAction
+from documents.models import WorkflowTrigger
 from documents.tasks import bulk_update_documents
+from documents.tasks import check_scheduled_workflows
 from documents.tasks import empty_trash
 from paperless.celery import app as celery_app
 from paperless.middleware import FileMoveMiddleware
@@ -838,6 +842,55 @@ class TestCeleryTaskScope:
         assert (paperless_dirs.originals_dir / "first-renamed.pdf").is_file()
         assert (paperless_dirs.originals_dir / "second-renamed.pdf").is_file()
 
+    def test_scheduled_workflows_move_each_document(
+        self,
+        document: Document,
+        paperless_dirs: PaperlessDirs,
+        mocker: MockerFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - Two documents matched by a scheduled workflow, where running the
+              workflow renames the document
+        WHEN:
+            - check_scheduled_workflows runs
+        THEN:
+            - The first document's file is already moved when the workflow
+              runs for the second, not only when the task ends
+        """
+        (paperless_dirs.originals_dir / "second.pdf").write_bytes(CONTENT)
+        second = DocumentFactory(
+            title="second",
+            filename="second.pdf",
+            mime_type="application/pdf",
+            checksum=hashlib.sha256(CONTENT).hexdigest(),
+        )
+        trigger = WorkflowTrigger.objects.create(
+            type=WorkflowTrigger.WorkflowTriggerType.SCHEDULED,
+            schedule_offset_days=0,
+            schedule_date_field=WorkflowTrigger.ScheduleDateField.ADDED,
+        )
+        workflow = Workflow.objects.create(name="scheduled", order=0)
+        workflow.triggers.add(trigger)
+        workflow.actions.add(WorkflowAction.objects.create(assign_title="unused"))
+
+        seen: dict[int, bool] = {}
+
+        def rename(*, document: Document, **kwargs: object) -> None:
+            seen[document.pk] = (paperless_dirs.originals_dir / "first.pdf").is_file()
+            document.title = "first" if document.pk == first_pk else "other"
+            document.save()
+
+        first_pk = min(document.pk, second.pk)
+        mocker.patch("documents.tasks.run_workflows", side_effect=rename)
+        mocker.patch("documents.tasks.send_websocket_document_updated")
+
+        check_scheduled_workflows.apply()
+
+        second_pk = max(document.pk, second.pk)
+        assert seen[first_pk] is False
+        assert seen[second_pk] is True
+
 
 @pytest.mark.django_db
 class TestEmptyTrashLockOrdering: