Apply to the rest of the operations

src/documents/bulk_edit.py

@@ -39,6 +39,8 @@ from documents.models import Tag
 from documents.permissions import PermissionsSpec
 from documents.permissions import set_permissions_for_objects
 from documents.plugins.helpers import DocumentsStatusManager
+from documents.task_tracking import queue_consumption
+from documents.task_tracking import queue_processing_tasks
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
 from documents.tasks import remove_document_from_index
@@ -536,6 +538,7 @@ def rotate(
     source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
     trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI,
+    parent_task_id: str | None = None,
 ) -> Literal["OK"]:
     logger.info(
         f"Attempting to rotate {len(doc_ids)} documents by {degrees} degrees.",
@@ -573,16 +576,15 @@ def rotate(
             if user is not None:
                 overrides.actor_id = user.id
 
-            consume_file.apply_async(
-                kwargs={
-                    "input_doc": ConsumableDocument(
-                        source=DocumentSource.ConsumeFolder,
-                        original_file=filepath,
-                        root_document_id=pair.root_doc.id,
-                    ),
-                    "overrides": overrides,
-                },
-                headers={"trigger_source": trigger_source},
+            queue_consumption(
+                input_doc=ConsumableDocument(
+                    source=DocumentSource.ConsumeFolder,
+                    original_file=filepath,
+                    root_document_id=pair.root_doc.id,
+                ),
+                overrides=overrides,
+                trigger_source=trigger_source,
+                parent_task_id=parent_task_id,
             )
             logger.info(
                 f"Queued new rotated version for document {pair.root_doc.id} by {degrees} degrees",
@@ -603,13 +605,15 @@ def merge(
     source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
     trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI,
-) -> Literal["OK"]:
+    parent_task_id: str | None = None,
+) -> Literal["OK"] | dict:
     logger.info(
         f"Attempting to merge {len(doc_ids)} documents into a single document.",
     )
     qs = Document.objects.select_related("root_document").filter(id__in=doc_ids)
     docs_by_id = {doc.id: doc for doc in qs}
     affected_docs: list[int] = []
+    failed_ids: list[int] = []
     handoff_asn: int | None = None
     with pdf_ops.PdfMerger() as merger:
         # use doc_ids to preserve order
@@ -635,11 +639,14 @@ def merge(
                 if handoff_asn is None and doc.archive_serial_number is not None:
                     handoff_asn = doc.archive_serial_number
             except Exception as e:
+                failed_ids.append(doc.id)
                 logger.exception(
                     f"Error merging document {doc.id}, it will not be included in the merge: {e}",
                 )
         if len(affected_docs) == 0:
             logger.warning("No documents were merged")
+            if parent_task_id is not None:
+                raise ValueError("None of the selected documents could be merged.")
             return "OK"
 
         filepath = (
@@ -682,6 +689,23 @@ def merge(
         overrides=overrides,
     ).set(headers={"trigger_source": trigger_source})
 
+    if parent_task_id is not None:
+        backup = (
+            release_archive_serial_numbers(affected_docs) if delete_originals else {}
+        )
+        queue_processing_tasks(
+            parent_task_id,
+            [consume_task],
+            delete_ids=affected_docs if delete_originals else None,
+            asn_backup=backup,
+        )
+        if failed_ids:
+            return {
+                "failed_document_ids": failed_ids,
+                "error_message": f"Documents {failed_ids} were excluded from the merge.",
+            }
+        return "OK"
+
     if delete_originals:
         backup = release_archive_serial_numbers(affected_docs)
         logger.info(
@@ -826,6 +850,7 @@ def split(
     source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
     trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI,
+    parent_task_id: str | None = None,
 ) -> Literal["OK"]:
     logger.info(
         f"Attempting to split document {doc_ids[0]} into {len(pages)} documents",
@@ -869,7 +894,17 @@ def split(
                 ).set(headers={"trigger_source": trigger_source}),
             )
 
-        if delete_originals:
+        if parent_task_id is not None:
+            backup = (
+                release_archive_serial_numbers([doc.id]) if delete_originals else {}
+            )
+            queue_processing_tasks(
+                parent_task_id,
+                consume_tasks,
+                delete_ids=[doc.id] if delete_originals else None,
+                asn_backup=backup,
+            )
+        elif delete_originals:
             backup = release_archive_serial_numbers([doc.id])
             logger.info(
                 "Queueing removal of original document after consumption of the split documents",
@@ -901,6 +936,7 @@ def delete_pages(
     source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
     trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI,
+    parent_task_id: str | None = None,
 ) -> Literal["OK"]:
     logger.info(
         f"Attempting to delete pages {pages} from {len(doc_ids)} documents",
@@ -919,16 +955,15 @@ def delete_pages(
         overrides = DocumentMetadataOverrides().from_document(pair.root_doc)
         if user is not None:
             overrides.actor_id = user.id
-        consume_file.apply_async(
-            kwargs={
-                "input_doc": ConsumableDocument(
-                    source=DocumentSource.ConsumeFolder,
-                    original_file=filepath,
-                    root_document_id=pair.root_doc.id,
-                ),
-                "overrides": overrides,
-            },
-            headers={"trigger_source": trigger_source},
+        queue_consumption(
+            input_doc=ConsumableDocument(
+                source=DocumentSource.ConsumeFolder,
+                original_file=filepath,
+                root_document_id=pair.root_doc.id,
+            ),
+            overrides=overrides,
+            trigger_source=trigger_source,
+            parent_task_id=parent_task_id,
         )
         logger.info(
             f"Queued new version for document {pair.root_doc.id} after deleting pages {pages}",
@@ -950,6 +985,7 @@ def edit_pdf(
     source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
     trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI,
+    parent_task_id: str | None = None,
 ) -> Literal["OK"]:
     """
     Operations is a list of dictionaries describing the final PDF pages.
@@ -994,16 +1030,15 @@ def edit_pdf(
             if user is not None:
                 overrides.owner_id = user.id
                 overrides.actor_id = user.id
-            consume_file.apply_async(
-                kwargs={
-                    "input_doc": ConsumableDocument(
-                        source=DocumentSource.ConsumeFolder,
-                        original_file=filepath,
-                        root_document_id=pair.root_doc.id,
-                    ),
-                    "overrides": overrides,
-                },
-                headers={"trigger_source": trigger_source},
+            queue_consumption(
+                input_doc=ConsumableDocument(
+                    source=DocumentSource.ConsumeFolder,
+                    original_file=filepath,
+                    root_document_id=pair.root_doc.id,
+                ),
+                overrides=overrides,
+                trigger_source=trigger_source,
+                parent_task_id=parent_task_id,
             )
         else:
             version_filepaths = pdf_ops.build_pdfs(
@@ -1043,7 +1078,17 @@ def edit_pdf(
                     ).set(headers={"trigger_source": trigger_source}),
                 )
 
-            if delete_original:
+            if parent_task_id is not None:
+                backup = (
+                    release_archive_serial_numbers([doc.id]) if delete_original else {}
+                )
+                queue_processing_tasks(
+                    parent_task_id,
+                    consume_tasks,
+                    delete_ids=[doc.id] if delete_original else None,
+                    asn_backup=backup,
+                )
+            elif delete_original:
                 backup = release_archive_serial_numbers([doc.id])
                 try:
                     chord(
@@ -1077,6 +1122,7 @@ def remove_password(
     source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
     trigger_source: PaperlessTask.TriggerSource = PaperlessTask.TriggerSource.WEB_UI,
+    parent_task_id: str | None = None,
     source_paths_by_id: Mapping[int, Path] | None = None,
 ) -> Literal["OK"]:
     """
@@ -1119,16 +1165,15 @@ def remove_password(
                 if user is not None:
                     overrides.owner_id = user.id
                     overrides.actor_id = user.id
-                consume_file.apply_async(
-                    kwargs={
-                        "input_doc": ConsumableDocument(
-                            source=DocumentSource.ConsumeFolder,
-                            original_file=filepath,
-                            root_document_id=pair.root_doc.id,
-                        ),
-                        "overrides": overrides,
-                    },
-                    headers={"trigger_source": trigger_source},
+                queue_consumption(
+                    input_doc=ConsumableDocument(
+                        source=DocumentSource.ConsumeFolder,
+                        original_file=filepath,
+                        root_document_id=pair.root_doc.id,
+                    ),
+                    overrides=overrides,
+                    trigger_source=trigger_source,
+                    parent_task_id=parent_task_id,
                 )
             else:
                 consume_tasks = []
@@ -1151,7 +1196,19 @@ def remove_password(
                     ).set(headers={"trigger_source": trigger_source}),
                 )
 
-                if delete_original:
+                if parent_task_id is not None:
+                    backup = (
+                        release_archive_serial_numbers([doc.id])
+                        if delete_original
+                        else {}
+                    )
+                    queue_processing_tasks(
+                        parent_task_id,
+                        consume_tasks,
+                        delete_ids=[doc.id] if delete_original else None,
+                        asn_backup=backup,
+                    )
+                elif delete_original:
                     chord(
                         header=consume_tasks,
                         body=delete.si([doc.id]),

src/documents/bulk_operations.py

@@ -115,7 +115,10 @@ def execute_bulk_operation(
 
     if operation not in MODIFIED_FIELDS:
         execution_parameters = parameters.copy()
-        if operation == "reprocess" and self.request.id is not None:
+        if (
+            operation in TRIGGER_SOURCE_OPERATIONS | {"reprocess"}
+            and self.request.id is not None
+        ):
             execution_parameters["parent_task_id"] = self.request.id
         if operation in USER_OPERATIONS:
             execution_parameters["user"] = user
@@ -128,7 +131,7 @@ def execute_bulk_operation(
                 if settings.AUDIT_LOG_ENABLED and modified_field
                 else {}
             )
-            method(document_ids, **execution_parameters)
+            result = method(document_ids, **execution_parameters)
             if settings.AUDIT_LOG_ENABLED and modified_field:
                 from auditlog.models import LogEntry
 
@@ -143,7 +146,11 @@ def execute_bulk_operation(
                         actor=user,
                         additional_data={"reason": f"Bulk edit: {operation}"},
                     )
-        return {"operation": operation, "document_count": len(document_ids)}
+        return {
+            "operation": operation,
+            "document_count": len(document_ids),
+            **(result if isinstance(result, dict) else {}),
+        }
 
     affected_ids: list[int] = []
     modified_field = MODIFIED_FIELDS[operation]

src/documents/signals/handlers.py

@@ -1079,6 +1079,7 @@ TRACKED_TASKS: dict[str, PaperlessTask.TaskType] = {
     "documents.tasks.update_document_content_maybe_archive_file": PaperlessTask.TaskType.REPROCESS_DOCUMENT,
     "documents.tasks.build_share_link_bundle": PaperlessTask.TaskType.BUILD_SHARE_LINK,
     "documents.bulk_edit.delete": PaperlessTask.TaskType.BULK_DELETE,
+    "documents.task_tracking.finish_file_operation": PaperlessTask.TaskType.BULK_DELETE,
     "documents.tasks.apply_ai_suggestions": PaperlessTask.TaskType.APPLY_AI_SUGGESTIONS,
     BULK_OPERATION_TASK_NAME: PaperlessTask.TaskType.BULK_UPDATE,
 }
@@ -1127,6 +1128,8 @@ def _extract_input_data(
                 data["overrides"] = override_dict
         return data
 
+    if task_type == PaperlessTask.TaskType.REPROCESS_DOCUMENT:
+        return {"document_id": task_kwargs["document_id"]}
     if task_type == PaperlessTask.TaskType.MAIL_FETCH:
         account_ids = task_kwargs.get("account_ids")
         if account_ids is not None:
@@ -1214,6 +1217,19 @@ def before_task_publish_handler(
             input_data = _extract_input_data(task_type, task_kwargs)
             owner_id = _extract_owner_id(task_type, task_kwargs)
         trigger_source = _determine_trigger_source(headers)
+        # Consumption can fan out again (e.g. barcode splitting). Keep those
+        # descendants attached so their caller cannot finish before them.
+        parent = None
+        if task_type == PaperlessTask.TaskType.CONSUME_FILE and headers.get(
+            "parent_id",
+        ):
+            parent = PaperlessTask.objects.filter(
+                task_id=headers["parent_id"],
+                parent__isnull=False,
+            ).first()
+            if parent is not None:
+                owner_id = parent.owner_id
+                trigger_source = parent.trigger_source
 
         # A retried task is republished with the same task_id, so this fires
         # again for it; get_or_create keeps the original PENDING record
@@ -1226,6 +1242,7 @@ def before_task_publish_handler(
                 "status": PaperlessTask.Status.PENDING,
                 "input_data": input_data,
                 "owner_id": owner_id,
+                "parent": parent,
             },
         )
     except Exception:  # pragma: no cover
@@ -1337,6 +1354,19 @@ def task_postrun_handler(
                 task_instance.status = PaperlessTask.Status.FAILURE
                 changed_fields.append("status")
 
+        if (
+            task_instance.parent_id is not None
+            and task_instance.task_type == PaperlessTask.TaskType.CONSUME_FILE
+            and (not isinstance(retval, dict) or "document_id" not in retval)
+            and not (isinstance(retval, dict) and "duplicate_of" in retval)
+        ):
+            task_instance.status = PaperlessTask.Status.FAILURE
+            task_instance.result_data = {
+                **(retval if isinstance(retval, dict) else {}),
+                "error_message": "Consumption stopped without creating a document.",
+            }
+            changed_fields.extend(["status", "result_data"])
+
         task_instance.save(update_fields=changed_fields)
         update_parent(task_instance)
         if task and task.name == BULK_OPERATION_TASK_NAME:

src/documents/task_tracking.py

@@ -1,12 +1,41 @@
 """Completion tracking for operations which dispatch document processing tasks."""
 
-import uuid
+from __future__ import annotations
 
+import uuid
+from typing import TYPE_CHECKING
+
+from celery import shared_task
 from django.db import transaction
 from django.utils import timezone
 
 from documents.models import PaperlessTask
 
+if TYPE_CHECKING:
+    from celery.canvas import Signature
+
+    from documents.data_models import ConsumableDocument
+    from documents.data_models import DocumentMetadataOverrides
+
+
+def queue_consumption(
+    *,
+    input_doc: ConsumableDocument,
+    overrides: DocumentMetadataOverrides,
+    trigger_source: str,
+    parent_task_id: str | None,
+) -> None:
+    from documents.tasks import consume_file
+
+    kwargs = {"input_doc": input_doc, "overrides": overrides}
+    if parent_task_id is None:
+        consume_file.apply_async(
+            kwargs=kwargs,
+            headers={"trigger_source": trigger_source},
+        )
+    else:
+        queue_processing_tasks(parent_task_id, [consume_file.s(**kwargs)])
+
 
 def queue_reprocessing_tasks(
     parent_task_id: str,
@@ -16,28 +45,63 @@ def queue_reprocessing_tasks(
 ) -> None:
     from documents.tasks import update_document_content_maybe_archive_file
 
+    queue_processing_tasks(
+        parent_task_id,
+        [
+            update_document_content_maybe_archive_file.s(
+                document_id=document_id,
+                remote_ocr=remote_ocr,
+            )
+            for document_id in document_ids
+        ],
+    )
+
+
+def queue_processing_tasks(
+    parent_task_id: str,
+    signatures: list[Signature],
+    *,
+    delete_ids: list[int] | None = None,
+    asn_backup: dict | None = None,
+) -> None:
+    from documents.signals.handlers import TRACKED_TASKS
+    from documents.signals.handlers import _extract_input_data
+
     parent = PaperlessTask.objects.get(task_id=parent_task_id)
     children = [
         PaperlessTask(
             task_id=str(uuid.uuid4()),
-            task_type=PaperlessTask.TaskType.REPROCESS_DOCUMENT,
+            task_type=TRACKED_TASKS[signature.task],
             trigger_source=parent.trigger_source,
             owner_id=parent.owner_id,
             parent=parent,
-            input_data={"document_id": document_id},
+            input_data=_extract_input_data(
+                TRACKED_TASKS[signature.task],
+                signature.kwargs,
+            ),
         )
-        for document_id in document_ids
+        for signature in signatures
     ]
-    # Register the complete set before any can finish
-    PaperlessTask.objects.bulk_create(children)
-    for index, child in enumerate(children):
-        try:
-            update_document_content_maybe_archive_file.apply_async(
-                task_id=child.task_id,
-                kwargs={
-                    "document_id": child.input_data["document_id"],
-                    "remote_ocr": remote_ocr,
+    # Dependencies and their final step must be registered together.
+    with transaction.atomic():
+        PaperlessTask.objects.bulk_create(children)
+        if delete_ids:
+            PaperlessTask.objects.create(
+                task_id=str(uuid.uuid4()),
+                task_type=PaperlessTask.TaskType.BULK_DELETE,
+                owner_id=parent.owner_id,
+                trigger_source=parent.trigger_source,
+                parent=parent,
+                input_data={
+                    "dependencies": [child.pk for child in children],
+                    "delete_ids": delete_ids,
+                    "asn_backup": asn_backup or {},
                 },
+            )
+    for index, (signature, child) in enumerate(zip(signatures, children, strict=True)):
+        try:
+            signature.apply_async(
+                task_id=child.task_id,
                 headers={"trigger_source": parent.trigger_source},
             )
         except Exception:
@@ -46,11 +110,75 @@ def queue_reprocessing_tasks(
             ).update(
                 status=PaperlessTask.Status.FAILURE,
                 date_done=timezone.now(),
-                result_data={"error_message": "Could not publish reprocessing task."},
+                result_data={"error_message": "Could not publish processing task."},
             )
             raise
 
 
+@shared_task
+def finish_file_operation(
+    dependencies: list[int],
+    delete_ids: list[int],
+    asn_backup: dict,
+) -> dict[str, str]:
+    from documents.bulk_edit import delete
+    from documents.bulk_edit import restore_archive_serial_numbers
+
+    statuses = list(
+        PaperlessTask.objects.filter(pk__in=dependencies).values_list(
+            "status",
+            flat=True,
+        ),
+    )
+    if len(statuses) != len(dependencies):
+        raise ValueError(
+            "Processing task records are missing; originals were preserved.",
+        )
+    if any(status != PaperlessTask.Status.SUCCESS for status in statuses):
+        restore_archive_serial_numbers(asn_backup)
+        return {"reason": "Originals preserved; archive serial numbers restored."}
+    delete(delete_ids)
+    return {"reason": "Originals moved to trash."}
+
+
+def _publish_finalizer(task: PaperlessTask) -> None:
+    try:
+        finish_file_operation.apply_async(
+            task_id=task.task_id,
+            kwargs=task.input_data,
+            headers={"trigger_source": task.trigger_source},
+        )
+    except Exception as error:
+        from documents.signals.handlers import task_failure_handler
+
+        task_failure_handler(
+            sender=finish_file_operation,
+            task_id=task.task_id,
+            exception=error,
+        )
+
+
+def _schedule_finalizers(parent: PaperlessTask) -> None:
+    for task in parent.children.filter(
+        task_type=PaperlessTask.TaskType.BULK_DELETE,
+        status=PaperlessTask.Status.PENDING,
+    ):
+        dependencies = task.input_data.get("dependencies")
+        if dependencies is None or task.result_data:
+            continue
+        if (
+            PaperlessTask.objects.filter(pk__in=dependencies)
+            .exclude(
+                status__in=PaperlessTask.COMPLETE_STATUSES,
+            )
+            .exists()
+        ):
+            continue
+        task.result_data = {"scheduled": True}
+        task.save(update_fields=["result_data"])
+        transaction.on_commit(lambda task=task: _publish_finalizer(task))
+
+
 def finish_dispatch(task: PaperlessTask, result_data: dict) -> None:
     """Record the scheduling outcome without completing an active parent."""
     with transaction.atomic():
@@ -74,6 +202,7 @@ def update_parent(task: PaperlessTask) -> None:
 def _finalize_parent(parent: PaperlessTask) -> None:
     if parent.is_complete or not parent.dispatched:
         return
+    _schedule_finalizers(parent)
     statuses = list(parent.children.values_list("status", flat=True))
     if any(status not in PaperlessTask.COMPLETE_STATUSES for status in statuses):
         return
@@ -83,7 +212,7 @@ def _finalize_parent(parent: PaperlessTask) -> None:
     result_data = parent.result_data or {}
     if failed and not result_data.get("error_message"):
         result_data["error_message"] = (
-            f"{succeeded} document(s) reprocessed; {failed} failed or were revoked. "
+            f"{succeeded} child task(s) succeeded; {failed} failed or were revoked. "
             "See child tasks for details."
         )
     parent.status = (
@@ -108,4 +237,6 @@ def _finalize_parent(parent: PaperlessTask) -> None:
     parent.save()
     from documents.signals.handlers import _send_bulk_operation_status
 
-    transaction.on_commit(lambda: _send_bulk_operation_status(parent))
+    if "operation" in parent.input_data:
+        transaction.on_commit(lambda: _send_bulk_operation_status(parent))
+    update_parent(parent)

src/documents/tests/test_task_tracking.py

@@ -1,3 +1,4 @@
+import shutil
 import uuid
 from unittest import mock
 
@@ -33,6 +34,20 @@ def operation(paperless_dirs):
     return parent, kwargs
 
 
+@pytest.fixture
+def pdf_documents(paperless_dirs, tmp_path):
+    from documents.models import Document
+    from paperless_testing.samples import SIMPLE_DIGITAL_PDF
+
+    documents = DocumentFactory.create_batch(2, mime_type="application/pdf")
+    for document in documents:
+        source = tmp_path / f"{document.pk}.pdf"
+        shutil.copy(SIMPLE_DIGITAL_PDF, source)
+        Document.objects.filter(pk=document.pk).update(filename=str(source))
+        document.refresh_from_db()
+    return documents
+
+
 def complete(child, status="SUCCESS"):
     if status == "FAILURE":
         task_failure_handler(
@@ -72,9 +87,7 @@ def test_parent_waits_for_all_children(operation, outcome):
         assert publish.call_count == 3
         assert all(child.owner_id == parent.owner_id for child in children)
         assert all(child.trigger_source == parent.trigger_source for child in children)
-        assert all(
-            call.kwargs["kwargs"]["remote_ocr"] for call in publish.call_args_list
-        )
+        assert all(call.args[1]["remote_ocr"] for call in publish.call_args_list)
         complete(children[0], outcome)
         complete(children[1])
         parent.refresh_from_db()
@@ -99,7 +112,7 @@ def test_parent_waits_for_all_children(operation, outcome):
 def test_fast_children_wait_for_dispatch_to_finish(operation, outcome):
     parent, kwargs = operation
 
-    def publish(*, task_id, **kwargs):
+    def publish(*args, task_id, **kwargs):
         # Every child is registered before the first publication.
         assert parent.children.count() == 3
         complete(PaperlessTask.objects.get(task_id=task_id), outcome)
@@ -174,3 +187,357 @@ def test_client_cannot_supply_parent_task_id(operation):
         )
     assert response.status_code == 400
     publish.assert_not_called()
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize(
+    "name,parameters",
+    [
+        ("rotate", {"degrees": 90}),
+        ("merge", {}),
+        ("split", {"pages": [[1]]}),
+        ("delete_pages", {"pages": [1]}),
+        ("edit_pdf", {"operations": [{"page": 1}], "update_document": True}),
+        ("edit_pdf", {"operations": [{"page": 1}]}),
+        ("remove_password", {"password": "secret", "update_document": True}),
+        ("remove_password", {"password": "secret"}),
+    ],
+)
+def test_file_operations_wait_for_consumption(
+    operation,
+    pdf_documents,
+    name,
+    parameters,
+):
+    from documents.tasks import consume_file
+
+    parent, kwargs = operation
+    documents = pdf_documents
+    kwargs.update(
+        operation=name,
+        document_ids=[doc.pk for doc in documents]
+        if name == "merge"
+        else [documents[0].pk],
+        parameters=parameters,
+    )
+    parent.input_data = {
+        "operation": name,
+        "document_count": len(kwargs["document_ids"]),
+    }
+    parent.save()
+    with (
+        mock.patch("documents.tasks.consume_file.apply_async") as publish,
+        mock.patch("documents.bulk_edit.pdf_ops.needs_decrypt", return_value=True),
+        mock.patch(
+            "documents.bulk_edit.pdf_ops.decrypt_pdf",
+            return_value=documents[0].source_path,
+        ),
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.STARTED
+    child = parent.children.get()
+    assert child.task_type == PaperlessTask.TaskType.CONSUME_FILE
+    publish.assert_called_once()
+    task_postrun_handler(
+        task=consume_file,
+        task_id=child.task_id,
+        state="SUCCESS",
+        retval={"document_id": 100},
+    )
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.SUCCESS
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize(
+    "outcome",
+    ["success", "failure", "duplicate", "stopped", "revoked"],
+)
+def test_replacement_waits_for_deletion_or_recovery(operation, pdf_documents, outcome):
+    from documents.models import Document
+    from documents.task_tracking import finish_file_operation
+    from documents.tasks import consume_file
+
+    parent, kwargs = operation
+    doc = pdf_documents[0]
+    doc.archive_serial_number = 123
+    Document.objects.filter(pk=doc.pk).update(archive_serial_number=123)
+    kwargs.update(
+        operation="edit_pdf",
+        document_ids=[doc.pk],
+        parameters={"operations": [{"page": 1}], "delete_original": True},
+    )
+    parent.input_data = {"operation": "edit_pdf", "document_count": 1}
+    parent.save()
+    with (
+        mock.patch("documents.tasks.consume_file.apply_async"),
+        mock.patch(
+            "documents.task_tracking.finish_file_operation.apply_async",
+        ) as finalize,
+        mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify,
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+        consume = parent.children.get(task_type=PaperlessTask.TaskType.CONSUME_FILE)
+        finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE)
+        finalize.assert_not_called()
+        if outcome == "failure":
+            complete(consume, "FAILURE")
+        elif outcome == "revoked":
+            complete(consume, "REVOKED")
+        else:
+            retval = (
+                {"document_id": 100}
+                if outcome == "success"
+                else (
+                    {"duplicate_of": 100}
+                    if outcome == "duplicate"
+                    else {"reason": "Stopped"}
+                )
+            )
+            task_postrun_handler(
+                task=consume_file,
+                task_id=consume.task_id,
+                state="SUCCESS",
+                retval=retval,
+            )
+        finalize.assert_called_once()
+        parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.STARTED
+        assert Document.objects.filter(pk=doc.pk).exists()
+        notify.assert_not_called()
+        finish_file_operation.apply(
+            kwargs=finalizer.input_data,
+            task_id=finalizer.task_id,
+            throw=True,
+        )
+        parent.refresh_from_db()
+        assert parent.status == ("success" if outcome == "success" else "failure")
+        notify.assert_called_once()
+        if outcome == "success":
+            assert not Document.objects.filter(pk=doc.pk).exists()
+        else:
+            doc.refresh_from_db()
+            assert not doc.is_deleted
+            assert doc.archive_serial_number == 123
+
+
+@pytest.mark.django_db(transaction=True)
+def test_consumption_descendants_keep_operation_active(operation):
+    from documents.data_models import ConsumableDocument
+    from documents.data_models import DocumentSource
+    from documents.signals.handlers import before_task_publish_handler
+    from documents.task_tracking import finish_dispatch
+    from documents.task_tracking import queue_processing_tasks
+    from documents.tasks import consume_file
+    from paperless_testing.samples import SIMPLE_DIGITAL_PDF
+
+    parent, _ = operation
+    parent.status = PaperlessTask.Status.STARTED
+    parent.save()
+    input_doc = ConsumableDocument(
+        source=DocumentSource.ConsumeFolder,
+        original_file=SIMPLE_DIGITAL_PDF,
+    )
+    with mock.patch("documents.tasks.consume_file.apply_async"):
+        queue_processing_tasks(parent.task_id, [consume_file.s(input_doc=input_doc)])
+    child = parent.children.get()
+    finish_dispatch(parent, {})
+    descendant_id = str(uuid.uuid4())
+    before_task_publish_handler(
+        headers={
+            "task": consume_file.name,
+            "id": descendant_id,
+            "parent_id": child.task_id,
+        },
+        body=((), {"input_doc": input_doc}, {}),
+    )
+    descendant = PaperlessTask.objects.get(task_id=descendant_id)
+    assert descendant.parent_id == child.pk
+    assert descendant.owner_id == parent.owner_id
+    task_postrun_handler(
+        task=consume_file,
+        task_id=child.task_id,
+        state="SUCCESS",
+        retval={"reason": "Barcode splitting complete"},
+    )
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.STARTED
+    task_postrun_handler(
+        task=consume_file,
+        task_id=descendant_id,
+        state="SUCCESS",
+        retval={"document_id": 100},
+    )
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.SUCCESS
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize("failure", ["delete", "restore", "publish"])
+def test_final_step_failures_reach_parent(operation, pdf_documents, failure):
+    from documents.task_tracking import finish_file_operation
+    from documents.tasks import consume_file
+
+    parent, kwargs = operation
+    doc = pdf_documents[0]
+    kwargs.update(
+        operation="edit_pdf",
+        document_ids=[doc.pk],
+        parameters={"operations": [{"page": 1}], "delete_original": True},
+    )
+    with (
+        mock.patch("documents.tasks.consume_file.apply_async"),
+        mock.patch(
+            "documents.task_tracking.finish_file_operation.apply_async",
+            side_effect=RuntimeError("Publish failed")
+            if failure == "publish"
+            else None,
+        ),
+        mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify,
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+        child = parent.children.get(task_type=PaperlessTask.TaskType.CONSUME_FILE)
+        if failure == "restore":
+            complete(child, "FAILURE")
+        else:
+            task_postrun_handler(
+                task=consume_file,
+                task_id=child.task_id,
+                state="SUCCESS",
+                retval={"document_id": 100},
+            )
+        finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE)
+        if failure != "publish":
+            parent.refresh_from_db()
+            assert parent.status == PaperlessTask.Status.STARTED
+            notify.assert_not_called()
+            target = (
+                "documents.bulk_edit.delete"
+                if failure == "delete"
+                else "documents.bulk_edit.restore_archive_serial_numbers"
+            )
+            with mock.patch(target, side_effect=RuntimeError("Final step failed")):
+                finish_file_operation.apply(
+                    kwargs=finalizer.input_data,
+                    task_id=finalizer.task_id,
+                    throw=False,
+                )
+        parent.refresh_from_db()
+        finalizer.refresh_from_db()
+        assert finalizer.status == PaperlessTask.Status.FAILURE
+        assert parent.status == PaperlessTask.Status.FAILURE
+        notify.assert_called_once()
+        doc.refresh_from_db()
+        assert not doc.is_deleted
+
+
+@pytest.mark.django_db(transaction=True)
+def test_partial_merge_preserves_skipped_original_and_reports_it(
+    operation,
+    pdf_documents,
+):
+    from documents.models import Document
+    from documents.task_tracking import finish_file_operation
+    from documents.tasks import consume_file
+
+    parent, kwargs = operation
+    included, excluded = pdf_documents
+    excluded.source_path.unlink()
+    kwargs.update(
+        operation="merge",
+        document_ids=[included.pk, excluded.pk],
+        parameters={"delete_originals": True},
+    )
+    parent.input_data = {"operation": "merge", "document_count": 2}
+    parent.save()
+    with (
+        mock.patch("documents.tasks.consume_file.apply_async") as consume,
+        mock.patch("documents.task_tracking.finish_file_operation.apply_async"),
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+        consume.assert_called_once()
+        parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.STARTED
+        assert parent.result_data["failed_document_ids"] == [excluded.pk]
+        child = parent.children.get(task_type=PaperlessTask.TaskType.CONSUME_FILE)
+        task_postrun_handler(
+            task=consume_file,
+            task_id=child.task_id,
+            state="SUCCESS",
+            retval={"document_id": 100},
+        )
+        finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE)
+        assert finalizer.input_data["delete_ids"] == [included.pk]
+        finish_file_operation.apply(
+            kwargs=finalizer.input_data,
+            task_id=finalizer.task_id,
+            throw=True,
+        )
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.FAILURE
+    assert parent.result_data["failed_document_ids"] == [excluded.pk]
+    assert not Document.objects.filter(pk=included.pk).exists()
+    assert Document.objects.filter(pk=excluded.pk).exists()
+
+
+@pytest.mark.django_db(transaction=True)
+def test_merge_with_no_valid_inputs_fails(operation, pdf_documents):
+    parent, kwargs = operation
+    for doc in pdf_documents:
+        doc.source_path.unlink()
+    kwargs.update(
+        operation="merge",
+        document_ids=[doc.pk for doc in pdf_documents],
+        parameters={},
+    )
+    execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=False)
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.FAILURE
+    assert not parent.children.exists()
+
+
+@pytest.mark.django_db(transaction=True)
+def test_split_waits_for_every_output_before_recovery(operation, pdf_documents):
+    from documents.task_tracking import finish_file_operation
+    from documents.tasks import consume_file
+
+    parent, kwargs = operation
+    doc = pdf_documents[0]
+    kwargs.update(
+        operation="split",
+        document_ids=[doc.pk],
+        parameters={"pages": [[1], [1]], "delete_originals": True},
+    )
+    with (
+        mock.patch("documents.tasks.consume_file.apply_async"),
+        mock.patch(
+            "documents.task_tracking.finish_file_operation.apply_async",
+        ) as finalize,
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+        children = list(
+            parent.children.filter(task_type=PaperlessTask.TaskType.CONSUME_FILE),
+        )
+        assert len(children) == 2
+        complete(children[0], "FAILURE")
+        finalize.assert_not_called()
+        parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.STARTED
+        task_postrun_handler(
+            task=consume_file,
+            task_id=children[1].task_id,
+            state="SUCCESS",
+            retval={"document_id": 100},
+        )
+        finalize.assert_called_once()
+        finalizer = parent.children.get(task_type=PaperlessTask.TaskType.BULK_DELETE)
+        finish_file_operation.apply(
+            kwargs=finalizer.input_data,
+            task_id=finalizer.task_id,
+            throw=True,
+        )
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.FAILURE
+    doc.refresh_from_db()
+    assert not doc.is_deleted