Fix test_bulk_edit_query_at_exactly_the_cap_is_accepted

Add back the queued notificaiton + link
prevent refreshes from changing destination
2026-11-08 07:33:38 +00:00 · 2026-11-07 08:22:34 -08:00 · 2026-11-07 08:20:54 -08:00 · 2026-11-07 08:10:04 -08:00 · 2026-11-07 08:10:04 -08:00 · 2026-11-07 08:10:03 -08:00
50 changed files with 1549 additions and 1034 deletions
@@ -493,7 +493,7 @@ fields and permissions, which will be merged.

 #### Types {#workflow-trigger-types}

-Currently, there are five events that correspond to workflow trigger 'types':
+Currently, there are four events that correspond to workflow trigger 'types':

 1. **Consumption Started**: _before_ a document is consumed, so events can include filters by source (mail, consumption
   folder or API), file path, file name, mail rule
@@ -505,10 +505,8 @@ Currently, there are five events that correspond to workflow trigger 'types':
 4. **Scheduled**: a scheduled trigger that can be used to run workflows at a specific time. The date used can be either the document
   added, created, updated date or you can specify a (date) custom field. You can also specify a day offset from the date (positive
   offsets will trigger after the date, negative offsets will trigger before).
-5. **Manual**: a trigger that only runs when you explicitly choose to run the workflow against one or more existing documents. See
-   [running a workflow manually](#running-a-workflow-manually) below.

-The following flow diagram illustrates the document trigger types:
+The following flow diagram illustrates the four document trigger types:

 ```mermaid
 flowchart TD
@@ -528,10 +526,6 @@ flowchart TD
    matching
    trigger(s)"}

-    manual{"Matching
-    'Manual'
-    trigger(s)"}
-
    A[New Document] --> consumption
    consumption --> |Yes| C[Workflow Actions Run]
    consumption --> |No| D
@@ -549,10 +543,6 @@ flowchart TD
    scheduled --> |Yes| N[Workflow Actions Run]
    scheduled --> |No| O[Document Saved]
    N --> O
-    P[User Runs Workflow] --> manual
-    manual --> |Yes| Q[Workflow Actions Run]
-    manual --> |No| R[Document Saved]
-    Q --> R
 ```

 #### Filters {#workflow-trigger-filters}
@@ -564,9 +554,9 @@ Workflows allow you to filter by:
 - File path, including wildcards. Note that enabling `PAPERLESS_CONSUMER_RECURSIVE` would allow, for
  example, automatically assigning documents to different owners based on the upload directory.
 - Mail rule. Choosing this option will force 'mail fetch' to be the workflow source.
- Content matching (`Added`, `Updated`, `Scheduled` and `Manual` triggers only). Filter document content using the matching settings.
+- Content matching (`Added`, `Updated` and `Scheduled` triggers only). Filter document content using the matching settings.

-There are also 'advanced' filters available for `Added`, `Updated`, `Scheduled` and `Manual` triggers:
+There are also 'advanced' filters available for `Added`, `Updated` and `Scheduled` triggers:

 - Any Tags: Filter for documents with any of the specified tags.
 - All Tags: Filter for documents with all of the specified tags.
@@ -579,21 +569,6 @@ There are also 'advanced' filters available for `Added`, `Updated`, `Scheduled`
 - Not Storage paths: Filter documents without any of these storage paths.
 - Custom field query: Filter documents with a custom field query (the same as used for the document list filters).

-#### Running a workflow manually {#running-a-workflow-manually}
-
-A workflow with a **Manual** trigger does not run automatically. Instead, it can be run on demand against existing
-documents from the web UI:
-
- From the document list, select one or more documents and choose **Run workflow** from the "Actions" menu.
- From an individual document, choose **Run workflow** from the "Actions" menu.
-
-You will be asked which workflow to run. By default the workflow's trigger filters still apply, so documents that do
-not match are skipped. You can instead choose to **run on all selected documents**, which ignores the trigger filters
-and runs the workflow's actions on every selected document.
-
-Manual runs happen in the background. As with scheduled workflows, the document list is not refreshed automatically when
-they finish, so you may need to reload the list to see the results.
-
 ### Workflow Actions

 #### Types {#workflow-action-types}
@@ -674,7 +649,7 @@ new version of the same document, this time with normal OCR/text extraction appl
 a password-protected file added with this trigger will briefly exist as an un-OCR'd version before
 the properly processed version is created.

-**Document Added**, **Document Updated**, **Scheduled**, **Manual**: these triggers run against a document that
+**Document Added**, **Document Updated**, **Scheduled**: these triggers run against a document that
 already exists, so password removal happens immediately: the decrypted content is queued for
 consumption as a new version right away. Note that if the document's initial consumption also
 happened while it was still encrypted, that original version will likewise be missing OCR text.
@@ -7,10 +7,15 @@ import {
 import { ComponentFixture, TestBed } from '@angular/core/testing'
 import { FormsModule } from '@angular/forms'
 import { By } from '@angular/platform-browser'
-import { Router } from '@angular/router'
+import {
+  ActivatedRoute,
+  convertToParamMap,
+  ParamMap,
+  Router,
+} from '@angular/router'
 import { RouterTestingModule } from '@angular/router/testing'
 import { NgbModal, NgbModalRef, NgbModule } from '@ng-bootstrap/ng-bootstrap'
-import { of, throwError } from 'rxjs'
+import { BehaviorSubject, of, throwError } from 'rxjs'
 import { routes } from 'src/app/app-routing.module'
 import {
  PaperlessTask,
@@ -159,6 +164,7 @@ const sectionCountResponse = {
 }

 describe('TasksComponent', () => {
+  let queryParams: BehaviorSubject<ParamMap>
  let component: TasksComponent
  let fixture: ComponentFixture<TasksComponent>
  let tasksService: TasksService
@@ -169,6 +175,7 @@ describe('TasksComponent', () => {
  let toastService: ToastService

  beforeEach(async () => {
+    queryParams = new BehaviorSubject(convertToParamMap({}))
    TestBed.configureTestingModule({
      imports: [
        NgbModule,
@@ -181,6 +188,17 @@ describe('TasksComponent', () => {
        ConfirmDialogComponent,
      ],
      providers: [
+        {
+          provide: ActivatedRoute,
+          useValue: {
+            queryParamMap: queryParams.asObservable(),
+            snapshot: {
+              get queryParamMap() {
+                return queryParams.value
+              },
+            },
+          },
+        },
        {
          provide: PermissionsService,
          useValue: {
@@ -238,6 +256,71 @@ describe('TasksComponent', () => {
    flushTaskLists(25, count)
  }

+  it('consumes a task link on load into the normal resettable Task ID filter', () => {
+    jest
+      .spyOn(tasksService, 'list')
+      .mockReturnValue(of({ count: 0, results: [] }))
+    jest
+      .spyOn(tasksService, 'statusCounts')
+      .mockReturnValue(of(sectionCountResponse))
+    jest.spyOn(router, 'navigate').mockResolvedValue(true)
+    queryParams.next(convertToParamMap({ task_id: 'linked-task' }))
+    fixture.destroy()
+    fixture = TestBed.createComponent(TasksComponent)
+    component = fixture.componentInstance
+    fixture.detectChanges()
+    expect(router.navigate).toHaveBeenCalledTimes(1)
+    expect(router.navigate).toHaveBeenCalledWith(
+      [],
+      expect.objectContaining({
+        queryParams: { task_id: null },
+        replaceUrl: true,
+      })
+    )
+    expect(component.filterTargetName).toBe('Task ID')
+    expect(component.filterText).toBe('linked-task')
+    expect(component.isFiltered).toBe(true)
+    component.setSection(TaskSection.InProgress)
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      task_id: 'linked-task',
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+    component.resetFilters()
+    expect(component.filterText).toBe('')
+    expect(component.isFiltered).toBe(false)
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      acknowledged: false,
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+    expect(router.navigate).toHaveBeenCalledTimes(1)
+  })
+
+  it('supports manually filtering by Task ID and clearing the text filter', () => {
+    jest
+      .spyOn(tasksService, 'list')
+      .mockReturnValue(of({ count: 0, results: [] }))
+    jest
+      .spyOn(tasksService, 'statusCounts')
+      .mockReturnValue(of(sectionCountResponse))
+    component.setSection(TaskSection.InProgress)
+    component.setFilterTarget(TaskFilterTargetID.TaskId)
+    component.filterText = tasks[0].task_id
+    jest.advanceTimersByTime(150)
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      task_id: tasks[0].task_id,
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+    component.pagedTasks.set(tasks)
+    expect(component.tasksForSection(TaskSection.NeedsAttention)).toEqual([
+      tasks[0],
+    ])
+    component.resetFilter()
+    expect(tasksService.list).toHaveBeenLastCalledWith(1, 25, {
+      acknowledged: false,
+      status: [PaperlessTaskStatus.Pending, PaperlessTaskStatus.Started],
+    })
+  })
+
  it('should display task sections with counts', () => {
    expect(component.selectedSection).toBe(TaskSection.All)
    expect(component.selectedTaskType).toBeNull()
@@ -1,7 +1,7 @@
 import { JsonPipe, NgTemplateOutlet } from '@angular/common'
 import { Component, inject, OnDestroy, OnInit, signal } from '@angular/core'
 import { FormsModule, ReactiveFormsModule } from '@angular/forms'
-import { Router, RouterLink } from '@angular/router'
+import { ActivatedRoute, Router, RouterLink } from '@angular/router'
 import {
  NgbCollapseModule,
  NgbDropdownModule,
@@ -47,11 +47,13 @@ export enum TaskSection {
 export enum TaskFilterTargetID {
  Name,
  Result,
+  TaskId,
 }

 const FILTER_TARGETS = [
  { id: TaskFilterTargetID.Name, name: $localize`Name` },
  { id: TaskFilterTargetID.Result, name: $localize`Result` },
+  { id: TaskFilterTargetID.TaskId, name: $localize`Task ID` },
 ]

 const SECTION_LABELS = {
@@ -161,6 +163,7 @@ export class TasksComponent
  tasksService = inject(TasksService)
  private modalService = inject(NgbModal)
  private readonly router = inject(Router)
+  private readonly route = inject(ActivatedRoute)
  private readonly toastService = inject(ToastService)

  readonly TaskSection = TaskSection
@@ -275,6 +278,17 @@ export class TasksComponent
  }

  ngOnInit() {
+    const taskId = this.route.snapshot.queryParamMap.get('task_id')
+    if (taskId) {
+      this.filterTargetID = TaskFilterTargetID.TaskId
+      this._filterText = taskId
+      this.router.navigate([], {
+        relativeTo: this.route,
+        queryParams: { task_id: null },
+        queryParamsHandling: 'merge',
+        replaceUrl: true,
+      })
+    }
    this.tasksService.reload()
    this.reloadPage()
    timer(5000, 5000)
@@ -636,6 +650,9 @@ export class TasksComponent

    const query = this._filterText.toLowerCase()

+    if (this.filterTargetID === TaskFilterTargetID.TaskId) {
+      return task.task_id.toLowerCase() === query
+    }
    if (this.filterTargetID == TaskFilterTargetID.Name) {
      return [
        this.taskDisplayName(task),
@@ -706,9 +723,15 @@ export class TasksComponent
    }

    if (this._filterText.length) {
-      params[
-        this.filterTargetID === TaskFilterTargetID.Name ? 'name' : 'result'
-      ] = this._filterText
+      const parameter = {
+        [TaskFilterTargetID.Name]: 'name',
+        [TaskFilterTargetID.Result]: 'result',
+        [TaskFilterTargetID.TaskId]: 'task_id',
+      }[this.filterTargetID]
+      params[parameter] = this._filterText
+      if (this.filterTargetID === TaskFilterTargetID.TaskId) {
+        delete params.acknowledged
+      }
    }

    return params
@@ -11,8 +11,10 @@ import { ActivatedRoute, Router } from '@angular/router'
 import { RouterTestingModule } from '@angular/router/testing'
 import { NgbModal, NgbModalModule, NgbModule } from '@ng-bootstrap/ng-bootstrap'
 import { provideUiTour } from 'ngx-ui-tour-ng-bootstrap'
-import { of, throwError } from 'rxjs'
+import { of, Subject, throwError } from 'rxjs'
 import { routes } from 'src/app/app-routing.module'
+import { BulkOperationStatus } from 'src/app/data/bulk-operation'
+import { PaperlessTaskStatus } from 'src/app/data/paperless-task'
 import { SavedView } from 'src/app/data/saved-view'
 import { HideableSidebarItemID, SETTINGS_KEYS } from 'src/app/data/ui-settings'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
@@ -21,10 +23,11 @@ import {
  DjangoMessageLevel,
  DjangoMessagesService,
 } from 'src/app/services/django-messages.service'
+import { DocumentListViewService } from 'src/app/services/document-list-view.service'
 import { OpenDocumentsService } from 'src/app/services/open-documents.service'
 import {
-  PermissionType,
  PermissionsService,
+  PermissionType,
 } from 'src/app/services/permissions.service'
 import { RemoteVersionService } from 'src/app/services/rest/remote-version.service'
 import { SavedViewService } from 'src/app/services/rest/saved-view.service'
@@ -95,6 +98,7 @@ describe('AppFrameComponent', () => {
  let savedViewSpy
  let modalService: NgbModal
  let maybeRefreshSpy
+  let bulkEvents: Subject<BulkOperationStatus>
  let tasksService: TasksService

  beforeEach(async () => {
@@ -182,6 +186,10 @@ describe('AppFrameComponent', () => {
    savedViewSpy = jest.spyOn(savedViewService, 'reload')
    maybeRefreshSpy = jest.spyOn(savedViewService, 'maybeRefreshDocumentCounts')

+    bulkEvents = new Subject<BulkOperationStatus>()
+    jest
+      .spyOn(TestBed.inject(WebsocketStatusService), 'onBulkOperationStatus')
+      .mockReturnValue(bulkEvents.asObservable())
    fixture = TestBed.createComponent(AppFrameComponent)
    component = fixture.componentInstance

@@ -190,6 +198,71 @@ describe('AppFrameComponent', () => {
    fixture.detectChanges()
  })

+  it('refreshes and notifies on completion without client-side task tracking', () => {
+    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 refresh = jest
+      .spyOn(openDocumentsService, 'refreshDocument')
+      .mockImplementation(() => {})
+    jest
+      .spyOn(openDocumentsService, 'getOpenDocuments')
+      .mockReturnValue([document])
+    tasksReload.mockClear()
+    maybeRefreshSpy.mockClear()
+    bulkEvents.next({
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: PaperlessTaskStatus.Started,
+    })
+    expect(reload).not.toHaveBeenCalled()
+    bulkEvents.next({
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(reload).toHaveBeenCalledTimes(1)
+    expect(reload).toHaveBeenCalledWith(null, false)
+    expect(refresh).toHaveBeenCalledWith(document.id)
+    expect(maybeRefreshSpy).toHaveBeenCalledTimes(1)
+    expect(tasksReload).toHaveBeenCalledTimes(1)
+    expect(success).toHaveBeenCalledWith('Bulk operation completed.')
+    fixture.destroy()
+    bulkEvents.next({
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: PaperlessTaskStatus.Success,
+    })
+    expect(reload).toHaveBeenCalledTimes(1)
+  })
+
+  it('refreshes and reports failed or revoked bulk operations', () => {
+    const list = TestBed.inject(DocumentListViewService)
+    const reload = jest.spyOn(list, 'reload').mockImplementation(() => {})
+    jest.spyOn(list, 'reduceSelectionToFilter').mockImplementation(() => {})
+    jest.spyOn(tasksService, 'reload').mockImplementation(() => {})
+    const error = jest.spyOn(toastService, 'showError')
+    const success = jest.spyOn(toastService, 'showSuccess')
+    for (const status of [
+      PaperlessTaskStatus.Failure,
+      PaperlessTaskStatus.Revoked,
+    ]) {
+      bulkEvents.next({
+        task_id: 'bulk-task',
+        operation: 'set_permissions',
+        status,
+        error: 'Could not update',
+      })
+    }
+    expect(reload).toHaveBeenCalledTimes(2)
+    expect(error).toHaveBeenCalledWith(expect.any(String), 'Could not update')
+    expect(success).not.toHaveBeenCalled()
+  })
+
  it('should initialize the saved view service', () => {
    expect(savedViewSpy).toHaveBeenCalled()
  })
@@ -8,12 +8,14 @@ import {
 import { NgClass } from '@angular/common'
 import {
  Component,
+  DestroyRef,
  HostListener,
  inject,
  OnInit,
  signal,
  viewChild,
 } from '@angular/core'
+import { takeUntilDestroyed } from '@angular/core/rxjs-interop'
 import { FormsModule } from '@angular/forms'
 import { ActivatedRoute, Router, RouterModule } from '@angular/router'
 import {
@@ -24,10 +26,11 @@ import {
  NgbPopoverModule,
 } from '@ng-bootstrap/ng-bootstrap'
 import { TourNgBootstrap } from 'ngx-ui-tour-ng-bootstrap'
-import { Observable } from 'rxjs'
+import { filter, Observable } from 'rxjs'
 import { first } from 'rxjs/operators'
 import { BootstrapIconComponent } from 'src/app/components/common/bootstrap-icon/bootstrap-icon.component'
 import { Document } from 'src/app/data/document'
+import { PaperlessTaskStatus } from 'src/app/data/paperless-task'
 import { SavedView } from 'src/app/data/saved-view'
 import {
  CollapsibleSection,
@@ -41,6 +44,7 @@ import {
  DjangoMessageLevel,
  DjangoMessagesService,
 } from 'src/app/services/django-messages.service'
+import { DocumentListViewService } from 'src/app/services/document-list-view.service'
 import { OpenDocumentsService } from 'src/app/services/open-documents.service'
 import {
  PermissionAction,
@@ -101,6 +105,8 @@ export class AppFrameComponent
  extends ComponentWithPermissions
  implements OnInit, ComponentCanDeactivate
 {
+  private destroyRef = inject(DestroyRef)
+  private documentList = inject(DocumentListViewService)
  router = inject(Router)
  private activatedRoute = inject(ActivatedRoute)
  private openDocumentsService = inject(OpenDocumentsService)
@@ -164,6 +170,44 @@ export class AppFrameComponent
  }

  ngOnInit(): void {
+    this.websocketStatusService
+      .onBulkOperationStatus()
+      .pipe(
+        filter(({ status }) =>
+          [
+            PaperlessTaskStatus.Success,
+            PaperlessTaskStatus.Failure,
+            PaperlessTaskStatus.Revoked,
+          ].includes(status)
+        ),
+        takeUntilDestroyed(this.destroyRef)
+      )
+      .subscribe(({ status, error }) => {
+        this.documentList.reload(null, false)
+        this.documentList.reduceSelectionToFilter()
+        // maybe someday we can optimize to only refresh affected documents
+        this.openDocumentsService.getOpenDocuments().forEach((doc) => {
+          this.openDocumentsService.refreshDocument(doc.id)
+        })
+        this.savedViewService.maybeRefreshDocumentCounts()
+        if (
+          this.permissionsService.currentUserCan(
+            PermissionAction.View,
+            PermissionType.PaperlessTask
+          )
+        ) {
+          this.tasksService.reload()
+        }
+        if (status === PaperlessTaskStatus.Success) {
+          this.toastService.showSuccess($localize`Bulk operation completed.`)
+        } else {
+          this.toastService.showError(
+            $localize`Bulk operation failed. Some changes may already have been applied.`,
+            error
+          )
+        }
+      })
+
    this.lastScrollY = window.scrollY
    this.detectClassicScrollbars()

@@ -118,7 +118,6 @@ import personFill from 'bootstrap-icons/icons/person-fill.svg'
 import personLock from 'bootstrap-icons/icons/person-lock.svg'
 import personSquare from 'bootstrap-icons/icons/person-square.svg'
 import person from 'bootstrap-icons/icons/person.svg'
-import playCircle from 'bootstrap-icons/icons/play-circle.svg'
 import playFill from 'bootstrap-icons/icons/play-fill.svg'
 import plusCircle from 'bootstrap-icons/icons/plus-circle.svg'
 import plus from 'bootstrap-icons/icons/plus.svg'
@@ -280,7 +279,6 @@ export const icons: Record<string, string> = {
  personFillLock,
  personLock,
  personSquare,
-  playCircle,
  playFill,
  plus,
  plusCircle,
@@ -1,32 +0,0 @@
-<div class="modal-header">
-  <h4 class="modal-title" id="modal-basic-title">{{title}}</h4>
-  <button type="button" class="btn-close" aria-label="Close" (click)="cancel()">
-  </button>
-</div>
-<div class="modal-body">
-  @if (message) {
-    <p class="text-break" [innerHTML]="message"></p>
-  }
-  <div class="mb-3">
-    <label class="form-label" for="runWorkflowSelect" i18n>Workflow</label>
-    <select class="form-select" id="runWorkflowSelect" [(ngModel)]="workflowId" (ngModelChange)="workflowSelected()">
-      <option [ngValue]="null" disabled i18n>Select a workflow</option>
-      @for (workflow of workflows; track workflow.id) {
-        <option [ngValue]="workflow.id">{{workflow.name}}</option>
-      }
-    </select>
-  </div>
-  <div class="form-check">
-    <input class="form-check-input" type="checkbox" id="runWorkflowForce" [(ngModel)]="force" />
-    <label class="form-check-label" for="runWorkflowForce" i18n>Run on all selected documents</label>
-    <div class="form-text" i18n>Ignore the workflow's trigger filters and run on every selected document.</div>
-  </div>
-</div>
-<div class="modal-footer">
-  <button type="button" class="btn" [class]="cancelBtnClass" (click)="cancel()" [disabled]="!buttonsEnabled">
-    <span class="d-inline-block" style="padding-bottom: 1px;">{{cancelBtnCaption}}</span>
-  </button>
-  <button type="button" class="btn" [class]="btnClass" (click)="confirm()" [disabled]="!confirmButtonEnabled || !buttonsEnabled">
-    {{btnCaption}}
-  </button>
-</div>
@@ -1,42 +0,0 @@
-import { ComponentFixture, TestBed } from '@angular/core/testing'
-import { NgbActiveModal } from '@ng-bootstrap/ng-bootstrap'
-import { RunWorkflowConfirmDialogComponent } from './run-workflow-confirm-dialog.component'
-
-describe('RunWorkflowConfirmDialogComponent', () => {
-  let component: RunWorkflowConfirmDialogComponent
-  let fixture: ComponentFixture<RunWorkflowConfirmDialogComponent>
-
-  beforeEach(async () => {
-    TestBed.configureTestingModule({
-      providers: [NgbActiveModal],
-      imports: [RunWorkflowConfirmDialogComponent],
-    }).compileComponents()
-
-    fixture = TestBed.createComponent(RunWorkflowConfirmDialogComponent)
-    component = fixture.componentInstance
-    component.workflows = [{ id: 5, name: 'Manual workflow' } as any]
-    fixture.detectChanges()
-  })
-
-  it('should disable confirm until a workflow is selected and not force by default', () => {
-    expect(component.confirmButtonEnabled).toBeFalsy()
-    expect(component.force).toBeFalsy()
-
-    component.workflowId = 5
-    component.workflowSelected()
-    expect(component.confirmButtonEnabled).toBeTruthy()
-
-    component.workflowId = null
-    component.workflowSelected()
-    expect(component.confirmButtonEnabled).toBeFalsy()
-  })
-
-  it('should render an option per workflow', () => {
-    const options = fixture.nativeElement.querySelectorAll(
-      '#runWorkflowSelect option'
-    )
-    // placeholder option + one per workflow
-    expect(options).toHaveLength(2)
-    expect(options[1].textContent).toContain('Manual workflow')
-  })
-})
@@ -1,24 +0,0 @@
-import { Component } from '@angular/core'
-import { FormsModule } from '@angular/forms'
-import { Workflow } from 'src/app/data/workflow'
-import { ConfirmDialogComponent } from '../confirm-dialog.component'
-
-@Component({
-  selector: 'pngx-run-workflow-confirm-dialog',
-  templateUrl: './run-workflow-confirm-dialog.component.html',
-  imports: [FormsModule],
-})
-export class RunWorkflowConfirmDialogComponent extends ConfirmDialogComponent {
-  public workflows: Workflow[] = []
-
-  public workflowId: number = null
-
-  // i.e. ignore the workflow's trigger filters
-  public force: boolean = false
-
-  override confirmButtonEnabled = false
-
-  public workflowSelected(): void {
-    this.confirmButtonEnabled = !!this.workflowId
-  }
-}
@@ -164,7 +164,7 @@
          <pngx-input-text i18n-title title="Filter path" formControlName="filter_path" horizontal="true" i18n-hint hint="Apply to documents that match this path. Wildcards specified as * are allowed. Case-normalized." [error]="error?.filter_path"></pngx-input-text>
          <pngx-input-select i18n-title title="Filter mail rule" [items]="mailRules()" horizontal="true" [allowNull]="true" formControlName="filter_mailrule" i18n-hint hint="Apply to documents consumed via this mail rule." [error]="error?.filter_mailrule"></pngx-input-select>
        }
-        @if (formGroup.get('type').value === WorkflowTriggerType.DocumentAdded || formGroup.get('type').value === WorkflowTriggerType.DocumentUpdated || formGroup.get('type').value === WorkflowTriggerType.Scheduled || formGroup.get('type').value === WorkflowTriggerType.Manual) {
+        @if (formGroup.get('type').value === WorkflowTriggerType.DocumentAdded || formGroup.get('type').value === WorkflowTriggerType.DocumentUpdated || formGroup.get('type').value === WorkflowTriggerType.Scheduled) {
          <pngx-input-select i18n-title title="Content matching algorithm" horizontal="true" [items]="getMatchingAlgorithms()" formControlName="matching_algorithm"></pngx-input-select>
          @if (matchingPatternRequired(formGroup)) {
            <pngx-input-text i18n-title title="Content matching pattern" horizontal="true" formControlName="match" [error]="error?.match"></pngx-input-text>
@@ -175,7 +175,7 @@
        }
      </div>
    </div>
-    @if (formGroup.get('type').value === WorkflowTriggerType.DocumentAdded || formGroup.get('type').value === WorkflowTriggerType.DocumentUpdated || formGroup.get('type').value === WorkflowTriggerType.Scheduled || formGroup.get('type').value === WorkflowTriggerType.Manual) {
+    @if (formGroup.get('type').value === WorkflowTriggerType.DocumentAdded || formGroup.get('type').value === WorkflowTriggerType.DocumentUpdated || formGroup.get('type').value === WorkflowTriggerType.Scheduled) {
      <div class="row mt-3">
        <div class="col">
          <div class="trigger-filters mb-3">
@@ -123,10 +123,6 @@ export const WORKFLOW_TYPE_OPTIONS = [
    id: WorkflowTriggerType.Scheduled,
    name: $localize`Scheduled`,
  },
-  {
-    id: WorkflowTriggerType.Manual,
-    name: $localize`Manual`,
-  },
 ]

 export const WORKFLOW_ACTION_OPTIONS = [
@@ -49,12 +49,6 @@
        <pngx-bs-icon width="1em" height="1em" name="arrow-counterclockwise" class="me-2"></pngx-bs-icon><span i18n>Reprocess</span>
      </button>

-      @if (manualWorkflows().length > 0) {
-        <button ngbDropdownItem (click)="runWorkflow()" [disabled]="!userCanEdit || !userIsOwner">
-          <pngx-bs-icon width="1em" height="1em" name="play-circle" class="me-2"></pngx-bs-icon><span i18n>Run workflow</span>
-        </button>
-      }
-
      <button ngbDropdownItem (click)="printDocument()" [hidden]="useNativePdfViewer || isMobile">
        <pngx-bs-icon width="1em" height="1em" name="printer" class="me-2"></pngx-bs-icon><span i18n>Print</span>
      </button>
@@ -1030,27 +1030,6 @@ describe('DocumentDetailComponent', () => {
    expect(modalCloseSpy).toHaveBeenCalled()
  })

-  it('should run a workflow on the document, confirm and close modal', () => {
-    initNormally()
-    const bulkEditSpy = jest.spyOn(documentService, 'bulkEdit')
-    bulkEditSpy.mockReturnValue(of(true))
-    let openModal: NgbModalRef
-    modalService.activeInstances.subscribe((modal) => (openModal = modal[0]))
-    const toastSpy = jest.spyOn(toastService, 'showInfo')
-    component.runWorkflow()
-    const modalCloseSpy = jest.spyOn(openModal, 'close')
-    openModal.componentInstance.workflowId = 5
-    openModal.componentInstance.force = true
-    openModal.componentInstance.confirmClicked.next()
-    expect(bulkEditSpy).toHaveBeenCalledWith(
-      { documents: [doc.id] },
-      'run_workflow',
-      { workflow: 5, filter_documents: false }
-    )
-    expect(toastSpy).toHaveBeenCalled()
-    expect(modalCloseSpy).toHaveBeenCalled()
-  })
-
  it('should pass remote OCR choice when reprocessing', () => {
    initNormally()
    const reprocessSpy = jest.spyOn(documentService, 'reprocessDocuments')
@@ -68,8 +68,6 @@ import { StoragePath } from 'src/app/data/storage-path'
 import { Tag } from 'src/app/data/tag'
 import { SETTINGS_KEYS, SuggestionSource } from 'src/app/data/ui-settings'
 import { User } from 'src/app/data/user'
-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 { CustomDatePipe } from 'src/app/pipes/custom-date.pipe'
 import { DocumentTitlePipe } from 'src/app/pipes/document-title.pipe'
@@ -95,7 +93,6 @@ import { SavedViewService } from 'src/app/services/rest/saved-view.service'
 import { StoragePathService } from 'src/app/services/rest/storage-path.service'
 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 { ToastService } from 'src/app/services/toast.service'
 import { TrashService } from 'src/app/services/trash.service'
@@ -107,7 +104,6 @@ import { DocumentDetailFieldID } from '../admin/settings/settings.component'
 import { ConfirmDialogComponent } from '../common/confirm-dialog/confirm-dialog.component'
 import { PasswordRemovalConfirmDialogComponent } from '../common/confirm-dialog/password-removal-confirm-dialog/password-removal-confirm-dialog.component'
 import { ReprocessConfirmDialogComponent } from '../common/confirm-dialog/reprocess-confirm-dialog/reprocess-confirm-dialog.component'
-import { RunWorkflowConfirmDialogComponent } from '../common/confirm-dialog/run-workflow-confirm-dialog/run-workflow-confirm-dialog.component'
 import { CustomFieldsDropdownComponent } from '../common/custom-fields-dropdown/custom-fields-dropdown.component'
 import { CorrespondentEditDialogComponent } from '../common/edit-dialog/correspondent-edit-dialog/correspondent-edit-dialog.component'
 import { DocumentTypeEditDialogComponent } from '../common/edit-dialog/document-type-edit-dialog/document-type-edit-dialog.component'
@@ -237,7 +233,6 @@ export class DocumentDetailComponent
  private componentRouterService = inject(ComponentRouterService)
  private deviceDetectorService = inject(DeviceDetectorService)
  private savedViewService = inject(SavedViewService)
-  private readonly workflowService = inject(WorkflowService)
  private readonly websocketStatusService = inject(WebsocketStatusService)
  private readonly useNativePdfViewerSetting = this.settings.getSignal<boolean>(
    SETTINGS_KEYS.USE_NATIVE_PDF_VIEWER
@@ -300,7 +295,6 @@ export class DocumentDetailComponent
  readonly correspondents = signal<Correspondent[]>(undefined)
  readonly documentTypes = signal<DocumentType[]>(undefined)
  readonly storagePaths = signal<StoragePath[]>(undefined)
-  readonly manualWorkflows = signal<Workflow[]>([])

  documentForm: FormGroup = new FormGroup({
    title: new FormControl(''),
@@ -842,19 +836,6 @@ export class DocumentDetailComponent
        .pipe(first(), takeUntil(this.unsubscribeNotifier))
        .subscribe((result) => this.users.set(result.results))
    }
-    if (
-      this.permissionsService.currentUserCan(
-        PermissionAction.View,
-        PermissionType.Workflow
-      )
-    ) {
-      this.workflowService
-        .list(1, 100000, null, null, {
-          trigger_type: WorkflowTriggerType.Manual,
-        })
-        .pipe(first(), takeUntil(this.unsubscribeNotifier))
-        .subscribe((result) => this.manualWorkflows.set(result.results))
-    }

    this.getCustomFields()

@@ -1578,39 +1559,6 @@ export class DocumentDetailComponent
    })
  }

-  runWorkflow() {
-    const modal = this.modalService.open(RunWorkflowConfirmDialogComponent, {
-      backdrop: 'static',
-    })
-    modal.componentInstance.title = $localize`Run workflow`
-    modal.componentInstance.message = $localize`Run a workflow on "${this.document().title}".`
-    modal.componentInstance.btnCaption = $localize`Run`
-    modal.componentInstance.workflows = this.manualWorkflows()
-    modal.componentInstance.confirmClicked.subscribe(() => {
-      modal.componentInstance.buttonsEnabled.set(false)
-      this.documentsService
-        .bulkEdit({ documents: [this.document().id] }, 'run_workflow', {
-          workflow: modal.componentInstance.workflowId,
-          filter_documents: !modal.componentInstance.force,
-        })
-        .subscribe({
-          next: () => {
-            this.toastService.showInfo(
-              $localize`Workflow will run on "${this.document().title}" in the background.`
-            )
-            modal.close()
-          },
-          error: (error) => {
-            modal.componentInstance.buttonsEnabled.set(true)
-            this.toastService.showError(
-              $localize`Error executing operation`,
-              error
-            )
-          },
-        })
-    })
-  }
-
  private getSelectedNonLatestVersionId(): number | null {
    const versions = this.document()?.versions ?? []
    if (!versions.length || !this.selectedVersionId()) {
@@ -98,11 +98,6 @@
          <button ngbDropdownItem (click)="mergeSelectedAsVersions()" [disabled]="!userOwnsAll || !userCanEditAll || !userCanDelete || list.allSelected || list.selectedCount < 2">
            <pngx-bs-icon name="journal-bookmark-fill" class="me-2"></pngx-bs-icon><ng-container i18n>Merge as versions</ng-container>
          </button>
-          @if (manualWorkflows.length > 0) {
-            <button ngbDropdownItem (click)="runWorkflowSelected()" [disabled]="!canRunWorkflow">
-              <pngx-bs-icon name="play-circle" class="me-2"></pngx-bs-icon><ng-container i18n>Run workflow</ng-container>
-            </button>
-          }
        </div>
      </div>
    </div>
@@ -17,7 +17,6 @@ import { FILTER_TITLE } from 'src/app/data/filter-rule-type'
 import { Results } 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'
 import { FilterPipe } from 'src/app/pipes/filter.pipe'
 import { DocumentListViewService } from 'src/app/services/document-list-view.service'
 import {
@@ -37,7 +36,6 @@ import { ShareLinkBundleService } from 'src/app/services/rest/share-link-bundle.
 import { StoragePathService } from 'src/app/services/rest/storage-path.service'
 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 { ToastService } from 'src/app/services/toast.service'
 import { TrashService } from 'src/app/services/trash.service'
@@ -85,7 +83,6 @@ describe('BulkEditorComponent', () => {
  let customFieldsService: CustomFieldsService
  let httpTestingController: HttpTestingController
  let shareLinkBundleService: ShareLinkBundleService
-  let workflowService: WorkflowService
  let router: Router

  beforeEach(async () => {
@@ -176,22 +173,6 @@ describe('BulkEditorComponent', () => {
            delete: jest.fn(),
          },
        },
-        {
-          provide: WorkflowService,
-          useValue: {
-            list: jest.fn(() =>
-              of({
-                results: [
-                  {
-                    id: 5,
-                    name: 'Manual workflow',
-                    triggers: [{ type: WorkflowTriggerType.Manual }],
-                  },
-                ],
-              })
-            ),
-          },
-        },
        {
          provide: Router,
          useValue: { navigate: jest.fn().mockResolvedValue(true) },
@@ -214,7 +195,6 @@ describe('BulkEditorComponent', () => {
    customFieldsService = TestBed.inject(CustomFieldsService)
    httpTestingController = TestBed.inject(HttpTestingController)
    shareLinkBundleService = TestBed.inject(ShareLinkBundleService)
-    workflowService = TestBed.inject(WorkflowService)
    router = TestBed.inject(Router)

    fixture = TestBed.createComponent(BulkEditorComponent)
@@ -579,18 +559,19 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: [3, 4],
      method: 'modify_tags',
      parameters: { add_tags: [101], remove_tags: [] },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should execute modify tags bulk operation for all filtered documents', () => {
@@ -627,17 +608,18 @@ describe('BulkEditorComponent', () => {
    const req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      all: true,
      filters: { title_search: 'apple' },
      excluded_documents: [4],
      method: 'modify_tags',
      parameters: { add_tags: [101], remove_tags: [] },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
+    )
  })

  it('should execute modify tags bulk operation with confirmation dialog if enabled', () => {
@@ -664,56 +646,12 @@ describe('BulkEditorComponent', () => {
    httpTestingController
      .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
      .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
-      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
-  })
-
-  it('should load manual workflows on init', () => {
-    jest.spyOn(permissionsService, 'currentUserCan').mockReturnValue(true)
-    fixture.detectChanges()
-    expect(workflowService.list).toHaveBeenCalledWith(1, 100000, null, null, {
-      trigger_type: WorkflowTriggerType.Manual,
-    })
-    expect(component.manualWorkflows).toHaveLength(1)
-    expect(component.manualWorkflows[0].id).toEqual(5)
-  })
-
-  it('should run a workflow on the selection via the run workflow dialog', () => {
-    let modal: NgbModalRef
-    modalService.activeInstances.subscribe((m) => (modal = m[0]))
-    jest.spyOn(permissionsService, 'currentUserCan').mockReturnValue(true)
-    jest
-      .spyOn(documentListViewService, 'documents', 'get')
-      .mockReturnValue([{ id: 3 }, { id: 4 }])
-    jest
-      .spyOn(documentListViewService, 'selected', 'get')
-      .mockReturnValue(new Set([3, 4]))
-    fixture.detectChanges()
-    component.runWorkflowSelected()
-    expect(modal).not.toBeUndefined()
-    expect(modal.componentInstance.workflows).toEqual(component.manualWorkflows)
-    modal.componentInstance.workflowId = 5
-    modal.componentInstance.force = true
-    modal.componentInstance.confirm()
-    const req = httpTestingController.expectOne(
-      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    expect(req.request.body.method).toEqual('run_workflow')
-    expect(req.request.body.parameters).toEqual({
-      workflow: 5,
-      filter_documents: false,
-    })
-    req.flush(true)
-    httpTestingController.match(
-      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should set modal dialog text accordingly for tag edit confirmation', () => {
@@ -793,18 +731,19 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: [3, 4],
      method: 'set_correspondent',
      parameters: { correspondent: 101 },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should execute modify correspondent bulk operation with confirmation dialog if enabled', () => {
@@ -831,12 +770,12 @@ describe('BulkEditorComponent', () => {
    httpTestingController
      .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
      .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should set modal dialog text accordingly for correspondent edit confirmation', () => {
@@ -891,18 +830,19 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: [3, 4],
      method: 'set_document_type',
      parameters: { document_type: 101 },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should execute modify document type bulk operation with confirmation dialog if enabled', () => {
@@ -929,12 +869,12 @@ describe('BulkEditorComponent', () => {
    httpTestingController
      .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
      .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should set modal dialog text accordingly for document type edit confirmation', () => {
@@ -989,18 +929,19 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: [3, 4],
      method: 'set_storage_path',
      parameters: { storage_path: 101 },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should execute modify storage path bulk operation with confirmation dialog if enabled', () => {
@@ -1027,12 +968,12 @@ describe('BulkEditorComponent', () => {
    httpTestingController
      .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
      .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should set modal dialog text accordingly for storage path edit confirmation', () => {
@@ -1087,18 +1028,19 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: [3, 4],
      method: 'modify_custom_fields',
      parameters: { add_custom_fields: [101], remove_custom_fields: [102] },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should execute modify custom fields bulk operation with confirmation dialog if enabled', () => {
@@ -1125,12 +1067,12 @@ describe('BulkEditorComponent', () => {
    httpTestingController
      .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
      .flush(true)
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )

    // coverage for modal messages
    component.setCustomFields({
@@ -1604,6 +1546,47 @@ describe('BulkEditorComponent', () => {
    )
  })

+  it.each([true, false])(
+    'shows queued feedback and gates the task link on permission (%s)',
+    (canViewTasks) => {
+      jest
+        .spyOn(permissionsService, 'currentUserCan')
+        .mockReturnValue(canViewTasks)
+      const queued = jest.spyOn(toastService, 'show')
+      let modal: NgbModalRef
+      modalService.activeInstances.subscribe((instances) => {
+        if (instances.length) modal = instances[0]
+      })
+      const reload = jest.spyOn(documentListViewService, 'reload')
+      component.setPermissions()
+      const close = jest.spyOn(modal, 'close')
+      modal.componentInstance.confirmClicked.emit({
+        permissions: {},
+        merge: true,
+      })
+      const req = httpTestingController.expectOne(
+        `${environment.apiBaseUrl}documents/bulk_edit/`
+      )
+      req.flush({ result: 'OK', task_id: 'queued-operation' })
+      expect(close).toHaveBeenCalledTimes(1)
+      const notification = queued.mock.calls.at(-1)[0]
+      expect(notification.content).toBe('Bulk operation queued.')
+      if (canViewTasks) {
+        expect(notification.actionName).toBe('View task')
+        notification.action()
+        expect(router.navigate).toHaveBeenCalledWith(['/tasks'], {
+          queryParams: { task_id: 'queued-operation' },
+        })
+      } else {
+        expect(notification.action).toBeUndefined()
+        expect(notification.actionName).toBeUndefined()
+      }
+      expect(reload).not.toHaveBeenCalled()
+      expect(documentListViewService.selected.size).toBe(0)
+      expect(documentListViewService.allSelected).toBe(false)
+    }
+  )
+
  it('should support bulk permissions update', () => {
    let modal: NgbModalRef
    modalService.activeInstances.subscribe((m) => (modal = m[0]))
@@ -1636,8 +1619,9 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush(true)
+    req.flush({ result: 'OK', task_id: 'queued-task' })
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: [3, 4],
      method: 'set_permissions',
      parameters: {
@@ -1645,12 +1629,12 @@ describe('BulkEditorComponent', () => {
        merge: true,
      },
    })
-    httpTestingController.match(
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should not attempt to retrieve objects if user does not have permissions', () => {
@@ -1951,15 +1935,18 @@ describe('BulkEditorComponent', () => {
    expect(toastServiceShowErrorSpy).toHaveBeenCalled()
    expect(listReloadSpy).not.toHaveBeenCalled()

-    modal.componentInstance.succeeded.emit()
-    expect(toastServiceShowInfoSpy).toHaveBeenCalled()
-    expect(listReloadSpy).toHaveBeenCalled()
-    httpTestingController.match(
+    modal.componentInstance.succeeded.emit({
+      result: 'OK',
+      task_id: 'custom-fields-task',
+    })
+    expect(toastServiceShowInfoSpy).not.toHaveBeenCalled()
+    expect(listReloadSpy).not.toHaveBeenCalled()
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    ) // list reload
-    httpTestingController.match(
+    )
+    httpTestingController.expectNone(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    ) // listAllFilteredIds
+    )
  })

  it('should create share link bundle and enable manage callback', () => {
@@ -34,7 +34,6 @@ 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 { 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'
@@ -54,7 +53,6 @@ import {
 } 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 { SettingsService } from 'src/app/services/settings.service'
@@ -66,7 +64,6 @@ import { MergeAsVersionsConfirmDialogComponent } from '../../common/confirm-dial
 import { MergeConfirmDialogComponent } from '../../common/confirm-dialog/merge-confirm-dialog/merge-confirm-dialog.component'
 import { ReprocessConfirmDialogComponent } from '../../common/confirm-dialog/reprocess-confirm-dialog/reprocess-confirm-dialog.component'
 import { RotateConfirmDialogComponent } from '../../common/confirm-dialog/rotate-confirm-dialog/rotate-confirm-dialog.component'
-import { RunWorkflowConfirmDialogComponent } from '../../common/confirm-dialog/run-workflow-confirm-dialog/run-workflow-confirm-dialog.component'
 import { CorrespondentEditDialogComponent } from '../../common/edit-dialog/correspondent-edit-dialog/correspondent-edit-dialog.component'
 import { CustomFieldEditDialogComponent } from '../../common/edit-dialog/custom-field-edit-dialog/custom-field-edit-dialog.component'
 import { DocumentTypeEditDialogComponent } from '../../common/edit-dialog/document-type-edit-dialog/document-type-edit-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',
@@ -108,21 +104,18 @@ export class BulkEditorComponent
  private correspondentService = inject(CorrespondentService)
  list = inject(DocumentListViewService)
  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 trashService = inject(TrashService)
  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)

-  public manualWorkflows: Workflow[] = []
-
  tagSelectionModel = new FilterableDropdownSelectionModel(true)
  correspondentSelectionModel = new FilterableDropdownSelectionModel()
  documentTypeSelectionModel = new FilterableDropdownSelectionModel()
@@ -265,19 +258,6 @@ export class BulkEditorComponent
          (result) => (this.customFieldsSelectionModel.items = result.results)
        )
    }
-    if (
-      this.permissionService.currentUserCan(
-        PermissionAction.View,
-        PermissionType.Workflow
-      )
-    ) {
-      this.workflowService
-        .list(1, 100000, null, null, {
-          trigger_type: WorkflowTriggerType.Manual,
-        })
-        .pipe(first())
-        .subscribe((result) => (this.manualWorkflows = result.results))
-    }

    this.downloadForm
      .get('downloadFileTypeArchive')
@@ -306,28 +286,6 @@ export class BulkEditorComponent
    this.unsubscribeNotifier.complete()
  }

-  get canRunWorkflow(): boolean {
-    return this.userCanEditAll && this.manualWorkflows.length > 0
-  }
-
-  runWorkflowSelected() {
-    const modal = this.modalService.open(RunWorkflowConfirmDialogComponent, {
-      backdrop: 'static',
-    })
-    modal.componentInstance.title = $localize`Run workflow`
-    modal.componentInstance.message = $localize`Run a workflow on ${this.getSelectionSize()} selected document(s).`
-    modal.componentInstance.btnCaption = $localize`Run`
-    modal.componentInstance.workflows = this.manualWorkflows
-    modal.componentInstance.confirmClicked
-      .pipe(takeUntil(this.unsubscribeNotifier))
-      .subscribe(() => {
-        this.executeBulkEditMethod(modal, 'run_workflow', {
-          workflow: modal.componentInstance.workflowId,
-          filter_documents: !modal.componentInstance.force,
-        })
-      })
-  }
-
  private executeBulkEditMethod(
    modal: NgbModalRef,
    method: DocumentBulkEditMethod,
@@ -341,11 +299,33 @@ export class BulkEditorComponent
      .bulkEdit(overrideSelection ?? this.getSelectionQuery(), method, args)
      .pipe(first())
      .subscribe({
-        next: () => this.handleOperationSuccess(modal),
+        next: ({ task_id }) => this.handleOperationQueued(modal, task_id),
        error: (error) => this.handleOperationError(modal, error),
      })
  }

+  private handleOperationQueued(modal: NgbModalRef, taskId: string) {
+    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 },
+              }),
+          }
+        : {}),
+    })
+  }
+
  private executeDocumentAction(
    modal: NgbModalRef,
    request: Observable<any>,
@@ -1130,14 +1110,9 @@ export class BulkEditorComponent

    dialog.selection = this.getSelectionQuery()
    dialog.selectionCount = this.getSelectionSize()
-    dialog.succeeded.subscribe((result) => {
-      this.toastService.showSuccess($localize`Custom fields updated.`)
-      this.list.reload()
-      this.list.reduceSelectionToFilter()
-      this.list.selected.forEach((id) => {
-        this.openDocumentService.refreshDocument(id)
-      })
-    })
+    dialog.succeeded.subscribe(({ task_id }) =>
+      this.handleOperationQueued(null, task_id)
+    )
    dialog.failed.subscribe((error) => {
      this.toastService.showError(
        $localize`Error updating custom fields.`,
@@ -59,7 +59,7 @@ describe('CustomFieldsBulkEditDialogComponent', () => {
  it('should emit succeeded event and close modal on successful save', () => {
    const editSpy = jest
      .spyOn(documentService, 'bulkEdit')
-      .mockReturnValue(of('Success'))
+      .mockReturnValue(of({ result: 'OK', task_id: 'custom-fields-task' }))
    const successSpy = jest.spyOn(component.succeeded, 'emit')

    component.selection = [1, 2]
@@ -16,6 +16,7 @@ import { NumberComponent } from 'src/app/components/common/input/number/number.c
 import { SelectComponent } from 'src/app/components/common/input/select/select.component'
 import { TextComponent } from 'src/app/components/common/input/text/text.component'
 import { UrlComponent } from 'src/app/components/common/input/url/url.component'
+import { BulkOperationResponse } from 'src/app/data/bulk-operation'
 import { CustomField, CustomFieldDataType } from 'src/app/data/custom-field'
 import {
  DocumentSelectionQuery,
@@ -49,7 +50,7 @@ export class CustomFieldsBulkEditDialogComponent {
  CustomFieldDataType = CustomFieldDataType

  @Output()
-  succeeded = new EventEmitter()
+  succeeded = new EventEmitter<BulkOperationResponse>()

  @Output()
  failed = new EventEmitter()
@@ -103,6 +104,8 @@ export class CustomFieldsBulkEditDialogComponent {
  }

  public save() {
+    if (this.networkActive) return
+    this.networkActive = true
    this.documentService
      .bulkEdit(this.selection, 'modify_custom_fields', {
        add_custom_fields: this.form.value,
@@ -110,11 +113,12 @@ export class CustomFieldsBulkEditDialogComponent {
      })
      .pipe(first())
      .subscribe({
-        next: () => {
+        next: (response) => {
          this.activeModal.close()
-          this.succeeded.emit()
+          this.succeeded.emit(response)
        },
        error: (error) => {
+          this.networkActive = false
          this.failed.emit(error)
        },
      })
@@ -0,0 +1,13 @@
+import { PaperlessTaskStatus } from './paperless-task'
+
+export interface BulkOperationResponse {
+  result: string
+  task_id: string
+}
+
+export interface BulkOperationStatus {
+  task_id: string
+  operation: string
+  status: PaperlessTaskStatus
+  error?: string
+}
@@ -12,7 +12,6 @@ export enum WorkflowTriggerType {
  DocumentAdded = 2,
  DocumentUpdated = 3,
  Scheduled = 4,
-  Manual = 5,
 }

 export enum ScheduleDateField {
@@ -124,6 +124,42 @@ describe('DocumentListViewService', () => {
    })
  })

+  it.each([false, true])(
+    'preserves the current route on a background refresh (saved view: %s)',
+    (savedView) => {
+      if (savedView) documentListViewService.activateSavedView(view)
+      const navigate = jest.mocked(router.navigate)
+      navigate.mockClear()
+      documentListViewService.reload(null, false)
+      httpTestingController
+        .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`)
+        .flush(full_results)
+      expect(documentListViewService.documents).toEqual(full_results.results)
+      expect(navigate).not.toHaveBeenCalled()
+    }
+  )
+
+  it('preserves the route when a background refresh falls back to the first page', () => {
+    jest
+      .spyOn(documentListViewService, 'saveDocumentListView')
+      .mockImplementation(() => {})
+    documentListViewService.currentPage = 2
+    httpTestingController
+      .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`)
+      .flush(full_results)
+    const navigate = jest.mocked(router.navigate)
+    navigate.mockClear()
+    documentListViewService.reload(null, false)
+    httpTestingController
+      .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`)
+      .flush('Not found', { status: 404, statusText: 'Not found' })
+    httpTestingController
+      .expectOne((req) => req.url === `${environment.apiBaseUrl}documents/`)
+      .flush(full_results)
+    expect(documentListViewService.currentPage).toBe(1)
+    expect(navigate).not.toHaveBeenCalled()
+  })
+
  it('should reload the list', () => {
    expect(documentListViewService.currentPage).toEqual(1)
    documentListViewService.reload()
@@ -380,7 +380,7 @@ export class DocumentListViewService {
              queryParams: paramsFromViewState(activeListViewState),
              replaceUrl: !this.router.routerState.snapshot.url.includes('?'), // in case navigating from params-less /documents
            })
-          } else if (this._activeSavedViewId) {
+          } else if (updateQueryParams && this._activeSavedViewId) {
            this.router.navigate([], {
              queryParams: paramsFromViewState(activeListViewState, true),
              queryParamsHandling: 'merge',
@@ -399,7 +399,7 @@ export class DocumentListViewService {
            // this happens when applying a filter: the current page might not be available anymore due to the reduced result set.
            activeListViewState.currentPage = 1
            this.markChanged()
-            this.reload()
+            this.reload(onFinish, updateQueryParams)
          } else if (
            activeListViewState.sortField.indexOf('custom_field') === 0 &&
            !this.settings
@@ -248,6 +248,7 @@ describe(`DocumentService`, () => {
    )
    expect(req.request.method).toEqual('POST')
    expect(req.request.body).toEqual({
+      from_webui: true,
      documents: ids,
      method,
      parameters,
@@ -271,6 +272,7 @@ describe(`DocumentService`, () => {
    )
    expect(req.request.method).toEqual('POST')
    expect(req.request.body).toEqual({
+      from_webui: true,
      all: true,
      filters: { title__icontains: 'apple' },
      excluded_documents: [2, 3],
@@ -14,6 +14,7 @@ import { DocumentSuggestions } from 'src/app/data/document-suggestions'
 import { FilterRule } from 'src/app/data/filter-rule'
 import { Results, SelectionData } from 'src/app/data/results'
 import { SETTINGS_KEYS } from 'src/app/data/ui-settings'
+import { BulkOperationResponse } from '../../data/bulk-operation'
 import { queryParamsFromFilterRules } from '../../utils/query-params'
 import {
  PermissionAction,
@@ -38,7 +39,6 @@ export type DocumentBulkEditMethod =
  | 'modify_tags'
  | 'modify_custom_fields'
  | 'set_permissions'
-  | 'run_workflow'

 export interface MergeDocumentsRequest {
  metadata_document_id?: number
@@ -338,11 +338,15 @@ export class DocumentService extends AbstractPaperlessService<Document> {
    method: DocumentBulkEditMethod,
    args: any
  ) {
-    return this.http.post(this.getResourceUrl(null, 'bulk_edit'), {
-      ...selection,
-      method: method,
-      parameters: args,
-    })
+    return this.http.post<BulkOperationResponse>(
+      this.getResourceUrl(null, 'bulk_edit'),
+      {
+        ...selection,
+        method: method,
+        from_webui: true,
+        parameters: args,
+      }
+    )
  }

  deleteDocuments(selection: DocumentSelectionQuery) {
@@ -438,6 +438,20 @@ describe('ConsumerStatusService', () => {
    expect(updated).toBeTruthy()
  })

+  it('forwards bulk operation completion events', () => {
+    const received = jest.fn()
+    websocketStatusService.onBulkOperationStatus().subscribe(received)
+    const data = {
+      task_id: 'bulk-task',
+      operation: 'set_permissions',
+      status: 'success',
+    }
+    websocketStatusService.connect()
+    server.send({ type: WebsocketStatusType.BULK_OPERATION_STATUS, data })
+    websocketStatusService.disconnect()
+    expect(received).toHaveBeenCalledWith(data)
+  })
+
  it('should ignore keep-alive heartbeat messages from the server', () => {
    let updated = false
    let deleted = false
@@ -1,6 +1,7 @@
 import { Injectable, inject, signal } from '@angular/core'
 import { Subject } from 'rxjs'
 import { environment } from 'src/environments/environment'
+import { BulkOperationStatus } from '../data/bulk-operation'
 import { User } from '../data/user'
 import { WebsocketDocumentUpdatedMessage } from '../data/websocket-document-updated-message'
 import { WebsocketDocumentsDeletedMessage } from '../data/websocket-documents-deleted-message'
@@ -12,6 +13,7 @@ export enum WebsocketStatusType {
  DOCUMENTS_DELETED = 'documents_deleted',
  DOCUMENT_UPDATED = 'document_updated',
  HEARTBEAT = 'heartbeat',
+  BULK_OPERATION_STATUS = 'bulk_operation_status',
 }

 // see ProgressStatusOptions in src/documents/plugins/helpers.py
@@ -116,6 +118,13 @@ export class WebsocketStatusService {
  private readonly documentDeletedSubject = new Subject<boolean>()
  private readonly documentUpdatedSubject =
    new Subject<WebsocketDocumentUpdatedMessage>()
+  private readonly bulkOperationStatusSubject =
+    new Subject<BulkOperationStatus>()
+
+  onBulkOperationStatus() {
+    return this.bulkOperationStatusSubject.asObservable()
+  }
+
  private readonly connectionStatusSubject = new Subject<boolean>()

  private get(taskId: string, filename?: string) {
@@ -192,6 +201,7 @@ export class WebsocketStatusService {
          | WebsocketProgressMessage
          | WebsocketDocumentsDeletedMessage
          | WebsocketDocumentUpdatedMessage
+          | BulkOperationStatus
      } = JSON.parse(ev.data)

      switch (type) {
@@ -209,6 +219,12 @@ export class WebsocketStatusService {
          this.handleProgressUpdate(messageData as WebsocketProgressMessage)
          break

+        case WebsocketStatusType.BULK_OPERATION_STATUS:
+          this.bulkOperationStatusSubject.next(
+            messageData as BulkOperationStatus
+          )
+          break
+
        case WebsocketStatusType.HEARTBEAT:
          // keep-alive from the server, see paperless.consumers.StatusConsumer
          break
@@ -42,13 +42,13 @@ from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
 from documents.tasks import remove_document_from_index
-from documents.tasks import run_workflows_for_documents
 from documents.tasks import update_document_content_maybe_archive_file
 from documents.versioning import get_latest_version_for_root
 from documents.versioning import get_root_document
 from documents.versioning import versions_newest_first

 if TYPE_CHECKING:
+    from collections.abc import Callable
    from collections.abc import Mapping

    from django.contrib.auth.models import User
@@ -132,9 +132,24 @@ def _scratch_path(name: str) -> Path:
    return Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR)) / name


+def _finish_metadata_edit(
+    document_ids: list[int],
+    update_callback: Callable[[list[int]], None] | None,
+) -> None:
+    if update_callback is not None:
+        update_callback(document_ids)
+    else:
+        bulk_update_documents.apply_async(
+            kwargs={"document_ids": document_ids},
+            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
+        )
+
+
 def set_correspondent(
    doc_ids: list[int],
    correspondent: Correspondent,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
    if correspondent:
        correspondent = Correspondent.objects.only("pk").get(id=correspondent)
@@ -147,15 +162,17 @@ def set_correspondent(
    affected_docs = list(qs.values_list("pk", flat=True))
    qs.update(correspondent=correspondent)

-    bulk_update_documents.apply_async(
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
+    _finish_metadata_edit(affected_docs, update_callback)

    return "OK"


-def set_storage_path(doc_ids: list[int], storage_path: StoragePath) -> Literal["OK"]:
+def set_storage_path(
+    doc_ids: list[int],
+    storage_path: StoragePath,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
    if storage_path:
        storage_path = StoragePath.objects.only("pk").get(id=storage_path)

@@ -169,15 +186,17 @@ def set_storage_path(doc_ids: list[int], storage_path: StoragePath) -> Literal["
    affected_docs = list(qs.values_list("pk", flat=True))
    qs.update(storage_path=storage_path)

-    bulk_update_documents.apply_async(
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
+    _finish_metadata_edit(affected_docs, update_callback)

    return "OK"


-def set_document_type(doc_ids: list[int], document_type: DocumentType) -> Literal["OK"]:
+def set_document_type(
+    doc_ids: list[int],
+    document_type: DocumentType,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
    if document_type:
        document_type = DocumentType.objects.only("pk").get(id=document_type)

@@ -189,15 +208,17 @@ def set_document_type(doc_ids: list[int], document_type: DocumentType) -> Litera
    affected_docs = list(qs.values_list("pk", flat=True))
    qs.update(document_type=document_type)

-    bulk_update_documents.apply_async(
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
+    _finish_metadata_edit(affected_docs, update_callback)

    return "OK"


-def add_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
+def add_tag(
+    doc_ids: list[int],
+    tag: int,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
    tag_obj = Tag.objects.get(pk=tag)
    tags_to_add = [tag_obj, *tag_obj.get_ancestors()]

@@ -218,15 +239,17 @@ def add_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
        DocumentTagRelationship.objects.bulk_create(to_create)

    if affected_docs:
-        bulk_update_documents.apply_async(
-            kwargs={"document_ids": list(affected_docs)},
-            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-        )
+        _finish_metadata_edit(list(affected_docs), update_callback)

    return "OK"


-def remove_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
+def remove_tag(
+    doc_ids: list[int],
+    tag: int,
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
+) -> Literal["OK"]:
    tag_obj = Tag.objects.get(pk=tag)
    tag_ids = [tag_obj.id, *tag_obj.get_descendants_pks()]

@@ -239,10 +262,7 @@ def remove_tag(doc_ids: list[int], tag: int) -> Literal["OK"]:
    qs.delete()

    if affected_docs:
-        bulk_update_documents.apply_async(
-            kwargs={"document_ids": affected_docs},
-            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-        )
+        _finish_metadata_edit(affected_docs, update_callback)

    return "OK"

@@ -251,6 +271,8 @@ def modify_tags(
    doc_ids: list[int],
    add_tags: list[int],
    remove_tags: list[int],
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
    qs = Document.objects.filter(id__in=doc_ids).only("pk")
    affected_docs = list(qs.values_list("pk", flat=True))
@@ -294,10 +316,7 @@ def modify_tags(
                )

    if affected_docs:
-        bulk_update_documents.apply_async(
-            kwargs={"document_ids": affected_docs},
-            headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-        )
+        _finish_metadata_edit(affected_docs, update_callback)

    return "OK"

@@ -306,6 +325,8 @@ def modify_custom_fields(
    doc_ids: list[int],
    add_custom_fields: list[int] | dict,
    remove_custom_fields: list[int],
+    *,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
    qs = Document.objects.filter(id__in=doc_ids).only("pk")
    affected_docs = list(qs.values_list("pk", flat=True))
@@ -368,10 +389,7 @@ def modify_custom_fields(
        field_id__in=remove_custom_fields,
    ).hard_delete()

-    bulk_update_documents.apply_async(
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
+    _finish_metadata_edit(affected_docs, update_callback)

    return "OK"

@@ -446,31 +464,13 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
    return "OK"


-def run_workflow(
-    doc_ids: list[int],
-    workflow: int,
-    *,
-    filter_documents: bool = True,
-) -> Literal["OK"]:
-    """
-    Manually run a single workflow against the given documents.
-
-    Done asynchronously via a Celery task.
-    """
-    run_workflows_for_documents.delay(
-        document_ids=doc_ids,
-        workflow_id=workflow,
-        filter_documents=filter_documents,
-    )
-    return "OK"
-
-
 def set_permissions(
    doc_ids: list[int],
    set_permissions: PermissionsSpec,
    *,
    owner: User | None = None,
    merge: bool = False,
+    update_callback: Callable[[list[int]], None] | None = None,
 ) -> Literal["OK"]:
    qs = Document.objects.filter(id__in=doc_ids).select_related("owner")

@@ -488,10 +488,7 @@ def set_permissions(
        merge=merge,
    )

-    bulk_update_documents.apply_async(
-        kwargs={"document_ids": affected_docs},
-        headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
-    )
+    _finish_metadata_edit(affected_docs, update_callback)

    return "OK"

@@ -0,0 +1,107 @@
+"""Queued document metadata edits."""
+
+from typing import Any
+
+from auditlog.context import set_actor
+from celery import shared_task
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.db import transaction
+from rest_framework.exceptions import PermissionDenied
+
+from documents import bulk_edit
+from documents.models import CustomFieldInstance
+from documents.models import Document
+from documents.permissions import has_document_operation_permissions
+from documents.tasks import bulk_update_documents
+
+MODIFIED_FIELDS = {
+    "set_correspondent": "correspondent",
+    "set_document_type": "document_type",
+    "set_storage_path": "storage_path",
+    "add_tag": "tags",
+    "remove_tag": "tags",
+    "modify_tags": "tags",
+    "modify_custom_fields": "custom_fields",
+    "set_permissions": None,
+}
+QUEUED_OPERATIONS = frozenset(MODIFIED_FIELDS)
+
+
+def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
+    """
+    Returns each document's current value of field, for the audit log.
+
+    Tags and custom fields are one row per value, so they are gathered
+    into a sorted list of pks per document (empty when there are none).
+    Reading them through Document.values() instead would join those rows
+    and return one arbitrary value per document.
+    """
+    if field == "tags":
+        rows = (
+            Document.tags.through.objects.filter(document_id__in=doc_ids)
+            .order_by("tag_id")
+            .values_list("document_id", "tag_id")
+        )
+    elif field == "custom_fields":
+        rows = (
+            CustomFieldInstance.objects.filter(document_id__in=doc_ids)
+            .order_by("pk")
+            .values_list("document_id", "pk")
+        )
+    else:
+        return dict(
+            Document.objects.filter(pk__in=doc_ids).values_list("pk", field),
+        )
+
+    values: dict[int, list[int]] = {doc_id: [] for doc_id in doc_ids}
+    for doc_id, pk in rows:
+        values[doc_id].append(pk)
+    return values
+
+
+@shared_task
+def execute_bulk_operation(
+    operation: str,
+    document_ids: list[int],
+    parameters: dict,
+    user_id: int,
+) -> dict:
+    if operation not in QUEUED_OPERATIONS:
+        raise ValueError(f"Unsupported bulk operation: {operation}")
+    method = getattr(bulk_edit, operation)
+    user = User.objects.get(pk=user_id)
+    if not has_document_operation_permissions(
+        user=user,
+        documents=document_ids,
+        method=method,
+        parameters=parameters,
+    ):
+        raise PermissionDenied("Insufficient permissions to execute bulk operation")
+
+    affected_ids: list[int] = []
+    modified_field = MODIFIED_FIELDS[operation]
+    # Keep metadata writes and their audit entries together. Follow-up processing
+    # runs after commit, within this same tracked task.
+    with set_actor(user), transaction.atomic():
+        old_values = (
+            snapshot_field(document_ids, modified_field)
+            if settings.AUDIT_LOG_ENABLED and modified_field
+            else {}
+        )
+        method(document_ids, **parameters, update_callback=affected_ids.extend)
+        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}"},
+                )
+    if affected_ids:
+        bulk_update_documents(list(dict.fromkeys(affected_ids)))
+    return {"operation": operation, "document_count": len(document_ids)}
@@ -35,7 +35,6 @@ from django_filters.rest_framework import DateTimeFilter
 from django_filters.rest_framework import Filter
 from django_filters.rest_framework import FilterSet
 from django_filters.rest_framework import MultipleChoiceFilter
-from django_filters.rest_framework import NumberFilter
 from drf_spectacular.utils import extend_schema_field
 from guardian.utils import get_group_obj_perms_model
 from guardian.utils import get_user_obj_perms_model
@@ -53,7 +52,6 @@ from documents.models import ShareLink
 from documents.models import ShareLinkBundle
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.models import Workflow
 from documents.permissions import annotate_authorizing_fields
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
@@ -251,16 +249,6 @@ class CustomFieldFilterSet(FilterSet):
        }


-class WorkflowFilterSet(FilterSet):
-    trigger_type = NumberFilter(field_name="triggers__type", distinct=True)
-
-    class Meta:
-        model = Workflow
-        fields = {
-            "id": ID_KWARGS,
-        }
-
-
@extend_schema_field(serializers.CharField)
 class CustomFieldsFilter(Filter):
    def filter(self, qs, value):
@@ -684,7 +684,6 @@ def document_matches_workflow(
                trigger_type == WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED
                or trigger_type == WorkflowTrigger.WorkflowTriggerType.DOCUMENT_UPDATED
                or trigger_type == WorkflowTrigger.WorkflowTriggerType.SCHEDULED
-                or trigger_type == WorkflowTrigger.WorkflowTriggerType.MANUAL
            ):
                trigger_matched, reason = existing_document_matches_workflow(
                    document,
@@ -1,43 +0,0 @@
-# Generated by Django 5.2.18 on 2026-11-06 17:12
-
-from django.db import migrations
-from django.db import models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("documents", "0028_replace_tag_tree_fields"),
-    ]
-
-    operations = [
-        migrations.AlterField(
-            model_name="workflowrun",
-            name="type",
-            field=models.PositiveSmallIntegerField(
-                choices=[
-                    (1, "Consumption Started"),
-                    (2, "Document Added"),
-                    (3, "Document Updated"),
-                    (4, "Scheduled"),
-                    (5, "Manual"),
-                ],
-                null=True,
-                verbose_name="workflow trigger type",
-            ),
-        ),
-        migrations.AlterField(
-            model_name="workflowtrigger",
-            name="type",
-            field=models.PositiveSmallIntegerField(
-                choices=[
-                    (1, "Consumption Started"),
-                    (2, "Document Added"),
-                    (3, "Document Updated"),
-                    (4, "Scheduled"),
-                    (5, "Manual"),
-                ],
-                default=1,
-                verbose_name="Workflow Trigger Type",
-            ),
-        ),
-    ]
@@ -1532,7 +1532,6 @@ class WorkflowTrigger(models.Model):
        DOCUMENT_ADDED = 2, _("Document Added")
        DOCUMENT_UPDATED = 3, _("Document Updated")
        SCHEDULED = 4, _("Scheduled")
-        MANUAL = 5, _("Manual")

    class DocumentSourceChoices(models.IntegerChoices):
        CONSUME_FOLDER = DocumentSource.ConsumeFolder.value, _("Consume Folder")
@@ -31,6 +31,7 @@ from guardian.core import ObjectPermissionChecker
 from guardian.models import GroupObjectPermission
 from guardian.models import UserObjectPermission
 from rest_framework.permissions import BasePermission
+from rest_framework.permissions import DjangoModelPermissions
 from rest_framework.permissions import DjangoObjectPermissions

 from documents.models import Document
@@ -63,6 +64,17 @@ class PermissionGranteesUpdate(TypedDict, total=False):
 PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGranteesUpdate]


+class PaperlessModelPermissions(DjangoModelPermissions):
+    """Require basic global model permissions."""
+
+    perms_map = {
+        **DjangoModelPermissions.perms_map,
+        "GET": ["%(app_label)s.view_%(model_name)s"],
+        "OPTIONS": ["%(app_label)s.view_%(model_name)s"],
+        "HEAD": ["%(app_label)s.view_%(model_name)s"],
+    }
+
+
 class PaperlessObjectPermissions(DjangoObjectPermissions):
    """
    A permissions backend that checks for object-level permissions
@@ -576,12 +588,9 @@ def annotate_document_count_by_ids(
                       or CustomFieldInstance)
        related_object_field: field on the relation pointing back to queryset pk
        document_ids: the document ids to count against -- a concrete list/set,
-                       or a simple (already resolved) queryset of ids. Callers
-                       that need this filtered by a complex condition (e.g. a
-                       permission check) should resolve it to a concrete list
-                       first if the same ids will be reused across multiple
-                       calls, rather than passing the complex queryset itself
-                       into each -- see ``_get_selection_data_for_queryset``.
+                       or a simple (already resolved) queryset of ids. Keep
+                       the list small: it is embedded in the statement as
+                       literal parameters.
        target_field: field on the relation pointing to Document id
    """

@@ -730,3 +739,100 @@ class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
        "OPTIONS": ["documents.change_paperlesstask"],
        "POST": ["documents.change_paperlesstask"],
    }
+
+
+def has_document_operation_permissions(
+    *,
+    user: User,
+    documents: list[int],
+    method,
+    parameters: dict[str, Any],
+) -> bool:
+    from documents import bulk_edit
+
+    if not user.is_active:
+        return False
+    if user.is_superuser:
+        return True
+
+    root_docs = {
+        get_root_document(doc)
+        for doc in Document.objects.select_related(
+            "owner",
+            "root_document__owner",
+        ).filter(pk__in=documents)
+    }
+    user_is_owner_of_all_documents = all(
+        (doc.owner == user or doc.owner is None) for doc in root_docs
+    )
+
+    # check global and object permissions for all documents
+    has_perms = (
+        user.has_perm(
+            "documents.change_document",
+        )
+        and not Document.global_objects.filter(pk__in=documents)
+        .exclude(pk__in=permitted_document_ids(user, perm="change_document"))
+        .exists()
+    )
+
+    # check ownership for methods that change original document
+    if (
+        (
+            has_perms
+            and method
+            in [
+                bulk_edit.set_permissions,
+                bulk_edit.delete,
+                bulk_edit.rotate,
+                bulk_edit.delete_pages,
+                bulk_edit.edit_pdf,
+                bulk_edit.merge_as_versions,
+                bulk_edit.remove_password,
+            ]
+        )
+        or (
+            method in [bulk_edit.merge, bulk_edit.split]
+            and parameters.get("delete_originals")
+        )
+        or (method == bulk_edit.edit_pdf and parameters.get("update_document"))
+    ):
+        has_perms = has_perms and user_is_owner_of_all_documents
+
+    # check global add permissions for methods that create documents
+    if (
+        has_perms
+        and (
+            method in [bulk_edit.split, bulk_edit.merge]
+            or (
+                method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
+                and not parameters.get("update_document")
+            )
+        )
+        and not user.has_perm("documents.add_document")
+    ):
+        has_perms = False
+
+    # check global delete permissions for methods that delete documents
+    if (
+        has_perms
+        and (
+            method == bulk_edit.delete
+            # Sources stop being documents of their own, and removing one
+            # again afterwards needs delete_document
+            or method == bulk_edit.merge_as_versions
+            or (
+                method in [bulk_edit.merge, bulk_edit.split]
+                and parameters.get("delete_originals")
+            )
+            or (
+                method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
+                and parameters.get("delete_original")
+                and not parameters.get("update_document")
+            )
+        )
+        and not user.has_perm("documents.delete_document")
+    ):
+        has_perms = False
+
+    return has_perms
@@ -1829,7 +1829,6 @@ class BulkEditSerializer(
            "modify_tags",
            "modify_custom_fields",
            "set_permissions",
-            "run_workflow",
            *LEGACY_DOCUMENT_ACTION_METHODS,
        ],
        label="Method",
@@ -1919,8 +1918,6 @@ class BulkEditSerializer(
            return bulk_edit.reprocess
        elif method == "set_permissions":
            return bulk_edit.set_permissions
-        elif method == "run_workflow":
-            return bulk_edit.run_workflow
        elif method == "rotate":
            return bulk_edit.rotate
        elif method == "merge":
@@ -2160,27 +2157,6 @@ class BulkEditSerializer(
        else:
            parameters["remote_ocr"] = False

-    def _validate_parameters_run_workflow(self, parameters) -> None:
-        if "workflow" not in parameters:
-            raise serializers.ValidationError("workflow not specified")
-        try:
-            workflow = Workflow.objects.get(id=parameters["workflow"])
-        except (Workflow.DoesNotExist, ValueError, TypeError):
-            raise serializers.ValidationError("Workflow does not exist")
-        if not workflow.triggers.filter(
-            type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
-        ).exists():
-            raise serializers.ValidationError(
-                "Workflow does not have a manual trigger",
-            )
-        if "filter_documents" in parameters:
-            if not isinstance(parameters["filter_documents"], bool):
-                raise serializers.ValidationError(
-                    "filter_documents must be a boolean",
-                )
-        else:
-            parameters["filter_documents"] = True
-
    def validate_parameters_remove_password(self, parameters):
        if "password" not in parameters:
            raise serializers.ValidationError("password not specified")
@@ -2247,8 +2223,6 @@ class BulkEditSerializer(
            self.validate_parameters_remove_password(parameters)
        elif method == bulk_edit.reprocess:
            self._validate_parameters_reprocess(parameters)
-        elif method == bulk_edit.run_workflow:
-            self._validate_parameters_run_workflow(parameters)

        return attrs

@@ -8,6 +8,7 @@ from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any

+from asgiref.sync import async_to_sync
 from celery import shared_task
 from celery.signals import before_task_publish
 from celery.signals import task_failure
@@ -16,6 +17,7 @@ from celery.signals import task_prerun
 from celery.signals import task_revoked
 from celery.signals import worker_process_init
 from celery.signals import worker_process_shutdown
+from channels.layers import get_channel_layer
 from django.conf import settings
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import User
@@ -862,8 +864,6 @@ def run_workflows(
    logging_group: uuid.UUID | None = None,
    overrides: DocumentMetadataOverrides | None = None,
    original_file: Path | None = None,
-    *,
-    skip_matching: bool = False,
 ) -> tuple[DocumentMetadataOverrides, str] | None:
    """
    Execute workflows matching a document for the given trigger. When `overrides` is provided
@@ -874,10 +874,6 @@ def run_workflows(
    `document.source_path` (Document) or `document.original_file` (ConsumableDocument).

    Passing `workflow_to_run` skips the workflow query (currently only used by scheduled runs).
-
-    Passing `skip_matching` runs the workflow's actions without evaluating its trigger
-    filters against the document. It is only meaningful together with `workflow_to_run`
-    (e.g. a manual run the user has explicitly forced onto the selected documents).
    """

    use_overrides = overrides is not None
@@ -927,11 +923,7 @@ def run_workflows(
                )
                break

-        if skip_matching or matching.document_matches_workflow(
-            document,
-            workflow,
-            trigger_type,
-        ):
+        if matching.document_matches_workflow(document, workflow, trigger_type):
            action: WorkflowAction
            has_move_to_trash_action = False
            for action in workflow.actions.order_by("order", "pk"):
@@ -1074,6 +1066,7 @@ TRACKED_TASKS: dict[str, PaperlessTask.TaskType] = {
    "documents.tasks.build_share_link_bundle": PaperlessTask.TaskType.BUILD_SHARE_LINK,
    "documents.bulk_edit.delete": PaperlessTask.TaskType.BULK_DELETE,
    "documents.tasks.apply_ai_suggestions": PaperlessTask.TaskType.APPLY_AI_SUGGESTIONS,
+    "documents.bulk_operations.execute_bulk_operation": PaperlessTask.TaskType.BULK_UPDATE,
 }

 _CELERY_STATE_TO_STATUS: dict[str, PaperlessTask.Status] = {
@@ -1091,8 +1084,8 @@ def _extract_input_data(

    For consume_file tasks this includes the filename, MIME type, and any
    non-null overrides from the DocumentMetadataOverrides object.  For
-    mail_fetch tasks it captures the account_ids list.  All other task
-    types store no input data and return {}.
+    mail_fetch tasks it captures the account_ids list. Bulk operations store
+    the operation name and document count, without copying the full selection.
    """
    if task_type == PaperlessTask.TaskType.CONSUME_FILE:
        input_doc = task_kwargs.get("input_doc")
@@ -1121,6 +1114,12 @@ def _extract_input_data(
                data["overrides"] = override_dict
        return data

+    if "operation" in task_kwargs and "user_id" in task_kwargs:
+        return {
+            "operation": task_kwargs["operation"],
+            "document_count": len(task_kwargs.get("document_ids", [])),
+        }
+
    if task_type == PaperlessTask.TaskType.MAIL_FETCH:
        account_ids = task_kwargs.get("account_ids")
        if account_ids is not None:
@@ -1157,7 +1156,9 @@ def _extract_owner_id(
    task_type: PaperlessTask.TaskType,
    task_kwargs: dict,
 ) -> int | None:
-    """Return the owner_id from consume_file overrides, or None for all other task types."""
+    """Use the initiator for bulk operations and the document owner for consumption."""
+    if "operation" in task_kwargs:
+        return task_kwargs.get("user_id")
    if task_type != PaperlessTask.TaskType.CONSUME_FILE:
        return None
    overrides = task_kwargs.get("overrides")
@@ -1238,6 +1239,29 @@ def task_prerun_handler(sender=None, task_id=None, task=None, **kwargs) -> None:
        logger.exception("Setting PaperlessTask started failed")


+def _send_bulk_operation_status(task_id: str) -> None:
+    """Notify the initiator, task records still happen."""
+    try:
+        task = PaperlessTask.objects.filter(task_id=task_id).first()
+        if task is None or task.owner_id is None or "operation" not in task.input_data:
+            return
+        async_to_sync(get_channel_layer().group_send)(
+            f"bulk_operations_{task.owner_id}",
+            {
+                "type": "bulk_operation_status",
+                "data": {
+                    "task_id": task.task_id,
+                    "status": task.status,
+                    "operation": task.input_data["operation"],
+                    "error": (task.result_data or {}).get("error_message", ""),
+                },
+            },
+        )
+    except Exception:
+        # Just a notification failure
+        logger.exception("Sending bulk operation status failed")
+
+
@task_postrun.connect
 def task_postrun_handler(
    sender=None,
@@ -1295,6 +1319,7 @@ def task_postrun_handler(
                changed_fields.append("status")

        task_instance.save(update_fields=changed_fields)
+        _send_bulk_operation_status(task_id)
    except Exception:  # pragma: no cover
        logger.exception("Updating PaperlessTask failed")

@@ -1360,6 +1385,7 @@ def task_failure_handler(
                    date_started - date_created
                ).total_seconds()
            task_qs.update(**update_fields)
+            _send_bulk_operation_status(task_id)
    except Exception:  # pragma: no cover
        logger.exception("Updating PaperlessTask on failure failed")

@@ -1403,6 +1429,7 @@ def task_revoked_handler(
                "reason": (reason + " Check worker logs for details."),
            },
        )
+        _send_bulk_operation_status(task_id)
    except Exception:  # pragma: no cover
        logger.exception("Updating PaperlessTask on revocation failed")

@@ -52,7 +52,6 @@ from documents.models import ShareLink
 from documents.models import ShareLinkBundle
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.models import Workflow
 from documents.models import WorkflowRun
 from documents.models import WorkflowTrigger
 from documents.plugins.base import ConsumeTaskPlugin
@@ -691,47 +690,6 @@ def check_scheduled_workflows() -> None:
                        )


-@shared_task
-def run_workflows_for_documents(
-    document_ids: list[int],
-    workflow_id: int,
-    *,
-    filter_documents: bool = True,
-) -> None:
-    """
-    Run a single workflow manually against the given documents.
-
-    ``filter_documents`` controls whether the workflow's trigger filters are applied.
-    When True, only documents matching the existing filters will have the workflow run.
-    When False, the workflow will be run on all specified documents regardless of the filters.
-    ``
-    """
-    try:
-        workflow = Workflow.objects.get(id=workflow_id)
-    except Workflow.DoesNotExist:
-        logger.warning(
-            f"Manual workflow run requested for missing workflow {workflow_id}",
-        )
-        return
-
-    for document in Document.objects.filter(
-        id__in=document_ids,
-        root_document__isnull=True,
-    ):
-        run_workflows(
-            trigger_type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
-            document=document,
-            workflow_to_run=workflow,
-            skip_matching=not filter_documents,
-        )
-        # Manual runs dont send a document_updated signal, so push a websocket
-        # update here to ensure clients are refreshed
-        send_websocket_document_updated(
-            sender=None,
-            document=document,
-        )
-
-
 def update_document_parent_tags(tag: Tag, new_parent: Tag) -> None:
    """
    When a tag's parent changes, ensure all documents containing the tag also have
@@ -15,6 +15,7 @@ from paperless.models import ColorConvertChoices
 from paperless_testing.dirs import DirectoriesMixin
 from paperless_testing.factories import UserFactory
 from paperless_testing.http import read_streaming_response
+from paperless_testing.permissions import grant_global


 class TestApiAppConfig(DirectoriesMixin, APITestCase):
@@ -26,6 +27,67 @@ class TestApiAppConfig(DirectoriesMixin, APITestCase):
        user = UserFactory(username="temp_admin", superuser=True)
        self.client.force_authenticate(user=user)

+    def test_api_read_config_requires_view_permission(self) -> None:
+        for permissions in (
+            (),
+            ("change_applicationconfiguration",),
+            ("view_document",),
+        ):
+            user = UserFactory()
+            grant_global(user, *permissions)
+            self.client.force_authenticate(user=user)
+
+            for endpoint in (self.ENDPOINT, f"{self.ENDPOINT}1/"):
+                for method in ("get", "head", "options"):
+                    with self.subTest(
+                        permissions=permissions,
+                        endpoint=endpoint,
+                        method=method,
+                    ):
+                        response = getattr(self.client, method)(endpoint)
+                        self.assertEqual(
+                            response.status_code,
+                            status.HTTP_403_FORBIDDEN,
+                        )
+
+    def test_api_read_config_with_view_permission(self) -> None:
+        user = UserFactory()
+        grant_global(user, "view_applicationconfiguration")
+        self.client.force_authenticate(user=user)
+
+        for endpoint in (self.ENDPOINT, f"{self.ENDPOINT}1/"):
+            for method in ("get", "head", "options"):
+                with self.subTest(endpoint=endpoint, method=method):
+                    response = getattr(self.client, method)(endpoint)
+                    self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+    def test_api_view_config_permission_does_not_allow_writes(self) -> None:
+        user = UserFactory()
+        grant_global(user, "view_applicationconfiguration")
+        self.client.force_authenticate(user=user)
+
+        for method in ("put", "patch", "delete"):
+            with self.subTest(method=method):
+                response = getattr(self.client, method)(f"{self.ENDPOINT}1/")
+                self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_api_change_config_permission_allows_updates(self) -> None:
+        user = UserFactory()
+        grant_global(user, "change_applicationconfiguration")
+        self.client.force_authenticate(user=user)
+
+        response = self.client.patch(
+            f"{self.ENDPOINT}1/",
+            {"app_title": "Updated title"},
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            ApplicationConfiguration.objects.get().app_title,
+            "Updated title",
+        )
+
    def test_api_get_config(self) -> None:
        """
        GIVEN:
@@ -7,6 +7,7 @@ from django.test import override_settings
 from rest_framework import status
 from rest_framework.test import APITestCase

+from documents.bulk_operations import execute_bulk_operation
 from documents.models import Correspondent
 from documents.models import CustomField
 from documents.models import CustomFieldInstance
@@ -14,11 +15,7 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowTrigger
 from paperless_testing.dirs import DirectoriesMixin
-from paperless_testing.factories import DocumentFactory
 from paperless_testing.factories import UserFactory
 from paperless_testing.permissions import grant_all_global
 from paperless_testing.permissions import grant_global
@@ -33,9 +30,18 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.user = user
        self.client.force_authenticate(user=user)

-        patcher = mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+        patcher = mock.patch("documents.bulk_operations.bulk_update_documents")
        self.async_task = patcher.start()
        self.addCleanup(patcher.stop)
+        # Exercise accepted requests and worker execution together in these
+        # operation/validation tests. Queue-only behavior is tested separately.
+        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)
+
        self.c1 = Correspondent.objects.create(name="c1")
        self.c2 = Correspondent.objects.create(name="c2")
        self.dt1 = DocumentType.objects.create(name="dt1")
@@ -65,11 +71,16 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.cf1 = CustomField.objects.create(name="cf1", data_type="string")
        self.cf2 = CustomField.objects.create(name="cf2", data_type="string")

+    @staticmethod
+    def run_operation(kwargs):
+        execute_bulk_operation(**kwargs)
+        return mock.Mock(id="test-bulk-operation")
+
    def setup_mock(self, m, method_name, return_value="OK") -> None:
        m.return_value = return_value
        m.__name__ = method_name

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
    def test_api_set_correspondent(self, bulk_update_task_mock) -> None:
        self.assertNotEqual(self.doc1.correspondent, self.c1)
        response = self.client.post(
@@ -88,11 +99,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(self.doc1.correspondent, self.c1)
        bulk_update_task_mock.assert_called_once()
        self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
            [self.doc1.pk],
        )

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
    def test_api_unset_correspondent(self, bulk_update_task_mock) -> None:
        self.doc1.correspondent = self.c1
        self.doc1.save()
@@ -114,7 +125,7 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.doc1.refresh_from_db()
        self.assertIsNone(self.doc1.correspondent)

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
    def test_api_set_type(self, bulk_update_task_mock) -> None:
        self.assertNotEqual(self.doc1.document_type, self.dt1)
        response = self.client.post(
@@ -133,11 +144,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(self.doc1.document_type, self.dt1)
        bulk_update_task_mock.assert_called_once()
        self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
            [self.doc1.pk],
        )

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
    def test_api_unset_type(self, bulk_update_task_mock) -> None:
        self.doc1.document_type = self.dt1
        self.doc1.save()
@@ -158,11 +169,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertIsNone(self.doc1.document_type)
        bulk_update_task_mock.assert_called_once()
        self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
            [self.doc1.pk],
        )

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
    def test_api_add_tag(self, bulk_update_task_mock) -> None:
        self.assertFalse(self.doc1.tags.filter(pk=self.t1.pk).exists())

@@ -184,11 +195,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):

        bulk_update_task_mock.assert_called_once()
        self.assertCountEqual(
-            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
+            bulk_update_task_mock.call_args.args[0],
            [self.doc1.pk],
        )

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.bulk_operations.bulk_update_documents")
    def test_api_remove_tag(self, bulk_update_task_mock) -> None:
        self.doc1.tags.add(self.t1)

@@ -227,6 +238,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertListEqual(args[0], [self.doc1.id, self.doc3.id])
        self.assertEqual(kwargs["add_tags"], [self.t1.id])
        self.assertEqual(kwargs["remove_tags"], [self.t2.id])
@@ -281,6 +295,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertListEqual(args[0], [self.doc1.id, self.doc3.id])
        self.assertEqual(kwargs["add_custom_fields"], [self.cf1.id])
        self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id])
@@ -349,6 +366,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertListEqual(args[0], [self.doc1.id, self.doc3.id])
        self.assertEqual(kwargs["add_custom_fields"], {self.cf1.id: "foo"})
        self.assertEqual(kwargs["remove_custom_fields"], [self.cf2.id])
@@ -510,6 +530,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertEqual(args[0], [self.doc1.id])
        self.assertEqual(len(kwargs), 0)

@@ -524,6 +547,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertEqual(args[0], [self.doc1.id])
        self.assertEqual(len(kwargs), 0)

@@ -548,6 +574,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertEqual(args[0], [self.doc1.id])
        self.assertEqual(len(kwargs), 0)

@@ -562,6 +591,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertEqual(args[0], [self.doc1.id])
        self.assertEqual(kwargs, {"remote_ocr": False})

@@ -584,88 +616,12 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertEqual(args[0], [self.doc1.id])
        self.assertEqual(kwargs, {"remote_ocr": True})

-    def _make_manual_workflow(self) -> Workflow:
-        trigger = WorkflowTrigger.objects.create(
-            type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
-        )
-        action = WorkflowAction.objects.create()
-        workflow = Workflow.objects.create(name="Manual workflow", order=0)
-        workflow.triggers.add(trigger)
-        workflow.actions.add(action)
-        return workflow
-
-    def _run_workflow(self, parameters, documents=None):
-        return self.client.post(
-            "/api/documents/bulk_edit/",
-            json.dumps(
-                {
-                    "documents": documents or [self.doc1.id],
-                    "method": "run_workflow",
-                    "parameters": parameters,
-                },
-            ),
-            content_type="application/json",
-        )
-
-    @mock.patch("documents.views.bulk_edit.run_workflow")
-    def test_run_workflow(self, m) -> None:
-        """A valid run_workflow call passes the docs, workflow id and filter flag through."""
-        self.setup_mock(m, "run_workflow")
-        workflow = self._make_manual_workflow()
-
-        response = self._run_workflow({"workflow": workflow.id})
-
-        self.assertEqual(response.status_code, status.HTTP_200_OK)
-        args, kwargs = m.call_args
-        self.assertEqual(args[0], [self.doc1.id])
-        self.assertEqual(kwargs, {"workflow": workflow.id, "filter_documents": True})
-
-        self._run_workflow({"workflow": workflow.id, "filter_documents": False})
-        self.assertFalse(m.call_args.kwargs["filter_documents"])
-
-    @mock.patch("documents.views.bulk_edit.run_workflow")
-    def test_run_workflow_invalid_parameters(self, m) -> None:
-        """Invalid run_workflow parameters are rejected before dispatch."""
-        self.setup_mock(m, "run_workflow")
-        manual = self._make_manual_workflow()
-        non_manual = Workflow.objects.create(name="Non-manual", order=0)
-        non_manual.triggers.add(
-            WorkflowTrigger.objects.create(
-                type=WorkflowTrigger.WorkflowTriggerType.DOCUMENT_ADDED,
-            ),
-        )
-        for label, parameters in (
-            ("missing workflow", {}),
-            ("nonexistent workflow", {"workflow": 9999}),
-            ("no manual trigger", {"workflow": non_manual.id}),
-            ("non-bool filter", {"workflow": manual.id, "filter_documents": "nope"}),
-        ):
-            with self.subTest(label):
-                response = self._run_workflow(parameters)
-                self.assertEqual(
-                    response.status_code,
-                    status.HTTP_400_BAD_REQUEST,
-                )
-        m.assert_not_called()
-
-    @mock.patch("documents.views.bulk_edit.run_workflow")
-    def test_run_workflow_requires_ownership(self, m) -> None:
-        """A non-owner without document ownership is forbidden (actions mutate docs)."""
-        self.setup_mock(m, "run_workflow")
-        user = UserFactory(username="no_perms")
-        grant_all_global(user)
-        doc = DocumentFactory(owner=UserFactory(username="other"))
-        workflow = self._make_manual_workflow()
-        self.client.force_authenticate(user=user)
-
-        response = self._run_workflow({"workflow": workflow.id}, documents=[doc.id])
-
-        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
-        m.assert_not_called()
-
    @mock.patch("documents.serialisers.bulk_edit.set_storage_path")
    def test_api_set_storage_path(self, m) -> None:
        """
@@ -692,6 +648,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }

        self.assertListEqual(args[0], [self.doc1.id])
        self.assertEqual(kwargs["storage_path"], self.sp1.id)
@@ -722,6 +681,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }

        self.assertListEqual(args[0], [self.doc1.id])
        self.assertEqual(kwargs["storage_path"], None)
@@ -844,6 +806,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertEqual(args[0], [self.doc2.id])
        self.assertEqual(kwargs["storage_path"], self.sp1.id)

@@ -867,6 +832,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [self.doc1.id, self.doc3.id, self.doc5.id])
        self.assertEqual(kwargs["storage_path"], self.sp1.id)

@@ -903,6 +871,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [first_duplicate.id, second_duplicate.id])
        self.assertEqual(kwargs["storage_path"], self.sp1.id)

@@ -939,6 +910,11 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
                self.assertEqual(response.status_code, status.HTTP_200_OK)
                get_backend.return_value.search_ids.assert_called_once()
                args, kwargs = m.call_args
+                kwargs = {
+                    key: value
+                    for key, value in kwargs.items()
+                    if key != "update_callback"
+                }
                self.assertEqual(args[0], [self.doc2.id])
                self.assertEqual(kwargs["storage_path"], self.sp1.id)

@@ -1224,8 +1200,8 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertCountEqual(
            response.data["selected_document_types"],
            [
-                {"id": self.c1.id, "document_count": 1},
-                {"id": self.c2.id, "document_count": 0},
+                {"id": self.dt1.id, "document_count": 1},
+                {"id": self.dt2.id, "document_count": 0},
            ],
        )

@@ -1315,6 +1291,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):

        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
        self.assertEqual(len(kwargs["set_permissions"]["view"]["users"]), 2)

@@ -1727,6 +1706,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
        self.assertEqual(kwargs["degrees"], 90)
        self.assertEqual(kwargs["source_mode"], "latest_version")
@@ -2013,6 +1995,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [self.doc2.id, self.doc3.id])
        self.assertEqual(kwargs["metadata_document_id"], self.doc3.id)
        self.assertEqual(kwargs["source_mode"], "latest_version")
@@ -2220,6 +2205,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [self.doc2.id])
        self.assertEqual(kwargs["operations"], [{"page": 1}])
        self.assertEqual(kwargs["source_mode"], "explicit_selection")
@@ -2565,6 +2553,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        m.assert_called_once()
        args, kwargs = m.call_args
+        kwargs = {
+            key: value for key, value in kwargs.items() if key != "update_callback"
+        }
        self.assertCountEqual(args[0], [self.doc2.id])
        self.assertEqual(kwargs["password"], "secret")
        self.assertTrue(kwargs["update_document"])
@@ -49,6 +49,7 @@ from documents.models import WorkflowTrigger
 from documents.search import get_backend
 from documents.signals.handlers import run_workflows
 from documents.tests.utils import ConsumeTaskMixin
+from documents.views import DocumentViewSet
 from paperless_testing.dirs import DirectoriesMixin
 from paperless_testing.factories import DocumentFactory
 from paperless_testing.factories import TagFactory
@@ -1755,12 +1756,113 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
        }
        self.assertEqual(counts_by_field[custom_field.id], 1)

+    def test_selection_data_counts_with_shared_by_filter(self) -> None:
+        """
+        GIVEN:
+            - Several documents owned by one user with the same correspondent
+              and document type, only one of which is shared
+        WHEN:
+            - Requesting the document list filtered by shared_by__id with
+              include_selection_data=true
+        THEN:
+            - Counts reflect only the shared document, not every document the
+              owner has with that correspondent or document type, regardless of
+              the batch size
+        """
+        owner = User.objects.create_user(username="owner")
+        other = User.objects.create_user(username="other")
+        correspondent = Correspondent.objects.create(name="c1")
+        doc_type = DocumentType.objects.create(name="dt1")
+        tag = Tag.objects.create(name="tag")
+        docs = [
+            Document.objects.create(
+                checksum=str(i),
+                owner=owner,
+                correspondent=correspondent,
+                document_type=doc_type,
+            )
+            for i in range(3)
+        ]
+        for doc in docs:
+            doc.tags.add(tag)
+        grant_object(other, docs[0], "view_document")
+
+        for batch_size in (10_000, 1):
+            with (
+                self.subTest(batch_size=batch_size),
+                mock.patch.object(
+                    DocumentViewSet,
+                    "selection_data_batch_size",
+                    batch_size,
+                ),
+            ):
+                response = self.client.get(
+                    f"/api/documents/?shared_by__id={owner.id}"
+                    "&include_selection_data=true",
+                )
+
+                self.assertEqual(response.status_code, status.HTTP_200_OK)
+                self.assertEqual(response.data["count"], 1)
+                selection_data = response.data["selection_data"]
+                for key, obj in (
+                    ("selected_correspondents", correspondent),
+                    ("selected_document_types", doc_type),
+                    ("selected_tags", tag),
+                ):
+                    counts = {
+                        item["id"]: item["document_count"]
+                        for item in selection_data[key]
+                    }
+                    self.assertEqual(counts[obj.id], 1, key)
+
+    def test_selection_data_counts_documents_once_when_filter_joins(self) -> None:
+        """
+        GIVEN:
+            - A document matching a multi-valued filter through two tags
+        WHEN:
+            - Requesting the document list with include_selection_data=true
+        THEN:
+            - The document is counted once per correspondent, not once per
+              matching tag, regardless of the batch size
+        """
+        correspondent = Correspondent.objects.create(name="c1")
+        tag_a = Tag.objects.create(name="a")
+        tag_b = Tag.objects.create(name="b")
+        both = Document.objects.create(checksum="1", correspondent=correspondent)
+        both.tags.add(tag_a, tag_b)
+        only_a = Document.objects.create(checksum="2", correspondent=correspondent)
+        only_a.tags.add(tag_a)
+
+        for batch_size in (10_000, 1):
+            with (
+                self.subTest(batch_size=batch_size),
+                mock.patch.object(
+                    DocumentViewSet,
+                    "selection_data_batch_size",
+                    batch_size,
+                ),
+            ):
+                response = self.client.get(
+                    f"/api/documents/?tags__id__in={tag_a.id},{tag_b.id}"
+                    "&include_selection_data=true",
+                )
+
+                self.assertEqual(response.status_code, status.HTTP_200_OK)
+                counts = {
+                    item["id"]: item["document_count"]
+                    for item in response.data["selection_data"][
+                        "selected_correspondents"
+                    ]
+                }
+                self.assertEqual(counts[correspondent.id], 2)
+
    def test_statistics(self) -> None:
        doc1 = Document.objects.create(
            title="none1",
            checksum="A",
            mime_type="application/pdf",
            content="abc",
+            archive_serial_number=42,
        )
        Document.objects.create(
            title="none2",
@@ -1804,6 +1906,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
        self.assertEqual(response.data["correspondent_count"], 2)
        self.assertEqual(response.data["document_type_count"], 1)
        self.assertEqual(response.data["storage_path_count"], 2)
+        self.assertEqual(response.data["current_asn"], 42)

    def test_statistics_excludes_document_versions(self) -> None:
        root = Document.objects.create(
@@ -1811,6 +1914,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
            checksum="A",
            mime_type="application/pdf",
            content="root",
+            archive_serial_number=42,
        )
        version = Document.objects.create(
            title="version",
@@ -1819,6 +1923,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
            content="version",
            root_document=root,
            version_index=1,
+            archive_serial_number=99,
        )
        tag_inbox = Tag.objects.create(name="t1", is_inbox_tag=True)
        version.tags.add(tag_inbox)
@@ -1829,6 +1934,7 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
        self.assertEqual(response.data["documents_total"], 1)
        self.assertEqual(response.data["documents_inbox"], 0)
        self.assertEqual(response.data["character_count"], 4)
+        self.assertEqual(response.data["current_asn"], 42)
        self.assertEqual(
            response.data["document_file_type_counts"][0]["mime_type_count"],
            1,
@@ -1842,6 +1948,45 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
        self.assertEqual(response.data["documents_inbox"], None)
        self.assertEqual(response.data["inbox_tags"], None)

+    def test_statistics_current_asn_respects_document_permissions(self) -> None:
+        viewer = UserFactory()
+        grant_global(viewer, "view_document")
+        DocumentFactory(owner=viewer, archive_serial_number=12)
+        shared = DocumentFactory(owner=self.user, archive_serial_number=23)
+        grant_object(viewer, shared, "view_document")
+        DocumentFactory(owner=self.user, archive_serial_number=99)
+
+        self.client.force_authenticate(user=viewer)
+        response = self.client.get("/api/statistics/")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["current_asn"], 23)
+
+    def test_statistics_current_asn_without_visible_documents(self) -> None:
+        DocumentFactory(owner=self.user, archive_serial_number=99)
+
+        for permissions in ((), ("view_document",)):
+            viewer = UserFactory()
+            grant_global(viewer, *permissions)
+            self.client.force_authenticate(user=viewer)
+
+            with self.subTest(permissions=permissions):
+                response = self.client.get("/api/statistics/")
+
+                self.assertEqual(response.status_code, status.HTTP_200_OK)
+                self.assertEqual(response.data["current_asn"], 0)
+
+    def test_statistics_current_asn_with_global_statistics_permission(self) -> None:
+        viewer = UserFactory()
+        grant_global(viewer, "view_global_statistics")
+        DocumentFactory(owner=self.user, archive_serial_number=99)
+
+        self.client.force_authenticate(user=viewer)
+        response = self.client.get("/api/statistics/")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["current_asn"], 99)
+
    def test_statistics_multiple_users(self) -> None:
        """
        GIVEN:
@@ -25,6 +25,7 @@ from documents.models import Tag
 from documents.models import Workflow
 from documents.search import get_backend
 from documents.search import reset_backend
+from documents.views import DocumentViewSet
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
 from paperless_testing.dirs import DirectoriesMixin
@@ -345,38 +346,39 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
        storage_path = StoragePath.objects.create(name="sp1")
        tag = Tag.objects.create(name="tag")

-        matching_doc = Document.objects.create(
-            title="bank statement",
-            content="bank content",
-            checksum="A",
-            correspondent=correspondent,
-            document_type=doc_type,
-            storage_path=storage_path,
-        )
-        matching_doc.tags.add(tag)
+        for checksum, title in (
+            ("A", "bank statement"),
+            ("B", "bank account"),
+            ("C", "credit card"),
+        ):
+            doc = Document.objects.create(
+                title=title,
+                checksum=checksum,
+                correspondent=correspondent,
+                document_type=doc_type,
+                storage_path=storage_path,
+            )
+            doc.tags.add(tag)
+            get_backend().add_or_update(doc)

-        get_backend().add_or_update(matching_doc)
-
-        response = self.client.get(
-            "/api/documents/?query=bank&include_selection_data=true",
-        )
+        with mock.patch.object(DocumentViewSet, "selection_data_batch_size", 1):
+            response = self.client.get(
+                "/api/documents/?query=bank&page_size=1&include_selection_data=true",
+            )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertIn("selection_data", response.data)
-
-        selected_correspondent = next(
-            item
-            for item in response.data["selection_data"]["selected_correspondents"]
-            if item["id"] == correspondent.id
-        )
-        selected_tag = next(
-            item
-            for item in response.data["selection_data"]["selected_tags"]
-            if item["id"] == tag.id
-        )
-
-        self.assertEqual(selected_correspondent["document_count"], 1)
-        self.assertEqual(selected_tag["document_count"], 1)
+        self.assertEqual(response.data["count"], 2)
+        self.assertEqual(len(response.data["results"]), 1)
+        for key, pk in (
+            ("selected_correspondents", correspondent.pk),
+            ("selected_document_types", doc_type.pk),
+            ("selected_storage_paths", storage_path.pk),
+            ("selected_tags", tag.pk),
+        ):
+            self.assertEqual(
+                response.data["selection_data"][key],
+                [{"id": pk, "document_count": 2}],
+            )

    def test_search_custom_field_ordering(self) -> None:
        custom_field = CustomField.objects.create(
@@ -149,10 +149,10 @@ class TestPostSelectionPathsEnforceTheCap:
        assert str(_MAX_QUERY_LENGTH + 1) in message
        parse_spy.assert_not_called()

-    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
+    @mock.patch("documents.views.execute_bulk_operation.apply_async")
    def test_bulk_edit_query_at_exactly_the_cap_is_accepted(
        self,
-        bulk_update_task_mock: mock.MagicMock,
+        enqueue_mock: mock.MagicMock,
        admin_client: APIClient,
        searchable_document: Document,
    ) -> None:
@@ -163,12 +163,12 @@ class TestPostSelectionPathsEnforceTheCap:
            - Its `filters.query` is exactly `_MAX_QUERY_LENGTH` characters
              long
        THEN:
-            - The cap check accepts it and the request reaches the real
-              bulk-edit method (its Celery dispatch is mocked out here,
-              same as every other bulk-edit test, since nothing here is
-              testing that method itself)
+            - The cap check accepts it and queues the metadata operation.
+              Mock the executor dispatch so this exercises query validation
+              without requiring a broker.
        """
        query = "a" * _MAX_QUERY_LENGTH
+        enqueue_mock.return_value.id = "query-cap-task"

        response = admin_client.post(
            "/api/documents/bulk_edit/",
@@ -183,6 +183,11 @@ class TestPostSelectionPathsEnforceTheCap:
        )

        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {"result": "OK", "task_id": "query-cap-task"}
+        enqueue_mock.assert_called_once()
+        assert (
+            enqueue_mock.call_args.kwargs["kwargs"]["operation"] == "set_document_type"
+        )

    def test_bulk_download_query_one_over_the_cap_is_a_400(
        self,
@@ -100,30 +100,6 @@ class TestApiWorkflows(DirectoriesMixin, APITestCase):
            self.action.assign_correspondent.pk,
        )

-    def test_api_get_workflows_filter_trigger_type(self) -> None:
-        """
-        GIVEN:
-            - A consumption workflow and a manual workflow
-        WHEN:
-            - Workflows are requested filtered by trigger_type
-        THEN:
-            - Only workflows with a matching trigger are returned
-        """
-        manual_trigger = WorkflowTrigger.objects.create(
-            type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
-        )
-        manual_workflow = Workflow.objects.create(name="Manual", order=1)
-        manual_workflow.triggers.add(manual_trigger)
-
-        response = self.client.get(
-            f"{self.ENDPOINT}?trigger_type={WorkflowTrigger.WorkflowTriggerType.MANUAL}",
-            format="json",
-        )
-
-        self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.data["count"], 1)
-        self.assertEqual(response.data["results"][0]["id"], manual_workflow.id)
-
    def test_api_get_workflow_actions_ordered(self) -> None:
        """
        GIVEN:
@@ -0,0 +1,281 @@
+"""Request acceptance, worker execution, and task tracking."""
+
+import uuid
+from unittest import mock
+
+import pytest
+from auditlog.models import LogEntry
+from django.contrib.auth.models import User
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.test import APIClient
+
+from documents.bulk_operations import execute_bulk_operation
+from documents.models import Correspondent
+from documents.models import Document
+from documents.models import PaperlessTask
+from documents.signals.handlers import before_task_publish_handler
+from documents.signals.handlers import task_failure_handler
+from documents.signals.handlers import task_postrun_handler
+from paperless_testing.factories import UserFactory
+
+
+@pytest.fixture
+def bulk_setup(paperless_dirs):
+    user = UserFactory(superuser=True)
+    client = APIClient()
+    client.force_authenticate(user)
+    doc = Document.objects.create(title="Example", checksum="bulk-op-example")
+    correspondent = Correspondent.objects.create(name="Example correspondent")
+    return user, client, doc, correspondent
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "endpoint,operation,parameters",
+    [
+        ("bulk_edit", "set_correspondent", {"correspondent": None}),
+    ],
+)
+def test_request_only_queues(bulk_setup, endpoint, operation, parameters):
+    user, client, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    data = {"documents": [doc.pk], "from_webui": True}
+    if endpoint == "bulk_edit":
+        data.update(method=operation, parameters=parameters)
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = task_id
+        response = client.post(f"/api/documents/{endpoint}/", data, format="json")
+    assert response.status_code == 200
+    assert response.data == {"result": "OK", "task_id": task_id}
+    assert Document.objects.filter(pk=doc.pk).exists()
+    assert enqueue.call_args.kwargs["kwargs"] == {
+        "operation": operation,
+        "document_ids": [doc.pk],
+        "parameters": parameters,
+        "user_id": user.pk,
+    }
+    assert enqueue.call_args.kwargs["headers"]["trigger_source"] == "web_ui"
+
+
+@pytest.mark.django_db
+def test_metadata_task_includes_audit_and_followup(bulk_setup, settings):
+    user, _, doc, correspondent = bulk_setup
+    settings.AUDIT_LOG_ENABLED = True
+    with (
+        mock.patch("documents.bulk_operations.bulk_update_documents") as followup,
+        mock.patch(
+            "documents.bulk_edit.bulk_update_documents.apply_async",
+        ) as extra_task,
+    ):
+
+        def check_followup(ids):
+            doc.refresh_from_db()
+            assert doc.correspondent_id == correspondent.pk
+            assert LogEntry.objects.filter(object_pk=str(doc.pk), actor=user).exists()
+            assert ids == [doc.pk]
+
+        followup.side_effect = check_followup
+        result = execute_bulk_operation(
+            "set_correspondent",
+            [doc.pk],
+            {"correspondent": correspondent.pk},
+            user.pk,
+        )
+    followup.assert_called_once()
+    extra_task.assert_not_called()
+    assert result["document_count"] == 1
+
+
+@pytest.mark.django_db
+def test_permissions_rechecked_on_worker(bulk_setup):
+    user, _, doc, correspondent = bulk_setup
+    User.objects.filter(pk=user.pk).update(is_superuser=False)
+    with pytest.raises(PermissionDenied):
+        execute_bulk_operation(
+            "set_correspondent",
+            [doc.pk],
+            {"correspondent": correspondent.pk},
+            user.pk,
+        )
+    doc.refresh_from_db()
+    assert doc.correspondent_id is None
+
+
+@pytest.mark.django_db
+def test_queue_failure_does_not_change_documents(bulk_setup):
+    _, client, doc, correspondent = bulk_setup
+    with mock.patch(
+        "documents.views.execute_bulk_operation.apply_async",
+        side_effect=RuntimeError("broker unavailable"),
+    ):
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "documents": [doc.pk],
+                "method": "set_correspondent",
+                "parameters": {"correspondent": correspondent.pk},
+            },
+            format="json",
+        )
+    assert response.status_code == 400
+    doc.refresh_from_db()
+    assert doc.correspondent_id is None
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("failed", [False, True])
+def test_tracked_outcome_records_initiator_and_result(bulk_setup, failed):
+    user, _, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    before_task_publish_handler(
+        headers={
+            "task": "documents.bulk_operations.execute_bulk_operation",
+            "id": task_id,
+        },
+        body=(
+            (),
+            {
+                "operation": "set_permissions",
+                "document_ids": [doc.pk],
+                "user_id": user.pk,
+            },
+            {},
+        ),
+    )
+    task = PaperlessTask.objects.get(task_id=task_id)
+    assert task.owner_id == user.pk
+    assert task.task_type == PaperlessTask.TaskType.BULK_UPDATE
+    if failed:
+        task_failure_handler(task_id=task_id, exception=ValueError("Could not update"))
+    else:
+        task_postrun_handler(
+            task_id=task_id,
+            state="SUCCESS",
+            retval={"document_count": 1},
+        )
+    task.refresh_from_db()
+    assert task.status == ("failure" if failed else "success")
+    if failed:
+        assert task.result_data["error_message"] == "Could not update"
+    else:
+        assert task.result_data == {"document_count": 1}
+
+
+@pytest.mark.django_db
+def test_large_filtered_selection_is_one_task(bulk_setup):
+    _, client, _, _ = bulk_setup
+    Document.objects.bulk_create(
+        [
+            Document(title="bulk-large-selection", checksum=f"large-{i}")
+            for i in range(10_000)
+        ],
+    )
+    with mock.patch("documents.views.execute_bulk_operation.apply_async") as enqueue:
+        enqueue.return_value.id = str(uuid.uuid4())
+        response = client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "all": True,
+                "filters": {"title__icontains": "bulk-large-selection"},
+                "method": "set_correspondent",
+                "parameters": {"correspondent": None},
+            },
+            format="json",
+        )
+    assert response.status_code == 200
+    enqueue.assert_called_once()
+    assert len(enqueue.call_args.kwargs["kwargs"]["document_ids"]) == 10_000
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("failed", [False, True])
+def test_tracked_outcome_notifies_only_initiator(bulk_setup, failed):
+    user, _, doc, _ = bulk_setup
+    task_id = str(uuid.uuid4())
+    before_task_publish_handler(
+        headers={
+            "task": "documents.bulk_operations.execute_bulk_operation",
+            "id": task_id,
+        },
+        body=(
+            (),
+            {
+                "operation": "set_permissions",
+                "document_ids": [doc.pk],
+                "user_id": user.pk,
+            },
+            {},
+        ),
+    )
+    task = PaperlessTask.objects.get(task_id=task_id)
+    assert task.owner_id == user.pk
+    assert task.task_type == PaperlessTask.TaskType.BULK_UPDATE
+    with mock.patch("documents.signals.handlers.async_to_sync") as send:
+        if failed:
+            task_failure_handler(
+                task_id=task_id,
+                exception=ValueError("Could not update"),
+            )
+        else:
+            task_postrun_handler(
+                task_id=task_id,
+                state="SUCCESS",
+                retval={"document_count": 1},
+            )
+    group, event = send.return_value.call_args.args
+    assert group == f"bulk_operations_{user.pk}"
+    assert event["data"]["status"] == ("failure" if failed else "success")
+    assert event["data"]["task_id"] == task_id
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio"])
+async def test_completion_socket_is_private(anyio_backend):
+    from types import SimpleNamespace
+
+    from channels.layers import get_channel_layer
+    from channels.testing import WebsocketCommunicator
+
+    from paperless.consumers import StatusConsumer
+
+    own = WebsocketCommunicator(StatusConsumer.as_asgi(), "/ws/status/")
+    other = WebsocketCommunicator(StatusConsumer.as_asgi(), "/ws/status/")
+    own.scope["user"] = SimpleNamespace(id=1, is_authenticated=True)
+    other.scope["user"] = SimpleNamespace(id=2, is_authenticated=True)
+    assert (await own.connect())[0]
+    assert (await other.connect())[0]
+    try:
+        event = {
+            "type": "bulk_operation_status",
+            "data": {"task_id": "example", "status": "success"},
+        }
+        await get_channel_layer().group_send("bulk_operations_1", event)
+        assert await own.receive_json_from() == event
+        assert await other.receive_nothing(timeout=0.02)
+    finally:
+        await own.disconnect()
+        await other.disconnect()
+
+
+@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"),
+    ):
+        response = client.post(
+            f"/api/documents/{endpoint}/",
+            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()
@@ -2288,56 +2288,6 @@ class TestWorkflows(
        doc.refresh_from_db()
        self.assertEqual(doc.owner, self.user2)

-    def _make_manual_workflow(self, *, require_tag: Tag | None = None) -> Workflow:
-        trigger = WorkflowTrigger.objects.create(
-            type=WorkflowTrigger.WorkflowTriggerType.MANUAL,
-        )
-        if require_tag is not None:
-            trigger.filter_has_all_tags.add(require_tag)
-        action = WorkflowAction.objects.create(assign_owner=self.user2)
-        workflow = Workflow.objects.create(name="Manual workflow", order=0)
-        workflow.triggers.add(trigger)
-        workflow.actions.add(action)
-        return workflow
-
-    @mock.patch("documents.tasks.send_websocket_document_updated")
-    def test_manual_workflow_run_applies_and_notifies(self, mock_ws) -> None:
-        """A manual run applies the workflow's actions and pushes a websocket update."""
-        workflow = self._make_manual_workflow()
-        doc = DocumentFactory()
-
-        tasks.run_workflows_for_documents([doc.pk], workflow.pk)
-
-        doc.refresh_from_db()
-        self.assertEqual(doc.owner, self.user2)
-        self.assertEqual(mock_ws.call_args.kwargs["document"].pk, doc.pk)
-
-    def test_manual_workflow_run_respects_filters(self) -> None:
-        """With filter_documents=True, a document failing the trigger filters is skipped."""
-        workflow = self._make_manual_workflow(require_tag=self.t1)
-        doc = DocumentFactory()
-
-        tasks.run_workflows_for_documents([doc.pk], workflow.pk, filter_documents=True)
-
-        doc.refresh_from_db()
-        self.assertIsNone(doc.owner)
-
-    def test_manual_workflow_run_force_ignores_filters(self) -> None:
-        """With filter_documents=False, actions run regardless of the trigger filters."""
-        workflow = self._make_manual_workflow(require_tag=self.t1)
-        doc = DocumentFactory()
-
-        tasks.run_workflows_for_documents([doc.pk], workflow.pk, filter_documents=False)
-
-        doc.refresh_from_db()
-        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."""
-        doc = DocumentFactory()
-        with self.assertLogs("paperless.tasks", level="WARNING"):
-            tasks.run_workflows_for_documents([doc.pk], 9999)
-
    @mock.patch("documents.tasks.send_websocket_document_updated")
    def test_workflow_scheduled_trigger_sends_websocket_update(
        self,
@@ -5,11 +5,14 @@ import platform
 import re
 import tempfile
 import zipfile
+from collections import Counter
 from collections import deque
 from collections.abc import Callable
 from datetime import datetime
 from datetime import timedelta
 from http import HTTPStatus
+from inspect import signature
+from itertools import islice
 from pathlib import Path
 from time import mktime
 from time import sleep
@@ -104,6 +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 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
@@ -140,7 +146,6 @@ from documents.filters import ShareLinkFilterSet
 from documents.filters import StoragePathFilterSet
 from documents.filters import TagFilterSet
 from documents.filters import TitleContentFilter
-from documents.filters import WorkflowFilterSet
 from documents.mail import EmailAttachment
 from documents.mail import send_email
 from documents.matching import match_correspondents
@@ -169,10 +174,10 @@ from documents.permissions import PaperlessNotePermissions
 from documents.permissions import PaperlessObjectPermissions
 from documents.permissions import TrashPermissions
 from documents.permissions import ViewDocumentsPermissions
-from documents.permissions import annotate_document_count_by_ids
 from documents.permissions import annotate_document_count_for_related_queryset
 from documents.permissions import get_document_count_filter_for_user
 from documents.permissions import get_permission_grantees
+from documents.permissions import has_document_operation_permissions
 from documents.permissions import has_global_statistics_permission
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
@@ -1015,71 +1020,80 @@ class DocumentViewSet(
        "custom_field_",
    )

-    def _get_selection_data_for_queryset(self, queryset):
-        # Resolve once instead of once per model below. `queryset` can carry an
-        # arbitrarily expensive WHERE clause (user filters plus the permission
-        # filter); re-embedding it as a subquery inside 5 separate Count(...)
-        # calls forces the database to re-evaluate that whole thing 5 times, and
-        # -- for FK relations especially -- can defeat semi-join planning
-        # entirely at scale. A concrete id list is cheap to reuse.
-        # order_by() drops the default/user ordering -- irrelevant for a plain
-        # id list, but left in place it forces a sort over the full filtered
-        # set before the ids can even be collected.
-        document_ids = list(queryset.order_by().values_list("pk", flat=True))
+    # Bound the IN clauses used for search results and through-table lookups.
+    selection_data_batch_size = 10_000

-        correspondents = Correspondent.objects.annotate(
-            document_count=Count(
-                "documents",
-                filter=Q(documents__id__in=document_ids),
-                distinct=True,
-            ),
-        )
-        document_types = DocumentType.objects.annotate(
-            document_count=Count(
-                "documents",
-                filter=Q(documents__id__in=document_ids),
-                distinct=True,
-            ),
-        )
-        storage_paths = StoragePath.objects.annotate(
-            document_count=Count(
-                "documents",
-                filter=Q(documents__id__in=document_ids),
-                distinct=True,
-            ),
-        )
-        # Tag and CustomField reach Document through an M2M/through-model table;
-        # a plain Count(filter=...) there is a much more expensive plan than the
-        # FK relations above once the bridge table is large -- see
-        # annotate_document_count_by_ids() for why.
-        tags = annotate_document_count_by_ids(
-            Tag.objects.all(),
-            through_model=Document.tags.through,
-            related_object_field="tag_id",
-            document_ids=document_ids,
-        )
-        custom_fields = annotate_document_count_by_ids(
-            CustomField.objects.all(),
-            through_model=CustomFieldInstance,
-            related_object_field="field_id",
-            document_ids=document_ids,
+    def _get_selection_data_for_queryset(
+        self,
+        queryset: QuerySet[Document],
+        *,
+        search_ids: list[int] | None = None,
+    ) -> dict[str, list[dict[str, int]]]:
+        # Keep permission and filter predicates at document granularity.
+        # Reusing the filtered queryset as a subquery for each count is slow.
+        rows = (
+            queryset.order_by()
+            .prefetch_related(None)
+            .values_list(
+                "pk",
+                "correspondent_id",
+                "document_type_id",
+                "storage_path_id",
+            )
        )
+        batch_size = self.selection_data_batch_size
+
+        def document_rows():
+            if search_ids is None:
+                yield from rows.iterator()
+            else:
+                for start in range(0, len(search_ids), batch_size):
+                    yield from rows.filter(
+                        pk__in=search_ids[start : start + batch_size],
+                    ).iterator()
+
+        correspondents: Counter[int | None] = Counter()
+        document_types: Counter[int | None] = Counter()
+        storage_paths: Counter[int | None] = Counter()
+        document_ids: set[int] = set()
+        for pk, correspondent, document_type, storage_path in document_rows():
+            # Filters that join multi-valued relations can repeat a document.
+            if pk in document_ids:
+                continue
+            document_ids.add(pk)
+            correspondents[correspondent] += 1
+            document_types[document_type] += 1
+            storage_paths[storage_path] += 1
+
+        tags: Counter[int | None] = Counter()
+        custom_fields: Counter[int | None] = Counter()
+        ids = iter(document_ids)
+        while batch := list(islice(ids, batch_size)):
+            for model, field, counts in (
+                (Document.tags.through, "tag_id", tags),
+                (CustomFieldInstance, "field_id", custom_fields),
+            ):
+                counts.update(
+                    dict(
+                        model.objects.filter(document_id__in=batch)
+                        .order_by()
+                        .values_list(field)
+                        .annotate(document_count=Count("document_id")),
+                    ),
+                )
+
+        def selection(model: type[Model], counts: Counter[int | None]):
+            return [
+                {"id": pk, "document_count": counts[pk]}
+                for pk in model._default_manager.values_list("pk", flat=True)
+            ]
+
        return {
-            "selected_correspondents": [
-                {"id": t.id, "document_count": t.document_count} for t in correspondents
-            ],
-            "selected_tags": [
-                {"id": t.id, "document_count": t.document_count} for t in tags
-            ],
-            "selected_document_types": [
-                {"id": t.id, "document_count": t.document_count} for t in document_types
-            ],
-            "selected_storage_paths": [
-                {"id": t.id, "document_count": t.document_count} for t in storage_paths
-            ],
-            "selected_custom_fields": [
-                {"id": t.id, "document_count": t.document_count} for t in custom_fields
-            ],
+            "selected_correspondents": selection(Correspondent, correspondents),
+            "selected_tags": selection(Tag, tags),
+            "selected_document_types": selection(DocumentType, document_types),
+            "selected_storage_paths": selection(StoragePath, storage_paths),
+            "selected_custom_fields": selection(CustomField, custom_fields),
        }

    @classmethod
@@ -2726,13 +2740,10 @@ class UnifiedSearchViewSet(DocumentViewSet):
                if get_boolean(
                    str(request.query_params.get("include_selection_data", "false")),
                ):
-                    # NOTE: pk__in=ordered_ids generates a large SQL IN clause
-                    # for big result sets.  Acceptable today but may need a temp
-                    # table or chunked approach if selection_data becomes slow
-                    # at scale (tens of thousands of matching documents).
                    response.data["selection_data"] = (
                        self._get_selection_data_for_queryset(
-                            filtered_qs.filter(pk__in=result.ordered_ids),
+                            filtered_qs,
+                            search_ids=result.ordered_ids,
                        )
                    )
                return response
@@ -2976,99 +2987,29 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
        "merge_as_versions",
    }

-    def _has_document_permissions(
-        self,
-        *,
-        user: User,
-        documents: list[int],
-        method,
-        parameters: dict[str, Any],
-    ) -> bool:
-        if user.is_superuser:
-            return True
+    def _has_document_permissions(self, **kwargs) -> bool:
+        return has_document_operation_permissions(**kwargs)

-        root_docs = {
-            get_root_document(doc)
-            for doc in Document.objects.select_related(
-                "owner",
-                "root_document__owner",
-            ).filter(pk__in=documents)
-        }
-        user_is_owner_of_all_documents = all(
-            (doc.owner == user or doc.owner is None) for doc in root_docs
+    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)
+        task = execute_bulk_operation.apply_async(
+            kwargs={
+                "operation": method.__name__,
+                "document_ids": documents,
+                "parameters": parameters,
+                "user_id": self.request.user.pk,
+            },
+            headers={
+                "trigger_source": (
+                    PaperlessTask.TriggerSource.WEB_UI
+                    if from_webui
+                    else PaperlessTask.TriggerSource.API_UPLOAD
+                ),
+            },
        )
-
-        # check global and object permissions for all documents
-        has_perms = (
-            user.has_perm(
-                "documents.change_document",
-            )
-            and not Document.global_objects.filter(pk__in=documents)
-            .exclude(pk__in=permitted_document_ids(user, perm="change_document"))
-            .exists()
-        )
-
-        # check ownership for methods that change original document
-        if (
-            (
-                has_perms
-                and method
-                in [
-                    bulk_edit.set_permissions,
-                    bulk_edit.delete,
-                    bulk_edit.rotate,
-                    bulk_edit.delete_pages,
-                    bulk_edit.edit_pdf,
-                    bulk_edit.merge_as_versions,
-                    bulk_edit.remove_password,
-                    bulk_edit.run_workflow,
-                ]
-            )
-            or (
-                method in [bulk_edit.merge, bulk_edit.split]
-                and parameters.get("delete_originals")
-            )
-            or (method == bulk_edit.edit_pdf and parameters.get("update_document"))
-        ):
-            has_perms = has_perms and user_is_owner_of_all_documents
-
-        # check global add permissions for methods that create documents
-        if (
-            has_perms
-            and (
-                method in [bulk_edit.split, bulk_edit.merge]
-                or (
-                    method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
-                    and not parameters.get("update_document")
-                )
-            )
-            and not user.has_perm("documents.add_document")
-        ):
-            has_perms = False
-
-        # check global delete permissions for methods that delete documents
-        if (
-            has_perms
-            and (
-                method == bulk_edit.delete
-                # Sources stop being documents of their own, and removing one
-                # again afterwards needs delete_document
-                or method == bulk_edit.merge_as_versions
-                or (
-                    method in [bulk_edit.merge, bulk_edit.split]
-                    and parameters.get("delete_originals")
-                )
-                or (
-                    method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
-                    and parameters.get("delete_original")
-                    and not parameters.get("update_document")
-                )
-            )
-            and not user.has_perm("documents.delete_document")
-        ):
-            has_perms = False
-
-        return has_perms
+        return Response({"result": "OK", "task_id": task.id})

    def _execute_document_action(
        self,
@@ -3136,6 +3077,7 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
                name="BulkEditDocumentsResult",
                fields={
                    "result": serializers.CharField(),
+                    "task_id": serializers.CharField(required=False),
                },
            ),
        },
@@ -3161,42 +3103,11 @@ class BulkEditView(DocumentOperationPermissionMixin):
        "edit_pdf": None,
        "reprocess": "checksum",
        "remove_password": None,
-        "run_workflow": None,
    }

    serializer_class = BulkEditSerializer

-    @staticmethod
-    def _snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]:
-        """
-        Returns each document's current value of field, for the audit log.
-
-        Tags and custom fields are one row per value, so they are gathered
-        into a sorted list of pks per document (empty when there are none).
-        Reading them through Document.values() instead would join those rows
-        and return one arbitrary value per document.
-        """
-        if field == "tags":
-            rows = (
-                Document.tags.through.objects.filter(document_id__in=doc_ids)
-                .order_by("tag_id")
-                .values_list("document_id", "tag_id")
-            )
-        elif field == "custom_fields":
-            rows = (
-                CustomFieldInstance.objects.filter(document_id__in=doc_ids)
-                .order_by("pk")
-                .values_list("document_id", "pk")
-            )
-        else:
-            return dict(
-                Document.objects.filter(pk__in=doc_ids).values_list("pk", field),
-            )
-
-        values: dict[int, list[int]] = {doc_id: [] for doc_id in doc_ids}
-        for doc_id, pk in rows:
-            values[doc_id].append(pk)
-        return values
+    _snapshot_field = staticmethod(snapshot_field)

    def post(self, request, *args, **kwargs):
        request_method = request.data.get("method")
@@ -3242,6 +3153,13 @@ 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.MODIFIED_FIELD_BY_METHOD.get(method.__name__, None)
            if settings.AUDIT_LOG_ENABLED and modified_field:
                old_values = self._snapshot_field(documents, modified_field)
@@ -4074,7 +3992,7 @@ class StatisticsView(GenericAPIView[Any]):
            for row in mime_type_stats
        ]

-        current_asn = Document.objects.aggregate(
+        current_asn = documents.aggregate(
            Max("archive_serial_number", default=0),
        ).get(
            "archive_serial_number__max",
@@ -5192,8 +5110,6 @@ class WorkflowViewSet(ModelViewSet[Workflow]):

    serializer_class = WorkflowSerializer
    pagination_class = StandardPagination
-    filter_backends = (DjangoFilterBackend,)
-    filterset_class = WorkflowFilterSet

    model = Workflow

@@ -2,7 +2,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: paperless-ngx\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-11-06 20:19+0000\n"
+"POT-Creation-Date: 2026-11-07 02:45+0000\n"
 "PO-Revision-Date: 2022-02-17 04:17\n"
 "Last-Translator: \n"
 "Language-Team: English\n"
@@ -1661,8 +1661,8 @@ msgid "workflow runs"
 msgstr ""

 #: documents/serialisers.py:483 documents/serialisers.py:844
-#: documents/serialisers.py:2850 documents/views.py:348 documents/views.py:614
-#: documents/views.py:2748 paperless_mail/serialisers.py:156
+#: documents/serialisers.py:2850 documents/views.py:349 documents/views.py:615
+#: documents/views.py:2755 paperless_mail/serialisers.py:156
 msgid "Insufficient permissions."
 msgstr ""

@@ -1702,7 +1702,7 @@ msgstr ""
 msgid "Duplicate document identifiers are not allowed."
 msgstr ""

-#: documents/serialisers.py:2936 documents/views.py:4798
+#: documents/serialisers.py:2936 documents/views.py:4805
 #, python-format
 msgid "Documents not found: %(ids)s"
 msgstr ""
@@ -1953,40 +1953,40 @@ msgstr ""
 msgid ", "
 msgstr ""

-#: documents/views.py:303
+#: documents/views.py:304
 msgid "Specify only one of text, title_search, query, or more_like_id."
 msgstr ""

-#: documents/views.py:341 documents/views.py:2745
+#: documents/views.py:342 documents/views.py:2752
 msgid "Invalid more_like_id"
 msgstr ""

-#: documents/views.py:1669
+#: documents/views.py:1679
 msgid "Invalid AI configuration."
 msgstr ""

-#: documents/views.py:1680
+#: documents/views.py:1690
 msgid "AI backend request timed out."
 msgstr ""

-#: documents/views.py:1692
+#: documents/views.py:1702
 msgid "AI backend rejected the request. Check logs for details."
 msgstr ""

-#: documents/views.py:4814
+#: documents/views.py:4821
 #, python-format
 msgid "Insufficient permissions to share document %(id)s."
 msgstr ""

-#: documents/views.py:4861
+#: documents/views.py:4868
 msgid "Bundle is already being processed."
 msgstr ""

-#: documents/views.py:4927
+#: documents/views.py:4934
 msgid "The share link bundle is still being prepared. Please try again later."
 msgstr ""

-#: documents/views.py:4941
+#: documents/views.py:4948
 msgid "The share link bundle is unavailable."
 msgstr ""

@@ -45,12 +45,21 @@ class StatusConsumer(AsyncWebsocketConsumer):
            await self.close()
            return
        await self.channel_layer.group_add("status_updates", self.channel_name)
+        await self.channel_layer.group_add(
+            f"bulk_operations_{self.scope['user'].id}",
+            self.channel_name,
+        )
        await self.accept()
        self._start_heartbeat()

    async def disconnect(self, code: int) -> None:
        await self._stop_heartbeat()
        await self.channel_layer.group_discard("status_updates", self.channel_name)
+        if self._authenticated():
+            await self.channel_layer.group_discard(
+                f"bulk_operations_{self.scope['user'].id}",
+                self.channel_name,
+            )

    def _start_heartbeat(self) -> None:
        self.heartbeat_task = asyncio.create_task(self._heartbeat_loop())
@@ -84,3 +93,9 @@ class StatusConsumer(AsyncWebsocketConsumer):
            await self.close()
        elif await self._can_view(event["data"]):
            await self.send(json.dumps(event))
+
+    async def bulk_operation_status(self, event: dict) -> None:
+        if not self._authenticated():
+            await self.close()
+        else:
+            await self.send(json.dumps(event))
@@ -690,7 +690,7 @@ logging.config.dictConfig(LOGGING)
 # https://docs.celeryq.dev/en/stable/userguide/configuration.html

 # Tasks outside tasks.py must be loaded even when Django skips checks
-CELERY_IMPORTS = ("documents.bulk_edit",)
+CELERY_IMPORTS = ("documents.bulk_edit", "documents.bulk_operations")

 CELERY_BROKER_URL = _CELERY_REDIS_URL
 CELERY_RESULT_BACKEND = _CELERY_REDIS_URL
@@ -32,13 +32,13 @@ from rest_framework.fields import BooleanField
 from rest_framework.filters import OrderingFilter
 from rest_framework.generics import GenericAPIView
 from rest_framework.pagination import PageNumberPagination
-from rest_framework.permissions import DjangoModelPermissions
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.throttling import ScopedRateThrottle
 from rest_framework.viewsets import ModelViewSet

 from documents.models import PaperlessTask
+from documents.permissions import PaperlessModelPermissions
 from documents.permissions import PaperlessObjectPermissions
 from documents.tasks import llmindex_index
 from paperless.filters import GroupFilterSet
@@ -414,7 +414,7 @@ class ApplicationConfigurationViewSet(ModelViewSet[ApplicationConfiguration]):
    queryset = ApplicationConfiguration.objects

    serializer_class = ApplicationConfigurationSerializer
-    permission_classes = (IsAuthenticated, DjangoModelPermissions)
+    permission_classes = (IsAuthenticated, PaperlessModelPermissions)

    @extend_schema(exclude=True)
    def create(self, request, *args, **kwargs):
Author	SHA1	Message	Date
shamoon	e427818360	Fix test_bulk_edit_query_at_exactly_the_cap_is_accepted	2026-11-07 08:22:34 -08:00
shamoon	2f623bad7b	Add back the queued notificaiton + link	2026-11-07 08:20:54 -08:00
shamoon	6a162abbfa	prevent refreshes from changing destination	2026-11-07 08:10:04 -08:00
shamoon	0bd07aa61d	This was a leftover	2026-11-07 08:10:04 -08:00
shamoon	358a291455	Shit, lets move delete back	2026-11-07 08:10:03 -08:00
shamoon	cfa997b167	Update app-frame.component.ts	2026-11-07 08:10:03 -08:00
shamoon	e52956c382	Hook up the refresh stuff on ws for bulk operations	2026-11-07 08:10:03 -08:00
shamoon	f13f83aff5	Track task_id query param in tasks, add filtering by id	2026-11-07 08:10:02 -08:00
shamoon	631b19ef56	Actually we always get the task_id	2026-11-07 08:10:02 -08:00
shamoon	14ec3a2582	Update frontend bulk operatoins to pass from_webui, get the task id	2026-11-07 08:10:02 -08:00
shamoon	02a3d5ebe1	Ping via ws after bulk operations complete	2026-11-07 08:10:01 -08:00
shamoon	a07a2f15fb	Backend tests for the bulk operatoins stuff, task tracking	2026-11-07 08:10:01 -08:00
shamoon	570a4af601	Include bulk_operations for celery	2026-11-07 08:10:01 -08:00
shamoon	dc02ca4707	Include from_webui for delete serializer	2026-11-07 08:10:00 -08:00
shamoon	ee55d0adca	Task tracking for bulk operations	2026-11-07 08:10:00 -08:00
shamoon	bc89e00771	Make bulk edit use the bulk_operations callbacks, and unify the finish	2026-11-07 08:10:00 -08:00
shamoon	27bf4b05fa	Create a bulk_operations module, move snapshot_field and has_document_operation_permissions	2026-11-07 08:09:59 -08:00
shamoon	0736c07d8e	Fix: use filtered documents for statistics ASN max (#14710 )	2026-11-07 07:52:23 -08:00
shamoon	53d9aa34a1	FIx: make config view permissions consistent (#14709 )	2026-11-07 07:52:12 -08:00
GitHub Actions	19fdc3b815	Auto translate strings	2026-11-07 02:45:43 +00:00
Trenton H Claude Sonnet 5.5 Codex shamoon	5357be2d43	Performance: stop embedding every document ID in selection_data queries (#14597 ) * Fix: count selection_data from one pass over the filtered documents Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> * Chore: use document type ids in the bulk edit selection_data test Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> * Batch search selection data queries Co-Authored-By: Codex <267193182+codex@users.noreply.github.com> --------- Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com> Co-authored-by: shamoon <4887959+shamoon@users.noreply.github.com> Co-authored-by: Codex <267193182+codex@users.noreply.github.com>	2026-11-06 18:44:18 -08:00