Buld this parent task into the model, dispatched. Hook up for reprocess

docs/api.md

@@ -290,9 +290,14 @@ a json payload of the format:
 
 Document bulk operations, including the dedicated document action endpoints, return
 `{"result":"OK","task_id":"…"}` when queued. Use `/api/tasks/?task_id={uuid}`
-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
-were scheduled, not that their file processing has finished.
+to check the task outcome. Reprocessing operations remain started until all of their
+individual document tasks have finished. Their results include `succeeded_count` and
+`failed_count`; any failed or revoked child makes the operation fail, while successful
+changes remain applied. A scheduling failure is also reported, after any tasks already
+published have finished. Child tasks belong to the initiating user.
+
+Other actions that create files may start further consumption tasks. Completion of
+those bulk tasks means consumption was scheduled, not that file processing has finished.
 
 ```json
 {

src/documents/bulk_edit.py

@@ -432,7 +432,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 +462,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},

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,8 @@ def execute_bulk_operation(
 
     if operation not in MODIFIED_FIELDS:
         execution_parameters = parameters.copy()
+        if operation == "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:

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
@@ -1309,6 +1311,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"]
@@ -1332,6 +1338,7 @@ def task_postrun_handler(
                 changed_fields.append("status")
 
         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 +1395,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 +1410,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 +1457,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,111 @@
+"""Completion tracking for operations which dispatch document processing tasks."""
+
+import uuid
+
+from django.db import transaction
+from django.utils import timezone
+
+from documents.models import PaperlessTask
+
+
+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
+
+    parent = PaperlessTask.objects.get(task_id=parent_task_id)
+    children = [
+        PaperlessTask(
+            task_id=str(uuid.uuid4()),
+            task_type=PaperlessTask.TaskType.REPROCESS_DOCUMENT,
+            trigger_source=parent.trigger_source,
+            owner_id=parent.owner_id,
+            parent=parent,
+            input_data={"document_id": document_id},
+        )
+        for document_id in document_ids
+    ]
+    # 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,
+                },
+                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 reprocessing task."},
+            )
+            raise
+
+
+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=["dispatch_complete", "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
+    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} document(s) reprocessed; {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
+
+    transaction.on_commit(lambda: _send_bulk_operation_status(parent))

src/documents/tests/test_task_tracking.py

@@ -0,0 +1,176 @@
+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
+
+
+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.kwargs["kwargs"]["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(*, 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.dispatch_complete
+
+    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()

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:
-            raise ValueError("Unsupported bulk operation parameter: update_callback")
+        for parameter in ("update_callback", "parent_task_id"):
+            if parameter in parameters:
+                raise ValueError(f"Unsupported bulk operation parameter: {parameter}")
         trigger_source = (
             PaperlessTask.TriggerSource.WEB_UI
             if from_webui