Update api.md

docs/api.md

@@ -285,8 +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
-a json payload of the format:
+Document bulk operations return `{"result":"OK","task_id":"…"}` when queued.
+Check `/api/tasks/?task_id={uuid}` for the outcome, including any follow-up processing,
+deletion, or recovery. Failed operations may leave partial changes applied.
+
+For operations with child tasks, `result_data` includes `succeeded_count` and
+`failed_count` (task counts). Merges report skipped inputs in `failed_document_ids`.
+
+The `/api/documents/bulk_edit/` endpoint accepts:
 
 ```json
 {

src-ui/messages.xlf

@@ -1940,6 +1940,10 @@
           <context context-type="sourcefile">src/app/components/common/permissions-dialog/permissions-dialog.component.ts</context>
           <context context-type="linenumber">45</context>
         </context-group>
+        <context-group purpose="location">
+          <context context-type="sourcefile">src/app/components/document-list/save-view-config-dialog/save-view-config-dialog.component.ts</context>
+          <context context-type="linenumber">59</context>
+        </context-group>
       </trans-unit>
       <trans-unit id="3066660568529853846" datatype="html">
         <source>Error retrieving groups</source>

src-ui/src/app/components/admin/tasks/tasks.component.html

@@ -175,7 +175,13 @@
               </div>
             </td>
             <td class="overflow-auto name-col">
-              <div>{{ taskDisplayName(task) }}</div>
+              @let documentCount = taskDocumentCount(task);
+              <div>
+                {{ taskDisplayName(task) }}
+                @if (documentCount !== null && documentCount > 1) {
+                  <small class="text-muted ms-1" i18n>({documentCount, plural, =1 {1 document} other {{{ documentCount }} documents}})</small>
+                }
+              </div>
               <div class="small text-muted">
                 @if (taskShowsSeparateTypeLabel(task)) {
                   <span class="me-2">{{ task.task_type_display }}</span>

src-ui/src/app/components/admin/tasks/tasks.component.spec.ts

@@ -895,6 +895,117 @@ describe('TasksComponent', () => {
     expect(sanityTaskRow.nativeElement.textContent).toContain('System')
   })
 
+  it.each([
+    ['modify_tags', 12, 'Bulk operation: Update tags', '(12 documents)'],
+    ['run_workflow', 1, 'Run workflow', null],
+    ['delete', 3, 'Bulk operation: Move to trash', '(3 documents)'],
+    ['reprocess', 2, 'Bulk operation: Document reprocessing', '(2 documents)'],
+  ])(
+    'shows the operation, document count and outcome for %s',
+    (operation, count, label, summary) => {
+      const task: PaperlessTask = {
+        ...tasks[4],
+        task_type:
+          operation === 'delete'
+            ? PaperlessTaskType.BulkDelete
+            : PaperlessTaskType.BulkUpdate,
+        task_type_display:
+          operation === 'delete' ? 'Bulk Delete' : 'Bulk Update',
+        trigger_source: PaperlessTaskTriggerSource.WebUI,
+        trigger_source_display: 'Web UI',
+        input_data: { operation, document_count: count },
+        result_data: { operation, document_count: count },
+      }
+      component.setSection(TaskSection.Completed)
+      flushTaskLists(25, 1, [task])
+      fixture.detectChanges()
+
+      const name = fixture.nativeElement.querySelector('tbody .name-col')
+      expect(name.textContent).toContain(label)
+      const countLabel = name.querySelector('small.text-muted')
+      if (summary) {
+        expect(countLabel.textContent).toContain(summary)
+      } else {
+        expect(countLabel).toBeNull()
+        expect(name.textContent).not.toContain('Bulk operation:')
+      }
+      expect(name.textContent).not.toContain('·')
+      expect(name.textContent).toContain('Web UI')
+      expect(name.textContent).not.toContain(task.task_type_display)
+      expect(
+        name.closest('tr').querySelector('.results-col').textContent
+      ).toContain(
+        [
+          'reprocess',
+          'rotate',
+          'merge',
+          'split',
+          'delete_pages',
+          'edit_pdf',
+          'remove_password',
+        ].includes(operation)
+          ? 'Submitted for processing'
+          : 'Completed'
+      )
+    }
+  )
+
+  it('shows operation context while work is pending without reporting completion', () => {
+    const task = {
+      ...tasks[2],
+      task_type: PaperlessTaskType.BulkUpdate,
+      task_type_display: 'Bulk Update',
+      input_data: { operation: 'reprocess', document_count: 1 },
+    }
+    component.setSection(TaskSection.InProgress)
+    flushTaskLists(25, 1, [task])
+    fixture.detectChanges()
+
+    const name = fixture.nativeElement.querySelector('tbody .name-col')
+    expect(name.textContent).toContain('Document reprocessing')
+    expect(name.textContent).not.toContain('Bulk operation:')
+    expect(name.querySelector('small')).toBeNull()
+    expect(component.taskResultMessage(task)).toBeNull()
+  })
+
+  it.each([{}, { operation: 'future_operation' }])(
+    'preserves the fallback for legacy or unknown operation data: %j',
+    (input_data) => {
+      const task = {
+        ...tasks[4],
+        task_type: PaperlessTaskType.BulkUpdate,
+        task_type_display: 'Bulk Update',
+        input_data,
+        result_data: {},
+      }
+      expect(component.taskDisplayName(task)).toBe('Bulk Update')
+      expect(component.taskDocumentCount(task)).toBeNull()
+      expect(component.taskResultMessage(task)).toBeNull()
+    }
+  )
+
+  it.each([-1, 1.5, '2', null, undefined])(
+    'does not display an invalid document count: %s',
+    (document_count) => {
+      expect(
+        component.taskDocumentCount({
+          ...tasks[4],
+          input_data: { operation: 'modify_tags', document_count },
+        })
+      ).toBeNull()
+    }
+  )
+
+  it('retains error details for failed operation tasks', () => {
+    expect(
+      component.taskResultMessage({
+        ...tasks[0],
+        input_data: { operation: 'modify_tags', document_count: 12 },
+        result_data: { error_message: 'Insufficient permissions' },
+      })
+    ).toBe('Insufficient permissions')
+  })
+
   it('should filter tasks by result', () => {
     component.setSection(TaskSection.NeedsAttention)
     component.filterTargetID = 1

src-ui/src/app/components/admin/tasks/tasks.component.ts

@@ -33,6 +33,10 @@ import { IfPermissionsDirective } from 'src/app/directives/if-permissions.direct
 import { CustomDatePipe } from 'src/app/pipes/custom-date.pipe'
 import { TasksService } from 'src/app/services/tasks.service'
 import { ToastService } from 'src/app/services/toast.service'
+import {
+  documentOperationHasFollowUpTasks,
+  getDocumentOperationLabel,
+} from 'src/app/utils/document-operation'
 import { ConfirmDialogComponent } from '../../common/confirm-dialog/confirm-dialog.component'
 import { PageHeaderComponent } from '../../common/page-header/page-header.component'
 import { LoadingComponentWithPermissions } from '../../loading-component/loading.component'
@@ -414,11 +418,30 @@ export class TasksComponent
   }
 
   taskDisplayName(task: PaperlessTask): string {
-    return task.input_data?.filename?.toString() || task.task_type_display
+    const filename = task.input_data?.filename?.toString()
+    if (filename) return filename
+    const operation = this.taskOperationLabel(task)
+    const count = this.taskDocumentCount(task)
+    if (operation && count !== null && count > 1) {
+      return $localize`Bulk operation: ${operation}`
+    }
+    return operation || task.task_type_display
   }
 
   taskShowsSeparateTypeLabel(task: PaperlessTask): boolean {
-    return this.taskDisplayName(task) !== task.task_type_display
+    return (
+      !this.taskOperationLabel(task) &&
+      this.taskDisplayName(task) !== task.task_type_display
+    )
+  }
+
+  private taskOperationLabel(task: PaperlessTask): string | null {
+    return getDocumentOperationLabel(task.input_data?.operation)
+  }
+
+  taskDocumentCount(task: PaperlessTask): number | null {
+    const count = task.input_data?.document_count
+    return Number.isInteger(count) && count >= 0 ? count : null
   }
 
   taskResultMessage(task: PaperlessTask): string | null {
@@ -446,6 +469,15 @@ export class TasksComponent
       return errorMessage
     }
 
+    if (
+      task.status === PaperlessTaskStatus.Success &&
+      this.taskOperationLabel(task)
+    ) {
+      return documentOperationHasFollowUpTasks(task.input_data?.operation)
+        ? $localize`Submitted for processing`
+        : $localize`Completed`
+    }
+
     return null
   }
 

src-ui/src/app/components/app-frame/app-frame.component.spec.ts

@@ -33,6 +33,7 @@ import { RemoteVersionService } from 'src/app/services/rest/remote-version.servi
 import { SavedViewService } from 'src/app/services/rest/saved-view.service'
 import { SearchService } from 'src/app/services/rest/search.service'
 import { SettingsService } from 'src/app/services/settings.service'
+import { TaskNotificationService } from 'src/app/services/task-notification.service'
 import { TasksService } from 'src/app/services/tasks.service'
 import { ToastService } from 'src/app/services/toast.service'
 import { UploadDocumentsService } from 'src/app/services/upload-documents.service'
@@ -198,14 +199,14 @@ describe('AppFrameComponent', () => {
     fixture.detectChanges()
   })
 
-  it('refreshes and notifies on completion without client-side task tracking', () => {
+  it('refreshes and updates the task notification on completion', () => {
     const list = TestBed.inject(DocumentListViewService)
     const reload = jest.spyOn(list, 'reload').mockImplementation(() => {})
     jest.spyOn(list, 'reduceSelectionToFilter').mockImplementation(() => {})
     const tasksReload = jest
       .spyOn(tasksService, 'reload')
       .mockImplementation(() => {})
-    const success = jest.spyOn(toastService, 'showSuccess')
+    const success = jest.spyOn(toastService, 'show')
     const refresh = jest
       .spyOn(openDocumentsService, 'refreshDocument')
       .mockImplementation(() => {})
@@ -230,7 +231,12 @@ describe('AppFrameComponent', () => {
     expect(refresh).toHaveBeenCalledWith(document.id)
     expect(maybeRefreshSpy).toHaveBeenCalledTimes(1)
     expect(tasksReload).toHaveBeenCalledTimes(1)
-    expect(success).toHaveBeenCalledWith('Bulk operation completed.')
+    expect(success).toHaveBeenCalledWith(
+      expect.objectContaining({
+        id: 'task-bulk-task',
+        content: "Operation 'Set permissions' completed.",
+      })
+    )
     fixture.destroy()
     bulkEvents.next({
       task_id: 'bulk-task',
@@ -245,14 +251,14 @@ describe('AppFrameComponent', () => {
     const reload = jest.spyOn(list, 'reload').mockImplementation(() => {})
     jest.spyOn(list, 'reduceSelectionToFilter').mockImplementation(() => {})
     jest.spyOn(tasksService, 'reload').mockImplementation(() => {})
-    const error = jest.spyOn(toastService, 'showError')
+    const error = jest.spyOn(toastService, 'show')
     const success = jest.spyOn(toastService, 'showSuccess')
     for (const status of [
       PaperlessTaskStatus.Failure,
       PaperlessTaskStatus.Revoked,
     ]) {
       bulkEvents.next({
-        task_id: 'bulk-task',
+        task_id: `${status}-task`,
         operation: 'set_permissions',
         status,
         error:
@@ -262,11 +268,36 @@ describe('AppFrameComponent', () => {
       })
     }
     expect(reload).toHaveBeenCalledTimes(2)
-    expect(error).toHaveBeenCalledWith(expect.any(String), 'Could not update')
     expect(error).toHaveBeenCalledWith(
-      'Bulk operation revoked.',
-      'Task expired before execution.'
+      expect.objectContaining({
+        content:
+          "Operation 'Set permissions' failed. Some changes may already have been applied.",
+        error: 'Could not update',
+      })
     )
+    expect(error).toHaveBeenCalledWith(
+      expect.objectContaining({
+        content: "Operation 'Set permissions' revoked.",
+        error: 'Task expired before execution.',
+      })
+    )
+    expect(success).not.toHaveBeenCalled()
+  })
+
+  it('passes delete completion to the shared task notification handler', () => {
+    const success = jest.spyOn(toastService, 'showSuccess')
+    const notifications = TestBed.inject(TaskNotificationService)
+    const handleStatus = jest.spyOn(notifications, 'complete')
+    bulkEvents.next({
+      task_id: 'delete-1',
+      operation: 'delete',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(handleStatus).toHaveBeenCalledWith({
+      task_id: 'delete-1',
+      operation: 'delete',
+      status: PaperlessTaskStatus.Success,
+    })
     expect(success).not.toHaveBeenCalled()
   })
 

src-ui/src/app/components/app-frame/app-frame.component.ts

@@ -57,6 +57,7 @@ import {
 } from 'src/app/services/rest/remote-version.service'
 import { SavedViewService } from 'src/app/services/rest/saved-view.service'
 import { SettingsService } from 'src/app/services/settings.service'
+import { TaskNotificationService } from 'src/app/services/task-notification.service'
 import { TasksService } from 'src/app/services/tasks.service'
 import { ToastService } from 'src/app/services/toast.service'
 import { UploadDocumentsService } from 'src/app/services/upload-documents.service'
@@ -115,6 +116,7 @@ export class AppFrameComponent
   settingsService = inject(SettingsService)
   tasksService = inject(TasksService)
   private readonly toastService = inject(ToastService)
+  private readonly taskNotifications = inject(TaskNotificationService)
   private modalService = inject(NgbModal)
   permissionsService = inject(PermissionsService)
   private djangoMessagesService = inject(DjangoMessagesService)
@@ -182,7 +184,7 @@ export class AppFrameComponent
         ),
         takeUntilDestroyed(this.destroyRef)
       )
-      .subscribe(({ status, error }) => {
+      .subscribe((result) => {
         this.documentList.reload(null, false)
         this.documentList.reduceSelectionToFilter()
         // maybe someday we can optimize to only refresh affected documents
@@ -198,16 +200,7 @@ export class AppFrameComponent
         ) {
           this.tasksService.reload()
         }
-        if (status === PaperlessTaskStatus.Success) {
-          this.toastService.showSuccess($localize`Bulk operation completed.`)
-        } else if (status === PaperlessTaskStatus.Revoked) {
-          this.toastService.showError($localize`Bulk operation revoked.`, error)
-        } else {
-          this.toastService.showError(
-            $localize`Bulk operation failed. Some changes may already have been applied.`,
-            error
-          )
-        }
+        this.taskNotifications.complete(result)
       })
 
     this.lastScrollY = window.scrollY

src-ui/src/app/components/common/toast/toast.component.scss

@@ -7,6 +7,7 @@
     background-color: var(--pngx-toast-bg);
     color: var(--bs-body-color);
     overflow: hidden;
+    transition: background-color .18s ease, border-color .18s ease;
 
     &.toast-success {
         --pngx-toast-accent: var(--pngx-toast-success);
@@ -23,6 +24,13 @@
 
 .toast-icon {
     color: var(--pngx-toast-accent);
+    transition: color .18s ease;
+}
+
+@media (prefers-reduced-motion: reduce) {
+    .toast, .toast-icon {
+        transition: none;
+    }
 }
 
 .toast-content {

src-ui/src/app/components/common/toast/toast.component.spec.ts

@@ -65,6 +65,36 @@ describe('ToastComponent', () => {
     expect(component.delayRemaining()).toEqual(3000)
   })
 
+  it('updates its color and icon in place and restarts the countdown', () => {
+    jest.useFakeTimers()
+    fixture.componentRef.setInput('toast', {
+      id: 'task-1',
+      content: 'Queued',
+      delay: 5000,
+    })
+    fixture.detectChanges()
+    component.onShown(component.toast)
+    jest.advanceTimersByTime(4500)
+    const element = fixture.nativeElement.querySelector('ngb-toast')
+    fixture.componentRef.setInput('toast', {
+      id: 'task-1',
+      content: 'Completed',
+      delay: 5000,
+      level: ToastLevel.Success,
+    })
+    fixture.detectChanges()
+    expect(fixture.nativeElement.querySelector('ngb-toast')).toBe(element)
+    expect(element.classList).toContain('toast-success')
+    expect(element.querySelector('[name="check-circle"]')).not.toBeNull()
+    expect(component.delayRemaining()).toBe(5000)
+    const hide = jest.spyOn(component.ngbToast, 'hide')
+    jest.advanceTimersByTime(1000)
+    expect(hide).not.toHaveBeenCalled()
+    jest.advanceTimersByTime(4200)
+    expect(hide).toHaveBeenCalled()
+    jest.useRealTimers()
+  })
+
   it('should hide the toast when the countdown ends', () => {
     jest.useFakeTimers()
     const toast = { content: 'Info', delay: 3000 }

src-ui/src/app/components/common/toast/toast.component.ts

@@ -5,8 +5,10 @@ import {
   DestroyRef,
   EventEmitter,
   Input,
+  OnChanges,
   OnInit,
   Output,
+  SimpleChanges,
   ViewChild,
   inject,
   signal,
@@ -34,10 +36,10 @@ import { Toast, ToastLevel } from 'src/app/services/toast.service'
   styleUrl: './toast.component.scss',
   providers: [DatePipe, CustomDatePipe],
 })
-export class ToastComponent implements OnInit {
-  private clipboard = inject(Clipboard)
-  private customDatePipe = inject(CustomDatePipe)
-  private destroyRef = inject(DestroyRef)
+export class ToastComponent implements OnInit, OnChanges {
+  private readonly clipboard = inject(Clipboard)
+  private readonly customDatePipe = inject(CustomDatePipe)
+  private readonly destroyRef = inject(DestroyRef)
   readonly copied = signal(false)
   readonly ToastLevel = ToastLevel
   readonly delayRemaining = signal<number>(null)
@@ -69,6 +71,12 @@ export class ToastComponent implements OnInit {
     if (this.autohide) this.delayRemaining.set(this.toast.delay)
   }
 
+  ngOnChanges(changes: SimpleChanges): void {
+    if (changes.toast && !changes.toast.firstChange && this.autohide) {
+      this.onShown(this.toast)
+    }
+  }
+
   get relativeTimestamp(): string {
     return this.customDatePipe.transform(this.toast.timestamp, 'relative')
   }

src-ui/src/app/components/common/toasts/toasts.component.spec.ts

@@ -68,9 +68,9 @@ describe('ToastsComponent', () => {
     toasts.forEach((t) => toastSubject.next(t))
     expect(component.toasts().map((t) => t.id)).toEqual(['3', '2', '1'])
 
-    // showing a visible toast again moves it to the top without duplicating
+    // Updating a visible toast preserves its position without duplicating it.
     toastSubject.next(toasts[1])
-    expect(component.toasts().map((t) => t.id)).toEqual(['1', '3', '2'])
+    expect(component.toasts().map((t) => t.id)).toEqual(['3', '2', '1'])
   })
 
   it('should remove a popup when it hides without closing it', () => {

src-ui/src/app/components/common/toasts/toasts.component.ts

@@ -32,10 +32,11 @@ export class ToastsComponent {
           return
         }
         this.toasts.update((toasts) =>
-          [toast, ...toasts.filter((t) => t.id !== toast.id)].slice(
-            0,
-            MAX_VISIBLE_TOASTS
-          )
+          toasts.some((existing) => existing.id === toast.id)
+            ? toasts.map((existing) =>
+                existing.id === toast.id ? toast : existing
+              )
+            : [toast, ...toasts].slice(0, MAX_VISIBLE_TOASTS)
         )
       })
   }

src-ui/src/app/components/document-detail/document-detail.component.spec.ts

@@ -1013,30 +1013,45 @@ describe('DocumentDetailComponent', () => {
     ])
   })
 
-  it('should support reprocess, confirm and close modal after started', () => {
+  it('should queue reprocess, show the task link and close the modal', () => {
+    jest.useFakeTimers()
     initNormally()
     const reprocessSpy = jest.spyOn(documentService, 'reprocessDocuments')
-    reprocessSpy.mockReturnValue(of(true))
+    reprocessSpy.mockReturnValue(of({ result: 'OK', task_id: 'reprocess-1' }))
     let openModal: NgbModalRef
     modalService.activeInstances.subscribe((modal) => (openModal = modal[0]))
     const modalSpy = jest.spyOn(modalService, 'open')
-    const toastSpy = jest.spyOn(toastService, 'showInfo')
+    const toastSpy = jest.spyOn(toastService, 'show')
     component.reprocess()
     const modalCloseSpy = jest.spyOn(openModal, 'close')
     openModal.componentInstance.confirmClicked.next()
     expect(reprocessSpy).toHaveBeenCalledWith({ documents: [doc.id] }, false)
     expect(modalSpy).toHaveBeenCalled()
-    expect(toastSpy).toHaveBeenCalled()
+    expect(toastSpy).not.toHaveBeenCalled()
     expect(modalCloseSpy).toHaveBeenCalled()
+    jest.advanceTimersByTime(400)
+    expect(toastSpy).toHaveBeenCalledWith(
+      expect.objectContaining({
+        content: 'Document reprocessing queued.',
+        actionName: 'View task',
+      })
+    )
+    toastSpy.mock.calls.at(-1)[0].action()
+    expect(router.navigate).toHaveBeenCalledWith(['/tasks'], {
+      queryParams: { task_id: 'reprocess-1' },
+    })
+    expect(modalCloseSpy).toHaveBeenCalled()
+    jest.useRealTimers()
   })
 
   it('should run a workflow on the document, confirm and close modal', () => {
+    jest.useFakeTimers()
     initNormally()
     const bulkEditSpy = jest.spyOn(documentService, 'bulkEdit')
-    bulkEditSpy.mockReturnValue(of(true))
+    bulkEditSpy.mockReturnValue(of({ result: 'OK', task_id: 'workflow-1' }))
     let openModal: NgbModalRef
     modalService.activeInstances.subscribe((modal) => (openModal = modal[0]))
-    const toastSpy = jest.spyOn(toastService, 'showInfo')
+    const toastSpy = jest.spyOn(toastService, 'show')
     component.runWorkflow()
     expect(openModal.componentInstance.singleDocument).toBe(true)
     expect(openModal.componentInstance.force).toBe(true)
@@ -1048,14 +1063,18 @@ describe('DocumentDetailComponent', () => {
       'run_workflow',
       { workflow: 5, filter_documents: false }
     )
-    expect(toastSpy).toHaveBeenCalled()
+    jest.advanceTimersByTime(400)
+    expect(toastSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ content: 'Workflow queued.' })
+    )
     expect(modalCloseSpy).toHaveBeenCalled()
+    jest.useRealTimers()
   })
 
   it('should apply workflow filters when the single-document checkbox is cleared', () => {
     initNormally()
     const bulkEditSpy = jest.spyOn(documentService, 'bulkEdit')
-    bulkEditSpy.mockReturnValue(of(true))
+    bulkEditSpy.mockReturnValue(of({ result: 'OK', task_id: 'workflow-2' }))
     let openModal: NgbModalRef
     modalService.activeInstances.subscribe((modal) => (openModal = modal[0]))
     component.runWorkflow()
@@ -1072,7 +1091,7 @@ describe('DocumentDetailComponent', () => {
   it('should pass remote OCR choice when reprocessing', () => {
     initNormally()
     const reprocessSpy = jest.spyOn(documentService, 'reprocessDocuments')
-    reprocessSpy.mockReturnValue(of(true))
+    reprocessSpy.mockReturnValue(of({ result: 'OK', task_id: 'reprocess-2' }))
     let openModal: NgbModalRef
     modalService.activeInstances.subscribe((modal) => (openModal = modal[0]))
     component.reprocess()
@@ -2001,10 +2020,12 @@ describe('DocumentDetailComponent', () => {
   })
 
   it('should support pdf editor, handle error', () => {
+    jest.useFakeTimers()
     let modal: NgbModalRef
     modalService.activeInstances.subscribe((m) => (modal = m[0]))
     const closeSpy = jest.spyOn(openDocumentsService, 'closeDocument')
     const errorSpy = jest.spyOn(toastService, 'showError')
+    const queuedToastSpy = jest.spyOn(toastService, 'show')
     initNormally()
     component.selectedVersionId.set(10)
     component.editPdf()
@@ -2023,6 +2044,7 @@ describe('DocumentDetailComponent', () => {
       update_document: false,
       include_metadata: true,
       source_mode: 'explicit_selection',
+      from_webui: true,
     })
     req.error(new ErrorEvent('failed'))
     expect(errorSpy).toHaveBeenCalled()
@@ -2035,13 +2057,21 @@ describe('DocumentDetailComponent', () => {
     req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/edit_pdf/`
     )
