Fix migration

docs
Add to doc details
2026-11-08 07:33:38 +00:00 · 2026-11-06 14:15:05 -08:00 · 2026-11-06 14:14:29 -08:00 · 2026-11-06 14:14:29 -08:00 · 2026-11-06 14:14:29 -08:00 · 2026-11-06 14:14:28 -08:00
50 changed files with 1034 additions and 1549 deletions
@@ -493,7 +493,7 @@ fields and permissions, which will be merged.

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

-Currently, there are four events that correspond to workflow trigger 'types':
+Currently, there are five 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,8 +505,10 @@ Currently, there are four 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 four document trigger types:
+The following flow diagram illustrates the document trigger types:

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

+    manual{"Matching
+    'Manual'
+    trigger(s)"}
+
    A[New Document] --> consumption
    consumption --> |Yes| C[Workflow Actions Run]
    consumption --> |No| D
@@ -543,6 +549,10 @@ 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}
@@ -554,9 +564,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` and `Scheduled` triggers only). Filter document content using the matching settings.
+- Content matching (`Added`, `Updated`, `Scheduled` and `Manual` triggers only). Filter document content using the matching settings.

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

 - Any Tags: Filter for documents with any of the specified tags.
 - All Tags: Filter for documents with all of the specified tags.
@@ -569,6 +579,21 @@ There are also 'advanced' filters available for `Added`, `Updated` and `Schedule
 - 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}
