Update api.md

docs/api.md

@@ -285,14 +285,14 @@ The API supports various bulk-editing operations which are executed asynchronous
 
 ### Documents
 
-For bulk operations on documents, use the endpoint `/api/documents/bulk_edit/` which accepts
+Document bulk operations return `{"result":"OK","task_id":"…"}` when queued.
-a json payload of the format:
+Check `/api/tasks/?task_id={uuid}` for the outcome, including any follow-up processing,
+deletion, or recovery. Failed operations may leave partial changes applied.
 
-Document bulk operations, including the dedicated document action endpoints, return
+For operations with child tasks, `result_data` includes `succeeded_count` and
-`{"result":"OK","task_id":"…"}` when queued. Use `/api/tasks/?task_id={uuid}`
+`failed_count` (task counts). Merges report skipped inputs in `failed_document_ids`.
-to check the task outcome. Actions that create or reprocess files may start further
+
-consumption or reprocessing tasks and completion of the bulk task means those tasks
+The `/api/documents/bulk_edit/` endpoint accepts:
-were scheduled, not that their file processing has finished.
 
 ```json
 {

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
@@ -432,7 +434,12 @@ def delete(doc_ids: list[int]) -> Literal["OK"]:
     return "OK"
 
 
-def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
+def reprocess(
+    doc_ids: list[int],
+    *,
+    remote_ocr: bool = False,
+    parent_task_id: str | None = None,
+) -> Literal["OK"]:
     """
     Re-run parsing for the given documents.
 
@@ -457,6 +464,16 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
         .values_list("source_id", flat=True)
         .distinct()
     )
+    if parent_task_id is not None:
+        from documents.task_tracking import queue_reprocessing_tasks
+
+        queue_reprocessing_tasks(
+            parent_task_id,
+            list(source_ids),
+            remote_ocr=remote_ocr,
+        )
+        return "OK"
+
     for source_id in source_ids:
         update_document_content_maybe_archive_file.apply_async(
             kwargs={"document_id": source_id, "remote_ocr": remote_ocr},
@@ -521,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.",
@@ -558,16 +576,15 @@ def rotate(
             if user is not None:
                 overrides.actor_id = user.id
 
-            consume_file.apply_async(
+            queue_consumption(
-                kwargs={
+                input_doc=ConsumableDocument(
-                    "input_doc": ConsumableDocument(
+                    source=DocumentSource.ConsumeFolder,
-                        source=DocumentSource.ConsumeFolder,
+                    original_file=filepath,
-                        original_file=filepath,
+                    root_document_id=pair.root_doc.id,
-                        root_document_id=pair.root_doc.id,
+                ),
-                    ),
+                overrides=overrides,
-                    "overrides": overrides,
+                trigger_source=trigger_source,
-                },
+                parent_task_id=parent_task_id,
-                headers={"trigger_source": trigger_source},
             )
             logger.info(
                 f"Queued new rotated version for document {pair.root_doc.id} by {degrees} degrees",
@@ -588,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
@@ -620,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 = (
@@ -667,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(
@@ -811,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",
@@ -854,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",
@@ -886,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",
@@ -904,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(
+        queue_consumption(
-            kwargs={
+            input_doc=ConsumableDocument(
-                "input_doc": ConsumableDocument(
+                source=DocumentSource.ConsumeFolder,
-                    source=DocumentSource.ConsumeFolder,
+                original_file=filepath,
-                    original_file=filepath,
+                root_document_id=pair.root_doc.id,
-                    root_document_id=pair.root_doc.id,
+            ),
-                ),
+            overrides=overrides,
-                "overrides": overrides,
+            trigger_source=trigger_source,
-            },
+            parent_task_id=parent_task_id,
-            headers={"trigger_source": trigger_source},
         )
         logger.info(
             f"Queued new version for document {pair.root_doc.id} after deleting pages {pages}",
@@ -935,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.
@@ -979,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(
+            queue_consumption(
-                kwargs={
+                input_doc=ConsumableDocument(
-                    "input_doc": ConsumableDocument(
+                    source=DocumentSource.ConsumeFolder,
-                        source=DocumentSource.ConsumeFolder,
+                    original_file=filepath,
-                        original_file=filepath,
+                    root_document_id=pair.root_doc.id,
-                        root_document_id=pair.root_doc.id,
+                ),
-                    ),
+                overrides=overrides,
-                    "overrides": overrides,
+                trigger_source=trigger_source,
-                },
+                parent_task_id=parent_task_id,
-                headers={"trigger_source": trigger_source},
             )
         else:
             version_filepaths = pdf_ops.build_pdfs(
@@ -1028,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(
@@ -1062,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"]:
     """
@@ -1104,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(
+                queue_consumption(
-                    kwargs={
+                    input_doc=ConsumableDocument(
-                        "input_doc": ConsumableDocument(
+                        source=DocumentSource.ConsumeFolder,
-                            source=DocumentSource.ConsumeFolder,
+                        original_file=filepath,
-                            original_file=filepath,
+                        root_document_id=pair.root_doc.id,
-                            root_document_id=pair.root_doc.id,
+                    ),
-                        ),
+                    overrides=overrides,
-                        "overrides": overrides,
+                    trigger_source=trigger_source,
-                    },
+                    parent_task_id=parent_task_id,
-                    headers={"trigger_source": trigger_source},
                 )
             else:
                 consume_tasks = []
@@ -1136,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

@@ -89,8 +89,9 @@ def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
     return values
 
 
-@shared_task
+@shared_task(bind=True)
 def execute_bulk_operation(
+    self,
     operation: str,
     document_ids: list[int],
     parameters: dict,
@@ -114,6 +115,11 @@ def execute_bulk_operation(
 
     if operation not in MODIFIED_FIELDS:
         execution_parameters = parameters.copy()
+        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
         if operation in TRIGGER_SOURCE_OPERATIONS:
@@ -125,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
 
@@ -140,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/migrations/0031_task_parent_tracking.py

@@ -0,0 +1,30 @@
+# Generated by Django 5.2.18 on 2026-11-14 19:59
+
+import django.db.models.deletion
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("documents", "0030_note_format"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="paperlesstask",
+            name="dispatched",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name="paperlesstask",
+            name="parent",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="children",
+                to="documents.paperlesstask",
+            ),
+        ),
+    ]

src/documents/models.py

@@ -970,6 +970,16 @@ class PaperlessTask(ModelWithOwner):
         help_text=_("Structured result data from task execution"),
     )
 
+    parent = models.ForeignKey(
+        "self",
+        null=True,
+        blank=True,
+        on_delete=models.SET_NULL,
+        related_name="children",
+    )
+
+    dispatched = models.BooleanField(default=False)
+
     # Acknowledgment
     acknowledged = models.BooleanField(
         default=False,

src/documents/signals/handlers.py

@@ -56,6 +56,8 @@ from documents.models import WorkflowRun
 from documents.models import WorkflowTrigger
 from documents.permissions import restrict_queryset_to_visible
 from documents.plugins.helpers import DocumentsStatusManager
+from documents.task_tracking import finish_dispatch
+from documents.task_tracking import update_parent
 from documents.templating.utils import convert_format_str_to_template_format
 from documents.utils import compute_checksum
 from documents.utils import copy_file_with_basic_stats
@@ -1077,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,
 }
@@ -1125,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:
@@ -1212,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
@@ -1224,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
@@ -1309,6 +1328,10 @@ def task_postrun_handler(
         except PaperlessTask.DoesNotExist:
             return
 
+        if task_instance.children.exists():
+            finish_dispatch(task_instance, retval if isinstance(retval, dict) else {})
+            return
+
         task_instance.status = new_status
         task_instance.date_done = now
         changed_fields = ["status", "date_done"]
@@ -1331,7 +1354,21 @@ 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:
             _send_bulk_operation_status(task_instance)
     except Exception:  # pragma: no cover
@@ -1388,6 +1425,10 @@ def task_failure_handler(
         }
 
         task_qs = PaperlessTask.objects.filter(task_id=task_id)
+        parent = task_qs.first()
+        if parent is not None and parent.children.exists():
+            finish_dispatch(parent, result_data)
+            return
         task_instance = task_qs.values("date_started", "date_created").first()
         if task_instance:
             date_started = task_instance["date_started"]
@@ -1399,6 +1440,7 @@ def task_failure_handler(
                     date_started - date_created
                 ).total_seconds()
             task_qs.update(**update_fields)
+            update_parent(task_qs.get())
             if sender and sender.name == BULK_OPERATION_TASK_NAME:
                 _send_bulk_operation_status(task_qs.first())
     except Exception:  # pragma: no cover
@@ -1445,6 +1487,8 @@ def task_revoked_handler(
                 "reason": (reason + " Check worker logs for details."),
             },
         )
+        if task_instance := task_qs.first():
+            update_parent(task_instance)
         if sender and sender.name == BULK_OPERATION_TASK_NAME:
             _send_bulk_operation_status(task_qs.first())
     except Exception:  # pragma: no cover

src/documents/task_tracking.py

@@ -0,0 +1,242 @@
+"""Completion tracking for operations which dispatch document processing tasks."""
+
+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,
+    document_ids: list[int],
+    *,
+    remote_ocr: bool,
+) -> 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=TRACKED_TASKS[signature.task],
+            trigger_source=parent.trigger_source,
+            owner_id=parent.owner_id,
+            parent=parent,
+            input_data=_extract_input_data(
+                TRACKED_TASKS[signature.task],
+                signature.kwargs,
+            ),
+        )
+        for signature in signatures
+    ]
+    # 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:
+            PaperlessTask.objects.filter(
+                pk__in=[task.pk for task in children[index:]],
+            ).update(
+                status=PaperlessTask.Status.FAILURE,
+                date_done=timezone.now(),
+                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():
+        parent = PaperlessTask.objects.select_for_update().get(pk=task.pk)
+        if parent.is_complete:
+            return
+        parent.dispatched = True
+        parent.result_data = result_data
+        parent.save(update_fields=["dispatched", "result_data"])
+        _finalize_parent(parent)
+
+
+def update_parent(task: PaperlessTask) -> None:
+    if task.parent_id is None:
+        return
+    with transaction.atomic():
+        parent = PaperlessTask.objects.select_for_update().get(pk=task.parent_id)
+        _finalize_parent(parent)
+
+
+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
+
+    succeeded = statuses.count(PaperlessTask.Status.SUCCESS)
+    failed = len(statuses) - succeeded
+    result_data = parent.result_data or {}
+    if failed and not result_data.get("error_message"):
+        result_data["error_message"] = (
+            f"{succeeded} child task(s) succeeded; {failed} failed or were revoked. "
+            "See child tasks for details."
+        )
+    parent.status = (
+        PaperlessTask.Status.FAILURE
+        if result_data.get("error_message")
+        else PaperlessTask.Status.SUCCESS
+    )
+    parent.result_data = {
+        **parent.input_data,
+        **result_data,
+        "succeeded_count": succeeded,
+        "failed_count": failed,
+    }
+    parent.date_done = timezone.now()
+    if parent.date_started:
+        parent.duration_seconds = (
+            parent.date_done - parent.date_started
+        ).total_seconds()
+        parent.wait_time_seconds = (
+            parent.date_started - parent.date_created
+        ).total_seconds()
+    parent.save()
+    from documents.signals.handlers import _send_bulk_operation_status
+
+    if "operation" in parent.input_data:
+        transaction.on_commit(lambda: _send_bulk_operation_status(parent))
+    update_parent(parent)

src/documents/tasks.py

@@ -394,11 +394,10 @@ def update_document_content_maybe_archive_file(
     )
 
     if not parser_class:
-        logger.error(
+        raise ValueError(
             f"No parser found for mime type {mime_type}, cannot "
-            f"archive document {document} (ID: {document_id})",
+            f"reprocess document {document} (ID: {document_id})",
         )
-        return
 
     with parser_class() as parser:
         parser.configure(ParserContext())
@@ -512,6 +511,7 @@ def update_document_content_maybe_archive_file(
             logger.exception(
                 f"Error while parsing document {document} (ID: {document_id})",
             )
+            raise
 
 
 @shared_task

src/documents/tests/test_management.py

@@ -69,7 +69,8 @@ class TestArchiver(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
         doc.save()
         shutil.copy(sample_file, doc.source_path)
 
-        update_document_content_maybe_archive_file(doc.pk)
+        with self.assertRaisesMessage(ValueError, "No parser found"):
+            update_document_content_maybe_archive_file(doc.pk)
 
         doc = Document.objects.get(id=doc.id)
 

src/documents/tests/test_task_tracking.py

@@ -0,0 +1,543 @@
+import shutil
+import uuid
+from unittest import mock
+
+import pytest
+
+from documents.bulk_operations import execute_bulk_operation
+from documents.models import PaperlessTask
+from documents.signals.handlers import task_failure_handler
+from documents.signals.handlers import task_postrun_handler
+from documents.signals.handlers import task_revoked_handler
+from documents.tasks import update_document_content_maybe_archive_file
+from paperless_testing.factories import DocumentFactory
+from paperless_testing.factories import UserFactory
+
+
+@pytest.fixture
+def operation(paperless_dirs):
+    user = UserFactory(superuser=True)
+    documents = DocumentFactory.create_batch(3)
+    parent = PaperlessTask.objects.create(
+        task_id=str(uuid.uuid4()),
+        task_type=PaperlessTask.TaskType.BULK_UPDATE,
+        owner=user,
+        trigger_source=PaperlessTask.TriggerSource.WEB_UI,
+        input_data={"operation": "reprocess", "document_count": 3},
+    )
+    kwargs = {
+        "operation": "reprocess",
+        "document_ids": [document.pk for document in documents],
+        "parameters": {"remote_ocr": True},
+        "user_id": user.pk,
+    }
+    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(
+            sender=update_document_content_maybe_archive_file,
+            task_id=child.task_id,
+            exception=RuntimeError("Parser failed"),
+        )
+    elif status == "REVOKED":
+        task_revoked_handler(
+            sender=update_document_content_maybe_archive_file,
+            request=mock.Mock(id=child.task_id),
+        )
+    else:
+        task_postrun_handler(
+            task=update_document_content_maybe_archive_file,
+            task_id=child.task_id,
+            state=status,
+        )
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize("outcome", ["SUCCESS", "FAILURE", "REVOKED"])
+def test_parent_waits_for_all_children(operation, outcome):
+    parent, kwargs = operation
+    with (
+        mock.patch(
+            "documents.tasks.update_document_content_maybe_archive_file.apply_async",
+        ) as publish,
+        mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify,
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+        parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.STARTED
+        assert parent.date_done is None
+        children = list(parent.children.order_by("pk"))
+        assert len(children) == 3
+        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.args[1]["remote_ocr"] for call in publish.call_args_list)
+        complete(children[0], outcome)
+        complete(children[1])
+        parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.STARTED
+        notify.assert_not_called()
+        complete(children[2])
+        parent.refresh_from_db()
+        assert parent.status == ("success" if outcome == "SUCCESS" else "failure")
+        assert parent.result_data["succeeded_count"] == (
+            3 if outcome == "SUCCESS" else 2
+        )
+        assert parent.result_data["failed_count"] == (0 if outcome == "SUCCESS" else 1)
+        assert parent.date_done is not None
+        assert parent.duration_seconds >= 0
+        notify.assert_called_once()
+        complete(children[2])
+        notify.assert_called_once()
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize("outcome", ["SUCCESS", "FAILURE"])
+def test_fast_children_wait_for_dispatch_to_finish(operation, outcome):
+    parent, kwargs = operation
+
+    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)
+        parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.STARTED
+        assert not parent.dispatched
+
+    with (
+        mock.patch(
+            "documents.tasks.update_document_content_maybe_archive_file.apply_async",
+            side_effect=publish,
+        ),
+        mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify,
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+        parent.refresh_from_db()
+        assert parent.status == ("success" if outcome == "SUCCESS" else "failure")
+        notify.assert_called_once()
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize("published_count", [0, 1])
+def test_publish_failure_waits_for_published_children(operation, published_count):
+    parent, kwargs = operation
+    with (
+        mock.patch(
+            "documents.tasks.update_document_content_maybe_archive_file.apply_async",
+            side_effect=[None] * published_count + [RuntimeError("Broker unavailable")],
+        ),
+        mock.patch("documents.signals.handlers._send_bulk_operation_status") as notify,
+    ):
+        execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=False)
+        parent.refresh_from_db()
+        if published_count:
+            assert parent.status == PaperlessTask.Status.STARTED
+            notify.assert_not_called()
+            complete(parent.children.get(status=PaperlessTask.Status.PENDING))
+            parent.refresh_from_db()
+        assert parent.status == PaperlessTask.Status.FAILURE
+        assert parent.result_data["error_message"] == "Broker unavailable"
+        assert parent.result_data["succeeded_count"] == published_count
+        assert parent.result_data["failed_count"] == 3 - published_count
+        notify.assert_called_once()
+
+
+@pytest.mark.django_db(transaction=True)
+def test_empty_selection_completes_without_children(operation):
+    parent, kwargs = operation
+    kwargs["document_ids"] = []
+    execute_bulk_operation.apply(kwargs=kwargs, task_id=parent.task_id, throw=True)
+    parent.refresh_from_db()
+    assert parent.status == PaperlessTask.Status.SUCCESS
+    assert not parent.children.exists()
+
+
+@pytest.mark.django_db
+def test_client_cannot_supply_parent_task_id(operation):
+    from rest_framework.test import APIClient
+
+    parent, kwargs = operation
+    client = APIClient()
+    client.force_authenticate(parent.owner)
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as publish:
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "documents": kwargs["document_ids"],
+                "method": "set_correspondent",
+                "parameters": {"correspondent": None, "parent_task_id": parent.task_id},
+            },
+            format="json",
+        )
+    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

src/documents/tests/test_tasks.py

@@ -1,4 +1,5 @@
 import shutil
+import uuid
 from datetime import timedelta
 from pathlib import Path
 from unittest import mock
@@ -16,6 +17,7 @@ from documents import tasks
 from documents.models import Correspondent
 from documents.models import Document
 from documents.models import DocumentType
+from documents.models import PaperlessTask
 from documents.models import Tag
 from documents.models import WorkflowAction
 from documents.sanity_checker import SanityCheckFailedException
@@ -280,6 +282,60 @@ class TestUpdateContent(DirectoriesMixin, TestCase):
         tasks.update_document_content_maybe_archive_file(doc.pk)
         self.assertNotEqual(Document.objects.get(pk=doc.pk).content, "test")
 
+    @mock.patch("documents.tasks.get_parser_registry")
+    def test_missing_parser_records_failure(self, registry) -> None:
+        doc = DocumentFactory(content="Original content", mime_type="application/pdf")
+        registry.return_value.get_parser_for_file.return_value = None
+        task = PaperlessTask.objects.create(
+            task_id=str(uuid.uuid4()),
+            task_type=PaperlessTask.TaskType.REPROCESS_DOCUMENT,
+        )
+
+        result = tasks.update_document_content_maybe_archive_file.apply(
+            args=[doc.pk],
+            task_id=task.task_id,
+            throw=False,
+        )
+
+        self.assertTrue(result.failed())
+        task.refresh_from_db()
+        self.assertEqual(task.status, PaperlessTask.Status.FAILURE)
+        self.assertIn("No parser found", task.result_data["error_message"])
+        doc.refresh_from_db()
+        self.assertEqual(doc.content, "Original content")
+
+    @mock.patch("documents.tasks.should_produce_archive", return_value=False)
+    @mock.patch("documents.tasks.get_parser_registry")
+    def test_parser_failure_records_failure(self, registry, _produce_archive) -> None:
+        for method in ("parse", "get_thumbnail"):
+            with self.subTest(method=method):
+                doc = DocumentFactory(
+                    content="Original content",
+                    mime_type="application/pdf",
+                )
+                parser = mock.MagicMock()
+                getattr(parser, method).side_effect = RuntimeError("Processing failed")
+                parser_class = mock.MagicMock()
+                parser_class.return_value.__enter__.return_value = parser
+                registry.return_value.get_parser_for_file.return_value = parser_class
+                task = PaperlessTask.objects.create(
+                    task_id=str(uuid.uuid4()),
+                    task_type=PaperlessTask.TaskType.REPROCESS_DOCUMENT,
+                )
+
+                result = tasks.update_document_content_maybe_archive_file.apply(
+                    args=[doc.pk],
+                    task_id=task.task_id,
+                    throw=False,
+                )
+
+                self.assertTrue(result.failed())
+                task.refresh_from_db()
+                self.assertEqual(task.status, PaperlessTask.Status.FAILURE)
+                self.assertEqual(task.result_data["error_message"], "Processing failed")
+                doc.refresh_from_db()
+                self.assertEqual(doc.content, "Original content")
+
     def _create_root_with_version(self) -> tuple[Document, Document]:
         sample1 = self.dirs.scratch_dir / "sample.pdf"
         shutil.copy(SIMPLE_DIGITAL_PDF, sample1)
@@ -370,7 +426,8 @@ class TestUpdateContentRemoteOCR(DirectoriesMixin, TestCase):
         )
 
     def _allow_remote(self, **kwargs) -> bool:
-        tasks.update_document_content_maybe_archive_file(self.doc.pk, **kwargs)
+        with self.assertRaisesMessage(ValueError, "No parser found"):
+            tasks.update_document_content_maybe_archive_file(self.doc.pk, **kwargs)
         _, call_kwargs = self.mock_registry.return_value.get_parser_for_file.call_args
         return call_kwargs["allow_remote"]
 

src/documents/views.py

@@ -2976,8 +2976,9 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
     parser_classes = (parsers.JSONParser,)
 
     def _queue_bulk_operation(self, method, documents, parameters, from_webui):
-        if "update_callback" in parameters:
+        for parameter in ("update_callback", "parent_task_id"):
-            raise ValueError("Unsupported bulk operation parameter: update_callback")
+            if parameter in parameters:
+                raise ValueError(f"Unsupported bulk operation parameter: {parameter}")
         trigger_source = (
             PaperlessTask.TriggerSource.WEB_UI
             if from_webui