-    req.flush(true)
-    expect(closeSpy).toHaveBeenCalled()
+    req.flush({ result: 'OK', task_id: 'edit-pdf-1' })
+    jest.advanceTimersByTime(400)
+    expect(closeSpy).not.toHaveBeenCalled()
+    expect(queuedToastSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ content: 'PDF edit queued.' })
+    )
+    jest.useRealTimers()
   })
 
   it('should support removing password protection from pdfs', () => {
+    jest.useFakeTimers()
     let modal: NgbModalRef
     modalService.activeInstances.subscribe((m) => (modal = m[0]))
+    const closeSpy = jest.spyOn(openDocumentsService, 'closeDocument')
+    const queuedToastSpy = jest.spyOn(toastService, 'show')
     initNormally()
     component.selectedVersionId.set(10)
     component.password = 'secret'
@@ -2062,8 +2092,15 @@ describe('DocumentDetailComponent', () => {
       include_metadata: false,
       delete_original: true,
       source_mode: 'explicit_selection',
+      from_webui: true,
     })
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'remove-password-1' })
+    jest.advanceTimersByTime(400)
+    expect(closeSpy).not.toHaveBeenCalled()
+    expect(queuedToastSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ content: 'Password removal queued.' })
+    )
+    jest.useRealTimers()
   })
 
   it('should require the current password before removing it', () => {
@@ -2115,9 +2152,9 @@ describe('DocumentDetailComponent', () => {
     const req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/remove_password/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'remove-password-2' })
 
-    expect(refreshSpy).toHaveBeenCalledWith(doc.id)
+    expect(refreshSpy).not.toHaveBeenCalled()
   })
 
   it('should support keyboard shortcuts', () => {

src-ui/src/app/components/document-detail/document-detail.component.ts

@@ -97,6 +97,7 @@ import { TagService } from 'src/app/services/rest/tag.service'
 import { UserService } from 'src/app/services/rest/user.service'
 import { WorkflowService } from 'src/app/services/rest/workflow.service'
 import { SettingsService } from 'src/app/services/settings.service'
+import { TaskNotificationService } from 'src/app/services/task-notification.service'
 import { ToastService } from 'src/app/services/toast.service'
 import { TrashService } from 'src/app/services/trash.service'
 import { WebsocketStatusService } from 'src/app/services/websocket-status.service'
@@ -226,6 +227,7 @@ export class DocumentDetailComponent
   private documentListViewService = inject(DocumentListViewService)
   private documentTitlePipe = inject(DocumentTitlePipe)
   private toastService = inject(ToastService)
+  private taskNotifications = inject(TaskNotificationService)
   private trashService = inject(TrashService)
   private settings = inject(SettingsService)
   private storagePathService = inject(StoragePathService)
@@ -1558,9 +1560,11 @@ export class DocumentDetailComponent
           modal.componentInstance.remoteOcr
         )
         .subscribe({
-          next: () => {
-            this.toastService.showInfo(
-              $localize`Reprocess operation for "${this.document().title}" will begin in the background.`
+          next: ({ task_id }) => {
+            this.taskNotifications.showQueued(
+              task_id,
+              $localize`Document reprocessing queued.`,
+              { content: $localize`Document submitted for reprocessing.` }
             )
             if (modal) {
               modal.close()
@@ -1597,9 +1601,13 @@ export class DocumentDetailComponent
           filter_documents: !modal.componentInstance.force,
         })
         .subscribe({
-          next: () => {
-            this.toastService.showInfo(
-              $localize`Workflow will run on "${this.document().title}" in the background.`
+          next: ({ task_id }) => {
+            this.taskNotifications.showQueued(
+              task_id,
+              $localize`Workflow queued.`,
+              {
+                content: $localize`Workflow completed.`,
+              }
             )
             modal.close()
           },
@@ -1989,14 +1997,15 @@ export class DocumentDetailComponent
           })
           .pipe(first(), takeUntil(this.unsubscribeNotifier))
           .subscribe({
-            next: () => {
-              this.toastService.showInfo(
-                $localize`PDF edit operation for "${this.document().title}" will begin in the background.`
+            next: ({ task_id }) => {
+              this.taskNotifications.showQueued(
+                task_id,
+                $localize`PDF edit queued.`,
+                {
+                  content: $localize`PDF edit submitted for processing.`,
+                }
               )
               modal.close()
-              if (modal.componentInstance.deleteOriginal) {
-                this.openDocumentService.closeDocument(this.document())
-              }
             },
             error: (error) => {
               if (modal) {
@@ -2046,17 +2055,16 @@ export class DocumentDetailComponent
           })
           .pipe(first(), takeUntil(this.unsubscribeNotifier))
           .subscribe({
-            next: () => {
-              this.toastService.showInfo(
-                $localize`Password removal operation for "${this.document().title}" will begin in the background.`
+            next: ({ task_id }) => {
+              this.taskNotifications.showQueued(
+                task_id,
+                $localize`Password removal queued.`,
+                {
+                  content: $localize`Password removal submitted for processing.`,
+                }
               )
               this.networkActive.set(false)
               modal.close()
-              if (!dialog.updateDocument && dialog.deleteOriginal) {
-                this.openDocumentService.closeDocument(this.document())
-              } else if (dialog.updateDocument) {
-                this.openDocumentService.refreshDocument(this.documentId())
-              }
             },
             error: (error) => {
               dialog.buttonsEnabled.set(true)

src-ui/src/app/components/document-detail/document-version-dropdown/document-version-dropdown.component.spec.ts

@@ -2,6 +2,7 @@ import { DatePipe } from '@angular/common'
 import { SimpleChange, signal } from '@angular/core'
 import { ComponentFixture, TestBed } from '@angular/core/testing'
 import { By } from '@angular/platform-browser'
+import { Router } from '@angular/router'
 import { NgbModal } from '@ng-bootstrap/ng-bootstrap'
 import { Subject, of, throwError } from 'rxjs'
 import { DocumentVersionInfo } from 'src/app/data/document'
@@ -33,7 +34,7 @@ describe('DocumentVersionDropdownComponent', () => {
     >
   >
   let toastService: jest.Mocked<
-    Pick<ToastService, 'showError' | 'showInfo' | 'showSuccess'>
+    Pick<ToastService, 'show' | 'showError' | 'showInfo' | 'showSuccess'>
   >
   let finished$: Subject<{ taskId: string }>
   let failed$: Subject<{ taskId: string; message?: string }>
@@ -54,6 +55,7 @@ describe('DocumentVersionDropdownComponent', () => {
     }
     modalService = { open: jest.fn() }
     toastService = {
+      show: jest.fn(),
       showError: jest.fn(),
       showInfo: jest.fn(),
       showSuccess: jest.fn(),
@@ -88,6 +90,10 @@ describe('DocumentVersionDropdownComponent', () => {
           provide: PermissionsService,
           useValue: permissionsService,
         },
+        {
+          provide: Router,
+          useValue: { navigate: jest.fn() },
+        },
         {
           provide: WebsocketStatusService,
           useValue: {
@@ -377,7 +383,8 @@ describe('DocumentVersionDropdownComponent', () => {
     expect(component.versionLabelDraft).toEqual('')
   })
 
-  it('addExistingDocumentAsVersion should merge with a label and refresh versions', () => {
+  it('addExistingDocumentAsVersion should queue with a label and wait to refresh versions', () => {
+    jest.useFakeTimers()
     const confirmClicked = new Subject<number>()
     const modal = {
       componentInstance: {
@@ -388,14 +395,9 @@ describe('DocumentVersionDropdownComponent', () => {
       close: jest.fn(),
     }
     modalService.open.mockReturnValue(modal as any)
-    documentService.mergeDocumentsAsVersions.mockReturnValue(of({} as any))
-    // Newest first, as the API returns them. The merged document has a lower id
-    // than the root, which is the whole point of merging an existing document.
-    const versions: DocumentVersionInfo[] = [
-      { id: 2, is_root: false, checksum: 'cccc' },
-      { id: 3, is_root: true, checksum: 'aaaa' },
-    ]
-    documentService.getVersions.mockReturnValue(of({ id: 3, versions } as any))
+    documentService.mergeDocumentsAsVersions.mockReturnValue(
+      of({ result: 'OK', task_id: 'versions-1' })
+    )
     component.newVersionLabel = '  Imported  '
     const versionsEmitSpy = jest.spyOn(component.versionsUpdated, 'emit')
     const selectedEmitSpy = jest.spyOn(component.versionSelected, 'emit')
@@ -410,11 +412,18 @@ describe('DocumentVersionDropdownComponent', () => {
       'Imported'
     )
     expect(documentService.updateVersionLabel).not.toHaveBeenCalled()
-    expect(documentService.getVersions).toHaveBeenCalledWith(3)
-    expect(versionsEmitSpy).toHaveBeenCalledWith(versions)
-    expect(selectedEmitSpy).toHaveBeenCalledWith(2)
+    expect(documentService.getVersions).not.toHaveBeenCalled()
+    expect(versionsEmitSpy).not.toHaveBeenCalled()
+    expect(selectedEmitSpy).not.toHaveBeenCalled()
     expect(component.newVersionLabel).toEqual('')
     expect(modal.close).toHaveBeenCalled()
-    expect(toastService.showSuccess).toHaveBeenCalled()
+    jest.advanceTimersByTime(400)
+    expect(toastService.show).toHaveBeenCalledWith(
+      expect.objectContaining({
+        content: 'Existing document queued to be added as a version.',
+        actionName: 'View task',
+      })
+    )
+    jest.useRealTimers()
   })
 })

src-ui/src/app/components/document-detail/document-version-dropdown/document-version-dropdown.component.ts

@@ -28,6 +28,7 @@ import { DocumentVersionInfo } from 'src/app/data/document'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
 import { CustomDatePipe } from 'src/app/pipes/custom-date.pipe'
 import { DocumentService } from 'src/app/services/rest/document.service'
+import { TaskNotificationService } from 'src/app/services/task-notification.service'
 import { ToastService } from 'src/app/services/toast.service'
 import {
   UploadState,
@@ -77,6 +78,7 @@ export class DocumentVersionDropdownComponent
   private readonly toastService = inject(ToastService)
   private readonly websocketStatusService = inject(WebsocketStatusService)
   private readonly modalService = inject(NgbModal)
+  private readonly taskNotifications = inject(TaskNotificationService)
   private readonly destroy$ = new Subject<void>()
   private readonly documentChange$ = new Subject<void>()
 
@@ -305,23 +307,19 @@ export class DocumentVersionDropdownComponent
             versionLabel
           )
           .pipe(
-            switchMap(() => this.documentsService.getVersions(this.documentId)),
             first(),
             finalize(() => dialog.buttonsEnabled.set(true)),
             takeUntil(this.destroy$),
             takeUntil(this.documentChange$)
           )
           .subscribe({
-            next: (document) => {
-              if (document?.versions?.length) {
-                this.versionsUpdated.emit(document.versions)
-                // The API returns versions newest first
-                this.versionSelected.emit(document.versions[0].id)
-              }
+            next: ({ task_id }) => {
               this.newVersionLabel = ''
               modal.close()
-              this.toastService.showSuccess(
-                $localize`Existing document added as a version.`
+              this.taskNotifications.showQueued(
+                task_id,
+                $localize`Existing document queued to be added as a version.`,
+                { content: $localize`Existing document added as a version.` }
               )
             },
             error: (error) => {

src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.spec.ts

@@ -14,7 +14,8 @@ import { Correspondent } from 'src/app/data/correspondent'
 import { CustomField, CustomFieldDataType } from 'src/app/data/custom-field'
 import { DocumentType } from 'src/app/data/document-type'
 import { FILTER_TITLE } from 'src/app/data/filter-rule-type'
-import { Results } from 'src/app/data/results'
+import { PaperlessTaskStatus } from 'src/app/data/paperless-task'
+import { Results, SelectionData } from 'src/app/data/results'
 import { StoragePath } from 'src/app/data/storage-path'
 import { Tag } from 'src/app/data/tag'
 import { WorkflowTriggerType } from 'src/app/data/workflow-trigger'
@@ -28,10 +29,7 @@ import {
 import { CorrespondentService } from 'src/app/services/rest/correspondent.service'
 import { CustomFieldsService } from 'src/app/services/rest/custom-fields.service'
 import { DocumentTypeService } from 'src/app/services/rest/document-type.service'
-import {
-  DocumentService,
-  SelectionData,
-} from 'src/app/services/rest/document.service'
+import { DocumentService } from 'src/app/services/rest/document.service'
 import { GroupService } from 'src/app/services/rest/group.service'
 import { ShareLinkBundleService } from 'src/app/services/rest/share-link-bundle.service'
 import { StoragePathService } from 'src/app/services/rest/storage-path.service'
@@ -39,6 +37,7 @@ import { TagService } from 'src/app/services/rest/tag.service'
 import { UserService } from 'src/app/services/rest/user.service'
 import { WorkflowService } from 'src/app/services/rest/workflow.service'
 import { SettingsService } from 'src/app/services/settings.service'
+import { TaskNotificationService } from 'src/app/services/task-notification.service'
 import { ToastService } from 'src/app/services/toast.service'
 import { TrashService } from 'src/app/services/trash.service'
 import { environment } from 'src/environments/environment'
@@ -1267,14 +1266,15 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/delete/`
     )
-    const notifySpy = jest
-      .spyOn(trashService, 'notifyMovedToTrash')
+    const trackSpy = jest
+      .spyOn(trashService, 'trackMovedToTrash')
       .mockImplementation(() => {})
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'delete-1' })
     expect(req.request.body).toEqual({
       documents: [3, 4],
+      from_webui: true,
     })
-    expect(notifySpy).toHaveBeenCalledWith(expect.any(String), [3, 4])
+    expect(trackSpy).toHaveBeenCalledWith('delete-1', 2, [3, 4])
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
     ) // list reload
@@ -1291,9 +1291,9 @@ describe('BulkEditorComponent', () => {
     req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/delete/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'delete-2' })
     // ids aren't known when all matching documents are selected, so no undo
-    expect(notifySpy).toHaveBeenLastCalledWith(expect.any(String), null)
+    expect(trackSpy).toHaveBeenLastCalledWith('delete-2', 0, null)
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
     ) // list reload
@@ -1373,10 +1373,11 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/reprocess/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'reprocess-1' })
     expect(req.request.body).toEqual({
       documents: [3, 4],
       remote_ocr: false,
+      from_webui: true,
     })
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
@@ -1387,6 +1388,7 @@ describe('BulkEditorComponent', () => {
   })
 
   it('should support rotate', () => {
+    const toast = jest.spyOn(toastService, 'show')
     let modal: NgbModalRef
     modalService.activeInstances.subscribe((m) => (modal = m[0]))
     jest.spyOn(permissionsService, 'currentUserCan').mockReturnValue(true)
@@ -1407,11 +1409,20 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/rotate/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'rotate-1' })
+    TestBed.inject(TaskNotificationService).complete({
+      task_id: 'rotate-1',
+      operation: 'rotate',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(toast.mock.calls.at(-1)[0].content).toBe(
+      "Bulk operation 'Rotate documents' submitted for processing."
+    )
     expect(req.request.body).toEqual({
       documents: [3, 4],
       degrees: 90,
       source_mode: 'latest_version',
+      from_webui: true,
     })
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
@@ -1449,10 +1460,11 @@ describe('BulkEditorComponent', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/merge/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'merge-1' })
     expect(req.request.body).toEqual({
       documents: [3, 4],
       metadata_document_id: 3,
+      from_webui: true,
     })
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
@@ -1467,11 +1479,12 @@ describe('BulkEditorComponent', () => {
     req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/merge/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'merge-2' })
     expect(req.request.body).toEqual({
       documents: [3, 4],
       metadata_document_id: 3,
       delete_originals: true,
+      from_webui: true,
     })
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
@@ -1488,11 +1501,12 @@ describe('BulkEditorComponent', () => {
     req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/merge/`
     )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'merge-3' })
     expect(req.request.body).toEqual({
       documents: [3, 4],
       metadata_document_id: 3,
       archive_fallback: true,
+      from_webui: true,
     })
     httpTestingController.match(
       `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
@@ -1504,6 +1518,7 @@ describe('BulkEditorComponent', () => {
   })
 
   it('should support merging documents as versions', () => {
+    jest.useFakeTimers()
     let modal: NgbModalRef
     modalService.activeInstances.subscribe((m) => (modal = m[0]))
     jest.spyOn(permissionsService, 'currentUserCan').mockReturnValue(true)
@@ -1531,8 +1546,8 @@ describe('BulkEditorComponent', () => {
       .mockReturnValue(true)
     const mergeAsVersionsSpy = jest
       .spyOn(documentService, 'mergeDocumentsAsVersions')
-      .mockReturnValue(of(true))
-    const toastSuccessSpy = jest.spyOn(toastService, 'showSuccess')
+      .mockReturnValue(of({ result: 'OK', task_id: 'versions-1' }))
+    const toastSpy = jest.spyOn(toastService, 'show')
     fixture.detectChanges()
 
     component.mergeSelectedAsVersions()
@@ -1548,9 +1563,21 @@ describe('BulkEditorComponent', () => {
       `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
     )
     expect(documentListViewService.selected.size).toEqual(0)
-    expect(toastSuccessSpy).toHaveBeenCalledWith(
-      'Documents merged as versions.'
+    jest.advanceTimersByTime(400)
+    expect(toastSpy).toHaveBeenCalledWith(
+      expect.objectContaining({
+        content: "Bulk operation 'Merge as versions' queued.",
+      })
     )
+    TestBed.inject(TaskNotificationService).complete({
+      task_id: 'versions-1',
+      operation: 'merge_as_versions',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(toastSpy.mock.calls.at(-1)[0].content).toBe(
+      "Bulk operation 'Merge as versions' completed."
+    )
+    jest.useRealTimers()
   })
 
   it('should not report success when merging documents as versions fails', () => {
@@ -1641,12 +1668,22 @@ describe('BulkEditorComponent', () => {
     )
   })
 
-  it.each([true, false])(
-    'shows queued feedback and gates the task link on permission (%s)',
-    (canViewTasks) => {
+  it.each<[boolean, number]>([
+    [true, 2],
+    [false, 2],
+    [true, 1],
+  ])(
+    'names queued work and gates the task link (permission %s, documents %s)',
+    (canViewTasks, count) => {
+      jest.useFakeTimers()
       jest
         .spyOn(permissionsService, 'currentUserCan')
         .mockReturnValue(canViewTasks)
+      jest
+        .spyOn(documentListViewService, 'selected', 'get')
+        .mockReturnValue(
+          new Set(Array.from({ length: count }, (_, index) => index + 3))
+        )
       const queued = jest.spyOn(toastService, 'show')
       let modal: NgbModalRef
       modalService.activeInstances.subscribe((instances) => {
@@ -1664,8 +1701,11 @@ describe('BulkEditorComponent', () => {
       )
       req.flush({ result: 'OK', task_id: 'queued-operation' })
       expect(close).toHaveBeenCalledTimes(1)
+      expect(queued).not.toHaveBeenCalled()
+      jest.advanceTimersByTime(400)
       const notification = queued.mock.calls.at(-1)[0]
-      expect(notification.content).toBe('Bulk operation queued.')
+      const prefix = count > 1 ? 'Bulk operation' : 'Operation'
+      expect(notification.content).toBe(`${prefix} \'Set permissions\' queued.`)
       if (canViewTasks) {
         expect(notification.actionName).toBe('View task')
         notification.action()
@@ -1679,6 +1719,15 @@ describe('BulkEditorComponent', () => {
       expect(reload).not.toHaveBeenCalled()
       expect(documentListViewService.selected.size).toBe(0)
       expect(documentListViewService.allSelected).toBe(false)
+      TestBed.inject(TaskNotificationService).complete({
+        task_id: 'queued-operation',
+        operation: 'set_permissions',
+        status: PaperlessTaskStatus.Success,
+      })
+      const completed = queued.mock.calls.at(-1)[0]
+      expect(completed.content).toBe(`${prefix} \'Set permissions\' completed.`)
+      expect(completed.action).toBeUndefined()
+      jest.useRealTimers()
     }
   )
 
@@ -1759,7 +1808,6 @@ describe('BulkEditorComponent', () => {
         { id: 2, name: 'Tag 2' },
       ],
       count: 2,
-      all: [1, 2],
     }
 
     const modalInstance = {
@@ -1802,7 +1850,6 @@ describe('BulkEditorComponent', () => {
         { id: 2, name: 'Correspondent 2' },
       ],
       count: 2,
-      all: [1, 2],
     }
 
     const modalInstance = {
@@ -1851,7 +1898,6 @@ describe('BulkEditorComponent', () => {
         { id: 2, name: 'Document Type 2' },
       ],
       count: 2,
-      all: [1, 2],
     }
 
     const modalInstance = {
@@ -1897,7 +1943,6 @@ describe('BulkEditorComponent', () => {
         { id: 2, name: 'Storage Path 2' },
       ],
       count: 2,
-      all: [1, 2],
     }
 
     const modalInstance = {
@@ -1951,7 +1996,6 @@ describe('BulkEditorComponent', () => {
         },
       ],
       count: 2,
-      all: [1, 2],
     }
 
     const modalInstance = {

src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.ts

@@ -19,25 +19,18 @@ import {
   NgbModalRef,
 } from '@ng-bootstrap/ng-bootstrap'
 import { saveAs } from 'file-saver'
-import {
-  first,
-  map,
-  Observable,
-  Subject,
-  switchMap,
-  takeUntil,
-  tap,
-} from 'rxjs'
+import { first, map, Observable, Subject, switchMap, takeUntil } from 'rxjs'
 import { BootstrapIconComponent } from 'src/app/components/common/bootstrap-icon/bootstrap-icon.component'
 import { ConfirmDialogComponent } from 'src/app/components/common/confirm-dialog/confirm-dialog.component'
+import { BulkOperationResponse } from 'src/app/data/bulk-operation'
 import { CustomField } from 'src/app/data/custom-field'
 import { MatchingModel } from 'src/app/data/matching-model'
 import { SelectionDataItem } from 'src/app/data/results'
 import { SETTINGS_KEYS } from 'src/app/data/ui-settings'
 import { Workflow } from 'src/app/data/workflow'
+import { WorkflowTriggerType } from 'src/app/data/workflow-trigger'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
 import { DocumentListViewService } from 'src/app/services/document-list-view.service'
-import { OpenDocumentsService } from 'src/app/services/open-documents.service'
 import {
   PermissionAction,
   PermissionsService,
@@ -52,14 +45,18 @@ import {
   DocumentService,
   MergeDocumentsRequest,
 } from 'src/app/services/rest/document.service'
-import { SavedViewService } from 'src/app/services/rest/saved-view.service'
 import { ShareLinkBundleService } from 'src/app/services/rest/share-link-bundle.service'
-import { WorkflowService } from 'src/app/services/rest/workflow.service'
 import { StoragePathService } from 'src/app/services/rest/storage-path.service'
 import { TagService } from 'src/app/services/rest/tag.service'
+import { WorkflowService } from 'src/app/services/rest/workflow.service'
 import { SettingsService } from 'src/app/services/settings.service'
+import { TaskNotificationService } from 'src/app/services/task-notification.service'
 import { ToastService } from 'src/app/services/toast.service'
 import { TrashService } from 'src/app/services/trash.service'
+import {
+  documentOperationHasFollowUpTasks,
+  getDocumentOperationLabel,
+} from 'src/app/utils/document-operation'
 import { flattenTags } from 'src/app/utils/flatten-tags'
 import { queryParamsFromFilterRules } from 'src/app/utils/query-params'
 import { MergeAsVersionsConfirmDialogComponent } from '../../common/confirm-dialog/merge-as-versions-confirm-dialog/merge-as-versions-confirm-dialog.component'
@@ -84,7 +81,6 @@ import { PermissionsDialogComponent } from '../../common/permissions-dialog/perm
 import { ShareLinkBundleDialogComponent } from '../../common/share-link-bundle-dialog/share-link-bundle-dialog.component'
 import { ComponentWithPermissions } from '../../with-permissions/with-permissions.component'
 import { CustomFieldsBulkEditDialogComponent } from './custom-fields-bulk-edit-dialog/custom-fields-bulk-edit-dialog.component'
-import { WorkflowTriggerType } from 'src/app/data/workflow-trigger'
 
 @Component({
   selector: 'pngx-bulk-editor',
@@ -110,13 +106,12 @@ export class BulkEditorComponent
   private documentService = inject(DocumentService)
   private trashService = inject(TrashService)
   private modalService = inject(NgbModal)
-  private openDocumentService = inject(OpenDocumentsService)
   private settings = inject(SettingsService)
   private toastService = inject(ToastService)
+  private taskNotifications = inject(TaskNotificationService)
   private storagePathService = inject(StoragePathService)
   private customFieldService = inject(CustomFieldsService)
   public readonly permissionService = inject(PermissionsService)
-  private savedViewService = inject(SavedViewService)
   private readonly shareLinkBundleService = inject(ShareLinkBundleService)
   private readonly workflowService = inject(WorkflowService)
   private readonly router = inject(Router)
@@ -357,70 +352,62 @@ export class BulkEditorComponent
       .bulkEdit(overrideSelection ?? this.getSelectionQuery(), method, args)
       .pipe(first())
       .subscribe({
-        next: ({ task_id }) => this.handleOperationQueued(modal, task_id),
+        next: ({ task_id }) =>
+          this.handleOperationQueued(modal, task_id, method),
         error: (error) => this.handleOperationError(modal, error),
       })
   }
 
-  private handleOperationQueued(modal: NgbModalRef, taskId: string) {
+  private handleOperationQueued(
+    modal: NgbModalRef,
+    taskId: string,
+    operation: string,
+    onQueued?: (taskId: string) => void
+  ) {
+    const bulk = this.getSelectionSize() > 1
     modal?.close()
     this.list.selectNone()
-    const canViewTasks = this.permissionService.currentUserCan(
-      PermissionAction.View,
-      PermissionType.PaperlessTask
-    )
-    this.toastService.show({
-      content: $localize`Bulk operation queued.`,
-      delay: 5000,
-      ...(canViewTasks
-        ? {
-            actionName: $localize`View task`,
-            action: () =>
-              this.router.navigate(['/tasks'], {
-                queryParams: { task_id: taskId },
-              }),
-          }
-        : {}),
-    })
+    if (onQueued) {
+      onQueued(taskId)
+    } else {
+      const label = getDocumentOperationLabel(operation)
+      const submittedMessage = bulk
+        ? $localize`Bulk operation '${label}' submitted for processing.`
+        : $localize`Operation '${label}' submitted for processing.`
+      const completedMessage = bulk
+        ? $localize`Bulk operation '${label}' completed.`
+        : $localize`Operation '${label}' completed.`
+      this.taskNotifications.showQueued(
+        taskId,
+        bulk
+          ? $localize`Bulk operation '${label}' queued.`
+          : $localize`Operation '${label}' queued.`,
+        {
+          content: documentOperationHasFollowUpTasks(operation)
+            ? submittedMessage
+            : completedMessage,
+        }
+      )
+    }
   }
 
   private executeDocumentAction(
     modal: NgbModalRef,
-    request: Observable<any>,
-    options: { clearSelection?: boolean; successMessage?: string } = {}
+    operation: string,
+    request: Observable<BulkOperationResponse>,
+    onQueued?: (taskId: string) => void
   ) {
     if (modal) {
       modal.componentInstance.buttonsEnabled.set(false)
     }
     request.pipe(first()).subscribe({
-      next: () => {
-        this.handleOperationSuccess(modal, options.clearSelection ?? false)
-        if (options.successMessage) {
-          this.toastService.showSuccess(options.successMessage)
-        }
+      next: ({ task_id }) => {
+        this.handleOperationQueued(modal, task_id, operation, onQueued)
       },
       error: (error) => this.handleOperationError(modal, error),
     })
   }
 
-  private handleOperationSuccess(
-    modal: NgbModalRef,
-    clearSelection: boolean = false
-  ) {
-    if (clearSelection) {
-      this.list.selected.clear()
-    }
-    this.list.reload()
-    this.list.reduceSelectionToFilter()
-    this.list.selected.forEach((id) => {
-      this.openDocumentService.refreshDocument(id)
-    })
-    this.savedViewService.maybeRefreshDocumentCounts()
-    if (modal) {
-      modal.close()
-    }
-  }
-
   private handleOperationError(modal: NgbModalRef, error: any) {
     if (modal) {
       modal.componentInstance.buttonsEnabled.set(true)
@@ -988,16 +975,9 @@ export class BulkEditorComponent
     const count = this.getSelectionSize()
     this.executeDocumentAction(
       modal,
-      this.documentService
-        .deleteDocuments(this.getSelectionQuery())
-        .pipe(
-          tap(() =>
-            this.trashService.notifyMovedToTrash(
-              $localize`Moved ${count} document(s) to the trash`,
-              documents
-            )
-          )
-        )
+      'delete',
+      this.documentService.deleteDocuments(this.getSelectionQuery()),
+      (taskId) => this.trashService.trackMovedToTrash(taskId, count, documents)
     )
   }
 
@@ -1038,6 +1018,7 @@ export class BulkEditorComponent
         modal.componentInstance.buttonsEnabled.set(false)
         this.executeDocumentAction(
           modal,
+          'reprocess',
           this.documentService.reprocessDocuments(
             this.getSelectionQuery(),
             modal.componentInstance.remoteOcr
@@ -1078,6 +1059,7 @@ export class BulkEditorComponent
         rotateDialog.buttonsEnabled.set(false)
         this.executeDocumentAction(
           modal,
+          'rotate',
           this.documentService.rotateDocuments(
             this.getSelectionQuery(),
             rotateDialog.degrees
@@ -1111,11 +1093,8 @@ export class BulkEditorComponent
         mergeDialog.buttonsEnabled.set(false)
         this.executeDocumentAction(
           modal,
-          this.documentService.mergeDocuments(mergeDialog.documentIDs(), args),
-          { clearSelection: !!args.delete_originals }
-        )
-        this.toastService.showInfo(
-          $localize`Merged document will be queued for consumption.`
+          'merge',
+          this.documentService.mergeDocuments(mergeDialog.documentIDs(), args)
         )
       })
   }
@@ -1137,14 +1116,11 @@ export class BulkEditorComponent
       .subscribe(() => {
         this.executeDocumentAction(
           modal,
+          'merge_as_versions',
           this.documentService.mergeDocumentsAsVersions(
             mergeDialog.documentIDs(),
             mergeDialog.rootDocumentID()
-          ),
-          {
-            clearSelection: true,
-            successMessage: $localize`Documents merged as versions.`,
-          }
+          )
         )
       })
   }
@@ -1169,7 +1145,7 @@ export class BulkEditorComponent
     dialog.selection = this.getSelectionQuery()
     dialog.selectionCount = this.getSelectionSize()
     dialog.succeeded.subscribe(({ task_id }) =>
-      this.handleOperationQueued(null, task_id)
+      this.handleOperationQueued(null, task_id, 'modify_custom_fields')
     )
     dialog.failed.subscribe((error) => {
       this.toastService.showError(

src-ui/src/app/components/document-list/save-view-config-dialog/save-view-config-dialog.component.html

@@ -16,7 +16,7 @@
     </pngx-input-select>
     <pngx-input-check i18n-title title="Show in sidebar" formControlName="showInSideBar"></pngx-input-check>
     <pngx-input-check i18n-title title="Show on dashboard" formControlName="showOnDashboard"></pngx-input-check>
-    <pngx-permissions-form accordion="true" formControlName="permissions_form"></pngx-permissions-form>
+    <pngx-permissions-form [users]="users()" accordion="true" formControlName="permissions_form"></pngx-permissions-form>
     @if (error()?.filter_rules) {
       <div class="alert alert-danger" role="alert">
         <h6 class="alert-heading" i18n>Filter rules error occurred while saving this view</h6>

src-ui/src/app/components/document-list/save-view-config-dialog/save-view-config-dialog.component.spec.ts

@@ -3,6 +3,7 @@ import { FormsModule, ReactiveFormsModule } from '@angular/forms'
 import { By } from '@angular/platform-browser'
 import { NgbActiveModal, NgbModalModule } from '@ng-bootstrap/ng-bootstrap'
 import { of } from 'rxjs'
+import { User } from 'src/app/data/user'
 import { GroupService } from 'src/app/services/rest/group.service'
 import { UserService } from 'src/app/services/rest/user.service'
 import { CheckComponent } from '../../common/input/check/check.component'
@@ -14,6 +15,7 @@ import { TextComponent } from '../../common/input/text/text.component'
 import { SaveViewConfigDialogComponent } from './save-view-config-dialog.component'
 
 describe('SaveViewConfigDialogComponent', () => {
+  const users = [{ id: 10, username: 'user10' }] as User[]
   let component: SaveViewConfigDialogComponent
   let fixture: ComponentFixture<SaveViewConfigDialogComponent>
   let modal: NgbActiveModal
@@ -25,7 +27,7 @@ describe('SaveViewConfigDialogComponent', () => {
         {
           provide: UserService,
           useValue: {
-            listAll: () => of({ results: [] }),
+            listAll: () => of({ results: users }),
           },
         },
         {
@@ -124,6 +126,19 @@ describe('SaveViewConfigDialogComponent', () => {
     })
   })
 
+  it('should list users in the owner selector', () => {
+    fixture.debugElement
+      .query(By.css('button[ngbAccordionButton]'))
+      .nativeElement.click()
+    fixture.detectChanges()
+
+    const permissionsForm = fixture.debugElement.query(
+      By.directive(PermissionsFormComponent)
+    )
+    const ownerSelect = permissionsForm.query(By.directive(SelectComponent))
+    expect(ownerSelect.componentInstance.items).toEqual(users)
+  })
+
   it('should support default name', () => {
     const saveClickedSpy = jest.spyOn(component.saveClicked, 'emit')
     const modalCloseSpy = jest.spyOn(modal, 'close')

src-ui/src/app/components/document-list/save-view-config-dialog/save-view-config-dialog.component.ts

@@ -6,6 +6,7 @@ import {
   inject,
   signal,
 } from '@angular/core'
+import { toSignal } from '@angular/core/rxjs-interop'
 import {
   FormControl,
   FormGroup,
@@ -13,11 +14,14 @@ import {
   ReactiveFormsModule,
 } from '@angular/forms'
 import { NgbActiveModal } from '@ng-bootstrap/ng-bootstrap'
+import { catchError, map, of } from 'rxjs'
 import {
   DEFAULT_SAVED_VIEW_ICON,
   SAVED_VIEW_ICONS,
 } from 'src/app/data/saved-view-icons'
 import { User } from 'src/app/data/user'
+import { UserService } from 'src/app/services/rest/user.service'
+import { ToastService } from 'src/app/services/toast.service'
 import { CheckComponent } from '../../common/input/check/check.component'
 import { PermissionsFormComponent } from '../../common/input/permissions/permissions-form/permissions-form.component'
 import { SelectComponent } from '../../common/input/select/select.component'
@@ -38,6 +42,8 @@ import { TextComponent } from '../../common/input/text/text.component'
 })
 export class SaveViewConfigDialogComponent implements OnInit {
   private modal = inject(NgbActiveModal)
+  private userService = inject(UserService)
+  private toastService = inject(ToastService)
   readonly error = signal(undefined)
   readonly buttonsEnabled = signal(true)
   readonly defaultName = signal('')
@@ -46,7 +52,16 @@ export class SaveViewConfigDialogComponent implements OnInit {
   @Output()
   public saveClicked = new EventEmitter()
 
-  users: User[]
+  readonly users = toSignal(
+    this.userService.listAll().pipe(
+      map((result) => result.results),
+      catchError((error) => {
+        this.toastService.showError($localize`Error retrieving users`, error)
+        return of([])
+      })
+    ),
+    { initialValue: undefined as User[] }
+  )
   readonly savedViewIcons = SAVED_VIEW_ICONS
 
   setDefaultName(value: string) {

src-ui/src/app/services/open-documents.service.spec.ts

@@ -240,7 +240,7 @@ describe('OpenDocumentsService', () => {
     expect(refreshed.tags).toEqual([4])
   })
 
-  it('should handle error on refresh documents', () => {
+  it('keeps documents open after a network error during refresh', () => {
     subscriptions.push(
       openDocumentsService.openDocument(documents[1]).subscribe()
     )
@@ -250,7 +250,63 @@ describe('OpenDocumentsService', () => {
     )
     expect(req.request.method).toEqual('GET')
     req.error(new ErrorEvent('timeout'))
+    expect(openDocumentsService.getOpenDocument(documents[1].id)).toBe(
+      documents[1]
+    )
+  })
+
+  it.each([500, 403])(
+    'keeps documents open after HTTP %s during refresh',
+    (status) => {
+      const doc = { ...documents[0] }
+      openDocumentsService.openDocument(doc).subscribe()
+      openDocumentsService.refreshDocument(doc.id)
+      httpTestingController
+        .expectOne(
+          `${environment.apiBaseUrl}documents/${doc.id}/?full_perms=true`
+        )
+        .flush({}, { status, statusText: 'Error' })
+      expect(openDocumentsService.getOpenDocument(doc.id)).toBe(doc)
+    }
+  )
+
+  it('preserves unsaved edits when a refreshed document no longer exists', () => {
+    const doc = { ...documents[0], title: 'Unsaved title' }
+    openDocumentsService.openDocument(doc).subscribe()
+    openDocumentsService.setDirty(doc, true, { title: doc.title })
+    openDocumentsService.refreshDocument(doc.id)
+    httpTestingController
+      .expectOne(
+        `${environment.apiBaseUrl}documents/${doc.id}/?full_perms=true`
+      )
+      .flush({}, { status: 404, statusText: 'Not Found' })
+    expect(openDocumentsService.getOpenDocument(doc.id)).toBe(doc)
+    expect(openDocumentsService.isDirty(doc)).toBe(true)
+    expect(
+      JSON.parse(sessionStorage.getItem(OPEN_DOCUMENT_SERVICE.DOCUMENTS))[0]
+        .title
+    ).toBe('Unsaved title')
+  })
+
+  it('removes only the missing document when tab positions change during refresh', () => {
+    const first = { ...documents[0] }
+    const second = { ...documents[1] }
+    openDocumentsService.openDocument(first).subscribe()
+    openDocumentsService.openDocument(second).subscribe()
+    openDocumentsService.refreshDocument(first.id)
+    openDocumentsService.refreshDocument(second.id)
+    const firstRequest = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}documents/${first.id}/?full_perms=true`
+    )
+    const secondRequest = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}documents/${second.id}/?full_perms=true`
+    )
+    openDocumentsService.closeDocument(second).subscribe()
+    firstRequest.flush({}, { status: 404, statusText: 'Not Found' })
     expect(openDocumentsService.getOpenDocuments()).toHaveLength(0)
+    openDocumentsService.openDocument(first).subscribe()
+    secondRequest.flush({}, { status: 404, statusText: 'Not Found' })
+    expect(openDocumentsService.getOpenDocuments()).toEqual([first])
   })
 
   it('should log error on sessionStorage save', () => {

src-ui/src/app/services/open-documents.service.ts

@@ -1,3 +1,4 @@
+import { HttpErrorResponse } from '@angular/common/http'
 import { Injectable, inject, signal } from '@angular/core'
 import { NgbModal } from '@ng-bootstrap/ng-bootstrap'
 import { Observable, Subject, of } from 'rxjs'
@@ -61,7 +62,12 @@ export class OpenDocumentsService {
           Object.assign(openDoc, doc, unsavedEdits)
           this.save()
         },
-        error: () => {
+        error: (error: HttpErrorResponse) => {
+          if (error.status !== 404) return
+
+          const index = this.openDocuments.findIndex((doc) => doc.id === id)
+          if (index === -1 || this.isDirty(this.openDocuments[index])) return
+
           this.openDocuments.splice(index, 1)
           this.save()
         },

src-ui/src/app/services/rest/document.service.spec.ts

@@ -290,6 +290,7 @@ describe(`DocumentService`, () => {
     expect(req.request.method).toEqual('POST')
     expect(req.request.body).toEqual({
       documents: ids,
+      from_webui: true,
     })
   })
 
@@ -303,6 +304,7 @@ describe(`DocumentService`, () => {
     expect(req.request.body).toEqual({
       documents: ids,
       remote_ocr: false,
+      from_webui: true,
     })
   })
 
@@ -317,6 +319,7 @@ describe(`DocumentService`, () => {
     expect(req.request.body).toEqual({
       documents: ids,
       remote_ocr: true,
+      from_webui: true,
     })
   })
 
@@ -331,6 +334,7 @@ describe(`DocumentService`, () => {
       documents: ids,
       degrees: 90,
       source_mode: 'latest_version',
+      from_webui: true,
     })
   })
 
@@ -346,6 +350,7 @@ describe(`DocumentService`, () => {
       documents: ids,
       metadata_document_id: 1,
       delete_originals: true,
+      from_webui: true,
     })
   })
 
@@ -359,6 +364,7 @@ describe(`DocumentService`, () => {
     expect(req.request.body).toEqual({
       documents: ids,
       root_document_id: 2,
+      from_webui: true,
     })
   })
 
@@ -374,6 +380,7 @@ describe(`DocumentService`, () => {
       documents: ids,
       root_document_id: 2,
       version_label: 'Imported',
+      from_webui: true,
     })
   })
 
@@ -388,6 +395,7 @@ describe(`DocumentService`, () => {
     expect(req.request.body).toEqual({
       documents: ids,
       operations: [{ page: 1, rotate: 90, doc: 0 }],
+      from_webui: true,
     })
   })
 
@@ -403,6 +411,7 @@ describe(`DocumentService`, () => {
       documents: ids,
       password: 'secret',
       update_document: true,
+      from_webui: true,
     })
   })
 

src-ui/src/app/services/rest/document.service.ts

@@ -351,19 +351,24 @@ export class DocumentService extends AbstractPaperlessService<Document> {
   }
 
   deleteDocuments(selection: DocumentSelectionQuery) {
-    return this.http.post(this.getResourceUrl(null, 'delete'), {
-      ...selection,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'delete'),
+      { ...selection, from_webui: true }
+    )
   }
 
   reprocessDocuments(
     selection: DocumentSelectionQuery,
     remoteOcr: boolean = false
   ) {
-    return this.http.post(this.getResourceUrl(null, 'reprocess'), {
-      ...selection,
-      remote_ocr: remoteOcr,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'reprocess'),
+      {
+        ...selection,
+        remote_ocr: remoteOcr,
+        from_webui: true,
+      }
+    )
   }
 
   rotateDocuments(
@@ -371,18 +376,26 @@ export class DocumentService extends AbstractPaperlessService<Document> {
     degrees: number,
     sourceMode: BulkEditSourceMode = BulkEditSourceMode.LATEST_VERSION
   ) {
-    return this.http.post(this.getResourceUrl(null, 'rotate'), {
-      ...selection,
-      degrees,
-      source_mode: sourceMode,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'rotate'),
+      {
+        ...selection,
+        degrees,
+        source_mode: sourceMode,
+        from_webui: true,
+      }
+    )
   }
 
   mergeDocuments(ids: number[], request: MergeDocumentsRequest = {}) {
-    return this.http.post(this.getResourceUrl(null, 'merge'), {
-      documents: ids,
-      ...request,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'merge'),
+      {
+        documents: ids,
+        ...request,
+        from_webui: true,
+      }
+    )
   }
 
   mergeDocumentsAsVersions(
@@ -390,28 +403,40 @@ export class DocumentService extends AbstractPaperlessService<Document> {
     rootDocumentId: number,
     versionLabel?: string
   ) {
-    return this.http.post(this.getResourceUrl(null, 'merge_as_versions'), {
-      documents: ids,
-      root_document_id: rootDocumentId,
-      ...(versionLabel ? { version_label: versionLabel } : {}),
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'merge_as_versions'),
+      {
+        documents: ids,
+        root_document_id: rootDocumentId,
+        ...(versionLabel ? { version_label: versionLabel } : {}),
+        from_webui: true,
+      }
+    )
   }
 
   editPdfDocuments(ids: number[], request: EditPdfDocumentsRequest) {
-    return this.http.post(this.getResourceUrl(null, 'edit_pdf'), {
-      documents: ids,
-      ...request,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'edit_pdf'),
+      {
+        documents: ids,
+        ...request,
+        from_webui: true,
+      }
+    )
   }
 
   removePasswordDocuments(
     ids: number[],
     request: RemovePasswordDocumentsRequest
   ) {
-    return this.http.post(this.getResourceUrl(null, 'remove_password'), {
-      documents: ids,
-      ...request,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'remove_password'),
+      {
+        documents: ids,
+        ...request,
+        from_webui: true,
+      }
+    )
   }
 
   getSelectionData(

src-ui/src/app/services/task-notification.service.spec.ts

@@ -0,0 +1,186 @@
+import { TestBed } from '@angular/core/testing'
+import { Router } from '@angular/router'
+import { BulkOperationStatus } from '../data/bulk-operation'
+import { PaperlessTaskStatus } from '../data/paperless-task'
+import { PermissionsService } from './permissions.service'
+import { TaskNotificationService } from './task-notification.service'
+import { Toast, ToastLevel, ToastService } from './toast.service'
+
+describe('TaskNotificationService', () => {
+  let service: TaskNotificationService
+  let toastService: ToastService
+  let popups: Toast[]
+  let notifications: Toast[]
+  const navigate = jest.fn()
+  const currentUserCan = jest.fn()
+  const result = (
+    status = PaperlessTaskStatus.Success
+  ): BulkOperationStatus => ({
+    task_id: 'task-1',
+    operation: 'run_workflow',
+    status,
+  })
+
+  beforeEach(() => {
+    jest.useFakeTimers()
+    currentUserCan.mockReturnValue(true)
+    TestBed.configureTestingModule({
+      providers: [
+        { provide: Router, useValue: { navigate } },
+        { provide: PermissionsService, useValue: { currentUserCan } },
+      ],
+    })
+    service = TestBed.inject(TaskNotificationService)
+    toastService = TestBed.inject(ToastService)
+    popups = []
+    notifications = []
+    toastService.showToast.subscribe((toast) => popups.push(toast))
+    toastService
+      .getToasts()
+      .subscribe((toasts) => (notifications = [...toasts]))
+  })
+
+  afterEach(() => {
+    jest.clearAllTimers()
+    jest.useRealTimers()
+  })
+
+  it('replaces a queued notification with its result, removing the task link on success', () => {
+    service.showQueued('task-1', 'Workflow queued.', {
+      content: 'Workflow completed.',
+    })
+    jest.advanceTimersByTime(399)
+    expect(popups).toHaveLength(0)
+    jest.advanceTimersByTime(1)
+    expect(popups[0].content).toBe('Workflow queued.')
+    expect(popups[0].actionName).toBe('View task')
+    popups[0].action()
+    expect(navigate).toHaveBeenCalledWith(['/tasks'], {
+      queryParams: { task_id: 'task-1' },
+    })
+    service.complete(result())
+    expect(notifications).toHaveLength(1)
+    expect(popups[1]).toEqual(
+      expect.objectContaining({
+        id: popups[0].id,
+        content: 'Workflow completed.',
+        level: ToastLevel.Success,
+      })
+    )
+    expect(popups[1].action).toBeUndefined()
+    expect(popups[1].actionName).toBeUndefined()
+  })
+
+  it('shows only the result when a task completes during the grace period', () => {
+    service.showQueued('task-1', 'Queued')
+    jest.advanceTimersByTime(100)
+    service.complete(result())
+    jest.advanceTimersByTime(1000)
+    expect(popups.map((toast) => toast.content)).toEqual([
+      "Operation 'Run workflow' completed.",
+    ])
+  })
+
+  it.each([
+    'reprocess',
+    'rotate',
+    'merge',
+    'split',
+    'delete_pages',
+    'edit_pdf',
+    'remove_password',
+  ])('reports %s as submitted rather than completed', (operation) => {
+    service.showQueued('task-1', 'Queued')
+    service.complete({ ...result(), operation })
+    expect(notifications[0].content).toContain('submitted for processing.')
+    expect(notifications[0].content).not.toContain('completed')
+    expect(notifications[0].action).toBeUndefined()
+  })
+
+  it('does not show a stale queued state when completion beats the HTTP response', () => {
+    service.complete(result())
+    service.showQueued('task-1', 'Workflow queued.', {
+      content: 'Workflow completed.',
+    })
+    jest.advanceTimersByTime(1000)
+    expect(popups.every((toast) => toast.level === ToastLevel.Success)).toBe(
+      true
+    )
+    expect(notifications).toHaveLength(1)
+    expect(notifications[0].content).toBe('Workflow completed.')
+    expect(notifications[0].action).toBeUndefined()
+  })
+
+  it.each([PaperlessTaskStatus.Failure, PaperlessTaskStatus.Revoked])(
+    'replaces the queued state with an error for %s',
+    (status) => {
+      service.showQueued('task-1', 'Queued', {
+        content: 'Success',
+        actionName: 'Undo',
+      })
+      jest.advanceTimersByTime(400)
+      service.complete({ ...result(status), error: 'Task failed' })
+      expect(notifications).toHaveLength(1)
+      expect(notifications[0]).toEqual(
+        expect.objectContaining({
+          level: ToastLevel.Error,
+          error: 'Task failed',
+          actionName: 'View task',
+        })
+      )
+    }
+  )
+
+  it('ignores nonterminal and duplicate outcomes', () => {
+    service.showQueued('task-1', 'Queued')
+    service.complete(result(PaperlessTaskStatus.Started))
+    expect(popups).toHaveLength(0)
+    service.complete(result())
+    service.complete(result())
+    expect(popups).toHaveLength(1)
+  })
+
+  it('keeps concurrent task notifications separate', () => {
+    service.showQueued('task-1', 'First queued')
+    service.showQueued('task-2', 'Second queued')
+    jest.advanceTimersByTime(400)
+    service.complete(result())
+    expect(notifications).toHaveLength(2)
+    expect(notifications.map((toast) => toast.content)).toEqual([
+      'Second queued',
+      "Operation 'Run workflow' completed.",
+    ])
+  })
+
+  it('omits task links without task viewing permission', () => {
+    currentUserCan.mockReturnValue(false)
+    service.showQueued('task-1', 'Queued')
+    jest.advanceTimersByTime(400)
+    service.complete(result())
+    expect(popups.every((toast) => !toast.action && !toast.actionName)).toBe(
+      true
+    )
+  })
+
+  it('identifies concurrent operations even when completion arrives before their queue responses', () => {
+    service.complete({
+      task_id: 'tags',
+      operation: 'modify_tags',
+      status: PaperlessTaskStatus.Success,
+    })
+    service.complete({
+      task_id: 'correspondent',
+      operation: 'set_correspondent',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(notifications.map((toast) => toast.content)).toEqual([
+      "Operation 'Set correspondent' completed.",
+      "Operation 'Update tags' completed.",
+    ])
+  })
+
+  it('keeps the generic fallback for unknown operation names', () => {
+    service.complete({ ...result(), operation: 'future_operation' })
+    expect(notifications[0].content).toBe('Operation completed.')
+  })
+})

src-ui/src/app/services/task-notification.service.ts

@@ -0,0 +1,128 @@
+import { inject, Injectable, OnDestroy } from '@angular/core'
+import { Router } from '@angular/router'
+import { BulkOperationStatus } from '../data/bulk-operation'
+import { PaperlessTaskStatus } from '../data/paperless-task'
+import {
+  documentOperationHasFollowUpTasks,
+  getDocumentOperationLabel,
+} from '../utils/document-operation'
+import {
+  PermissionAction,
+  PermissionsService,
+  PermissionType,
+} from './permissions.service'
+import { Toast, ToastLevel, ToastService } from './toast.service'
+
+interface PendingNotification {
+  success: Partial<Toast>
+  timer: ReturnType<typeof setTimeout>
+}
+
+@Injectable({ providedIn: 'root' })
+export class TaskNotificationService implements OnDestroy {
+  private readonly toastService = inject(ToastService)
+  private readonly router = inject(Router)
+  private readonly permissionsService = inject(PermissionsService)
+  private readonly pending = new Map<string, PendingNotification>()
+  private readonly completed = new Map<string, BulkOperationStatus>()
+
+  ngOnDestroy() {
+    this.pending.forEach(({ timer }) => clearTimeout(timer))
+  }
+
+  showQueued(taskId: string, content: string, success: Partial<Toast> = {}) {
+    const completed = this.completed.get(taskId)
+    if (completed) {
+      this.showResult(completed, success)
+      return
+    }
+    if (this.pending.has(taskId)) return
+    this.pending.set(taskId, {
+      success,
+      // short delay in case the task finishes super fast
+      timer: setTimeout(() => this.show(taskId, { content, delay: 5000 }), 400),
+    })
+  }
+
+  complete(result: BulkOperationStatus) {
+    if (
+      ![
+        PaperlessTaskStatus.Success,
+        PaperlessTaskStatus.Failure,
+        PaperlessTaskStatus.Revoked,
+      ].includes(result.status) ||
+      this.completed.has(result.task_id)
+    )
+      return
+
+    this.completed.set(result.task_id, result)
+    // Retain recent outcomes in case task finishes quickly
+    if (this.completed.size > 100) {
+      this.completed.delete(this.completed.keys().next().value)
+    }
+    const pending = this.pending.get(result.task_id)
+    if (pending) {
+      clearTimeout(pending.timer)
+      this.pending.delete(result.task_id)
+    }
+    // Trash needs the caller's document IDs to offer Undo on success.
+    if (
+      pending ||
+      result.operation !== 'delete' ||
+      result.status !== PaperlessTaskStatus.Success
+    ) {
+      this.showResult(result, pending?.success)
+    }
+  }
+
+  private showResult(
+    result: BulkOperationStatus,
+    success: Partial<Toast> = {}
+  ) {
+    const succeeded = result.status === PaperlessTaskStatus.Success
+    const operation = getDocumentOperationLabel(result.operation)
+    let content: string
+    if (succeeded) {
+      let operationText = $localize`Operation`
+      if (operation) operationText += ` '${operation}'`
+      const resultText = documentOperationHasFollowUpTasks(result.operation)
+        ? $localize`submitted for processing`
+        : $localize`completed`
+      content = `${operationText} ${resultText}.`
+    } else if (result.status === PaperlessTaskStatus.Revoked) {
+      content = operation
+        ? $localize`Operation '${operation}' revoked.`
+        : $localize`Operation revoked.`
+    } else {
+      content = operation
+        ? $localize`Operation '${operation}' failed. Some changes may already have been applied.`
+        : $localize`Operation failed. Some changes may already have been applied.`
+    }
+    this.show(result.task_id, {
+      content,
+      delay: succeeded ? 5000 : 10000,
+      level: succeeded ? ToastLevel.Success : ToastLevel.Error,
+      ...(succeeded ? success : { error: result.error }),
+    })
+  }
+
+  private show(taskId: string, toast: Toast) {
+    const canViewTasks = this.permissionsService.currentUserCan(
+      PermissionAction.View,
+      PermissionType.PaperlessTask
+    )
+    this.toastService.show({
+      id: `task-${taskId}`,
+      ...(canViewTasks && toast.level !== ToastLevel.Success
+        ? {
+            actionName: $localize`View task`,
+            action: () =>
+              this.router.navigate(['/tasks'], {
+                queryParams: { task_id: taskId },
+              }),
+          }
+        : {}),
+      ...toast,
+    })
+  }
+}

src-ui/src/app/services/toast.service.spec.ts

@@ -38,6 +38,29 @@ describe('ToastService', () => {
     })
   })
 
+  it('updates a notification by ID without duplicating or moving it', () => {
+    let notifications: Toast[] = []
+    toastService
+      .getToasts()
+      .subscribe((toasts) => (notifications = [...toasts]))
+    toastService.show({ id: 'task-1', content: 'Queued', delay: 5000 })
+    toastService.show({
+      id: 'other',
+      content: 'Another notification',
+      delay: 5000,
+    })
+    toastService.show({
+      id: 'task-1',
+      content: 'Completed',
+      delay: 5000,
+      level: ToastLevel.Success,
+    })
+    expect(notifications.map((toast) => toast.content)).toEqual([
+      'Another notification',
+      'Completed',
+    ])
+  })
+
   it('parses error string to object on show', () => {
     const toast = {
       title: 'Title',

src-ui/src/app/services/toast.service.ts

@@ -75,7 +75,14 @@ export class ToastService {
     if (persistent) {
       // the dropdown is open, so it's seen as it arrives
       toast.read = !!this._suppressPopupToasts
-      this.toasts.unshift(toast)
+      const index = this.toasts.findIndex(
+        (existing) => existing.id === toast.id
+      )
+      if (index === -1) {
+        this.toasts.unshift(toast)
+      } else {
+        this.toasts[index] = toast
+      }
       this.toastsSubject.next(this.toasts)
     }
     // Popups are suppressed while the dropdown is open

src-ui/src/app/services/trash.service.spec.ts

@@ -5,8 +5,10 @@ import {
   HttpTestingController,
 } from '@angular/common/http/testing'
 import { environment } from 'src/environments/environment'
+import { PaperlessTaskStatus } from '../data/paperless-task'
 import { DocumentListViewService } from './document-list-view.service'
-import { Toast, ToastService } from './toast.service'
+import { TaskNotificationService } from './task-notification.service'
+import { Toast, ToastLevel, ToastService } from './toast.service'
 import { TrashService } from './trash.service'
 
 describe('TrashService', () => {
@@ -105,4 +107,52 @@ describe('TrashService', () => {
     expect(toast.content).toEqual('Moved 500 document(s) to the trash')
     expect(toast.action).toBeUndefined()
   })
+
+  it('offers Undo only after the matching delete task succeeds', () => {
+    let toasts: Toast[] = []
+    toastService.getToasts().subscribe((t) => (toasts = [...t]))
+    service.trackMovedToTrash('delete-1', 2, [1, 2])
+    expect(toasts).toHaveLength(0)
+    const notifications = TestBed.inject(TaskNotificationService)
+    notifications.complete({
+      task_id: 'other-delete',
+      operation: 'delete',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(toasts).toHaveLength(0)
+    notifications.complete({
+      task_id: 'delete-1',
+      operation: 'delete',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(toasts).toHaveLength(1)
+    expect(toasts[0].actionName).toBe('Undo')
+    toasts[0].action()
+    const req = httpTestingController.expectOne(
+      `${environment.apiBaseUrl}trash/`
+    )
+    expect(req.request.body).toEqual({ action: 'restore', documents: [1, 2] })
+  })
+
+  it('handles completion before the queue response and does not offer Undo on failure', () => {
+    let toasts: Toast[] = []
+    toastService.getToasts().subscribe((t) => (toasts = [...t]))
+    const notifications = TestBed.inject(TaskNotificationService)
+    notifications.complete({
+      task_id: 'delete-1',
+      operation: 'delete',
+      status: PaperlessTaskStatus.Success,
+    })
+    service.trackMovedToTrash('delete-1', 1, [1])
+    expect(toasts[0].actionName).toBe('Undo')
+    service.trackMovedToTrash('delete-2', 1, [2])
+    notifications.complete({
+      task_id: 'delete-2',
+      operation: 'delete',
+      status: PaperlessTaskStatus.Failure,
+    })
+    expect(toasts).toHaveLength(2)
+    expect(toasts[0].level).toBe(ToastLevel.Error)
+    expect(toasts[0].actionName).not.toBe('Undo')
+  })
 })

src-ui/src/app/services/trash.service.ts

@@ -5,7 +5,8 @@ import { environment } from 'src/environments/environment'
 import { Document } from '../data/document'
 import { Results } from '../data/results'
 import { DocumentListViewService } from './document-list-view.service'
-import { ToastLevel, ToastService } from './toast.service'
+import { TaskNotificationService } from './task-notification.service'
+import { Toast, ToastLevel, ToastService } from './toast.service'
 
 @Injectable({
   providedIn: 'root',
@@ -14,6 +15,24 @@ export class TrashService {
   private http = inject(HttpClient)
   private toastService = inject(ToastService)
   private documentListViewService = inject(DocumentListViewService)
+  private taskNotifications = inject(TaskNotificationService)
+
+  public trackMovedToTrash(
+    taskId: string,
+    count: number,
+    documents?: number[]
+  ) {
+    this.taskNotifications.showQueued(
+      taskId,
+      count > 1
+        ? $localize`Bulk move to trash queued.`
+        : $localize`Move to trash queued.`,
+      this.movedToTrashToast(
+        $localize`Moved ${count} document(s) to the trash`,
+        documents
+      )
+    )
+  }
 
   public getTrash(page: number = 1): Observable<Results<Document>> {
     const httpParams = new HttpParams().set('page', page.toString())
@@ -43,14 +62,22 @@ export class TrashService {
    * Shows a toast with an Undo action when possible
    */
   public notifyMovedToTrash(content: string, documents?: number[]) {
+    this.toastService.show(this.movedToTrashToast(content, documents))
+  }
+
+  private movedToTrashToast(content: string, documents?: number[]): Toast {
     const canUndo = documents?.length > 0
-    this.toastService.show({
+    return {
       content,
       delay: 10000,
       level: ToastLevel.Success,
-      actionName: canUndo ? $localize`Undo` : undefined,
-      action: canUndo ? () => this.undoMoveToTrash(documents) : undefined,
-    })
+      ...(canUndo
+        ? {
+            actionName: $localize`Undo`,
+            action: () => this.undoMoveToTrash(documents),
+          }
+        : {}),
+    }
   }
 
   private undoMoveToTrash(documents: number[]) {

src-ui/src/app/utils/document-operation.spec.ts

@@ -0,0 +1,11 @@
+import { getDocumentOperationLabel } from './document-operation'
+
+describe('Document Operation Utils', () => {
+  it('should retrieve the document operation label correctly for an existing operation, null otherwise', () => {
+    const result = getDocumentOperationLabel('modify_custom_fields')
+    expect(result).toBe('Update custom fields')
+
+    const resultNull = getDocumentOperationLabel('non_existing_operation')
+    expect(resultNull).toBeNull()
+  })
+})

src-ui/src/app/utils/document-operation.ts

@@ -0,0 +1,38 @@
+const OPERATION_LABELS: Record<string, string> = {
+  set_correspondent: $localize`Set correspondent`,
+  set_document_type: $localize`Set document type`,
+  set_storage_path: $localize`Set storage path`,
+  add_tag: $localize`Add tag`,
+  remove_tag: $localize`Remove tag`,
+  modify_tags: $localize`Update tags`,
+  modify_custom_fields: $localize`Update custom fields`,
+  set_permissions: $localize`Set permissions`,
+  delete: $localize`Move to trash`,
+  reprocess: $localize`Document reprocessing`,
+  run_workflow: $localize`Run workflow`,
+  rotate: $localize`Rotate documents`,
+  merge: $localize`Merge documents`,
+  merge_as_versions: $localize`Merge as versions`,
+  split: $localize`Split documents`,
+  delete_pages: $localize`Delete pages`,
+  edit_pdf: $localize`Edit PDF`,
+  remove_password: $localize`Remove password`,
+}
+
+export function getDocumentOperationLabel(operation?: string): string | null {
+  const label = operation ? OPERATION_LABELS[operation] : null
+  return label ?? null
+}
+
+// These operations queue processing in separate tasks
+export function documentOperationHasFollowUpTasks(operation?: string): boolean {
+  return [
+    'reprocess',
+    'rotate',
+    'merge',
+    'split',
+    'delete_pages',
+    'edit_pdf',
+    'remove_password',
+  ].includes(operation ?? '')
+}

src-ui/tsconfig.json

@@ -18,7 +18,8 @@
       "es2022",
       "dom"
     ],
-    "useDefineForClassFields": false
+    "useDefineForClassFields": false,
+    "types": ["jest", "node"]
   },
   "angularCompilerOptions": {
     "enableI18nLegacyMessageIdFormat": false,

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
@@ -427,11 +429,17 @@ def delete(doc_ids: list[int]) -> Literal["OK"]:
                 "Detected a possible incompatible database column. See https://docs.paperless-ngx.com/troubleshooting/#convert-uuid-field",
             )
         logger.error(f"Error deleting documents: {e!s}")
+        raise
 
     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.
 
@@ -456,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},
@@ -474,9 +492,9 @@ def run_workflow(
     """
     Manually run a single workflow against the given documents.
 
-    Done asynchronously via a Celery task.
+    Called from the tracked bulk operation task.
     """
-    run_workflows_for_documents.delay(
+    run_workflows_for_documents(
         document_ids=doc_ids,
         workflow_id=workflow,
         filter_documents=filter_documents,
@@ -520,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.",
@@ -557,22 +576,22 @@ 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",
             )
         except Exception as e:
             logger.exception(f"Error rotating document {pair.root_doc.id}: {e}")
+            raise
 
     return "OK"
 
@@ -586,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
@@ -618,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 = (
@@ -665,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(
@@ -701,7 +742,15 @@ def merge_as_versions(
         )
         documents_by_id = {document.id: document for document in documents}
 
+        if any(document.root_document_id is not None for document in documents):
+            raise ValueError("Only top-level documents can be merged as versions.")
+
         source_ids = [doc_id for doc_id in doc_ids if doc_id != root_document_id]
+        if Document.global_objects.filter(root_document_id__in=source_ids).exists():
+            raise ValueError(
+                "Documents with existing versions cannot be merged into another document.",
+            )
+
         root_document = documents_by_id[root_document_id]
         next_version_index = (
             Document.global_objects.filter(
@@ -801,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",
@@ -844,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",
@@ -864,6 +924,7 @@ def split(
 
     except Exception as e:
         logger.exception(f"Error splitting document {doc.id}: {e}")
+        raise
 
     return "OK"
 
@@ -875,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",
@@ -893,22 +955,22 @@ 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}",
         )
     except Exception as e:
         logger.exception(f"Error deleting pages from document {pair.root_doc.id}: {e}")
+        raise
 
     return "OK"
 
@@ -923,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.
@@ -967,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(
@@ -1016,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(
@@ -1050,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"]:
     """
@@ -1092,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 = []
@@ -1124,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

@@ -1,4 +1,4 @@
-"""Queued document metadata edits."""
+"""Queued document bulk operations."""
 
 from typing import Any
 
@@ -12,6 +12,7 @@ from rest_framework.exceptions import PermissionDenied
 from documents import bulk_edit
 from documents.models import CustomFieldInstance
 from documents.models import Document
+from documents.models import PaperlessTask
 from documents.permissions import has_document_operation_permissions
 from documents.tasks import bulk_update_documents
 
@@ -25,7 +26,35 @@ MODIFIED_FIELDS = {
     "modify_custom_fields": "custom_fields",
     "set_permissions": None,
 }
-QUEUED_OPERATIONS = frozenset(MODIFIED_FIELDS)
+LEGACY_MODIFIED_FIELDS = {"delete": "deleted_at", "reprocess": "checksum"}
+QUEUED_OPERATIONS = frozenset(
+    (
+        *MODIFIED_FIELDS,
+        "delete",
+        "reprocess",
+        "run_workflow",
+        "rotate",
+        "merge",
+        "merge_as_versions",
+        "split",
+        "delete_pages",
+        "edit_pdf",
+        "remove_password",
+    ),
+)
+USER_OPERATIONS = frozenset(
+    {
+        "split",
+        "merge",
+        "rotate",
+        "delete_pages",
+        "edit_pdf",
+        "remove_password",
+        "merge_as_versions",
+    },
+)
+# merge_as_versions doesn't queue any consume tasks
+TRIGGER_SOURCE_OPERATIONS = USER_OPERATIONS - {"merge_as_versions"}
 
 
 def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
@@ -60,12 +89,14 @@ 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,
     user_id: int,
+    trigger_source: str = PaperlessTask.TriggerSource.API_UPLOAD,
 ) -> dict:
     if operation not in QUEUED_OPERATIONS:
         raise ValueError(f"Unsupported bulk operation: {operation}")
@@ -79,6 +110,48 @@ def execute_bulk_operation(
     ):
         raise PermissionDenied("Insufficient permissions to execute bulk operation")
 
+    if operation == "run_workflow" and not user.has_perm("documents.view_workflow"):
+        raise PermissionDenied("Insufficient permissions to run this workflow")
+
+    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:
+            execution_parameters["trigger_source"] = trigger_source
+        modified_field = LEGACY_MODIFIED_FIELDS.get(operation)
+        with set_actor(user):
+            old_values = (
+                snapshot_field(document_ids, modified_field)
+                if settings.AUDIT_LOG_ENABLED and modified_field
+                else {}
+            )
+            result = method(document_ids, **execution_parameters)
+            if settings.AUDIT_LOG_ENABLED and modified_field:
+                from auditlog.models import LogEntry
+
+                new_values = snapshot_field(document_ids, modified_field)
+                for doc in Document.objects.filter(pk__in=document_ids):
+                    LogEntry.objects.log_create(
+                        instance=doc,
+                        changes={
+                            modified_field: [old_values[doc.pk], new_values[doc.pk]],
+                        },
+                        action=LogEntry.Action.UPDATE,
+                        actor=user,
+                        additional_data={"reason": f"Bulk edit: {operation}"},
+                    )
+        return {
+            "operation": operation,
+            "document_count": len(document_ids),
+            **(result if isinstance(result, dict) else {}),
+        }
+
     affected_ids: list[int] = []
     modified_field = MODIFIED_FIELDS[operation]
     # Keep metadata writes and their audit entries together. Follow-up processing

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/serialisers.py

@@ -1688,6 +1688,7 @@ class MergeDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin
 
 class MergeDocumentsAsVersionsSerializer(DocumentListSerializer):
     root_document_id = serializers.IntegerField(required=True)
+    from_webui = serializers.BooleanField(required=False, default=False)
     version_label = serializers.CharField(
         required=False,
         allow_blank=True,
@@ -1716,19 +1717,8 @@ class MergeDocumentsAsVersionsSerializer(DocumentListSerializer):
                 "root_document_id must be one of the selected documents.",
             )
 
-        selected_documents = Document.objects.filter(id__in=documents)
-        if selected_documents.filter(root_document__isnull=False).exists():
-            raise serializers.ValidationError(
-                "Only top-level documents can be merged as versions.",
-            )
-
-        source_document_ids = set(documents) - {attrs["root_document_id"]}
-        if Document.global_objects.filter(
-            root_document_id__in=source_document_ids,
-        ).exists():
-            raise serializers.ValidationError(
-                "Documents with existing versions cannot be merged into another document.",
-            )
+        # Version state checks (e.g., whether a document is already part of a version chain)
+        # moved to merge_as_versions() under row locks, since it could change between queueing and execution
         return attrs
 
 
@@ -1801,11 +1791,12 @@ class RemovePasswordDocumentsSerializer(
 
 
 class DeleteDocumentsSerializer(DocumentSelectionSerializer):
-    pass
+    from_webui = serializers.BooleanField(required=False, default=False)
 
 
 class ReprocessDocumentsSerializer(DocumentSelectionSerializer):
     remote_ocr = serializers.BooleanField(required=False, default=False)
+    from_webui = serializers.BooleanField(required=False, default=False)
 
 
 class BulkEditSerializer(

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:
@@ -1206,10 +1211,25 @@ def before_task_publish_handler(
                 "document_count": len(task_kwargs["document_ids"]),
             }
             owner_id = task_kwargs["user_id"]
+            if task_kwargs["operation"] == "delete":
+                task_type = PaperlessTask.TaskType.BULK_DELETE
         else:
             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
@@ -1222,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
@@ -1307,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"]
@@ -1329,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
@@ -1386,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"]
@@ -1397,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
@@ -1443,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
@@ -713,8 +713,9 @@ def run_workflows_for_documents(
         logger.warning(
             f"Manual workflow run requested for missing workflow {workflow_id}",
         )
-        return
+        raise
 
+    failed_ids = []
     for document in Document.objects.filter(
         id__in=document_ids,
         root_document__isnull=True,
@@ -732,10 +733,16 @@ def run_workflows_for_documents(
                 document=document,
             )
         except Exception:
+            failed_ids.append(document.pk)
             logger.exception(
                 f"Error running workflow {workflow} on document {document.pk}",
             )
 
+    if failed_ids:
+        raise RuntimeError(
+            f"Workflow {workflow_id} failed for document(s): {failed_ids}. Check logs for details.",
+        )
+
 
 def update_document_parent_tags(tag: Tag, new_parent: Tag) -> None:
     """

src/documents/tests/test_bulk_edit.py

@@ -698,10 +698,10 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
     def test_delete_documents_old_uuid_field(self, m) -> None:
         m.side_effect = Exception("Data too long for column 'transaction_id' at row 1")
         doc_ids = [self.doc1.id, self.doc2.id, self.doc3.id]
-        bulk_edit.delete(doc_ids)
         with self.assertLogs(level="WARNING") as cm:
-            bulk_edit.delete(doc_ids)
-            self.assertIn("possible incompatible database column", cm.output[0])
+            with self.assertRaisesRegex(Exception, "Data too long for column"):
+                bulk_edit.delete(doc_ids)
+        self.assertIn("possible incompatible database column", cm.output[0])
 
 
 class TestPDFActions(DirectoriesMixin, TestCase):
@@ -1038,8 +1038,9 @@ class TestPDFActions(DirectoriesMixin, TestCase):
 
         mock_consume_file.assert_not_called()
 
+    @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
-    def test_split(self, mock_consume_file) -> None:
+    def test_split(self, mock_consume_file, mock_group) -> None:
         """
         GIVEN:
             - Existing documents
@@ -1060,6 +1061,7 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertIsNone(mock_consume_file.call_args.kwargs["overrides"].asn)
 
         self.assertEqual(result, "OK")
+        mock_group.return_value.delay.assert_called_once()
 
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.pdf_ops.build_pdfs")
@@ -1168,8 +1170,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         sig.apply_async.side_effect = Exception("boom")
         mock_chord.return_value = sig
 
-        result = bulk_edit.split(doc_ids, pages, delete_originals=True)
-        self.assertEqual(result, "OK")
+        with self.assertRaisesMessage(Exception, "boom"):
+            bulk_edit.split(doc_ids, pages, delete_originals=True)
 
         self.doc2.refresh_from_db()
         self.assertEqual(self.doc2.archive_serial_number, 222)
@@ -1191,7 +1193,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         pages = [[1, 2], [3]]
 
         with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm:
-            bulk_edit.split(doc_ids, pages)
+            with self.assertRaisesMessage(Exception, "Error building PDFs"):
+                bulk_edit.split(doc_ids, pages)
             error_str = cm.output[0]
             expected_str = "Error splitting document 2"
             self.assertIn(expected_str, error_str)
@@ -1240,7 +1243,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         doc_ids = [self.doc2.id, self.doc3.id]
 
         with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm:
-            bulk_edit.rotate(doc_ids, 90)
+            with self.assertRaisesMessage(Exception, "Error rotating PDF"):
+                bulk_edit.rotate(doc_ids, 90)
             error_str = cm.output[0]
             expected_str = "Error rotating document"
             self.assertIn(expected_str, error_str)
@@ -1375,7 +1379,8 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         pages = [1, 3]
 
         with self.assertLogs("paperless.bulk_edit", level="ERROR") as cm:
-            bulk_edit.delete_pages(doc_ids, pages)
+            with self.assertRaisesMessage(Exception, "Error removing pages"):
+                bulk_edit.delete_pages(doc_ids, pages)
             error_str = cm.output[0]
             expected_str = "Error deleting pages from document"
             self.assertIn(expected_str, error_str)

src/documents/tests/test_bulk_operations.py

@@ -54,6 +54,7 @@ def test_request_only_queues(bulk_setup, endpoint, operation, parameters):
         "document_ids": [doc.pk],
         "parameters": parameters,
         "user_id": user.pk,
+        "trigger_source": "web_ui",
     }
     assert enqueue.call_args.kwargs["headers"]["trigger_source"] == "web_ui"
 
@@ -102,6 +103,119 @@ def test_permissions_rechecked_on_worker(bulk_setup):
     assert doc.correspondent_id is None
 
 
+@pytest.mark.django_db
+def test_document_action_runs_with_user_in_worker(bulk_setup):
+    user, _, doc, _ = bulk_setup
+    with mock.patch("documents.bulk_edit.rotate") as rotate:
+        result = execute_bulk_operation(
+            "rotate",
+            [doc.pk],
+            {"degrees": 90},
+            user.pk,
+            trigger_source="web_ui",
+        )
+    rotate.assert_called_once_with(
+        [doc.pk],
+        degrees=90,
+        user=user,
+        trigger_source="web_ui",
+    )
+    assert result == {"operation": "rotate", "document_count": 1}
+
+
+@pytest.mark.django_db
+def test_document_action_queue_contains_only_serializable_parameters(bulk_setup):
+    user, client, doc, _ = bulk_setup
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = "rotate-task"
+        response = client.post(
+            "/api/documents/rotate/",
+            {"documents": [doc.pk], "degrees": 90, "from_webui": True},
+            format="json",
+        )
+    assert response.data == {"result": "OK", "task_id": "rotate-task"}
+    assert enqueue.call_args.kwargs["kwargs"] == {
+        "operation": "rotate",
+        "document_ids": [doc.pk],
+        "parameters": {"degrees": 90, "source_mode": "latest_version"},
+        "user_id": user.pk,
+        "trigger_source": "web_ui",
+    }
+
+
+@pytest.mark.django_db
+def test_reprocess_keeps_bulk_audit_entry(bulk_setup, settings):
+    user, _, doc, _ = bulk_setup
+    settings.AUDIT_LOG_ENABLED = True
+    LogEntry.objects.all().delete()
+    with mock.patch("documents.bulk_edit.reprocess"):
+        execute_bulk_operation("reprocess", [doc.pk], {}, user.pk)
+    entry = LogEntry.objects.get_for_object(doc).get()
+    assert entry.changes_dict["checksum"] == [doc.checksum, doc.checksum]
+    assert entry.actor_id == user.pk
+
+
+@pytest.mark.django_db
+def test_workflow_runs_in_tracked_task_instead_of_dispatching_another(bulk_setup):
+    user, _, doc, _ = bulk_setup
+    with mock.patch("documents.bulk_edit.run_workflows_for_documents") as run:
+        execute_bulk_operation(
+            "run_workflow",
+            [doc.pk],
+            {"workflow": 7, "filter_documents": False},
+            user.pk,
+        )
+    run.assert_called_once_with(
+        document_ids=[doc.pk],
+        workflow_id=7,
+        filter_documents=False,
+    )
+
+
+@pytest.mark.django_db
+def test_delete_failure_propagates_from_worker(bulk_setup):
+    user, _, doc, _ = bulk_setup
+    with mock.patch(
+        "documents.models.Document.delete",
+        side_effect=RuntimeError("delete failed"),
+    ):
+        with pytest.raises(RuntimeError, match="delete failed"):
+            execute_bulk_operation("delete", [doc.pk], {}, user.pk)
+    assert Document.objects.filter(pk=doc.pk).exists()
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "operation,parameters,preparation",
+    [
+        ("rotate", {"degrees": 90}, "rotate_pdf"),
+        ("split", {"pages": [[1]]}, "build_pdfs"),
+        ("delete_pages", {"pages": [1]}, "remove_pages"),
+    ],
+)
+def test_preparation_failure_propagates_from_worker(
+    bulk_setup,
+    operation,
+    parameters,
+    preparation,
+):
+    user, _, doc, _ = bulk_setup
+    doc.mime_type = "application/pdf"
+    doc.save()
+    with (
+        mock.patch(
+            f"documents.bulk_edit.pdf_ops.{preparation}",
+            side_effect=RuntimeError("preparation failed"),
+        ),
+        mock.patch("documents.bulk_edit.consume_file.apply_async") as consume,
+        mock.patch("documents.bulk_edit.consume_file.s") as consume_signature,
+        pytest.raises(RuntimeError, match="preparation failed"),
+    ):
+        execute_bulk_operation(operation, [doc.pk], parameters, user.pk)
+    consume.assert_not_called()
+    consume_signature.assert_not_called()
+
+
 @pytest.mark.django_db
 def test_queue_failure_does_not_change_documents(bulk_setup):
     _, client, doc, correspondent = bulk_setup
@@ -162,6 +276,22 @@ def test_tracked_outcome_records_initiator_and_result(bulk_setup, failed):
         assert task.result_data == {"document_count": 1}
 
 
+@pytest.mark.django_db
+def test_queued_delete_is_tracked_as_bulk_delete(bulk_setup):
+    user, _, doc, _ = bulk_setup
+    before_task_publish_handler(
+        headers={"task": execute_bulk_operation.name, "id": "delete-task"},
+        body=(
+            (),
+            {"operation": "delete", "document_ids": [doc.pk], "user_id": user.pk},
+            {},
+        ),
+    )
+    task = PaperlessTask.objects.get(task_id="delete-task")
+    assert task.task_type == PaperlessTask.TaskType.BULK_DELETE
+    assert task.owner_id == user.pk
+
+
 @pytest.mark.django_db
 def test_large_filtered_selection_is_one_task(bulk_setup):
     _, client, _, _ = bulk_setup
@@ -284,24 +414,46 @@ async def test_completion_socket_is_private(anyio_backend):
 
 
 @pytest.mark.django_db
-@pytest.mark.parametrize("endpoint", ["delete", "bulk_edit"])
-def test_trash_remains_synchronous(bulk_setup, endpoint):
-    _, client, doc, _ = bulk_setup
-    data = {"documents": [doc.pk]}
-    if endpoint == "bulk_edit":
-        data.update(method="delete", parameters={})
-    with (
-        mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue,
-        mock.patch("documents.search.get_backend"),
-    ):
+def test_legacy_trash_queues_without_deleting_in_request(bulk_setup):
+    user, client, doc, _ = bulk_setup
+    data = {"documents": [doc.pk], "method": "delete", "parameters": {}}
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = "delete-task"
         response = client.post(
-            f"/api/documents/{endpoint}/",
+            "/api/documents/bulk_edit/",
             data,
             format="json",
             HTTP_ACCEPT="application/json; version=9",
         )
     assert response.status_code == 200
-    assert response.data == {"result": "OK"}
-    assert not Document.objects.filter(pk=doc.pk).exists()
-    assert Document.deleted_objects.filter(pk=doc.pk).exists()
-    enqueue.assert_not_called()
+    assert response.data == {"result": "OK", "task_id": "delete-task"}
+    assert Document.objects.filter(pk=doc.pk).exists()
+    assert enqueue.call_args.kwargs["kwargs"] == {
+        "operation": "delete",
+        "document_ids": [doc.pk],
+        "parameters": {},
+        "user_id": user.pk,
+        "trigger_source": "api_upload",
+    }
+
+
+@pytest.mark.django_db
+def test_dedicated_trash_queues_without_deleting_in_request(bulk_setup):
+    user, client, doc, _ = bulk_setup
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = "delete-task"
+        response = client.post(
+            "/api/documents/delete/",
+            {"documents": [doc.pk]},
+            format="json",
+        )
+    assert response.status_code == 200
+    assert response.data == {"result": "OK", "task_id": "delete-task"}
+    assert Document.objects.filter(pk=doc.pk).exists()
+    assert enqueue.call_args.kwargs["kwargs"] == {
+        "operation": "delete",
+        "document_ids": [doc.pk],
+        "parameters": {},
+        "user_id": user.pk,
+        "trigger_source": "api_upload",
+    }

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_merge_documents_as_versions.py

@@ -8,6 +8,7 @@ from rest_framework import status
 from rest_framework.test import APITestCase
 
 from documents.bulk_edit import merge_as_versions
+from documents.bulk_operations import execute_bulk_operation
 from documents.models import Document
 from documents.serialisers import MergeDocumentsAsVersionsSerializer
 from paperless_testing.factories import UserFactory
@@ -34,6 +35,7 @@ class TestMergeDocumentsAsVersionsSerializer(TestCase):
             {
                 "documents": [self.doc1.id, self.doc2.id, self.doc3.id],
                 "root_document_id": self.doc2.id,
+                "from_webui": False,
             },
         )
 
@@ -103,67 +105,6 @@ class TestMergeDocumentsAsVersionsSerializer(TestCase):
         self.assertFalse(serializer.is_valid())
         self.assertIn("documents", serializer.errors)
 
-    def test_rejects_selected_version(self) -> None:
-        version = Document.objects.create(
-            checksum="D",
-            title="D",
-            root_document=self.doc1,
-            version_index=1,
-        )
-        serializer = MergeDocumentsAsVersionsSerializer(
-            data={
-                "documents": [version.id, self.doc2.id],
-                "root_document_id": self.doc2.id,
-            },
-        )
-
-        self.assertFalse(serializer.is_valid())
-        self.assertEqual(
-            serializer.errors["non_field_errors"][0],
-            "Only top-level documents can be merged as versions.",
-        )
-
-    def test_rejects_source_document_with_versions(self) -> None:
-        Document.objects.create(
-            checksum="D",
-            title="D",
-            root_document=self.doc1,
-            version_index=1,
-        )
-        serializer = MergeDocumentsAsVersionsSerializer(
-            data={
-                "documents": [self.doc1.id, self.doc2.id],
-                "root_document_id": self.doc2.id,
-            },
-        )
-
-        self.assertFalse(serializer.is_valid())
-        self.assertEqual(
-            serializer.errors["non_field_errors"][0],
-            "Documents with existing versions cannot be merged into another document.",
-        )
-
-    def test_rejects_source_document_with_trashed_versions(self) -> None:
-        version = Document.objects.create(
-            checksum="D",
-            title="D",
-            root_document=self.doc1,
-            version_index=1,
-        )
-        version.delete()  # trashed, but still points at doc1
-        serializer = MergeDocumentsAsVersionsSerializer(
-            data={
-                "documents": [self.doc1.id, self.doc2.id],
-                "root_document_id": self.doc2.id,
-            },
-        )
-
-        self.assertFalse(serializer.is_valid())
-        self.assertEqual(
-            serializer.errors["non_field_errors"][0],
-            "Documents with existing versions cannot be merged into another document.",
-        )
-
     def test_allows_root_document_with_versions(self) -> None:
         Document.objects.create(
             checksum="D",
@@ -182,6 +123,95 @@ class TestMergeDocumentsAsVersionsSerializer(TestCase):
 
 
 class TestMergeDocumentsAsVersions(TestCase):
+    def setUp(self) -> None:
+        self.doc1 = Document.objects.create(checksum="test-A", title="A")
+        self.doc2 = Document.objects.create(checksum="test-B", title="B")
+
+    def test_rejects_selected_version(self) -> None:
+        version = Document.objects.create(
+            checksum="D",
+            title="D",
+            root_document=self.doc1,
+            version_index=1,
+        )
+        with self.assertRaisesMessage(
+            ValueError,
+            "Only top-level documents can be merged as versions.",
+        ):
+            merge_as_versions(
+                [version.id, self.doc2.id],
+                root_document_id=self.doc2.id,
+            )
+
+    def test_rejects_source_document_with_versions(self) -> None:
+        Document.objects.create(
+            checksum="D",
+            title="D",
+            root_document=self.doc1,
+            version_index=1,
+        )
+        with self.assertRaisesMessage(
+            ValueError,
+            "Documents with existing versions cannot be merged into another document.",
+        ):
+            merge_as_versions(
+                [self.doc1.id, self.doc2.id],
+                root_document_id=self.doc2.id,
+            )
+
+    def test_rejects_source_document_with_trashed_versions(self) -> None:
+        version = Document.objects.create(
+            checksum="D",
+            title="D",
+            root_document=self.doc1,
+            version_index=1,
+        )
+        version.delete()  # trashed, but still points at doc1
+        with self.assertRaisesMessage(
+            ValueError,
+            "Documents with existing versions cannot be merged into another document.",
+        ):
+            merge_as_versions(
+                [self.doc1.id, self.doc2.id],
+                root_document_id=self.doc2.id,
+            )
+
+    @mock.patch("documents.bulk_edit.DocumentsStatusManager")
+    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.search.get_backend")
+    def test_rejects_source_that_gained_versions_after_validation(
+        self,
+        _get_backend_mock,
+        _bulk_update_mock,
+        _status_manager_mock,
+    ) -> None:
+        destination = Document.objects.create(checksum="destination")
+        serializer = MergeDocumentsAsVersionsSerializer(
+            data={
+                "documents": [self.doc1.id, destination.id],
+                "root_document_id": destination.id,
+            },
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+
+        merge_as_versions(
+            [self.doc1.id, self.doc2.id],
+            root_document_id=self.doc1.id,
+        )
+        with self.assertRaisesMessage(
+            ValueError,
+            "Documents with existing versions cannot be merged into another document.",
+        ):
+            merge_as_versions(
+                serializer.validated_data["documents"],
+                root_document_id=serializer.validated_data["root_document_id"],
+            )
+
+        self.doc1.refresh_from_db()
+        self.doc2.refresh_from_db()
+        self.assertIsNone(self.doc1.root_document_id)
+        self.assertEqual(self.doc2.root_document_id, self.doc1.id)
+
     @mock.patch("documents.bulk_edit.DocumentsStatusManager")
     @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
     @mock.patch("documents.search.get_backend")
@@ -348,6 +378,17 @@ class TestMergeDocumentsAsVersionsAPI(APITestCase):
             owner=self.user,
         )
         self.client.force_authenticate(user=self.user)
+        enqueue = mock.patch(
+            "documents.views.execute_bulk_operation.apply_async",
+            side_effect=lambda *, kwargs, headers: self.run_operation(kwargs),
+        )
+        enqueue.start()
+        self.addCleanup(enqueue.stop)
+
+    @staticmethod
+    def run_operation(kwargs):
+        execute_bulk_operation(**kwargs)
+        return mock.Mock(id="test-bulk-operation")
 
     @mock.patch("documents.views.bulk_edit.merge_as_versions")
     def test_merges_documents_as_versions(self, merge_mock) -> None:
@@ -367,7 +408,10 @@ class TestMergeDocumentsAsVersionsAPI(APITestCase):
         )
 
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.data, {"result": "OK"})
+        self.assertEqual(
+            response.data,
+            {"result": "OK", "task_id": "test-bulk-operation"},
+        )
         merge_mock.assert_called_once_with(
             [self.doc1.id, self.doc2.id],
             root_document_id=self.doc2.id,
@@ -440,8 +484,9 @@ class TestMergeDocumentsAsVersionsAPI(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
         merge_mock.assert_not_called()
 
-    @mock.patch("documents.views.bulk_edit.merge_as_versions")
-    def test_rejects_source_document_with_versions(self, merge_mock) -> None:
+    @mock.patch("documents.views.execute_bulk_operation.apply_async")
+    def test_queues_source_document_with_versions(self, enqueue_mock) -> None:
+        enqueue_mock.return_value.id = "queued-merge"
         Document.objects.create(
             checksum="C",
             title="C",
@@ -459,8 +504,9 @@ class TestMergeDocumentsAsVersionsAPI(APITestCase):
             format="json",
         )
 
-        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
-        merge_mock.assert_not_called()
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        enqueue_mock.assert_called_once()
+        self.assertEqual(response.data["task_id"], "queued-merge")
         self.doc1.refresh_from_db()
         self.assertIsNone(self.doc1.root_document_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/tests/test_workflows.py

@@ -2333,9 +2333,12 @@ class TestWorkflows(
         self.assertEqual(doc.owner, self.user2)
 
     def test_manual_workflow_run_missing_workflow(self) -> None:
-        """A missing workflow id logs a warning rather than raising."""
+        """A workflow removed after queueing fails the task."""
         doc = DocumentFactory()
-        with self.assertLogs("paperless.tasks", level="WARNING"):
+        with (
+            self.assertLogs("paperless.tasks", level="WARNING"),
+            self.assertRaises(Workflow.DoesNotExist),
+        ):
             tasks.run_workflows_for_documents([doc.pk], 9999)
 
     @mock.patch("documents.tasks.send_websocket_document_updated")
@@ -2356,7 +2359,13 @@ class TestWorkflows(
 
         mock_run_workflows.side_effect = side_effect
 
-        with self.assertLogs("paperless.tasks", level="ERROR"):
+        with (
+            self.assertLogs("paperless.tasks", level="ERROR"),
+            self.assertRaisesMessage(
+                RuntimeError,
+                f"failed for document(s): [{doc1.pk}]",
+            ),
+        ):
             tasks.run_workflows_for_documents([doc1.pk, doc2.pk], workflow.pk)
 
         # Both documents were attempted; the websocket update only fired for the

src/documents/views.py

@@ -107,9 +107,9 @@ from documents import bulk_edit
 from documents.bulk_download import ArchiveOnlyStrategy
 from documents.bulk_download import OriginalAndArchiveStrategy
 from documents.bulk_download import OriginalsOnlyStrategy
-from documents.bulk_operations import QUEUED_OPERATIONS
+from documents.bulk_operations import TRIGGER_SOURCE_OPERATIONS
+from documents.bulk_operations import USER_OPERATIONS
 from documents.bulk_operations import execute_bulk_operation
-from documents.bulk_operations import snapshot_field
 from documents.caching import get_llm_suggestion_cache
 from documents.caching import get_metadata_cache
 from documents.caching import get_suggestion_cache
@@ -2974,38 +2974,31 @@ class DocumentSelectionMixin:
 class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
     permission_classes = (IsAuthenticated,)
     parser_classes = (parsers.JSONParser,)
-    METHOD_NAMES_REQUIRING_USER = {
-        "split",
-        "merge",
-        "rotate",
-        "delete_pages",
-        "edit_pdf",
-        "remove_password",
-        "merge_as_versions",
-    }
-    # merge_as_versions doesn't queue any consume tasks
-    METHOD_NAMES_REQUIRING_TRIGGER_SOURCE = METHOD_NAMES_REQUIRING_USER - {
-        "merge_as_versions",
-    }
 
     def _queue_bulk_operation(self, method, documents, parameters, from_webui):
-        if "update_callback" in parameters:
-            raise ValueError("Unsupported bulk operation parameter: update_callback")
-        signature(method).bind(documents, **parameters)
+        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
+            else PaperlessTask.TriggerSource.API_UPLOAD
+        )
+        final_parameters = parameters.copy()
+        if method.__name__ in USER_OPERATIONS:
+            final_parameters["user"] = self.request.user
+        if method.__name__ in TRIGGER_SOURCE_OPERATIONS:
+            final_parameters["trigger_source"] = trigger_source
+        signature(method).bind(documents, **final_parameters)
         task = execute_bulk_operation.apply_async(
             kwargs={
                 "operation": method.__name__,
                 "document_ids": documents,
                 "parameters": parameters,
                 "user_id": self.request.user.pk,
+                "trigger_source": trigger_source,
             },
-            headers={
-                "trigger_source": (
-                    PaperlessTask.TriggerSource.WEB_UI
-                    if from_webui
-                    else PaperlessTask.TriggerSource.API_UPLOAD
-                ),
-            },
+            headers={"trigger_source": trigger_source},
         )
         return Response({"result": "OK", "task_id": task.id})
 
@@ -3129,15 +3122,6 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
         user = self.request.user
         from_webui = validated_data.get("from_webui", False)
 
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_USER:
-            parameters["user"] = user
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_TRIGGER_SOURCE:
-            parameters["trigger_source"] = (
-                PaperlessTask.TriggerSource.WEB_UI
-                if from_webui
-                else PaperlessTask.TriggerSource.API_UPLOAD
-            )
-
         if not has_document_operation_permissions(
             user=user,
             documents=documents,
@@ -3147,8 +3131,12 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
             return HttpResponseForbidden("Insufficient permissions")
 
         try:
-            result = method(documents, **parameters)
-            return Response({"result": result})
+            return self._queue_bulk_operation(
+                method,
+                documents,
+                parameters,
+                from_webui,
+            )
         except Exception as e:
             logger.warning(f"An error occurred performing {operation_label}: {e!s}")
             return HttpResponseBadRequest(
@@ -3169,19 +3157,13 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
                 name="BulkEditDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
-                    "task_id": serializers.CharField(required=False),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
     ),
 )
 class BulkEditView(DocumentOperationPermissionMixin):
-    LEGACY_MODIFIED_FIELDS = {
-        "delete": "deleted_at",
-        "reprocess": "checksum",
-        "remove_password": None,
-    }
-
     serializer_class = BulkEditSerializer
 
     def post(self, request, *args, **kwargs):
@@ -3211,14 +3193,6 @@ class BulkEditView(DocumentOperationPermissionMixin):
             user=user,
             validated_data=serializer.validated_data,
         )
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_USER:
-            parameters["user"] = user
-        if method.__name__ in self.METHOD_NAMES_REQUIRING_TRIGGER_SOURCE:
-            parameters["trigger_source"] = (
-                PaperlessTask.TriggerSource.WEB_UI
-                if from_webui
-                else PaperlessTask.TriggerSource.API_UPLOAD
-            )
         if not has_document_operation_permissions(
             user=user,
             documents=documents,
@@ -3228,38 +3202,12 @@ class BulkEditView(DocumentOperationPermissionMixin):
             return HttpResponseForbidden("Insufficient permissions")
 
         try:
-            if method.__name__ in QUEUED_OPERATIONS:
-                return self._queue_bulk_operation(
-                    method,
-                    documents,
-                    parameters,
-                    from_webui,
-                )
-            modified_field = self.LEGACY_MODIFIED_FIELDS.get(method.__name__)
-            if settings.AUDIT_LOG_ENABLED and modified_field:
-                old_values = snapshot_field(documents, modified_field)
-
-            result = method(documents, **parameters)
-
-            if settings.AUDIT_LOG_ENABLED and modified_field:
-                new_values = snapshot_field(documents, modified_field)
-                for doc in Document.objects.filter(pk__in=documents):
-                    LogEntry.objects.log_create(
-                        instance=doc,
-                        changes={
-                            modified_field: [
-                                old_values[doc.pk],
-                                new_values[doc.pk],
-                            ],
-                        },
-                        action=LogEntry.Action.UPDATE,
-                        actor=user,
-                        additional_data={
-                            "reason": f"Bulk edit: {method.__name__}",
-                        },
-                    )
-
-            return Response({"result": result})
+            return self._queue_bulk_operation(
+                method,
+                documents,
+                parameters,
+                from_webui,
+            )
         except Exception as e:
             logger.warning(f"An error occurred performing bulk edit: {e!s}")
             return HttpResponseBadRequest(
@@ -3276,6 +3224,7 @@ class BulkEditView(DocumentOperationPermissionMixin):
                 name="RotateDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
@@ -3303,6 +3252,7 @@ class RotateDocumentsView(DocumentOperationPermissionMixin):
                 name="MergeDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
@@ -3330,6 +3280,7 @@ class MergeDocumentsView(DocumentOperationPermissionMixin):
                 name="MergeDocumentsAsVersionsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
@@ -3357,6 +3308,7 @@ class MergeDocumentsAsVersionsView(DocumentOperationPermissionMixin):
                 name="DeleteDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
@@ -3384,6 +3336,7 @@ class DeleteDocumentsView(DocumentOperationPermissionMixin):
                 name="ReprocessDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
@@ -3411,6 +3364,7 @@ class ReprocessDocumentsView(DocumentOperationPermissionMixin):
                 name="EditPdfDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },
@@ -3438,6 +3392,7 @@ class EditPdfDocumentsView(DocumentOperationPermissionMixin):
                 name="RemovePasswordDocumentsResult",
                 fields={
                     "result": serializers.CharField(),
+                    "task_id": serializers.CharField(),
                 },
             ),
         },