@@ -649,7 +674,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**: these triggers run against a document that
+**Document Added**, **Document Updated**, **Scheduled**, **Manual**: 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,15 +7,10 @@ import {
 import { ComponentFixture, TestBed } from '@angular/core/testing'
 import { FormsModule } from '@angular/forms'
 import { By } from '@angular/platform-browser'
-import {
-  ActivatedRoute,
-  convertToParamMap,
-  ParamMap,
-  Router,
-} from '@angular/router'
+import { Router } from '@angular/router'
 import { RouterTestingModule } from '@angular/router/testing'
 import { NgbModal, NgbModalRef, NgbModule } from '@ng-bootstrap/ng-bootstrap'
-import { BehaviorSubject, of, throwError } from 'rxjs'
+import { of, throwError } from 'rxjs'
 import { routes } from 'src/app/app-routing.module'
 import {
  PaperlessTask,
@@ -164,7 +159,6 @@ const sectionCountResponse = {
 }

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

  beforeEach(async () => {
-    queryParams = new BehaviorSubject(convertToParamMap({}))
    TestBed.configureTestingModule({
      imports: [
        NgbModule,
@@ -188,17 +181,6 @@ describe('TasksComponent', () => {
        ConfirmDialogComponent,
      ],
      providers: [
-        {
-          provide: ActivatedRoute,
-          useValue: {
-            queryParamMap: queryParams.asObservable(),
-            snapshot: {
-              get queryParamMap() {
-                return queryParams.value
-              },
-            },
-          },
-        },
        {
          provide: PermissionsService,
          useValue: {
@@ -256,71 +238,6 @@ 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 { ActivatedRoute, Router, RouterLink } from '@angular/router'
+import { Router, RouterLink } from '@angular/router'
 import {
  NgbCollapseModule,
  NgbDropdownModule,
@@ -47,13 +47,11 @@ 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 = {
@@ -163,7 +161,6 @@ 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
@@ -278,17 +275,6 @@ 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)
@@ -650,9 +636,6 @@ 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),
@@ -723,15 +706,9 @@ export class TasksComponent
    }

    if (this._filterText.length) {
-      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
-      }
+      params[
+        this.filterTargetID === TaskFilterTargetID.Name ? 'name' : 'result'
+      ] = this._filterText
    }

    return params
@@ -11,10 +11,8 @@ 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, Subject, throwError } from 'rxjs'
+import { of, 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'
@@ -23,11 +21,10 @@ 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 {
-  PermissionsService,
  PermissionType,
+  PermissionsService,
 } 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'
@@ -98,7 +95,6 @@ describe('AppFrameComponent', () => {
  let savedViewSpy
  let modalService: NgbModal
  let maybeRefreshSpy
-  let bulkEvents: Subject<BulkOperationStatus>
  let tasksService: TasksService

  beforeEach(async () => {
@@ -186,10 +182,6 @@ 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

@@ -198,71 +190,6 @@ 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,14 +8,12 @@ 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 {
@@ -26,11 +24,10 @@ import {
  NgbPopoverModule,
 } from '@ng-bootstrap/ng-bootstrap'
 import { TourNgBootstrap } from 'ngx-ui-tour-ng-bootstrap'
-import { filter, Observable } from 'rxjs'
+import { 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,
@@ -44,7 +41,6 @@ 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,
@@ -105,8 +101,6 @@ 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)
@@ -170,44 +164,6 @@ 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,6 +118,7 @@ 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'
@@ -279,6 +280,7 @@ export const icons: Record<string, string> = {
  personFillLock,
  personLock,
  personSquare,
+  playCircle,
  playFill,
  plus,
  plusCircle,
@@ -0,0 +1,32 @@
+<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>
@@ -0,0 +1,42 @@
+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')
+  })
+})
@@ -0,0 +1,24 @@
+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) {
+        @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) {
          <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) {
+    @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) {
      <div class="row mt-3">
        <div class="col">
          <div class="trigger-filters mb-3">
@@ -123,6 +123,10 @@ export const WORKFLOW_TYPE_OPTIONS = [
    id: WorkflowTriggerType.Scheduled,
    name: $localize`Scheduled`,
  },
+  {
+    id: WorkflowTriggerType.Manual,
+    name: $localize`Manual`,
+  },
 ]

 export const WORKFLOW_ACTION_OPTIONS = [
@@ -49,6 +49,12 @@
        <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,6 +1030,27 @@ 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,6 +68,8 @@ 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'
@@ -93,6 +95,7 @@ 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'
@@ -104,6 +107,7 @@ 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'
@@ -233,6 +237,7 @@ 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
@@ -295,6 +300,7 @@ 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(''),
@@ -836,6 +842,19 @@ 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()

@@ -1559,6 +1578,39 @@ 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,6 +98,11 @@
          <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,6 +17,7 @@ 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 {
@@ -36,6 +37,7 @@ 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'
@@ -83,6 +85,7 @@ describe('BulkEditorComponent', () => {
  let customFieldsService: CustomFieldsService
  let httpTestingController: HttpTestingController
  let shareLinkBundleService: ShareLinkBundleService
+  let workflowService: WorkflowService
  let router: Router

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

  it('should execute modify tags bulk operation for all filtered documents', () => {
@@ -608,18 +627,17 @@ describe('BulkEditorComponent', () => {
    const req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush({ result: 'OK', task_id: 'queued-task' })
+    req.flush(true)
    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.expectNone(
+    httpTestingController.match(
      `${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', () => {
@@ -646,12 +664,56 @@ describe('BulkEditorComponent', () => {
    httpTestingController
      .expectOne(`${environment.apiBaseUrl}documents/bulk_edit/`)
      .flush(true)
-    httpTestingController.expectNone(
+    httpTestingController.match(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    )
-    httpTestingController.expectNone(
+    ) // 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(
+      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
+    ) // listAllFilteredIds
  })

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

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

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

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

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

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

  it('should set modal dialog text accordingly for storage path edit confirmation', () => {
@@ -1028,19 +1087,18 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush({ result: 'OK', task_id: 'queued-task' })
+    req.flush(true)
    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.expectNone(
+    httpTestingController.match(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    )
-    httpTestingController.expectNone(
+    ) // list reload
+    httpTestingController.match(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    )
+    ) // listAllFilteredIds
  })

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

    // coverage for modal messages
    component.setCustomFields({
@@ -1546,47 +1604,6 @@ 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]))
@@ -1619,9 +1636,8 @@ describe('BulkEditorComponent', () => {
    let req = httpTestingController.expectOne(
      `${environment.apiBaseUrl}documents/bulk_edit/`
    )
-    req.flush({ result: 'OK', task_id: 'queued-task' })
+    req.flush(true)
    expect(req.request.body).toEqual({
-      from_webui: true,
      documents: [3, 4],
      method: 'set_permissions',
      parameters: {
@@ -1629,12 +1645,12 @@ describe('BulkEditorComponent', () => {
        merge: true,
      },
    })
-    httpTestingController.expectNone(
+    httpTestingController.match(
      `${environment.apiBaseUrl}documents/?page=1&page_size=50&ordering=-created&truncate_content=true&include_selection_data=true`
-    )
-    httpTestingController.expectNone(
+    ) // list reload
+    httpTestingController.match(
      `${environment.apiBaseUrl}documents/?page=1&page_size=100000&fields=id`
-    )
+    ) // listAllFilteredIds
  })

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

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

  it('should create share link bundle and enable manage callback', () => {
@@ -34,6 +34,7 @@ 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'
@@ -53,6 +54,7 @@ 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'
@@ -64,6 +66,7 @@ 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'
@@ -81,6 +84,7 @@ 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',
@@ -104,18 +108,21 @@ 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()
@@ -258,6 +265,19 @@ 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')
@@ -286,6 +306,28 @@ 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,
@@ -299,33 +341,11 @@ export class BulkEditorComponent
      .bulkEdit(overrideSelection ?? this.getSelectionQuery(), method, args)
      .pipe(first())
      .subscribe({
-        next: ({ task_id }) => this.handleOperationQueued(modal, task_id),
+        next: () => this.handleOperationSuccess(modal),
        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>,
@@ -1110,9 +1130,14 @@ export class BulkEditorComponent

    dialog.selection = this.getSelectionQuery()
    dialog.selectionCount = this.getSelectionSize()
-    dialog.succeeded.subscribe(({ task_id }) =>
-      this.handleOperationQueued(null, task_id)
-    )
+    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.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({ result: 'OK', task_id: 'custom-fields-task' }))
+      .mockReturnValue(of('Success'))
    const successSpy = jest.spyOn(component.succeeded, 'emit')

    component.selection = [1, 2]
@@ -16,7 +16,6 @@ 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,
@@ -50,7 +49,7 @@ export class CustomFieldsBulkEditDialogComponent {
  CustomFieldDataType = CustomFieldDataType

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

  @Output()
  failed = new EventEmitter()
@@ -104,8 +103,6 @@ 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,
@@ -113,12 +110,11 @@ export class CustomFieldsBulkEditDialogComponent {
      })
      .pipe(first())
      .subscribe({
-        next: (response) => {
+        next: () => {
          this.activeModal.close()
-          this.succeeded.emit(response)
+          this.succeeded.emit()
        },
        error: (error) => {
-          this.networkActive = false
          this.failed.emit(error)
        },
      })
@@ -1,13 +0,0 @@
-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,6 +12,7 @@ export enum WorkflowTriggerType {
  DocumentAdded = 2,
  DocumentUpdated = 3,
  Scheduled = 4,
+  Manual = 5,
 }

 export enum ScheduleDateField {
@@ -124,42 +124,6 @@ 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 (updateQueryParams && this._activeSavedViewId) {
+          } else if (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(onFinish, updateQueryParams)
+            this.reload()
          } else if (
            activeListViewState.sortField.indexOf('custom_field') === 0 &&
            !this.settings
@@ -248,7 +248,6 @@ describe(`DocumentService`, () => {
    )
    expect(req.request.method).toEqual('POST')
    expect(req.request.body).toEqual({
-      from_webui: true,
      documents: ids,
      method,
      parameters,
@@ -272,7 +271,6 @@ 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,7 +14,6 @@ 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,
@@ -39,6 +38,7 @@ export type DocumentBulkEditMethod =
  | 'modify_tags'
  | 'modify_custom_fields'
  | 'set_permissions'
+  | 'run_workflow'

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

  deleteDocuments(selection: DocumentSelectionQuery) {
@@ -438,20 +438,6 @@ 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,7 +1,6 @@
 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'
@@ -13,7 +12,6 @@ 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
@@ -118,13 +116,6 @@ 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) {
@@ -201,7 +192,6 @@ export class WebsocketStatusService {
          | WebsocketProgressMessage
          | WebsocketDocumentsDeletedMessage
          | WebsocketDocumentUpdatedMessage
-          | BulkOperationStatus
      } = JSON.parse(ev.data)

      switch (type) {
@@ -219,12 +209,6 @@ 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,24 +132,9 @@ 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)
@@ -162,17 +147,15 @@ def set_correspondent(
    affected_docs = list(qs.values_list("pk", flat=True))
    qs.update(correspondent=correspondent)

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

    return "OK"


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

@@ -186,17 +169,15 @@ def set_storage_path(
    affected_docs = list(qs.values_list("pk", flat=True))
    qs.update(storage_path=storage_path)

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

    return "OK"


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

@@ -208,17 +189,15 @@ def set_document_type(
    affected_docs = list(qs.values_list("pk", flat=True))
    qs.update(document_type=document_type)

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

    return "OK"


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

@@ -239,17 +218,15 @@ def add_tag(
        DocumentTagRelationship.objects.bulk_create(to_create)

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

    return "OK"


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

@@ -262,7 +239,10 @@ def remove_tag(
    qs.delete()

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

    return "OK"

@@ -271,8 +251,6 @@ 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))
@@ -316,7 +294,10 @@ def modify_tags(
                )

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

    return "OK"

@@ -325,8 +306,6 @@ 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))
@@ -389,7 +368,10 @@ def modify_custom_fields(
        field_id__in=remove_custom_fields,
    ).hard_delete()

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

    return "OK"

@@ -464,13 +446,31 @@ 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,7 +488,10 @@ def set_permissions(
        merge=merge,
    )

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

    return "OK"

@@ -1,107 +0,0 @@
-"""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,6 +35,7 @@ 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
@@ -52,6 +53,7 @@ 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
@@ -249,6 +251,16 @@ 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,6 +684,7 @@ 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,
@@ -0,0 +1,43 @@
+# 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,6 +1532,7 @@ 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,7 +31,6 @@ 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
@@ -64,17 +63,6 @@ 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
@@ -588,9 +576,12 @@ 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. Keep
-                       the list small: it is embedded in the statement as
-                       literal parameters.
+                       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``.
        target_field: field on the relation pointing to Document id
    """

@@ -739,100 +730,3 @@ 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,6 +1829,7 @@ class BulkEditSerializer(
            "modify_tags",
            "modify_custom_fields",
            "set_permissions",
+            "run_workflow",
            *LEGACY_DOCUMENT_ACTION_METHODS,
        ],
        label="Method",
@@ -1918,6 +1919,8 @@ 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":
@@ -2157,6 +2160,27 @@ 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")
@@ -2223,6 +2247,8 @@ 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,7 +8,6 @@ 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
@@ -17,7 +16,6 @@ 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
@@ -864,6 +862,8 @@ 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,6 +874,10 @@ 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
@@ -923,7 +927,11 @@ def run_workflows(
                )
                break

-        if matching.document_matches_workflow(document, workflow, trigger_type):
+        if skip_matching or 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"):
@@ -1066,7 +1074,6 @@ 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] = {
@@ -1084,8 +1091,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. Bulk operations store
-    the operation name and document count, without copying the full selection.
+    mail_fetch tasks it captures the account_ids list.  All other task
+    types store no input data and return {}.
    """
    if task_type == PaperlessTask.TaskType.CONSUME_FILE:
        input_doc = task_kwargs.get("input_doc")
@@ -1114,12 +1121,6 @@ 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:
@@ -1156,9 +1157,7 @@ def _extract_owner_id(
    task_type: PaperlessTask.TaskType,
    task_kwargs: dict,
 ) -> int | None:
-    """Use the initiator for bulk operations and the document owner for consumption."""
-    if "operation" in task_kwargs:
-        return task_kwargs.get("user_id")
+    """Return the owner_id from consume_file overrides, or None for all other task types."""
    if task_type != PaperlessTask.TaskType.CONSUME_FILE:
        return None
    overrides = task_kwargs.get("overrides")
@@ -1239,29 +1238,6 @@ 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,
@@ -1319,7 +1295,6 @@ 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")

@@ -1385,7 +1360,6 @@ 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")

@@ -1429,7 +1403,6 @@ 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,6 +52,7 @@ 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
@@ -690,6 +691,47 @@ 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,7 +15,6 @@ 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):
@@ -27,67 +26,6 @@ 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,7 +7,6 @@ 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
@@ -15,7 +14,11 @@ 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
@@ -30,18 +33,9 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
        self.user = user
        self.client.force_authenticate(user=user)

-        patcher = mock.patch("documents.bulk_operations.bulk_update_documents")
+        patcher = mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
        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")
@@ -71,16 +65,11 @@ 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_operations.bulk_update_documents")
+    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
    def test_api_set_correspondent(self, bulk_update_task_mock) -> None:
        self.assertNotEqual(self.doc1.correspondent, self.c1)
        response = self.client.post(
@@ -99,11 +88,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.args[0],
+            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
            [self.doc1.pk],
        )

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

-    @mock.patch("documents.bulk_operations.bulk_update_documents")
+    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
    def test_api_set_type(self, bulk_update_task_mock) -> None:
        self.assertNotEqual(self.doc1.document_type, self.dt1)
        response = self.client.post(
@@ -144,11 +133,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.args[0],
+            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
            [self.doc1.pk],
        )

-    @mock.patch("documents.bulk_operations.bulk_update_documents")
+    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
    def test_api_unset_type(self, bulk_update_task_mock) -> None:
        self.doc1.document_type = self.dt1
        self.doc1.save()
@@ -169,11 +158,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.args[0],
+            bulk_update_task_mock.call_args.kwargs["kwargs"]["document_ids"],
            [self.doc1.pk],
        )

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

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

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

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

@@ -238,9 +227,6 @@ 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])
@@ -295,9 +281,6 @@ 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])
@@ -366,9 +349,6 @@ 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])
@@ -530,9 +510,6 @@ 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)

@@ -547,9 +524,6 @@ 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)

@@ -574,9 +548,6 @@ 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)

@@ -591,9 +562,6 @@ 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})

@@ -616,12 +584,88 @@ 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:
        """
@@ -648,9 +692,6 @@ 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)
@@ -681,9 +722,6 @@ 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)
@@ -806,9 +844,6 @@ 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)

@@ -832,9 +867,6 @@ 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)

@@ -871,9 +903,6 @@ 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)

@@ -910,11 +939,6 @@ 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)

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

@@ -1291,9 +1315,6 @@ 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)

@@ -1706,9 +1727,6 @@ 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")
@@ -1995,9 +2013,6 @@ 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")
@@ -2205,9 +2220,6 @@ 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")
@@ -2553,9 +2565,6 @@ 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,7 +49,6 @@ 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
@@ -1756,113 +1755,12 @@ 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",
@@ -1906,7 +1804,6 @@ 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(
@@ -1914,7 +1811,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
            checksum="A",
            mime_type="application/pdf",
            content="root",
-            archive_serial_number=42,
        )
        version = Document.objects.create(
            title="version",
@@ -1923,7 +1819,6 @@ 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)
@@ -1934,7 +1829,6 @@ 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,
@@ -1948,45 +1842,6 @@ 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,7 +25,6 @@ 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
@@ -346,39 +345,38 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
        storage_path = StoragePath.objects.create(name="sp1")
        tag = Tag.objects.create(name="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)
+        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)

-        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",
-            )
+        get_backend().add_or_update(matching_doc)
+
+        response = self.client.get(
+            "/api/documents/?query=bank&include_selection_data=true",
+        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
-        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}],
-            )
+        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)

    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.views.execute_bulk_operation.apply_async")
+    @mock.patch("documents.bulk_edit.bulk_update_documents.apply_async")
    def test_bulk_edit_query_at_exactly_the_cap_is_accepted(
        self,
-        enqueue_mock: mock.MagicMock,
+        bulk_update_task_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 queues the metadata operation.
-              Mock the executor dispatch so this exercises query validation
-              without requiring a broker.
+            - 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)
        """
        query = "a" * _MAX_QUERY_LENGTH
-        enqueue_mock.return_value.id = "query-cap-task"

        response = admin_client.post(
            "/api/documents/bulk_edit/",
@@ -183,11 +183,6 @@ 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,6 +100,30 @@ 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:
@@ -1,281 +0,0 @@
-"""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,6 +2288,56 @@ 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,14 +5,11 @@ 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
@@ -107,9 +104,6 @@ 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
@@ -146,6 +140,7 @@ 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
@@ -174,10 +169,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
@@ -1020,80 +1015,71 @@ class DocumentViewSet(
        "custom_field_",
    )

-    # Bound the IN clauses used for search results and through-table lookups.
-    selection_data_batch_size = 10_000
+    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))

-    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",
-            )
+        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,
        )
-        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": 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),
+            "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
+            ],
        }

    @classmethod
@@ -2740,10 +2726,13 @@ 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,
-                            search_ids=result.ordered_ids,
+                            filtered_qs.filter(pk__in=result.ordered_ids),
                        )
                    )
                return response
@@ -2987,29 +2976,99 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
        "merge_as_versions",
    }

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

-    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
-                ),
-            },
+        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
        )
-        return Response({"result": "OK", "task_id": task.id})
+
+        # 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

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

    serializer_class = BulkEditSerializer

-    _snapshot_field = staticmethod(snapshot_field)
+    @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

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

-        current_asn = documents.aggregate(
+        current_asn = Document.objects.aggregate(
            Max("archive_serial_number", default=0),
        ).get(
            "archive_serial_number__max",
@@ -5110,6 +5192,8 @@ 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-07 02:45+0000\n"
+"POT-Creation-Date: 2026-11-06 20:19+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:349 documents/views.py:615
-#: documents/views.py:2755 paperless_mail/serialisers.py:156
+#: documents/serialisers.py:2850 documents/views.py:348 documents/views.py:614
+#: documents/views.py:2748 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:4805
+#: documents/serialisers.py:2936 documents/views.py:4798
 #, python-format
 msgid "Documents not found: %(ids)s"
 msgstr ""
@@ -1953,40 +1953,40 @@ msgstr ""
 msgid ", "
 msgstr ""

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

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

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

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

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

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

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

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

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

@@ -45,21 +45,12 @@ 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())
@@ -93,9 +84,3 @@ 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", "documents.bulk_operations")
+CELERY_IMPORTS = ("documents.bulk_edit",)

 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, PaperlessModelPermissions)
+    permission_classes = (IsAuthenticated, DjangoModelPermissions)

    @extend_schema(exclude=True)
    def create(self, request, *args, **kwargs):
Author	SHA1	Message	Date
shamoon	62648fffe8	Fix migration	2026-11-06 14:15:05 -08:00
shamoon	b0c8ce25fd	docs	2026-11-06 14:14:29 -08:00
shamoon	3c4f5d8959	Add to doc details	2026-11-06 14:14:29 -08:00
shamoon	85c52cc27f	tests	2026-11-06 14:14:29 -08:00
shamoon	9b3d484e24	Bulk run workflow + dialog	2026-11-06 14:14:28 -08:00
shamoon	3a5ba814b6	frotnend use the manual filter	2026-11-06 14:14:28 -08:00
shamoon	26552f54b4	Ah, need a backend filter for wf type	2026-11-06 14:14:28 -08:00
shamoon	de9d7e54bd	WorkflowTriggerType.Manual	2026-11-06 14:14:27 -08:00
shamoon	a5c460a567	Add some backend tests for the manual wf	2026-11-06 14:14:27 -08:00
shamoon	cfca79e7fc	api bulk edit run_workflow	2026-11-06 14:14:27 -08:00
shamoon	9d59f3aaa2	run_workflows task= with filter_documents option	2026-11-06 14:14:26 -08:00
shamoon	332c308f84	Add skip_matching to run_workflows	2026-11-06 14:14:26 -08:00
shamoon	a0b8da7687	Add manual trigger	2026-11-06 14:14:23 -08:00