Fix: correct text/stream compression workaround

[skip ci]

docker/rootfs/etc/s6-overlay/s6-rc.d/init-compile-bytecode/run

@@ -0,0 +1,58 @@
+#!/command/with-contenv /usr/bin/bash
+# shellcheck shell=bash
+declare -r log_prefix="[init-compile-bytecode]"
+
+# PYTHONDONTWRITEBYTECODE=1 is set for the whole container. This unit compiles a
+# scoped set of libraries anyway, to speed up startup without bloating image size.
+
+# Handle the people using a read only file system
+if [[ "${S6_READ_ONLY_ROOT}" == "1" ]]; then
+	echo "${log_prefix} S6_READ_ONLY_ROOT=1, skipping (nothing to write bytecode to)"
+	exit 0
+fi
+
+# When running as a non-root user, site-packages is still root-owned and unwritable,
+# so this step would just fail loudly on every container start. Skip it.
+if [[ -n "${USER_IS_NON_ROOT}" ]]; then
+	echo "${log_prefix} USER_IS_NON_ROOT is set, skipping (site-packages is not writable)"
+	exit 0
+fi
+
+declare -r site_packages="$(python3 -c 'import site; print(site.getsitepackages()[0])')"
+
+# Deliberately scoped to packages that paperless.settings/paperless/__init__.py import
+# unconditionally on every manage.py invocation (Django itself, the always-loaded
+# INSTALLED_APPS, and celery). This is NOT "compile everything" - the optional AI stack
+# (torch, llama-index, sentence-transformers, ...) is intentionally excluded since it is
+# lazy-imported and large.
+declare -a scope=(
+	"${PAPERLESS_SRC_DIR}"
+	"${site_packages}/django"
+	"${site_packages}/celery"
+	"${site_packages}/kombu"
+	"${site_packages}/rest_framework"
+	"${site_packages}/django_filters"
+	"${site_packages}/whitenoise"
+	"${site_packages}/corsheaders"
+	"${site_packages}/django_extensions"
+	"${site_packages}/guardian"
+	"${site_packages}/allauth"
+	"${site_packages}/drf_spectacular"
+	"${site_packages}/drf_spectacular_sidecar"
+	"${site_packages}/treenode"
+	"${site_packages}/compression_middleware"
+)
+
+declare -a existing_scope=()
+for path in "${scope[@]}"; do
+	[[ -d "${path}" ]] && existing_scope+=("${path}")
+done
+
+echo "${log_prefix} Compiling bytecode for: ${existing_scope[*]}"
+declare -r start_seconds=${SECONDS}
+
+if ! PYTHONDONTWRITEBYTECODE= python3 -m compileall -q "${existing_scope[@]}"; then
+	echo "${log_prefix} WARNING: compileall reported errors (read-only filesystem or unwritable site-packages?); continuing without a bytecode cache"
+fi
+
+echo "${log_prefix} Done in $((SECONDS - start_seconds))s"

docker/rootfs/etc/s6-overlay/s6-rc.d/init-compile-bytecode/type

@@ -0,0 +1 @@
+oneshot

docker/rootfs/etc/s6-overlay/s6-rc.d/init-compile-bytecode/up

@@ -0,0 +1 @@
+/etc/s6-overlay/s6-rc.d/init-compile-bytecode/run

docs/configuration.md

@@ -1200,6 +1200,15 @@ still perform some basic text pre-processing before matching.
 
     Defaults to true, enabling the feature.
 
+#### [`PAPERLESS_CLASSIFIER_MATCH_THRESHOLD=<float>`](#PAPERLESS_CLASSIFIER_MATCH_THRESHOLD) {#PAPERLESS_CLASSIFIER_MATCH_THRESHOLD}
+
+: Sets the minimum confidence score (0.0-1.0) required for the automatic
+classifier to assign a correspondent, document type, or storage path to a
+document. Predictions below this threshold are discarded and the field is
+left unassigned, preventing low-confidence guesses from being applied.
+
+    Defaults to 0.6.
+
 #### [`PAPERLESS_DATE_PARSER_LANGUAGES=<lang>`](#PAPERLESS_DATE_PARSER_LANGUAGES) {#PAPERLESS_DATE_PARSER_LANGUAGES}
 
 : Specifies which language Paperless should use when parsing dates from documents.

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

@@ -112,6 +112,22 @@
 
                 <pngx-input-check i18n-title title="Use 'slim' sidebar (icons only)" formControlName="slimSidebarEnabled"></pngx-input-check>
 
+                <p class="mb-2 mt-3" i18n>Sidebar items to show:</p>
+                @for (option of sidebarItemOptions; track option.id) {
+                  <div class="form-check">
+                    <input
+                      class="form-check-input"
+                      type="checkbox"
+                      [id]="'sidebar-item-setting-' + option.id"
+                      [checked]="isSidebarItemShown(option.id)"
+                      (change)="toggleSidebarItem(option.id, $event.target.checked)"
+                    />
+                    <label class="form-check-label" [for]="'sidebar-item-setting-' + option.id">
+                      {{ option.label }}
+                    </label>
+                  </div>
+                }
+
               </div>
             </div>
 

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

@@ -24,7 +24,7 @@ import {
   SystemStatus,
   SystemStatusItemStatus,
 } from 'src/app/data/system-status'
-import { SETTINGS_KEYS } from 'src/app/data/ui-settings'
+import { HideableSidebarItemID, SETTINGS_KEYS } from 'src/app/data/ui-settings'
 import { IfOwnerDirective } from 'src/app/directives/if-owner.directive'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
 import { PermissionsGuard } from 'src/app/guards/permissions.guard'
@@ -209,6 +209,45 @@ describe('SettingsComponent', () => {
     fixture.detectChanges()
   }
 
+  it('supports configuring sidebar items and canceling changes', () => {
+    completeSetup()
+
+    component.toggleSidebarItem(HideableSidebarItemID.Workflows, false)
+    fixture.detectChanges()
+
+    expect(component.settingsForm.value.sidebarHiddenItems).toContain(
+      HideableSidebarItemID.Workflows
+    )
+
+    settingsService.updateSidebarItemVisibility(
+      HideableSidebarItemID.Mail,
+      false
+    )
+
+    expect(component.settingsForm.value.sidebarHiddenItems).toContain(
+      HideableSidebarItemID.Mail
+    )
+
+    component.reset()
+
+    expect(component.settingsForm.value.sidebarHiddenItems).not.toContain(
+      HideableSidebarItemID.Workflows
+    )
+    expect(component.settingsForm.value.sidebarHiddenItems).not.toContain(
+      HideableSidebarItemID.Mail
+    )
+  })
+
+  it('enables sidebar item controls on general settings until destroyed', () => {
+    completeSetup()
+
+    expect(settingsService.organizingSidebarItems()).toBe(true)
+
+    component.ngOnDestroy()
+
+    expect(settingsService.organizingSidebarItems()).toBe(false)
+  })
+
   it('should support tabbed settings & change URL, prevent navigation if dirty confirmation rejected', async () => {
     completeSetup()
     const navigateSpy = jest.spyOn(router, 'navigate')
@@ -249,6 +288,7 @@ describe('SettingsComponent', () => {
 
   it('should support save local settings updating appearance settings and calling API, show error', () => {
     completeSetup()
+    component.toggleSidebarItem(HideableSidebarItemID.Workflows, false)
     const toastErrorSpy = jest.spyOn(toastService, 'showError')
     const toastSpy = jest.spyOn(toastService, 'show')
     const storeSpy = jest.spyOn(settingsService, 'storeSettings')
@@ -267,7 +307,10 @@ describe('SettingsComponent', () => {
     expect(toastErrorSpy).toHaveBeenCalled()
     expect(storeSpy).toHaveBeenCalled()
     expect(appearanceSettingsSpy).not.toHaveBeenCalled()
-    expect(setSpy).toHaveBeenCalledTimes(33)
+    expect(setSpy).toHaveBeenCalledTimes(34)
+    expect(setSpy).toHaveBeenCalledWith(SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS, [
+      HideableSidebarItemID.Workflows,
+    ])
 
     // succeed
     storeSpy.mockReturnValueOnce(of(true))

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

@@ -39,7 +39,12 @@ import {
   SystemStatus,
   SystemStatusItemStatus,
 } from 'src/app/data/system-status'
-import { GlobalSearchType, SETTINGS_KEYS } from 'src/app/data/ui-settings'
+import {
+  GlobalSearchType,
+  HIDEABLE_SIDEBAR_ITEM_IDS,
+  HideableSidebarItemID,
+  SETTINGS_KEYS,
+} from 'src/app/data/ui-settings'
 import { User } from 'src/app/data/user'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
 import { CustomDatePipe } from 'src/app/pipes/custom-date.pipe'
@@ -102,6 +107,14 @@ const documentDetailFieldOptions = [
   { id: DocumentDetailFieldID.Tags, label: $localize`Tags` },
 ]
 
+const sidebarItemLabels: Record<HideableSidebarItemID, string> = {
+  [HideableSidebarItemID.Dashboard]: $localize`Dashboard`,
+  [HideableSidebarItemID.SavedViews]: $localize`Saved Views`,
+  [HideableSidebarItemID.Workflows]: $localize`Workflows`,
+  [HideableSidebarItemID.Mail]: $localize`Mail`,
+  [HideableSidebarItemID.Documentation]: $localize`Documentation`,
+}
+
 @Component({
   selector: 'pngx-settings',
   templateUrl: './settings.component.html',
@@ -149,6 +162,7 @@ export class SettingsComponent
     bulkEditApplyOnClose: new FormControl(null),
     documentListItemPerPage: new FormControl(null),
     slimSidebarEnabled: new FormControl(null),
+    sidebarHiddenItems: new FormControl<HideableSidebarItemID[]>([]),
     darkModeUseSystem: new FormControl(null),
     darkModeEnabled: new FormControl(null),
     darkModeInvertThumbs: new FormControl(null),
@@ -186,6 +200,7 @@ export class SettingsComponent
 
   store: BehaviorSubject<any>
   storeSub: Subscription
+  sidebarItemsSub: Subscription
   isDirty$: Observable<boolean>
   isDirty: boolean = false
   unsubscribeNotifier: Subject<any> = new Subject()
@@ -203,6 +218,10 @@ export class SettingsComponent
   public readonly PdfEditorEditMode = PdfEditorEditMode
 
   public readonly documentDetailFieldOptions = documentDetailFieldOptions
+  public readonly sidebarItemOptions = HIDEABLE_SIDEBAR_ITEM_IDS.map((id) => ({
+    id,
+    label: sidebarItemLabels[id],
+  }))
 
   get systemStatusHasErrors(): boolean {
     const status = this.systemStatus()
@@ -230,6 +249,10 @@ export class SettingsComponent
 
   constructor() {
     super()
+    this.sidebarItemsSub =
+      this.settings.sidebarHiddenItemsEditingChanged.subscribe((hiddenItems) =>
+        this.settingsForm.controls.sidebarHiddenItems.setValue(hiddenItems)
+      )
     this.settings.settingsSaved.subscribe(() => {
       if (!this.savePending) this.initialize()
       this.savedViewsService.maybeRefreshDocumentCounts()
@@ -279,14 +302,21 @@ export class SettingsComponent
 
     this.activatedRoute.paramMap.subscribe((paramMap) => {
       const section = paramMap.get('section')
+      let navID = SettingsNavIDs.General
       if (section) {
         const navIDKey: string = Object.keys(SettingsNavIDs).find(
           (navID) => navID.toLowerCase() == section
         )
         if (navIDKey) {
-          this.activeNavID.set(SettingsNavIDs[navIDKey])
+          navID = SettingsNavIDs[navIDKey]
         }
       }
+      this.activeNavID.set(navID)
+      this.settings.sidebarHiddenItemsEditing.set(
+        navID === SettingsNavIDs.General
+          ? [...this.settingsForm.controls.sidebarHiddenItems.value]
+          : null
+      )
     })
   }
 
@@ -310,6 +340,7 @@ export class SettingsComponent
         SETTINGS_KEYS.DOCUMENT_LIST_SIZE
       ),
       slimSidebarEnabled: this.settings.get(SETTINGS_KEYS.SLIM_SIDEBAR),
+      sidebarHiddenItems: this.settings.get(SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS),
       darkModeUseSystem: this.settings.get(SETTINGS_KEYS.DARK_MODE_USE_SYSTEM),
       darkModeEnabled: this.settings.get(SETTINGS_KEYS.DARK_MODE_ENABLED),
       darkModeInvertThumbs: this.settings.get(
@@ -436,6 +467,12 @@ export class SettingsComponent
       this.settingsForm.patchValue(currentFormValue)
     }
 
+    if (this.settings.organizingSidebarItems()) {
+      this.settings.sidebarHiddenItemsEditing.set([
+        ...this.settingsForm.controls.sidebarHiddenItems.value,
+      ])
+    }
+
     if (this.canViewSystemStatus) {
       this.systemStatusService.get().subscribe((status) => {
         this.systemStatus.set(status)
@@ -444,8 +481,18 @@ export class SettingsComponent
   }
 
   ngOnDestroy() {
+    this.settings.sidebarHiddenItemsEditing.set(null)
     if (this.isDirty) this.settings.updateAppearanceSettings() // in case user changed appearance but didn't save
     this.storeSub && this.storeSub.unsubscribe()
+    this.sidebarItemsSub.unsubscribe()
+  }
+
+  isSidebarItemShown(item: HideableSidebarItemID): boolean {
+    return !(this.settingsForm.value.sidebarHiddenItems || []).includes(item)
+  }
+
+  toggleSidebarItem(item: HideableSidebarItemID, checked: boolean): void {
+    this.settings.updateSidebarItemVisibility(item, checked)
   }
 
   public saveSettings() {
@@ -473,6 +520,10 @@ export class SettingsComponent
       SETTINGS_KEYS.SLIM_SIDEBAR,
       this.settingsForm.value.slimSidebarEnabled
     )
+    this.settings.set(
+      SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS,
+      this.settingsForm.value.sidebarHiddenItems
+    )
     this.settings.set(
       SETTINGS_KEYS.DARK_MODE_USE_SYSTEM,
       this.settingsForm.value.darkModeUseSystem
@@ -632,6 +683,11 @@ export class SettingsComponent
 
   reset() {
     this.settingsForm.patchValue(this.store.getValue())
+    if (this.settings.organizingSidebarItems()) {
+      this.settings.sidebarHiddenItemsEditing.set([
+        ...this.settingsForm.controls.sidebarHiddenItems.value,
+      ])
+    }
   }
 
   clearThemeColor() {

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

@@ -86,12 +86,15 @@
       }
       <div class="sidebar-sticky pt-3 pb-1 d-flex flex-column justify-space-around">
         <ul class="nav flex-column">
-          <li class="nav-item app-link">
-            <a class="nav-link" routerLink="dashboard" routerLinkActive="active" (click)="closeMenu()"
+          <li class="nav-item app-link position-relative" [class.d-none]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Dashboard) && !settingsService.organizingSidebarItems()">
+            <a class="nav-link" [class.opacity-50]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Dashboard)" [class.pe-5]="settingsService.organizingSidebarItems() && !slimSidebarEnabled && !slimSidebarAnimating()" routerLink="dashboard" routerLinkActive="active" (click)="closeMenu()"
               ngbPopover="Dashboard" i18n-ngbPopover [disablePopover]="!slimSidebarPopoversEnabled" placement="end"
               container="body" triggers="mouseenter:mouseleave" popoverClass="popover-slim">
               <i-bs class="me-2" name="house"></i-bs><span class="nav-link-label"><ng-container i18n>Dashboard</ng-container></span>
             </a>
+            @if (settingsService.organizingSidebarItems()) {
+              <pngx-input-switch class="position-absolute top-50 end-0 translate-middle-y me-1" [class.d-none]="slimSidebarEnabled || slimSidebarAnimating()" [compact]="true" title="Dashboard" i18n-title [ngModel]="!settingsService.sidebarItemIsHidden(HideableSidebarItemID.Dashboard)" (ngModelChange)="toggleSidebarItem(HideableSidebarItemID.Dashboard, $event)"></pngx-input-switch>
+            }
           </li>
           <li class="nav-item app-link" *pngxIfPermissions="{ action: PermissionAction.View, type: PermissionType.Document }">
             <a class="nav-link" routerLink="documents" routerLinkActive="active"
@@ -237,29 +240,38 @@
                 </div>
               </li>
             }
-            <li class="nav-item app-link" *pngxIfPermissions="{ action: PermissionAction.View, type: PermissionType.SavedView }">
-              <a class="nav-link" routerLink="savedviews" routerLinkActive="active" (click)="closeMenu()"
+            <li class="nav-item app-link position-relative" [class.d-none]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.SavedViews) && !settingsService.organizingSidebarItems()" *pngxIfPermissions="{ action: PermissionAction.View, type: PermissionType.SavedView }">
+              <a class="nav-link" [class.opacity-50]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.SavedViews)" [class.pe-5]="settingsService.organizingSidebarItems() && !slimSidebarEnabled && !slimSidebarAnimating()" routerLink="savedviews" routerLinkActive="active" (click)="closeMenu()"
                 ngbPopover="Saved Views" i18n-ngbPopover [disablePopover]="!slimSidebarPopoversEnabled" placement="end"
                 container="body" triggers="mouseenter:mouseleave" popoverClass="popover-slim">
                 <i-bs class="me-2" name="window-stack"></i-bs><span class="nav-link-label"><ng-container i18n>Saved Views</ng-container></span>
               </a>
+              @if (settingsService.organizingSidebarItems()) {
+                <pngx-input-switch class="position-absolute top-50 end-0 translate-middle-y me-1" [class.d-none]="slimSidebarEnabled || slimSidebarAnimating()" [compact]="true" title="Saved Views" i18n-title [ngModel]="!settingsService.sidebarItemIsHidden(HideableSidebarItemID.SavedViews)" (ngModelChange)="toggleSidebarItem(HideableSidebarItemID.SavedViews, $event)"></pngx-input-switch>
+              }
             </li>
-            <li class="nav-item app-link"
+            <li class="nav-item app-link position-relative" [class.d-none]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Workflows) && !settingsService.organizingSidebarItems()"
               *pngxIfPermissions="{ action: PermissionAction.View, type: PermissionType.Workflow }"
               tourAnchor="tour.workflows">
-              <a class="nav-link" routerLink="workflows" routerLinkActive="active" (click)="closeMenu()"
+              <a class="nav-link" [class.opacity-50]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Workflows)" [class.pe-5]="settingsService.organizingSidebarItems() && !slimSidebarEnabled && !slimSidebarAnimating()" routerLink="workflows" routerLinkActive="active" (click)="closeMenu()"
                 ngbPopover="Workflows" i18n-ngbPopover [disablePopover]="!slimSidebarPopoversEnabled" placement="end"
                 container="body" triggers="mouseenter:mouseleave" popoverClass="popover-slim">
                 <i-bs class="me-2" name="boxes"></i-bs><span class="nav-link-label"><ng-container i18n>Workflows</ng-container></span>
               </a>
+              @if (settingsService.organizingSidebarItems()) {
+                <pngx-input-switch class="position-absolute top-50 end-0 translate-middle-y me-1" [class.d-none]="slimSidebarEnabled || slimSidebarAnimating()" [compact]="true" title="Workflows" i18n-title [ngModel]="!settingsService.sidebarItemIsHidden(HideableSidebarItemID.Workflows)" (ngModelChange)="toggleSidebarItem(HideableSidebarItemID.Workflows, $event)"></pngx-input-switch>
+              }
             </li>
-            <li class="nav-item app-link" *pngxIfPermissions="{ action: PermissionAction.View, type: PermissionType.MailAccount }"
+            <li class="nav-item app-link position-relative" [class.d-none]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Mail) && !settingsService.organizingSidebarItems()" *pngxIfPermissions="{ action: PermissionAction.View, type: PermissionType.MailAccount }"
               tourAnchor="tour.mail">
-              <a class="nav-link" routerLink="mail" routerLinkActive="active" (click)="closeMenu()" ngbPopover="Mail"
+              <a class="nav-link" [class.opacity-50]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Mail)" [class.pe-5]="settingsService.organizingSidebarItems() && !slimSidebarEnabled && !slimSidebarAnimating()" routerLink="mail" routerLinkActive="active" (click)="closeMenu()" ngbPopover="Mail"
                 i18n-ngbPopover [disablePopover]="!slimSidebarPopoversEnabled" placement="end" container="body"
                 triggers="mouseenter:mouseleave" popoverClass="popover-slim">
                 <i-bs class="me-2" name="envelope"></i-bs><span class="nav-link-label"><ng-container i18n>Mail</ng-container></span>
               </a>
+              @if (settingsService.organizingSidebarItems()) {
+                <pngx-input-switch class="position-absolute top-50 end-0 translate-middle-y me-1" [class.d-none]="slimSidebarEnabled || slimSidebarAnimating()" [compact]="true" title="Mail" i18n-title [ngModel]="!settingsService.sidebarItemIsHidden(HideableSidebarItemID.Mail)" (ngModelChange)="toggleSidebarItem(HideableSidebarItemID.Mail, $event)"></pngx-input-switch>
+              }
             </li>
             <li class="nav-item app-link" *pngxIfPermissions="{ action: PermissionAction.Delete, type: PermissionType.Document }">
               <a class="nav-link" routerLink="trash" routerLinkActive="active" (click)="closeMenu()" ngbPopover="Trash"
@@ -322,13 +334,16 @@
                 </a>
               </li>
             }
-            <li class="nav-item mt-2" tourAnchor="tour.outro">
-              <a class="text-muted small d-flex align-items-center flex-wrap text-decoration-none nav-anchor"
+            <li class="nav-item mt-2 position-relative" [class.d-none]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Documentation) && !settingsService.organizingSidebarItems()" tourAnchor="tour.outro">
+              <a class="text-muted small d-flex align-items-center flex-wrap text-decoration-none nav-anchor" [class.opacity-50]="settingsService.sidebarItemIsHidden(HideableSidebarItemID.Documentation)" [class.pe-5]="settingsService.organizingSidebarItems() && !slimSidebarEnabled && !slimSidebarAnimating()"
                 target="_blank" rel="noopener noreferrer" href="https://docs.paperless-ngx.com" ngbPopover="Documentation"
                 i18n-ngbPopover [disablePopover]="!slimSidebarPopoversEnabled" placement="end" container="body"
                 triggers="mouseenter:mouseleave" popoverClass="popover-slim">
                 <i-bs class="d-flex me-2" name="question-circle"></i-bs><span><ng-container i18n>Documentation</ng-container></span>
               </a>
+              @if (settingsService.organizingSidebarItems()) {
+                <pngx-input-switch class="position-absolute top-50 end-0 translate-middle-y me-1" [class.d-none]="slimSidebarEnabled || slimSidebarAnimating()" [compact]="true" title="Documentation" i18n-title [ngModel]="!settingsService.sidebarItemIsHidden(HideableSidebarItemID.Documentation)" (ngModelChange)="toggleSidebarItem(HideableSidebarItemID.Documentation, $event)"></pngx-input-switch>
+              }
             </li>
             <li class="nav-item" [class.visually-hidden]="slimSidebarEnabled">
               <div class="text-muted small d-flex align-items-center flex-wrap nav-label">

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

@@ -15,7 +15,7 @@ import { provideUiTour } from 'ngx-ui-tour-ng-bootstrap'
 import { of, throwError } from 'rxjs'
 import { routes } from 'src/app/app-routing.module'
 import { SavedView } from 'src/app/data/saved-view'
-import { SETTINGS_KEYS } from 'src/app/data/ui-settings'
+import { HideableSidebarItemID, SETTINGS_KEYS } from 'src/app/data/ui-settings'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
 import { PermissionsGuard } from 'src/app/guards/permissions.guard'
 import {
@@ -287,6 +287,82 @@ describe('AppFrameComponent', () => {
     jest.useRealTimers()
   })
 
+  it('should hide configured sidebar items', () => {
+    settingsService.set(SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS, [
+      HideableSidebarItemID.Dashboard,
+      HideableSidebarItemID.Workflows,
+    ])
+    fixture.detectChanges()
+
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="dashboard"]')
+        .parentElement.classList
+    ).toContain('d-none')
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="workflows"]')
+        .parentElement.classList
+    ).toContain('d-none')
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="mail"]').parentElement
+        .classList
+    ).not.toContain('d-none')
+  })
+
+  it('should show hidden items and visibility switches while customizing', () => {
+    settingsService.set(SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS, [
+      HideableSidebarItemID.Dashboard,
+    ])
+    settingsService.sidebarHiddenItemsEditing.set([
+      HideableSidebarItemID.Dashboard,
+    ])
+    fixture.detectChanges()
+
+    expect(
+      fixture.nativeElement.querySelectorAll('pngx-input-switch').length
+    ).toBe(5)
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="dashboard"]')
+        .parentElement.classList
+    ).not.toContain('d-none')
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="dashboard"]').classList
+    ).toContain('opacity-50')
+
+    settingsService.set(SETTINGS_KEYS.SLIM_SIDEBAR, true)
+    fixture.detectChanges()
+
+    expect(
+      Array.from(
+        fixture.nativeElement.querySelectorAll('pngx-input-switch')
+      ).every((toggle: HTMLElement) => toggle.classList.contains('d-none'))
+    ).toBe(true)
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="dashboard"]').classList
+    ).not.toContain('pe-5')
+
+    settingsService.set(SETTINGS_KEYS.SLIM_SIDEBAR, false)
+    component.slimSidebarAnimating.set(true)
+    fixture.detectChanges()
+
+    expect(
+      Array.from(
+        fixture.nativeElement.querySelectorAll('pngx-input-switch')
+      ).every((toggle: HTMLElement) => toggle.classList.contains('d-none'))
+    ).toBe(true)
+
+    component.slimSidebarAnimating.set(false)
+    fixture.detectChanges()
+
+    expect(
+      Array.from(
+        fixture.nativeElement.querySelectorAll('pngx-input-switch')
+      ).every((toggle: HTMLElement) => !toggle.classList.contains('d-none'))
+    ).toBe(true)
+    expect(
+      fixture.nativeElement.querySelector('[routerLink="dashboard"]').classList
+    ).toContain('pe-5')
+  })
+
   it('should show error on toggle slim sidebar if store settings fails', () => {
     jest.spyOn(console, 'warn').mockImplementation(() => {})
     const toastSpy = jest.spyOn(toastService, 'showError')

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

@@ -7,6 +7,7 @@ import {
 } from '@angular/cdk/drag-drop'
 import { NgClass } from '@angular/common'
 import { Component, HostListener, inject, OnInit, signal } from '@angular/core'
+import { FormsModule } from '@angular/forms'
 import { ActivatedRoute, Router, RouterModule } from '@angular/router'
 import {
   NgbCollapseModule,
@@ -21,7 +22,11 @@ import { Observable } from 'rxjs'
 import { first } from 'rxjs/operators'
 import { Document } from 'src/app/data/document'
 import { SavedView } from 'src/app/data/saved-view'
-import { CollapsibleSection, SETTINGS_KEYS } from 'src/app/data/ui-settings'
+import {
+  CollapsibleSection,
+  HideableSidebarItemID,
+  SETTINGS_KEYS,
+} from 'src/app/data/ui-settings'
 import { IfPermissionsDirective } from 'src/app/directives/if-permissions.directive'
 import { ComponentCanDeactivate } from 'src/app/guards/dirty-doc.guard'
 import { DocumentTitlePipe } from 'src/app/pipes/document-title.pipe'
@@ -48,6 +53,7 @@ import { ChatComponent } from '../chat/chat/chat.component'
 import { BrandMarkComponent } from '../common/logo/brand-mark/brand-mark.component'
 import { LogoComponent } from '../common/logo/logo.component'
 import { ProfileEditDialogComponent } from '../common/profile-edit-dialog/profile-edit-dialog.component'
+import { SwitchComponent } from '../common/input/switch/switch.component'
 import { DocumentDetailComponent } from '../document-detail/document-detail.component'
 import { ComponentWithPermissions } from '../with-permissions/with-permissions.component'
 import { GlobalSearchComponent } from './global-search/global-search.component'
@@ -76,6 +82,8 @@ const SCROLL_THRESHOLD = 16
     NgxBootstrapIconsModule,
     DragDropModule,
     TourNgBootstrap,
+    FormsModule,
+    SwitchComponent,
   ],
 })
 export class AppFrameComponent
@@ -98,6 +106,7 @@ export class AppFrameComponent
   readonly isMenuCollapsed = signal(true)
   readonly slimSidebarAnimating = signal(false)
   readonly mobileSearchHidden = signal(false)
+  readonly HideableSidebarItemID = HideableSidebarItemID
   private readonly versionSetting = this.settingsService.getSignal<string>(
     SETTINGS_KEYS.VERSION
   )
@@ -195,6 +204,10 @@ export class AppFrameComponent
     }, 200) // slightly longer than css animation for slim sidebar
   }
 
+  toggleSidebarItem(item: HideableSidebarItemID, visible: boolean): void {
+    this.settingsService.updateSidebarItemVisibility(item, visible)
+  }
+
   toggleAttributesSections(event?: Event): void {
     event?.preventDefault()
     event?.stopPropagation()

src-ui/src/app/components/common/input/switch/switch.component.html

@@ -1,6 +1,6 @@
-<div class="mb-3">
-  <div class="row">
-    @if (!horizontal) {
+<div [class.mb-3]="!compact">
+  <div [class.row]="!compact">
+    @if (!horizontal && !compact) {
       <div class="d-flex align-items-center position-relative hidden-button-container col-md-3">
         <label class="form-label" [for]="inputId" [ngbTooltip]="showUnsetNote && isUnset ? tipContent: null" placement="end">
           {{title}}
@@ -17,8 +17,8 @@
     }
     <div [ngClass]="{'align-items-center': horizontal, 'd-flex': horizontal}">
       <div class="form-check form-switch">
-        <input #inputField type="checkbox" class="form-check-input" [id]="inputId" [(ngModel)]="value" [ngModelOptions]="{standalone: true}" (change)="onChange(value)" (blur)="onTouched()" [disabled]="disabled">
-        @if (horizontal) {
+        <input #inputField type="checkbox" class="form-check-input" [id]="inputId" [(ngModel)]="value" [ngModelOptions]="{standalone: true}" (change)="onChange(value)" (blur)="onTouched()" [disabled]="disabled" [attr.aria-label]="compact ? title : null">
+        @if (horizontal && !compact) {
           <label class="form-check-label" [class.text-muted]="showUnsetNote && isUnset" [for]="inputId" [ngbTooltip]="showUnsetNote && isUnset ? tipContent: null" placement="end">
             {{title}}
             @if (showUnsetNote && isUnset) {

src-ui/src/app/components/common/input/switch/switch.component.spec.ts

@@ -48,4 +48,14 @@ describe('SwitchComponent', () => {
     component.value = undefined
     expect(component.isUnset).toBeTruthy()
   })
+
+  it('should support a compact layout', () => {
+    component.compact = true
+    component.title = 'Test switch'
+    fixture.detectChanges()
+
+    expect(fixture.nativeElement.querySelector('.mb-3')).toBeNull()
+    expect(fixture.nativeElement.querySelector('.row')).toBeNull()
+    expect(input.getAttribute('aria-label')).toEqual('Test switch')
+  })
 })

src-ui/src/app/components/common/input/switch/switch.component.ts

@@ -25,6 +25,9 @@ export class SwitchComponent extends AbstractInputComponent<boolean> {
   @Input()
   showUnsetNote: boolean = false
 
+  @Input()
+  compact: boolean = false
+
   constructor() {
     super()
   }

src-ui/src/app/data/ui-settings.ts

@@ -24,6 +24,16 @@ export enum CollapsibleSection {
   ATTRIBUTES = 'attributes',
 }
 
+export enum HideableSidebarItemID {
+  Dashboard = 'dashboard',
+  SavedViews = 'saved_views',
+  Workflows = 'workflows',
+  Mail = 'mail',
+  Documentation = 'documentation',
+}
+
+export const HIDEABLE_SIDEBAR_ITEM_IDS = Object.values(HideableSidebarItemID)
+
 export const PAPERLESS_GREEN_HEX = '#17541f'
 
 export const SETTINGS_KEYS = {
@@ -56,6 +66,7 @@ export const SETTINGS_KEYS = {
   NOTES_ENABLED: 'general-settings:notes-enabled',
   AUDITLOG_ENABLED: 'general-settings:auditlog-enabled',
   SLIM_SIDEBAR: 'general-settings:slim-sidebar',
+  SIDEBAR_HIDDEN_ITEMS: 'general-settings:sidebar:hidden-items',
   ATTRIBUTES_SECTIONS_COLLAPSED:
     'general-settings:attributes-sections-collapsed',
   UPDATE_CHECKING_ENABLED: 'general-settings:update-checking:enabled',
@@ -127,6 +138,11 @@ export const SETTINGS: UiSetting[] = [
     type: 'boolean',
     default: false,
   },
+  {
+    key: SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS,
+    type: 'array',
+    default: [],
+  },
   {
     key: SETTINGS_KEYS.ATTRIBUTES_SECTIONS_COLLAPSED,
     type: 'array',

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

@@ -14,7 +14,11 @@ import { CustomFieldDataType } from '../data/custom-field'
 import { DEFAULT_DISPLAY_FIELDS, DisplayField } from '../data/document'
 import { SavedView } from '../data/saved-view'
 import { RemoteOCRModeConfig } from '../data/paperless-config'
-import { SETTINGS_KEYS, UiSettings } from '../data/ui-settings'
+import {
+  HideableSidebarItemID,
+  SETTINGS_KEYS,
+  UiSettings,
+} from '../data/ui-settings'
 import { PermissionsService } from './permissions.service'
 import { CustomFieldsService } from './rest/custom-fields.service'
 import { SettingsService } from './settings.service'
@@ -230,6 +234,35 @@ describe('SettingsService', () => {
     expect(notesEnabled()).toBeFalsy()
   })
 
+  it('updates sidebar item visibility', () => {
+    httpTestingController
+      .expectOne(`${environment.apiBaseUrl}ui_settings/`)
+      .flush(ui_settings)
+
+    expect(
+      settingsService.sidebarItemIsHidden(HideableSidebarItemID.Workflows)
+    ).toBe(false)
+
+    settingsService.updateSidebarItemVisibility(
+      HideableSidebarItemID.Workflows,
+      false
+    )
+
+    expect(
+      settingsService.sidebarItemIsHidden(HideableSidebarItemID.Workflows)
+    ).toBe(true)
+    expect(settingsService.get(SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS)).toEqual([])
+
+    settingsService.updateSidebarItemVisibility(
+      HideableSidebarItemID.Workflows,
+      true
+    )
+
+    expect(
+      settingsService.sidebarItemIsHidden(HideableSidebarItemID.Workflows)
+    ).toBe(false)
+  })
+
   it('updates setting signals when settings are reinitialized', () => {
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}ui_settings/`

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

@@ -24,6 +24,7 @@ import { DEFAULT_DISPLAY_FIELDS, DisplayField } from '../data/document'
 import { RemoteOCRModeConfig } from '../data/paperless-config'
 import { SavedView } from '../data/saved-view'
 import {
+  HideableSidebarItemID,
   PAPERLESS_GREEN_HEX,
   SETTINGS,
   SETTINGS_KEYS,
@@ -313,6 +314,18 @@ export class SettingsService {
   readonly globalDropzoneEnabled = signal(true)
   readonly globalDropzoneActive = signal(false)
   readonly organizingSidebarSavedViews = signal(false)
+  readonly sidebarHiddenItemsEditing = signal<HideableSidebarItemID[] | null>(
+    null
+  )
+  readonly organizingSidebarItems = computed(
+    () => this.sidebarHiddenItemsEditing() !== null
+  )
+  readonly sidebarHiddenItemsEditingChanged = new EventEmitter<
+    HideableSidebarItemID[]
+  >()
+  readonly hiddenSidebarItems = this.getSignal<HideableSidebarItemID[]>(
+    SETTINGS_KEYS.SIDEBAR_HIDDEN_ITEMS
+  )
 
   readonly allDisplayFields = signal<Array<{ id: DisplayField; name: string }>>(
     DEFAULT_DISPLAY_FIELDS
@@ -749,6 +762,29 @@ export class SettingsService {
     return this.storeSettings()
   }
 
+  sidebarItemIsHidden(item: HideableSidebarItemID): boolean {
+    return (
+      this.sidebarHiddenItemsEditing() ?? this.hiddenSidebarItems()
+    ).includes(item)
+  }
+
+  updateSidebarItemVisibility(
+    item: HideableSidebarItemID,
+    visible: boolean
+  ): void {
+    const hiddenItems = new Set(
+      this.sidebarHiddenItemsEditing() ?? this.hiddenSidebarItems()
+    )
+    if (visible) {
+      hiddenItems.delete(item)
+    } else {
+      hiddenItems.add(item)
+    }
+    const updatedHiddenItems = [...hiddenItems]
+    this.sidebarHiddenItemsEditing.set(updatedHiddenItems)
+    this.sidebarHiddenItemsEditingChanged.emit(updatedHiddenItems)
+  }
+
   updateSavedViewsVisibility(
     dashboardVisibleViewIds: number[],
     sidebarVisibleViewIds: number[]

src/documents/apps.py

@@ -28,6 +28,7 @@ class DocumentsConfig(AppConfig):
         document_consumption_finished.connect(set_storage_path)
         document_consumption_finished.connect(add_to_index)
         document_consumption_finished.connect(run_workflows_added)
+        document_consumption_finished.connect(add_to_index)
         document_consumption_finished.connect(add_or_update_document_in_llm_index)
         document_updated.connect(run_workflows_updated)
         document_updated.connect(send_websocket_document_updated)

src/documents/bulk_edit.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import logging
 import tempfile
+import uuid
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Literal
@@ -27,7 +28,7 @@ from documents.models import DocumentType
 from documents.models import PaperlessTask
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import set_permissions_for_object
+from documents.permissions import set_permissions_for_objects
 from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
@@ -298,53 +299,55 @@ def modify_custom_fields(
 ) -> Literal["OK"]:
     qs = Document.objects.filter(id__in=doc_ids).only("pk")
     affected_docs = list(qs.values_list("pk", flat=True))
-    # Ensure add_custom_fields is a list of tuples, supports old API
+    # Ensure add_custom_fields is a list of (int, value) tuples, supports old API
     add_custom_fields = (
-        add_custom_fields.items()
+        [(int(field), value) for field, value in add_custom_fields.items()]
         if isinstance(add_custom_fields, dict)
-        else [(field, None) for field in add_custom_fields]
+        else [(int(field), None) for field in add_custom_fields]
     )
 
-    custom_fields = CustomField.objects.filter(
-        id__in=[int(field) for field, _ in add_custom_fields],
-    ).distinct()
+    # Resolved once, instead of re-querying the same field for every document
+    custom_fields_by_id: dict[int, CustomField] = CustomField.objects.in_bulk(
+        [field_id for field_id, _ in add_custom_fields],
+    )
+    # Passed to update_or_create() below rather than a bare id, so the FK is
+    # cached on the created instance and auditlog's post_save receiver does
+    # not reload it per row. Only needed for additions. content is deferred:
+    # the one field here that is both large and unused.
+    docs_by_id: dict[int, Document] = (
+        Document.objects.defer("content").in_bulk(affected_docs)
+        if add_custom_fields
+        else {}
+    )
     for field_id, value in add_custom_fields:
+        custom_field = custom_fields_by_id[field_id]
+        value_field = CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[
+            custom_field.data_type
+        ]
+        is_doclink = custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
         for doc_id in affected_docs:
-            defaults = {}
-            custom_field = custom_fields.get(id=field_id)
-            if custom_field:
-                value_field = CustomFieldInstance.TYPE_TO_DATA_STORE_NAME_MAP[
-                    custom_field.data_type
-                ]
-                defaults[value_field] = value
-                if (
-                    custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK
-                    and value
-                    and doc_id in value
-                ):
-                    # Prevent self-linking
-                    continue
+            if is_doclink and value and doc_id in value:
+                # Prevent self-linking
+                continue
             CustomFieldInstance.objects.update_or_create(
-                document_id=doc_id,
-                field_id=field_id,
-                defaults=defaults,
+                document=docs_by_id[doc_id],
+                field=custom_field,
+                defaults={value_field: value},
             )
-            if custom_field.data_type == CustomField.FieldDataType.DOCUMENTLINK:
-                doc = Document.objects.get(id=doc_id)
-                reflect_doclinks(doc, custom_field, value)
+            if is_doclink:
+                reflect_doclinks(docs_by_id[doc_id], custom_field, value)
 
-    # For doc link fields that are being removed, remove symmetrical links
+    # For doc link fields that are being removed, remove symmetrical links.
+    # select_related avoids a per-instance reload of the document and field.
     for doclink_being_removed_instance in CustomFieldInstance.objects.filter(
         document_id__in=affected_docs,
         field__id__in=remove_custom_fields,
         field__data_type=CustomField.FieldDataType.DOCUMENTLINK,
         value_document_ids__isnull=False,
-    ):
+    ).select_related("field", "document"):
         for target_doc_id in doclink_being_removed_instance.value:
             remove_doclink(
-                document=Document.objects.get(
-                    id=doclink_being_removed_instance.document.id,
-                ),
+                document=doclink_being_removed_instance.document,
                 field=doclink_being_removed_instance.field,
                 target_doc_id=target_doc_id,
             )
@@ -379,7 +382,7 @@ def delete(doc_ids: list[int]) -> Literal["OK"]:
         )
         delete_ids = list({*doc_ids, *version_ids})
 
-        Document.objects.filter(id__in=delete_ids).delete()
+        Document.objects.filter(id__in=delete_ids).delete(transaction_id=uuid.uuid4())
 
         from documents.search import get_backend
 
@@ -430,10 +433,13 @@ def set_permissions(
     else:
         qs.update(owner=owner)
 
-    for doc in qs:
-        set_permissions_for_object(permissions=set_permissions, object=doc, merge=merge)
-
     affected_docs = list(qs.values_list("pk", flat=True))
+    set_permissions_for_objects(
+        permissions=set_permissions,
+        model=Document,
+        pks=affected_docs,
+        merge=merge,
+    )
 
     bulk_update_documents.apply_async(
         kwargs={"document_ids": affected_docs},
@@ -1177,10 +1183,13 @@ def remove_doclink(
     """
     Removes a 'symmetrical' link to `document` from the target document's existing custom field instance
     """
-    target_doc_field_instance = CustomFieldInstance.objects.filter(
-        document_id=target_doc_id,
-        field=field,
-    ).first()
+    # select_related: a signal receiver (auditlog) touches .document/.field on
+    # the save() below, without this that is a per-call reload query
+    target_doc_field_instance = (
+        CustomFieldInstance.objects.filter(document_id=target_doc_id, field=field)
+        .select_related("document", "field")
+        .first()
+    )
     if (
         target_doc_field_instance is not None
         and document.id in target_doc_field_instance.value

src/documents/classifier.py

@@ -34,6 +34,27 @@ from paperless.signed_pickle import signed_pickle_loads
 
 logger = logging.getLogger("paperless.classifier")
 
+
+def _predict_with_threshold(classifier, X, threshold: float) -> int | None:
+    """
+    Return the predicted class id, or None if:
+    - the prediction is -1 (no match), or
+    - the winning class probability is below the configured threshold.
+
+    Using predict_proba() instead of predict() lets us apply a minimum-confidence
+    cutoff so that uncertain predictions are discarded rather than assigned.
+    """
+    probas = classifier.predict_proba(X)[0]
+    best_idx = int(probas.argmax())
+    best_class = int(classifier.classes_[best_idx])
+
+    if best_class == -1:
+        return None
+    if threshold > 0.0 and probas[best_idx] < threshold:
+        return None
+    return best_class
+
+
 ADVANCED_TEXT_PROCESSING_ENABLED = (
     settings.NLTK_LANGUAGE is not None and settings.NLTK_ENABLED
 )
@@ -102,7 +123,8 @@ class DocumentClassifier:
     # v8 - Added storage path classifier
     # v9 - Changed from hashing to time/ids for re-train check
     # v10 - HMAC-signed model file
-    FORMAT_VERSION = 10
+    # v11 - Use sample_weight for balanced training; predict_proba with threshold
+    FORMAT_VERSION = 11
 
     HMAC_SIZE = 32  # SHA-256 digest length
 
@@ -324,6 +346,13 @@ class DocumentClassifier:
         from sklearn.preprocessing import LabelBinarizer
         from sklearn.preprocessing import MultiLabelBinarizer
 
+        # MLPClassifier does not support class_weight directly
+        # (https://github.com/scikit-learn/scikit-learn/issues/9113), so we use
+        # compute_sample_weight to balance classes during training and prevent
+        # over-represented correspondents from dominating predictions.
+        # https://scikit-learn.org/stable/modules/generated/sklearn.utils.class_weight.compute_sample_weight.html
+        from sklearn.utils.class_weight import compute_sample_weight
+
         # Step 2: vectorize data
         logger.debug("Vectorizing data...")
         notify("Vectorizing document content...")
@@ -369,7 +398,7 @@ class DocumentClassifier:
                 self.tags_binarizer = MultiLabelBinarizer()
                 labels_tags_vectorized = self.tags_binarizer.fit_transform(labels_tags)
 
-            self.tags_classifier = MLPClassifier(tol=0.01)
+            self.tags_classifier = MLPClassifier(tol=0.01, random_state=0)
             self.tags_classifier.fit(data_vectorized, labels_tags_vectorized)
         else:
             self.tags_classifier = None
@@ -380,8 +409,12 @@ class DocumentClassifier:
             notify(
                 f"Training correspondent classifier ({num_correspondents} correspondent(s))...",
             )
-            self.correspondent_classifier = MLPClassifier(tol=0.01)
-            self.correspondent_classifier.fit(data_vectorized, labels_correspondent)
+            self.correspondent_classifier = MLPClassifier(tol=0.01, random_state=0)
+            self.correspondent_classifier.fit(
+                data_vectorized,
+                labels_correspondent,
+                sample_weight=compute_sample_weight("balanced", labels_correspondent),
+            )
         else:
             self.correspondent_classifier = None
             logger.debug(
@@ -393,8 +426,12 @@ class DocumentClassifier:
             notify(
                 f"Training document type classifier ({num_document_types} type(s))...",
             )
-            self.document_type_classifier = MLPClassifier(tol=0.01)
-            self.document_type_classifier.fit(data_vectorized, labels_document_type)
+            self.document_type_classifier = MLPClassifier(tol=0.01, random_state=0)
+            self.document_type_classifier.fit(
+                data_vectorized,
+                labels_document_type,
+                sample_weight=compute_sample_weight("balanced", labels_document_type),
+            )
         else:
             self.document_type_classifier = None
             logger.debug(
@@ -406,10 +443,11 @@ class DocumentClassifier:
                 "Training storage paths classifier...",
             )
             notify(f"Training storage path classifier ({num_storage_paths} path(s))...")
-            self.storage_path_classifier = MLPClassifier(tol=0.01)
+            self.storage_path_classifier = MLPClassifier(tol=0.01, random_state=0)
             self.storage_path_classifier.fit(
                 data_vectorized,
                 labels_storage_path,
+                sample_weight=compute_sample_weight("balanced", labels_storage_path),
             )
         else:
             self.storage_path_classifier = None
@@ -546,24 +584,24 @@ class DocumentClassifier:
     def predict_correspondent(self, content: str) -> int | None:
         if self.correspondent_classifier:
             X = self._vectorize(content)
-            correspondent_id = self.correspondent_classifier.predict(X)
-            if correspondent_id != -1:
-                return correspondent_id
-            else:
-                return None
-        else:
-            return None
+            predicted_id = _predict_with_threshold(
+                self.correspondent_classifier,
+                X,
+                settings.CLASSIFIER_MATCH_THRESHOLD,
+            )
+            return predicted_id
+        return None
 
     def predict_document_type(self, content: str) -> int | None:
         if self.document_type_classifier:
             X = self._vectorize(content)
-            document_type_id = self.document_type_classifier.predict(X)
-            if document_type_id != -1:
-                return document_type_id
-            else:
-                return None
-        else:
-            return None
+            predicted_id = _predict_with_threshold(
+                self.document_type_classifier,
+                X,
+                settings.CLASSIFIER_MATCH_THRESHOLD,
+            )
+            return predicted_id
+        return None
 
     def predict_tags(self, content: str) -> list[int]:
         from sklearn.utils.multiclass import type_of_target
@@ -589,10 +627,10 @@ class DocumentClassifier:
     def predict_storage_path(self, content: str) -> int | None:
         if self.storage_path_classifier:
             X = self._vectorize(content)
-            storage_path_id = self.storage_path_classifier.predict(X)
-            if storage_path_id != -1:
-                return storage_path_id
-            else:
-                return None
-        else:
-            return None
+            predicted_id = _predict_with_threshold(
+                self.storage_path_classifier,
+                X,
+                settings.CLASSIFIER_MATCH_THRESHOLD,
+            )
+            return predicted_id
+        return None

src/documents/management/commands/document_consumer.py

@@ -156,6 +156,15 @@ class FileStabilityTracker:
                     logger.debug(f"File disappeared during stability check: {path}")
                 continue
 
+            # Stable, but empty: some scanners create a zero byte placeholder
+            # and only write the page some time later. Consuming it now can
+            # only fail  so drop it and let the writer's next event
+            # (or the periodic rescan) bring it back once it has content
+            if not tracked.last_size:
+                to_remove.append(path)
+                logger.debug("Ignoring stable but empty file: %s", path)
+                continue
+
             # File is stable, we can return it
             to_yield.append(path)
             logger.info(f"File is stable: {path}")

src/documents/models.py

@@ -1,4 +1,5 @@
 import datetime
+import uuid
 from pathlib import Path
 from typing import Final
 
@@ -374,6 +375,7 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         If the queryset already annotated ``effective_content``, that value is used.
         """
         # Here to avoid circular import
+        from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
         from documents.versioning import sort_versions_newest_first
         from documents.versioning import versions_newest_first
 
@@ -383,6 +385,19 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
         if self.root_document_id is not None or self.pk is None:
             return self.content
 
+        latest_version_prefetch = getattr(
+            self,
+            LATEST_VERSION_CONTENT_PREFETCH_ATTR,
+            None,
+        )
+        if latest_version_prefetch is not None:
+            # Empty list means prefetch ran and found no versions — use own content.
+            return (
+                latest_version_prefetch[0].content
+                if latest_version_prefetch
+                else self.content
+            )
+
         prefetched_cache = getattr(self, "_prefetched_objects_cache", None)
         prefetched_versions = (
             prefetched_cache.get("versions")
@@ -514,13 +529,20 @@ class Document(SoftDeleteModel, ModelWithOwner):  # type: ignore[django-manager-
     def delete(
         self,
         *args,
+        transaction_id=None,
         **kwargs,
     ):
-        # If deleting a root document, move all its versions to trash as well.
+        # Versions must share the root's transaction ID so they are restored
+        # together by django-softdelete.
+        if transaction_id is None:
+            transaction_id = uuid.uuid4()
         if self.root_document_id is None:
-            Document.objects.filter(root_document=self).delete()
+            Document.objects.filter(root_document=self).delete(
+                transaction_id=transaction_id,
+            )
         return super().delete(
             *args,
+            transaction_id=transaction_id,
             **kwargs,
         )
 

src/documents/permissions.py

@@ -173,6 +173,179 @@ def set_permissions_for_object(
                         )
 
 
+def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permission]:
+    """
+    Resolves `codenames` to Permission rows, raising like the single-object
+    assign_perm() this bulk path replaces does (via a `.get()` internally)
+    if any codename doesn't exist -- e.g. a client-supplied action name that
+    was never validated (BulkEditObjectsSerializer._validate_permissions
+    calls validate_set_permissions() only for its side-effecting id checks
+    and discards the filtered dict it returns, so an unrecognized action key
+    reaches this function as-is). A plain `.filter()` with no existence
+    check would otherwise silently build zero rows and no-op instead of
+    reporting the bad input.
+    """
+    permission_objs = list(
+        Permission.objects.filter(content_type=ctype, codename__in=codenames),
+    )
+    missing = codenames - {p.codename for p in permission_objs}
+    if missing:
+        raise Permission.DoesNotExist(
+            f"Permission matching query does not exist for codename(s): "
+            f"{', '.join(sorted(missing))}",
+        )
+    return permission_objs
+
+
+def _apply_bulk_permission_entry(
+    *,
+    perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
+    identity_model: type[User] | type[Group],
+    identity_field: str,
+    ids: list[int],
+    codename: str,
+    permission_objs: list[Permission],
+    ctype: ContentType,
+    object_pks: list[str],
+    merge: bool,
+) -> None:
+    # Only the ids are needed to build permission rows (via `<field>_id=`),
+    # so avoid fetching full User/Group rows for identities that may not
+    # even end up being granted anything new.
+    add_ids = set(
+        identity_model.objects.filter(id__in=ids).values_list("id", flat=True),
+    )
+
+    if not merge:
+        existing_ids = set(
+            perm_model.objects.filter(
+                content_type=ctype,
+                object_pk__in=object_pks,
+                permission__codename=codename,
+            )
+            .values_list(f"{identity_field}_id", flat=True)
+            .distinct(),
+        )
+        remove_ids = existing_ids - add_ids
+        if remove_ids:
+            perm_model.objects.filter(
+                content_type=ctype,
+                object_pk__in=object_pks,
+                permission__codename=codename,
+                **{f"{identity_field}_id__in": remove_ids},
+            ).delete()
+
+    if not add_ids:
+        return
+
+    rows = [
+        perm_model(
+            content_type=ctype,
+            object_pk=pk,
+            permission=permission_obj,
+            **{f"{identity_field}_id": identity_id},
+        )
+        for permission_obj in permission_objs
+        for pk in object_pks
+        for identity_id in add_ids
+    ]
+    # ignore_conflicts skips only rows that already exist as an exact
+    # (identity, permission, object) match -- the same de-dup the
+    # underlying (user|group, permission, object_pk) unique constraint
+    # already enforces for the single-object assign_perm() this replaces,
+    # so it doesn't change what counts as "already granted". batch_size
+    # caps how many rows go into a single INSERT statement.
+    perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
+
+
+def set_permissions_for_objects(
+    permissions: dict,
+    model: type[Model],
+    pks: QuerySet | list,
+    *,
+    merge: bool = False,
+) -> None:
+    """
+    Bulk equivalent of set_permissions_for_object: applies the same
+    permission changes to every object identified by `pks` at once.
+
+    Takes a model + pks (rather than model instances) deliberately -- the
+    permission rows built below only ever need `pk`, `content_type`, and
+    identity ids, so callers shouldn't have to fetch full rows (with every
+    other field) just to hand them to this function.
+
+    Deliberately does not use guardian's queryset/list-aware assign_perm:
+    passing a list as the object routes to bulk_assign_perm, which skips
+    creating a direct permission row for anyone who already has the
+    permission via ANY group membership (it checks
+    ObjectPermissionChecker.has_perm, which is group-inheritance-aware) --
+    unlike the single-object assign_perm this replaces, which always
+    ensures a direct row via get_or_create regardless of group-derived
+    access. Losing that guarantee would mean a later revocation of the
+    group's grant silently strips access an admin explicitly asked to be
+    direct. Bulk-creating rows straight against the permission models
+    instead (see _apply_bulk_permission_entry) preserves the original
+    always-create-a-direct-row semantics while still batching every object
+    and every identity into one query per action, rather than one query per
+    (object, user) pair.
+    """
+    object_pks = [str(pk) for pk in pks]
+    if not object_pks:  # pragma: no cover
+        return
+
+    model_name = model.__name__.lower()
+    ctype = ContentType.objects.get_for_model(model)
+
+    # Every action is resolved up front, before anything is written, so an
+    # unrecognized action name (see _resolve_permissions) aborts the whole
+    # call instead of leaving the actions ahead of it already applied --
+    # BulkEditObjectsSerializer lets unknown keys through and its view turns
+    # the exception into a 400, so a half-applied change would otherwise be
+    # reported to the client as a failure.
+    permissions_by_action: dict[str, list[Permission]] = {}
+    for action, entry in permissions.items():
+        if "users" not in entry and "groups" not in entry:
+            continue
+        implied_codenames = {f"{action}_{model_name}"}
+        if action == "change":
+            # change gives view too
+            implied_codenames.add(f"view_{model_name}")
+        permissions_by_action[action] = _resolve_permissions(
+            implied_codenames,
+            ctype,
+        )
+
+    for action, entry in permissions.items():
+        codename = f"{action}_{model_name}"
+        permission_objs = permissions_by_action.get(action, [])
+
+        if "users" in entry:
+            _apply_bulk_permission_entry(
+                perm_model=UserObjectPermission,
+                identity_model=User,
+                identity_field="user",
+                ids=entry["users"],
+                codename=codename,
+                permission_objs=permission_objs,
+                ctype=ctype,
+                object_pks=object_pks,
+                merge=merge,
+            )
+
+        if "groups" in entry:
+            _apply_bulk_permission_entry(
+                perm_model=GroupObjectPermission,
+                identity_model=Group,
+                identity_field="group",
+                ids=entry["groups"],
+                codename=codename,
+                permission_objs=permission_objs,
+                ctype=ctype,
+                object_pks=object_pks,
+                merge=merge,
+            )
+
+
 def permitted_object_ids(
     user: User | None,
     model: type[Model],

src/documents/serialisers.py

@@ -89,6 +89,7 @@ from documents.templating.utils import convert_format_str_to_template_format
 from documents.templating.workflows import validate_workflow_template
 from documents.validators import uri_validator
 from documents.validators import url_validator
+from documents.versioning import has_prefetched_effective_content
 from documents.versioning import sort_versions_newest_first
 
 if TYPE_CHECKING:
@@ -1152,8 +1153,14 @@ class DocumentSerializer(
 
     def to_representation(self, instance):
         doc = super().to_representation(instance)
-        if "content" in self.fields and hasattr(instance, "effective_content"):
-            doc["content"] = getattr(instance, "effective_content") or ""
+        if "content" in self.fields and has_prefetched_effective_content(instance):
+            # Only resolve version-aware content when it's cheap: an SQL
+            # annotation or a versions prefetch is already on the instance.
+            # A caller that set up neither (e.g. TrashView, GlobalSearchView,
+            # which build their own querysets) gets the document's own,
+            # unresolved content instead of paying for an extra per-instance
+            # query -- same as before effective_content resolution existed.
+            doc["content"] = instance.get_effective_content() or ""
         if self.truncate_content and "content" in self.fields:
             doc["content"] = doc.get("content")[0:550]
         return doc
@@ -1247,30 +1254,31 @@ class DocumentSerializer(
 
             validated_data["tags"] = list(final_tags)
         if validated_data.get("remove_inbox_tags"):
-            tag_ids_being_added = (
-                [
-                    tag.id
-                    for tag in validated_data["tags"]
-                    if tag not in instance.tags.all()
-                ]
+            current_tag_ids = {t.pk for t in instance.tags.all()}
+            tags = (
+                validated_data["tags"]
                 if "tags" in validated_data
-                else []
+                else list(instance.tags.all())
             )
-            inbox_tags_not_being_added = Tag.objects.filter(is_inbox_tag=True).exclude(
-                id__in=tag_ids_being_added,
-            )
-            if "tags" in validated_data:
-                validated_data["tags"] = [
-                    tag
-                    for tag in validated_data["tags"]
-                    if tag not in inbox_tags_not_being_added
-                ]
-            else:
-                validated_data["tags"] = [
-                    tag
-                    for tag in instance.tags.all()
-                    if tag not in inbox_tags_not_being_added
-                ]
+
+            # Tags newly added in this update, plus their ancestors, are kept
+            keep_ids: set[int] = set()
+            for tag in tags:
+                if tag.pk not in current_tag_ids:
+                    keep_ids.add(tag.pk)
+                    keep_ids.update(int(pk) for pk in tag.get_ancestors_pks())
+
+            # Remove inbox tags and their descendants, except those being kept
+            remove_ids: set[int] = set()
+            for inbox_tag in (
+                Tag.objects.filter(is_inbox_tag=True)
+                .exclude(pk__in=keep_ids)
+                .only("pk", "tn_descendants_pks")
+            ):
+                remove_ids.add(inbox_tag.pk)
+                remove_ids.update(int(pk) for pk in inbox_tag.get_descendants_pks())
+
+            validated_data["tags"] = [t for t in tags if t.pk not in remove_ids]
 
         if settings.AUDIT_LOG_ENABLED:
             with set_actor(self.user):

src/documents/tests/test_api_chat.py

@@ -38,6 +38,42 @@ class TestChatStreamingViewInputValidation(APITestCase):
             )
         assert resp.status_code == status.HTTP_400_BAD_REQUEST
 
+    def test_answer_is_not_compressed(self) -> None:
+        """
+        GIVEN:
+            - A client that accepts compressed responses
+        WHEN:
+            - It asks the chat endpoint a question
+        THEN:
+            - The answer is streamed unencoded, chunk for chunk
+
+        The stream compressors buffer, so a compressed answer arrives in one
+        piece. The view cannot opt out by flagging the request: DRF's request
+        wrapper proxies reads but keeps writes to itself, so the flag never
+        reaches the Django request the middleware sees.
+        """
+        chunks = [f"token{i} " for i in range(40)]
+        with (
+            mock.patch(
+                "documents.views.AIConfig",
+                return_value=self._mock_ai_enabled(),
+            ),
+            mock.patch(
+                "documents.views.stream_chat_with_documents",
+                return_value=iter(chunks),
+            ),
+        ):
+            resp = self.client.post(
+                "/api/documents/chat/",
+                {"q": "What is in my archive?"},
+                format="json",
+                HTTP_ACCEPT_ENCODING="gzip, deflate, br, zstd",
+            )
+
+        assert resp.status_code == status.HTTP_200_OK
+        assert not resp.has_header("Content-Encoding")
+        assert list(resp.streaming_content) == [c.encode() for c in chunks]
+
     def test_missing_question_is_rejected(self) -> None:
         with mock.patch(
             "documents.views.AIConfig",

src/documents/tests/test_api_objects.py

@@ -2,10 +2,15 @@ import datetime
 import json
 from unittest import mock
 
+from django.contrib.auth.models import Group
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
+from django.db import connection
 from django.test import override_settings
+from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
+from guardian.shortcuts import get_groups_with_perms
+from guardian.shortcuts import get_users_with_perms
 from rest_framework import status
 from rest_framework.test import APITestCase
 
@@ -842,6 +847,66 @@ class TestBulkEditObjects(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(StoragePath.objects.count(), 0)
 
+    def test_bulk_objects_set_permissions_batched_across_object_count(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - Many tags are being bulk-edited to set permissions at once
+        WHEN:
+            - bulk_edit_objects API endpoint is called with set_permissions
+              operation over a small batch vs. a much larger one
+        THEN:
+            - Permissions are applied correctly at both scales
+            - Query count does not grow with the number of tags, i.e. each
+              user/group is applied across all tags with one batched call
+              rather than one call per (tag, identity) pair
+        """
+        group1 = Group.objects.create(name="perm-group")
+        permissions = {
+            "view": {"users": [self.user1.id, self.user2.id], "groups": [group1.id]},
+            "change": {"users": [self.user1.id], "groups": [group1.id]},
+        }
+
+        def run_with_n_tags(n: int) -> int:
+            tags = [Tag.objects.create(name=f"perm-tag-{n}-{i}") for i in range(n)]
+            with CaptureQueriesContext(connection) as ctx:
+                response = self.client.post(
+                    "/api/bulk_edit_objects/",
+                    json.dumps(
+                        {
+                            "objects": [t.id for t in tags],
+                            "object_type": "tags",
+                            "operation": "set_permissions",
+                            "permissions": permissions,
+                            "merge": False,
+                        },
+                    ),
+                    content_type="application/json",
+                )
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            for tag in tags:
+                self.assertEqual(get_users_with_perms(tag).count(), 2)
+                self.assertEqual(get_groups_with_perms(tag).count(), 1)
+            return len(ctx.captured_queries)
+
+        small_batch_queries = run_with_n_tags(5)
+        large_batch_queries = run_with_n_tags(50)
+
+        # A tolerance rather than equality, matching the N+1 check in
+        # test_views.py: bulk_create's batch_size caps rows per INSERT, so a
+        # large enough selection does legitimately add statements, and the
+        # per-process ContentType cache makes the first run carry an extra
+        # query. Neither can hide a regression to per-object assignment,
+        # which would be ~10x the small-batch count here.
+        self.assertLessEqual(
+            large_batch_queries,
+            small_batch_queries + 5,
+            "Permission assignment appears to scale with object count: "
+            f"{small_batch_queries} queries for 5 tags vs. "
+            f"{large_batch_queries} for 50",
+        )
+
     def test_bulk_objects_delete_all_filtered(self) -> None:
         """
         GIVEN:

src/documents/tests/test_api_trash.py

@@ -207,3 +207,65 @@ class TestTrashAPI(DirectoriesMixin, APITestCase):
         )
         self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
         self.assertIn("have not yet been deleted", resp.data["documents"][0])
+
+    def _make_versioned_document(self) -> tuple[Document, list[Document]]:
+        root = Document.objects.create(
+            title="root",
+            content="root-content",
+            checksum="root",
+            mime_type="application/pdf",
+        )
+        versions = [
+            Document.objects.create(
+                title=f"v{index}",
+                content=f"v{index}-content",
+                checksum=f"v{index}",
+                mime_type="application/pdf",
+                root_document=root,
+                version_index=index,
+            )
+            for index in range(1, 3)
+        ]
+        return root, versions
+
+    def test_api_trash_restore_document_restores_its_versions(self) -> None:
+        """
+        GIVEN:
+            - Existing document with two versions
+        WHEN:
+            - API request to delete the document
+            - API request to restore it from the trash
+        THEN:
+            - Only the document itself is listed in the trash
+            - A version cannot be restored without its root
+            - The document is restored together with all of its versions
+        """
+        root, versions = self._make_versioned_document()
+
+        self.client.force_login(user=self.user)
+        self.client.delete(f"/api/documents/{root.pk}/")
+        self.assertEqual(Document.deleted_objects.count(), 3)
+
+        resp = self.client.get("/api/trash/")
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(resp.data["count"], 1)
+        self.assertEqual(resp.data["results"][0]["id"], root.pk)
+
+        # A version cannot be restored while its root remains in the trash.
+        resp = self.client.post(
+            "/api/trash/",
+            {"action": "restore", "documents": [versions[0].pk]},
+        )
+        self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn("Restore the root document", resp.data["documents"][0])
+
+        resp = self.client.post(
+            "/api/trash/",
+            {"action": "restore", "documents": [root.pk]},
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(Document.deleted_objects.count(), 0)
+        self.assertCountEqual(
+            Document.objects.filter(root_document=root).values_list("id", flat=True),
+            [version.pk for version in versions],
+        )

src/documents/tests/test_bulk_edit.py

@@ -5,8 +5,11 @@ from unittest import mock
 
 import pikepdf
 from django.contrib.auth.models import Group
+from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
+from django.db import connection
 from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
 from guardian.shortcuts import assign_perm
 from guardian.shortcuts import get_groups_with_perms
 from guardian.shortcuts import get_users_with_perms
@@ -19,6 +22,7 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
+from documents.permissions import set_permissions_for_objects
 from documents.tests.utils import DirectoriesMixin
 
 
@@ -392,6 +396,11 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         self.assertFalse(Document.objects.filter(id=self.doc1.id).exists())
         self.assertFalse(Document.objects.filter(id=version.id).exists())
 
+        Document.deleted_objects.get(id=self.doc1.id).restore(strict=False)
+
+        self.assertTrue(Document.objects.filter(id=self.doc1.id).exists())
+        self.assertTrue(Document.objects.filter(id=version.id).exists())
+
     def test_delete_version_document_keeps_root(self) -> None:
         version = Document.objects.create(
             checksum="A-v1",
@@ -510,6 +519,178 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         )
         self.assertEqual(groups_with_perms.count(), 2)
 
+    @mock.patch("documents.tasks.bulk_update_documents.apply_async")
+    def test_set_permissions_batched_across_document_count(
+        self,
+        m,
+    ) -> None:
+        """
+        GIVEN:
+            - Many documents are being bulk-edited to set permissions at once
+        WHEN:
+            - set_permissions runs over a small batch vs. a much larger one
+        THEN:
+            - Permissions are applied correctly at both scales
+            - Query count does not grow with the number of documents, i.e.
+              each user/group is applied across all documents with one
+              batched call rather than one call per (document, identity)
+              pair
+        """
+        permissions = {
+            "view": {
+                "users": [self.user1.id, self.user2.id],
+                "groups": [self.group2.id],
+            },
+            "change": {
+                "users": [self.user1.id],
+                "groups": [self.group2.id],
+            },
+        }
+
+        def run_with_n_documents(n: int) -> int:
+            docs = [
+                Document.objects.create(checksum=f"perm-{n}-{i}", title=f"perm-{n}-{i}")
+                for i in range(n)
+            ]
+            with CaptureQueriesContext(connection) as ctx:
+                bulk_edit.set_permissions(
+                    [doc.id for doc in docs],
+                    set_permissions=permissions,
+                    owner=self.owner,
+                    merge=False,
+                )
+            for doc in docs:
+                self.assertEqual(get_users_with_perms(doc).count(), 2)
+                self.assertEqual(get_groups_with_perms(doc).count(), 1)
+            return len(ctx.captured_queries)
+
+        small_batch_queries = run_with_n_documents(5)
+        large_batch_queries = run_with_n_documents(50)
+
+        # A tolerance rather than equality, matching the N+1 check in
+        # test_views.py: bulk_create's batch_size caps rows per INSERT, so a
+        # large enough selection does legitimately add statements, and the
+        # per-process ContentType cache makes the first run carry an extra
+        # query. Neither can hide a regression to per-document assignment,
+        # which would be ~10x the small-batch count here.
+        self.assertLessEqual(
+            large_batch_queries,
+            small_batch_queries + 5,
+            "Permission assignment appears to scale with document count: "
+            f"{small_batch_queries} queries for 5 documents vs. "
+            f"{large_batch_queries} for 50",
+        )
+
+    @mock.patch("documents.tasks.bulk_update_documents.apply_async")
+    def test_set_permissions_grants_direct_perm_even_if_already_granted_via_group(
+        self,
+        m,
+    ) -> None:
+        """
+        GIVEN:
+            - A user already has view access to a document via group
+              membership, with no direct grant of their own
+        WHEN:
+            - set_permissions explicitly grants that same user direct view
+              access via bulk_edit
+        THEN:
+            - A direct permission grant is created for the user, not skipped
+              because they already have equivalent access via the group
+
+        Regression test: guardian's queryset-aware assign_perm() (routed to
+        when the target is a list/queryset) skips creating a direct row for
+        anyone whose ObjectPermissionChecker.has_perm() already returns True
+        -- which includes group-derived access. The single-object assign_perm
+        this bulk path replaces has no such check; it always ensures a
+        direct row via get_or_create. Losing that guarantee would mean
+        revoking the group's grant later silently strips access that was
+        supposed to be explicit.
+        """
+        self.doc1.owner = self.user1
+        self.doc1.save()
+        self.user1.groups.add(self.group1)
+        assign_perm("view_document", self.group1, self.doc1)
+
+        bulk_edit.set_permissions(
+            [self.doc1.id],
+            set_permissions={
+                "view": {"users": [self.user1.id], "groups": []},
+            },
+            merge=True,
+        )
+
+        direct_users = get_users_with_perms(
+            self.doc1,
+            only_with_perms_in=["view_document"],
+            with_group_users=False,
+        )
+        self.assertIn(self.user1, direct_users)
+
+    def test_set_permissions_for_objects_raises_for_unknown_action(self) -> None:
+        """
+        GIVEN:
+            - An unrecognized permission action name with users to grant it
+              to
+        WHEN:
+            - set_permissions_for_objects is called
+        THEN:
+            - Permission.DoesNotExist is raised, not a silent no-op
+
+        Regression test: the endpoint that calls this
+        (BulkEditObjectPermissionsView) never actually validates action
+        names against the raw client-supplied permissions dict --
+        BulkEditObjectsSerializer._validate_permissions calls
+        validate_set_permissions() only for its side-effecting user/group id
+        checks and discards the filtered dict it returns -- so a bogus
+        action key reaches this function as-is. Resolving the Permission via
+        a bare `.filter()` (which returns empty instead of raising) would
+        silently drop the grant and report success.
+        """
+        with self.assertRaises(Permission.DoesNotExist):
+            set_permissions_for_objects(
+                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},
+                Document,
+                [self.doc1.pk],
+            )
+
+    def test_set_permissions_for_objects_unknown_action_applies_nothing(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - A permissions dict with a valid action ordered ahead of an
+              unrecognized one
+        WHEN:
+            - set_permissions_for_objects is called
+        THEN:
+            - Permission.DoesNotExist is raised
+            - The valid action ahead of it is not applied either
+
+        Every action is resolved before any row is written, so a bad action
+        name cannot leave a half-applied change behind. That matters because
+        BulkEditObjectsView turns this exception into a 400: without the
+        up-front resolution the client would be told the request failed
+        while the leading action had already been committed.
+        """
+        with self.assertRaises(Permission.DoesNotExist):
+            set_permissions_for_objects(
+                {
+                    "view": {"users": [self.user1.id], "groups": []},
+                    "not_a_real_action": {"users": [self.user1.id], "groups": []},
+                },
+                Document,
+                [self.doc1.pk],
+            )
+
+        self.assertNotIn(
+            self.user1,
+            get_users_with_perms(
+                self.doc1,
+                only_with_perms_in=["view_document"],
+                with_group_users=False,
+            ),
+        )
+
     @mock.patch("documents.models.Document.delete")
     def test_delete_documents_old_uuid_field(self, m) -> None:
         m.side_effect = Exception("Data too long for column 'transaction_id' at row 1")

src/documents/tests/test_classifier.py

@@ -3,6 +3,7 @@ import warnings
 from pathlib import Path
 from unittest import mock
 
+import numpy as np
 import pytest
 from django.conf import settings
 from django.test import TestCase
@@ -11,6 +12,7 @@ from django.test import override_settings
 from documents.classifier import ClassifierModelCorruptError
 from documents.classifier import DocumentClassifier
 from documents.classifier import IncompatibleClassifierVersionError
+from documents.classifier import _predict_with_threshold
 from documents.classifier import load_classifier
 from documents.models import Correspondent
 from documents.models import Document
@@ -625,6 +627,103 @@ class TestClassifier(DirectoriesMixin, TestCase):
         self.assertEqual(self.classifier.predict_storage_path(doc1.content), sp.pk)
         self.assertIsNone(self.classifier.predict_storage_path(doc2.content))
 
+    def test_predict_rejects_prediction_below_match_threshold(self) -> None:
+        """
+        GIVEN:
+            - Classifiers trained against test data with confident predictions
+        WHEN:
+            - CLASSIFIER_MATCH_THRESHOLD exceeds the model's confidence
+        THEN:
+            - Every predict_* method discards the match in favor of no match
+        """
+        c1 = Correspondent.objects.create(
+            name="c1",
+            matching_algorithm=Correspondent.MATCH_AUTO,
+        )
+        dt1 = DocumentType.objects.create(
+            name="dt1",
+            matching_algorithm=DocumentType.MATCH_AUTO,
+        )
+        sp1 = StoragePath.objects.create(
+            name="sp1",
+            matching_algorithm=StoragePath.MATCH_AUTO,
+        )
+
+        doc1 = Document.objects.create(
+            title="doc1",
+            content="this is a document from c1",
+            correspondent=c1,
+            document_type=dt1,
+            storage_path=sp1,
+            checksum="A",
+        )
+        Document.objects.create(
+            title="doc2",
+            content="this is a document from no one",
+            checksum="B",
+        )
+
+        self.classifier.train()
+
+        predictors = {
+            "correspondent": self.classifier.predict_correspondent,
+            "document_type": self.classifier.predict_document_type,
+            "storage_path": self.classifier.predict_storage_path,
+        }
+        # No real prediction can reach a confidence this high, so this
+        # isolates the threshold check from the model's actual output.
+        with override_settings(CLASSIFIER_MATCH_THRESHOLD=0.999999):
+            for name, predict in predictors.items():
+                with self.subTest(field=name):
+                    self.assertIsNone(predict(doc1.content))
+
+    def test_train_uses_balanced_sample_weight(self) -> None:
+        """
+        GIVEN:
+            - A training set with correspondents, document types and storage paths
+        WHEN:
+            - The classifier is trained
+        THEN:
+            - Each MLP classifier is fit with balanced sample weights, so that
+              over-represented classes don't dominate predictions
+        """
+        c1 = Correspondent.objects.create(
+            name="c1",
+            matching_algorithm=Correspondent.MATCH_AUTO,
+        )
+        dt1 = DocumentType.objects.create(
+            name="dt1",
+            matching_algorithm=DocumentType.MATCH_AUTO,
+        )
+        sp1 = StoragePath.objects.create(
+            name="sp1",
+            matching_algorithm=StoragePath.MATCH_AUTO,
+        )
+
+        Document.objects.create(
+            title="doc1",
+            content="this is a document from c1",
+            correspondent=c1,
+            document_type=dt1,
+            storage_path=sp1,
+            checksum="A",
+        )
+        Document.objects.create(
+            title="doc2",
+            content="this is a document from no one",
+            checksum="B",
+        )
+
+        with mock.patch(
+            "sklearn.utils.class_weight.compute_sample_weight",
+            return_value=None,
+        ) as mocked_compute_sample_weight:
+            self.classifier.train()
+
+        self.assertEqual(mocked_compute_sample_weight.call_count, 3)
+        for call in mocked_compute_sample_weight.call_args_list:
+            self.assertEqual(call.args[0], "balanced")
+
     def test_one_tag_predict(self) -> None:
         t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
 
@@ -810,6 +909,52 @@ class TestClassifier(DirectoriesMixin, TestCase):
             load_classifier(raise_exception=True)
 
 
+class _StubProbaClassifier:
+    """
+    A fake scikit-learn classifier exposing just enough of the API for
+    `_predict_with_threshold`: `classes_` and `predict_proba`.
+    """
+
+    def __init__(self, classes: list[int], probabilities: list[float]) -> None:
+        self.classes_ = np.array(classes)
+        self._probabilities = np.array([probabilities])
+
+    def predict_proba(self, X) -> np.ndarray:
+        return self._probabilities
+
+
+@pytest.mark.parametrize(
+    ("classes", "probabilities", "threshold", "expected"),
+    [
+        # confident prediction above the threshold is returned
+        ([-1, 3], [0.1, 0.9], 0.6, 3),
+        # prediction below the threshold is discarded
+        ([-1, 3], [0.45, 0.55], 0.6, None),
+        # boundary: exactly at the threshold is accepted, not discarded
+        ([-1, 3], [0.4, 0.6], 0.6, 3),
+        # the winning class is the "no match" pseudo-class, regardless of its
+        # own confidence
+        ([-1, 3], [0.99, 0.01], 0.0, None),
+        # threshold of 0.0 disables the confidence check entirely
+        ([-1, 3], [0.45, 0.55], 0.0, 3),
+    ],
+)
+def test_predict_with_threshold(classes, probabilities, threshold, expected) -> None:
+    classifier = _StubProbaClassifier(classes, probabilities)
+    result = _predict_with_threshold(classifier, X=None, threshold=threshold)
+    assert result == expected
+
+
+def test_classifier_match_threshold_default() -> None:
+    """
+    GIVEN:
+        - No PAPERLESS_CLASSIFIER_MATCH_THRESHOLD environment variable is set
+    THEN:
+        - The classifier match threshold defaults to 0.6
+    """
+    assert settings.CLASSIFIER_MATCH_THRESHOLD == 0.6
+
+
 def test_preprocess_content() -> None:
     """
     GIVEN:

src/documents/tests/test_document_list_effective_content_annotation.py

@@ -0,0 +1,457 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import TYPE_CHECKING
+
+import pytest
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+from rest_framework import status
+
+from documents.models import Document
+from documents.tests.factories import DocumentFactory
+from documents.versioning import LATEST_VERSION_CONTENT_PREFETCH_ATTR
+from documents.versioning import has_prefetched_effective_content
+from documents.versioning import latest_version_content_prefetch
+from documents.views import DocumentViewSet
+
+if TYPE_CHECKING:
+    from rest_framework.test import APIClient
+
+
+class TestNeedsEffectiveContentAnnotation:
+    """
+    DocumentViewSet._needs_effective_content_annotation() decides whether
+    the effective_content correlated subquery is worth attaching to the
+    queryset at all -- see TestDocumentListEffectiveContentAnnotation below
+    for why. This only checks that decision's own logic (a plain query-param
+    membership test), not that Django/DRF's filtering machinery works.
+    """
+
+    @pytest.mark.parametrize(
+        ("params", "expected"),
+        [
+            ({}, False),
+            ({"ordering": "-added"}, False),
+            ({"tags__id__in": "1,2"}, False),
+            ({"search": ""}, False),
+            ({"search": "   "}, False),
+            ({"content__icontains": ""}, False),
+            ({"search": "foo"}, True),
+            ({"title_content": "foo"}, True),
+            ({"content__istartswith": "foo"}, True),
+            ({"content__iendswith": "foo"}, True),
+            ({"content__icontains": "foo"}, True),
+            ({"content__iexact": "foo"}, True),
+        ],
+    )
+    def test_detects_content_filter_params(
+        self,
+        params: dict[str, str],
+        expected: bool,  # noqa: FBT001
+    ) -> None:
+        """
+        GIVEN:
+            - A view bound to a request carrying the given query params
+        WHEN:
+            - Checking whether the effective_content annotation is needed
+        THEN:
+            - It is needed only for requests that actually filter on it
+        """
+        view = DocumentViewSet()
+        view.request = SimpleNamespace(query_params=params)
+
+        assert view._needs_effective_content_annotation() is expected
+
+
+class TestNeedsEffectiveContentPrefetch:
+    """
+    DocumentViewSet._needs_effective_content_prefetch() decides whether the
+    single-version content prefetch is worth attaching. It has to read the
+    `fields` param exactly the way get_serializer() does, or a request whose
+    response includes content ends up without the prefetch and pays
+    get_effective_content()'s per-instance fallback instead.
+    """
+
+    @pytest.mark.parametrize(
+        ("params", "expected"),
+        [
+            pytest.param({}, True, id="no-fields-param-keeps-every-field"),
+            pytest.param({"fields": ""}, True, id="blank-fields-keeps-every-field"),
+            pytest.param(
+                {"fields": "id,content"},
+                True,
+                id="content-among-requested-fields",
+            ),
+            pytest.param({"fields": "content"}, True, id="content-only"),
+            pytest.param({"fields": "id"}, False, id="content-not-requested"),
+            pytest.param(
+                {"fields": "id,title"},
+                False,
+                id="several-fields-without-content",
+            ),
+        ],
+    )
+    def test_detects_whether_content_can_reach_the_response(
+        self,
+        params: dict[str, str],
+        expected: bool,  # noqa: FBT001
+    ) -> None:
+        """
+        GIVEN:
+            - A view bound to a request carrying the given query params
+        WHEN:
+            - Checking whether the content prefetch is needed
+        THEN:
+            - It is needed exactly when get_serializer() would emit content,
+              which treats a blank `fields` the same as an absent one
+        """
+        view = DocumentViewSet()
+        view.request = SimpleNamespace(query_params=params)
+
+        assert view._needs_effective_content_prefetch() is expected
+
+
+@pytest.mark.django_db
+class TestDocumentListEffectiveContentAnnotation:
+    """
+    DocumentViewSet.get_queryset() only attaches the effective_content
+    correlated subquery when a request actually filters on it. Attaching it
+    unconditionally re-executes it once per candidate row before the page's
+    LIMIT is applied -- fine on SQLite/Postgres, but pathological on
+    MariaDB's default cardinality estimation for the root_document_id
+    self-join once candidate counts get large (see the root_document_id /
+    effective_content perf investigation).
+    """
+
+    def test_list_without_content_filter_skips_annotation_but_returns_latest_content(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document whose latest version has different content
+        WHEN:
+            - Listing documents with no search/content-filter param
+        THEN:
+            - The response still reflects the latest version's content
+            - The database never evaluates effective_content per row
+        """
+        root = DocumentFactory(content="old-root-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="new-version-content",
+        )
+
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get("/api/documents/?fields=id,content")
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["results"] == [
+            {"id": root.id, "content": "new-version-content"},
+        ]
+        assert not any(
+            "effective_content" in query["sql"] for query in ctx.captured_queries
+        )
+
+    @pytest.mark.parametrize(
+        "fields_param",
+        [
+            pytest.param("", id="blank-fields"),
+            pytest.param("id,content", id="content-requested"),
+        ],
+    )
+    def test_content_resolves_without_a_query_per_document(
+        self,
+        admin_client: APIClient,
+        fields_param: str,
+    ) -> None:
+        """
+        GIVEN:
+            - One versioned root document, then two more
+        WHEN:
+            - Listing documents with a `fields` param that keeps content
+        THEN:
+            - Every root's content resolves to its latest version's
+            - The query count does not grow with the number of documents,
+              i.e. a blank `fields` does not skip the prefetch and fall back
+              to loading each root's deferred version content
+        """
+        first = DocumentFactory(content="first-root-content")
+        DocumentFactory(
+            root_document=first,
+            version_index=1,
+            content="first-version-content",
+        )
+
+        with CaptureQueriesContext(connection) as one_document:
+            response = admin_client.get(f"/api/documents/?fields={fields_param}")
+
+        assert response.status_code == status.HTTP_200_OK
+        assert [r["content"] for r in response.data["results"]] == [
+            "first-version-content",
+        ]
+
+        for index in range(2):
+            root = DocumentFactory(content=f"root-content-{index}")
+            DocumentFactory(
+                root_document=root,
+                version_index=1,
+                content=f"version-content-{index}",
+            )
+        with CaptureQueriesContext(connection) as three_documents:
+            response = admin_client.get(f"/api/documents/?fields={fields_param}")
+
+        assert response.status_code == status.HTTP_200_OK
+        assert sorted(r["content"] for r in response.data["results"]) == [
+            "first-version-content",
+            "version-content-0",
+            "version-content-1",
+        ]
+        assert len(_get_document_queries(three_documents)) == len(
+            _get_document_queries(one_document),
+        )
+
+    def test_list_without_content_field_skips_prefetch_and_omits_content(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        """
+        GIVEN:
+            - A versioned root document
+        WHEN:
+            - Listing documents without asking for content
+        THEN:
+            - Content is neither serialized nor resolved
+            - Nothing pays for the prefetch or the per-instance fallback
+        """
+        root = DocumentFactory(content="root-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="version-content",
+        )
+
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get("/api/documents/?fields=id")
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["results"] == [{"id": root.id}]
+        assert _get_effective_content_fallback_queries(ctx) == []
+        # Only the list query itself reads a content column: no extra query
+        # for the skipped prefetch, none for a per-instance fallback
+        content_queries = [
+            query
+            for query in ctx.captured_queries
+            if '"documents_document"."content"' in query["sql"]
+        ]
+        assert len(content_queries) == 1
+
+    def test_latest_version_content_prefetch_carries_only_the_newest_version(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document with two versions
+        WHEN:
+            - Fetching the root through latest_version_content_prefetch()
+        THEN:
+            - The prefetch carries only the single newest version, not every
+              historical version's content (the whole point of not reusing
+              the metadata-only "versions" prefetch for this)
+        """
+        root = DocumentFactory(content="root-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="older-version-content",
+        )
+        DocumentFactory(
+            root_document=root,
+            version_index=2,
+            content="newest-version-content",
+        )
+
+        fetched_root = (
+            Document.objects.filter(pk=root.pk)
+            .prefetch_related(
+                latest_version_content_prefetch(),
+            )
+            .get()
+        )
+
+        latest = getattr(fetched_root, LATEST_VERSION_CONTENT_PREFETCH_ATTR)
+        assert [v.content for v in latest] == ["newest-version-content"]
+
+
+class TestHasPrefetchedEffectiveContent:
+    """
+    DocumentSerializer.to_representation() only calls get_effective_content()
+    when has_prefetched_effective_content() says it's cheap -- otherwise a
+    caller that never set up an annotation or prefetch (TrashView,
+    GlobalSearchView, which build their own querysets and don't display
+    content at all) would pay for a per-instance query nobody asked for.
+    """
+
+    def test_false_with_no_annotation_or_prefetch(self) -> None:
+        """
+        GIVEN:
+            - A document the ORM never annotated or prefetched for
+        WHEN:
+            - Asking whether its effective content is already resolved
+        THEN:
+            - It is not, so the serializer must leave it alone
+        """
+        document = DocumentFactory.build()
+
+        assert has_prefetched_effective_content(document) is False
+
+    def test_true_with_effective_content_annotation(self) -> None:
+        """
+        GIVEN:
+            - A document carrying the queryset's effective_content annotation
+        WHEN:
+            - Asking whether its effective content is already resolved
+        THEN:
+            - It is, straight off the annotation
+        """
+        document = DocumentFactory.build()
+        document.effective_content = "resolved"
+
+        assert has_prefetched_effective_content(document) is True
+
+    def test_true_with_lean_prefetch_attr_even_when_empty(self) -> None:
+        """
+        GIVEN:
+            - A document the lean content prefetch ran for, finding no versions
+        WHEN:
+            - Asking whether its effective content is already resolved
+        THEN:
+            - It is: an empty prefetch is an answer, not a missing one
+        """
+        document = DocumentFactory.build()
+        setattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, [])
+
+        assert has_prefetched_effective_content(document) is True
+
+    def test_true_with_metadata_versions_prefetch_cache(self) -> None:
+        """
+        GIVEN:
+            - A document carrying only the metadata "versions" prefetch
+        WHEN:
+            - Asking whether its effective content is already resolved
+        THEN:
+            - It is, via get_effective_content()'s prefetch-cache branch
+        """
+        document = DocumentFactory.build()
+        document._prefetched_objects_cache = {"versions": []}
+
+        assert has_prefetched_effective_content(document) is True
+
+
+def _get_document_queries(
+    ctx: CaptureQueriesContext,
+) -> list[dict[str, str]]:
+    """
+    The queries a list request spends on the documents themselves, i.e.
+    everything but the one-time django_content_type lookup guardian's
+    permission filtering makes. That lookup is process-cached, and the
+    autouse fixture in conftest clears the cache before every test, so it
+    lands in whichever request happens to run first and never repeats --
+    counting it makes a request look like it costs one query more than the
+    identical request after it.
+    """
+    return [q for q in ctx.captured_queries if '"django_content_type"' not in q["sql"]]
+
+
+def _get_effective_content_fallback_queries(
+    ctx: CaptureQueriesContext,
+) -> list[dict[str, str]]:
+    """
+    Document.get_effective_content()'s per-instance fallback (no annotation,
+    no prefetch) is a `.values_list("content", flat=True).first()` query --
+    a SELECT of just the content column. Distinct from get_versions()'s own,
+    unrelated per-instance metadata query (id/checksum/added/etc, no
+    content) run to build the "versions" response field, which isn't part
+    of what this test file covers.
+    """
+    return [
+        q
+        for q in ctx.captured_queries
+        if q["sql"].startswith('SELECT "documents_document"."content" FROM')
+    ]
+
+
+@pytest.mark.django_db
+class TestTrashAndGlobalSearchEffectiveContentIsNeverPerInstance:
+    """
+    TrashView and GlobalSearchView serialize Document instances with
+    DocumentSerializer too, but build their querysets independently of
+    DocumentViewSet.get_queryset(). TrashView doesn't display content at all,
+    so it keeps the document's own unresolved content; GlobalSearchView
+    annotates effective_content itself, so it shows the latest version's.
+    Neither should ever fall back to a per-instance query.
+    """
+
+    def test_trash_list_shows_unresolved_content_with_no_extra_query(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        """
+        GIVEN:
+            - A trashed root document whose own content differs from what a
+              version would have had (also trashed, deletion cascades)
+        WHEN:
+            - Listing trash
+        THEN:
+            - The response shows the document's own content
+            - Nothing ever queries for versions to resolve it
+        """
+        root = DocumentFactory(content="own-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="version-content",
+        )
+        root.delete()
+
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get("/api/trash/")
+
+        assert response.status_code == status.HTTP_200_OK
+        [result] = [r for r in response.data["results"] if r["id"] == root.id]
+        assert result["content"] == "own-content"
+        assert _get_effective_content_fallback_queries(ctx) == []
+
+    def test_global_search_db_only_shows_latest_version_content_with_no_extra_query(
+        self,
+        admin_client: APIClient,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document, findable by title, whose own content differs
+              from its latest version's
+        WHEN:
+            - Using the global search endpoint's db_only mode
+        THEN:
+            - The response shows the latest version's content, resolved by
+              GlobalSearchView's own effective_content annotation
+            - There is no per-instance fallback query
+        """
+        root = DocumentFactory(title="findme", content="own-content")
+        DocumentFactory(
+            root_document=root,
+            version_index=1,
+            content="version-content",
+        )
+
+        with CaptureQueriesContext(connection) as ctx:
+            response = admin_client.get(
+                "/api/search/?query=findme&db_only=true",
+            )
+
+        assert response.status_code == status.HTTP_200_OK
+        [result] = [d for d in response.data["documents"] if d["id"] == root.id]
+        assert result["content"] == "version-content"
+        assert _get_effective_content_fallback_queries(ctx) == []

src/documents/tests/test_document_model.py

@@ -110,7 +110,7 @@ class TestDocument(TestCase):
             checksum="checksum",
             mime_type="application/pdf",
         )
-        Document.objects.create(
+        version = Document.objects.create(
             root_document=root,
             correspondent=root.correspondent,
             title="Version",
@@ -124,6 +124,10 @@ class TestDocument(TestCase):
         self.assertEqual(Document.objects.count(), 0)
         self.assertEqual(Document.deleted_objects.count(), 2)
 
+        root.restore(strict=False)
+
+        self.assertTrue(Document.objects.filter(pk=version.pk).exists())
+
     def test_file_name(self) -> None:
         doc = Document(
             mime_type="application/pdf",

src/documents/tests/test_management_consumer.py

@@ -136,6 +136,23 @@ def wait_for_mock_call(
     return False
 
 
+def sleep_past_stability(
+    owner: FileStabilityTracker | ConsumerThread,
+    *,
+    windows: float = 1.5,
+) -> None:
+    """
+    Block until a tracked file's stability window has certainly elapsed.
+
+    Args:
+        owner: The tracker, or the consumer thread running one, whose
+               configured stability delay sets the wait.
+        windows: How many stability windows to wait, giving slop for a slow
+                 or loaded test runner.
+    """
+    sleep(owner.stability_delay * windows)
+
+
 class TestTrackedFile:
     """Tests for the TrackedFile dataclass."""
 
@@ -261,6 +278,56 @@ class TestFileStabilityTracker:
         assert len(stable) == 0
         assert stability_tracker.pending_count == 1
 
+    def test_get_stable_files_skips_empty_file(
+        self,
+        stability_tracker: FileStabilityTracker,
+        tmp_path: Path,
+    ) -> None:
+        """
+        GIVEN:
+            - A zero byte file, tracked and past its stability delay
+        WHEN:
+            - Stable files are collected
+        THEN:
+            - The file is not yielded for consumption
+            - The file is dropped from tracking rather than held, so an
+              abandoned placeholder does not keep the watch loop awake
+        """
+        empty = tmp_path / "scan.pdf"
+        empty.write_bytes(b"")
+        stability_tracker.track(empty, Change.added)
+        sleep_past_stability(stability_tracker)
+
+        stable = list(stability_tracker.get_stable_files())
+
+        assert stable == []
+        assert stability_tracker.pending_count == 0
+
+    def test_empty_file_is_yielded_once_content_arrives(
+        self,
+        stability_tracker: FileStabilityTracker,
+        tmp_path: Path,
+    ) -> None:
+        """
+        GIVEN:
+            - A zero byte file which was dropped from tracking while empty
+        WHEN:
+            - The writer fills the file and a new event re-tracks it
+        THEN:
+            - The file is yielded for consumption once it is stable
+        """
+        target = tmp_path / "scan.pdf"
+        target.write_bytes(b"")
+        stability_tracker.track(target, Change.added)
+        sleep_past_stability(stability_tracker)
+        assert list(stability_tracker.get_stable_files()) == []
+
+        target.write_bytes(b"%PDF-1.4 content")
+        stability_tracker.track(target, Change.modified)
+        sleep_past_stability(stability_tracker)
+
+        assert list(stability_tracker.get_stable_files()) == [target]
+
     def test_get_stable_files_deleted_during_check(self, temp_file: Path) -> None:
         """Test deleted file is not returned during stability check."""
         tracker = FileStabilityTracker(stability_delay=0.1)
@@ -879,6 +946,51 @@ class TestCommandWatch:
 
         mock_consume_file_delay.apply_async.assert_called()
 
+    def test_scanner_placeholder_is_not_consumed_while_empty(
+        self,
+        consumption_dir: Path,
+        sample_pdf: Path,
+        mock_consume_file_delay: MagicMock,
+        start_consumer: Callable[..., ConsumerThread],
+    ) -> None:
+        """
+        GIVEN:
+            - A scanner which creates a zero byte placeholder and only writes
+              the page some time later (GH discussion #13969)
+        WHEN:
+            - The placeholder sits untouched well past the stability delay
+            - The scanner then writes the real content
+        THEN:
+            - The empty placeholder is never queued, as it could only fail
+              with "Unsupported mime type inode/x-empty"
+            - The file is queued exactly once, when the content lands
+        """
+        thread = start_consumer(stability_delay=0.2)
+
+        target = consumption_dir / "scan.pdf"
+        target.write_bytes(b"")  # the scanner's placeholder
+
+        # Well past the stability delay: the old behaviour queued it here.
+        sleep_past_stability(thread, windows=5)
+        if thread.exception:
+            raise thread.exception
+        assert mock_consume_file_delay.apply_async.call_count == 0
+
+        shutil.copy(sample_pdf, target)  # the scanner finishes the page
+
+        assert wait_for_mock_call(
+            mock_consume_file_delay.apply_async,
+            timeout_s=5.0,
+        )
+        if thread.exception:
+            raise thread.exception
+
+        assert mock_consume_file_delay.apply_async.call_count == 1
+        queued_doc = mock_consume_file_delay.apply_async.call_args.kwargs["kwargs"][
+            "input_doc"
+        ]
+        assert queued_doc.original_file.name == "scan.pdf"
+
     def test_ignores_macos_files(
         self,
         consumption_dir: Path,

src/documents/tests/test_tag_hierarchy.py

@@ -2,6 +2,7 @@ from unittest import mock
 
 from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
+from rest_framework import status
 from rest_framework.test import APITestCase
 
 from documents import bulk_edit
@@ -108,6 +109,44 @@ class TestTagHierarchy(DirectoriesMixin, APITestCase):
         self.document.refresh_from_db()
         assert self.document.tags.count() == 0
 
+    def test_remove_inbox_tags_removes_nested_children(self) -> None:
+        inbox = Tag.objects.create(name="Inbox", is_inbox_tag=True)
+        nested = Tag.objects.create(name="Nested", tn_parent=inbox)
+        self.document.add_nested_tags([nested])
+
+        resp = self.client.patch(
+            f"/api/documents/{self.document.pk}/",
+            {"title": "new title", "remove_inbox_tags": True},
+            format="json",
+        )
+        assert resp.status_code == status.HTTP_200_OK
+        self.document.refresh_from_db()
+        assert self.document.tags.count() == 0
+
+        # A subsequent save must not re-add the inbox tag as an ancestor
+        resp = self.client.patch(
+            f"/api/documents/{self.document.pk}/",
+            {"title": "another title", "tags": [], "remove_inbox_tags": True},
+            format="json",
+        )
+        assert resp.status_code == status.HTTP_200_OK
+        self.document.refresh_from_db()
+        assert self.document.tags.count() == 0
+
+    def test_remove_inbox_tags_keeps_inbox_when_nested_child_added(self) -> None:
+        inbox = Tag.objects.create(name="Inbox", is_inbox_tag=True)
+        nested = Tag.objects.create(name="Nested", tn_parent=inbox)
+        self.document.add_nested_tags([inbox])
+
+        self.client.patch(
+            f"/api/documents/{self.document.pk}/",
+            {"tags": [nested.pk], "remove_inbox_tags": True},
+            format="json",
+        )
+        self.document.refresh_from_db()
+        tags = set(self.document.tags.values_list("pk", flat=True))
+        assert tags == {inbox.pk, nested.pk}
+
     def test_bulk_edit_respects_hierarchy(self) -> None:
         bulk_edit.add_tag([self.document.pk], self.child.pk)
         self.document.refresh_from_db()

src/documents/tests/test_views.py

@@ -32,6 +32,7 @@ from documents.signals.handlers import update_llm_suggestions_cache
 from documents.tests.utils import DirectoriesMixin
 from documents.tests.utils import read_streaming_response
 from paperless.models import ApplicationConfiguration
+from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 
 
@@ -737,6 +738,38 @@ class TestAISuggestions(DirectoriesMixin, TestCase):
             get_llm_suggestion_cache(self.document.pk, backend="openai-like"),
         )
 
+    @patch("documents.views.get_ai_document_classification")
+    @override_settings(
+        AI_ENABLED=True,
+        LLM_BACKEND="openai-like",
+    )
+    def test_ai_suggestions_with_llm_provider_error(
+        self,
+        mock_get_ai_classification,
+    ) -> None:
+        mock_get_ai_classification.side_effect = LLMProviderError(
+            "confidential provider response",
+        )
+
+        self.client.force_login(user=self.user)
+        response = self.client.get(
+            f"/api/documents/{self.document.pk}/ai_suggestions/",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
+        self.assertEqual(
+            response.json(),
+            {
+                "ai": [
+                    "AI backend rejected the request. Check logs for details.",
+                ],
+            },
+        )
+        self.assertNotIn("confidential provider response", response.content.decode())
+        self.assertIsNone(
+            get_llm_suggestion_cache(self.document.pk, backend="openai-like"),
+        )
+
     @patch("documents.views.get_ai_document_classification")
     @override_settings(
         AI_ENABLED=True,

src/documents/versioning.py

@@ -7,9 +7,12 @@ from typing import Any
 
 from django.db.models import F
 from django.db.models import OuterRef
+from django.db.models import Prefetch
 from django.db.models import QuerySet
 from django.db.models import Subquery
+from django.db.models import Window
 from django.db.models.functions import Coalesce
+from django.db.models.functions import RowNumber
 
 from documents.models import Document
 
@@ -46,6 +49,68 @@ def annotate_effective_content(documents: QuerySet[Document]) -> QuerySet[Docume
     )
 
 
+LATEST_VERSION_CONTENT_PREFETCH_ATTR = "_latest_version_content_prefetch"
+
+
+def latest_version_content_prefetch() -> Prefetch:
+    """
+    A Prefetch for Document.versions scoped to just the newest version's
+    content, for get_effective_content()'s fallback when no SQL annotation
+    is present.
+
+    Deliberately not merged into a metadata-only "versions" prefetch (the one
+    used for the serialized versions list): that one fetches every historical
+    version of every document, and pulling full OCR content for versions
+    nobody will read wastes DB transfer/memory at scale. This one is windowed
+    down to a single row per root, then bounded by Prefetch's own IN-list to
+    whatever page/result set it's attached to -- one cheap bulk query total,
+    not one per document and not one per version.
+    """
+    return Prefetch(
+        "versions",
+        queryset=(
+            Document.objects.filter(
+                root_document_id__isnull=False,
+                deleted_at__isnull=True,
+            )
+            .annotate(
+                rn=Window(
+                    RowNumber(),
+                    partition_by=F("root_document_id"),
+                    order_by=[
+                        F("version_index").desc(nulls_last=True),
+                        F("id").desc(),
+                    ],
+                ),
+            )
+            .filter(rn=1)
+            .only("id", "root_document_id", "content")
+        ),
+        to_attr=LATEST_VERSION_CONTENT_PREFETCH_ATTR,
+    )
+
+
+def has_prefetched_effective_content(document: Document) -> bool:
+    """
+    True if document.get_effective_content() can answer without an extra
+    per-instance query -- an SQL ``effective_content`` annotation, the lean
+    latest_version_content_prefetch(), or the metadata-only "versions"
+    prefetch is already present on the instance.
+
+    Callers that haven't set any of those up (e.g. views that build their
+    own querysets independently of DocumentViewSet.get_queryset(), like
+    TrashView or GlobalSearchView) intentionally don't pay for version-aware
+    content resolution -- see DocumentSerializer.to_representation(), which
+    uses this to decide whether to call get_effective_content() at all.
+    """
+    if hasattr(document, "effective_content"):
+        return True
+    if getattr(document, LATEST_VERSION_CONTENT_PREFETCH_ATTR, None) is not None:
+        return True
+    prefetched_cache = getattr(document, "_prefetched_objects_cache", None)
+    return isinstance(prefetched_cache, dict) and "versions" in prefetched_cache
+
+
 def sort_versions_newest_first(documents: list[Document]) -> list[Document]:
     """
     Same sorting as versions_newest_first()

src/documents/views.py

@@ -36,7 +36,6 @@ from django.db.migrations.recorder import MigrationRecorder
 from django.db.models import Avg
 from django.db.models import Case
 from django.db.models import Count
-from django.db.models import F
 from django.db.models import IntegerField
 from django.db.models import Max
 from django.db.models import Model
@@ -137,12 +136,14 @@ from documents.filters import CustomFieldFilterSet
 from documents.filters import DocumentFilterSet
 from documents.filters import DocumentsOrderingFilter
 from documents.filters import DocumentTypeFilterSet
+from documents.filters import EffectiveContentFilter
 from documents.filters import PaperlessTaskFilterSet
 from documents.filters import PermittedObjectsFilter
 from documents.filters import ShareLinkBundleFilterSet
 from documents.filters import ShareLinkFilterSet
 from documents.filters import StoragePathFilterSet
 from documents.filters import TagFilterSet
+from documents.filters import TitleContentFilter
 from documents.mail import EmailAttachment
 from documents.mail import send_email
 from documents.matching import match_correspondents
@@ -179,7 +180,7 @@ from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
-from documents.permissions import set_permissions_for_object
+from documents.permissions import set_permissions_for_objects
 from documents.permissions import user_is_unrestricted
 from documents.plugins.date_parsing import get_date_parser
 from documents.schema import generate_object_with_permissions_schema
@@ -236,6 +237,7 @@ from documents.versioning import annotate_effective_content
 from documents.versioning import get_latest_version_for_root
 from documents.versioning import get_request_version_param
 from documents.versioning import get_root_document
+from documents.versioning import latest_version_content_prefetch
 from documents.versioning import resolve_requested_version_for_root
 from documents.versioning import versions_newest_first
 from paperless import version
@@ -252,6 +254,7 @@ from paperless.views import StandardPagination
 from paperless_ai.ai_classifier import get_ai_document_classification
 from paperless_ai.ai_classifier import get_llm_output_language
 from paperless_ai.chat import stream_chat_with_documents
+from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 from paperless_ai.matching import extract_unmatched_names
 from paperless_ai.matching import match_correspondents_by_name
@@ -1083,12 +1086,59 @@ class DocumentViewSet(
             ],
         }
 
-    def get_queryset(self):
-        latest_version_content = Subquery(
-            versions_newest_first(
-                Document.objects.filter(root_document=OuterRef("pk")),
-            ).values("content")[:1],
+    @classmethod
+    def _content_filter_params(cls) -> tuple[str, ...]:
+        """
+        Query params whose filtering needs effective_content evaluated in SQL
+        against every candidate row -- see
+        _needs_effective_content_annotation(). Derived rather than
+        hand-maintained so a new content-filtering param counts automatically.
+        """
+        params = [
+            name
+            for name, f in DocumentFilterSet.declared_filters.items()
+            if isinstance(f, (TitleContentFilter, EffectiveContentFilter))
+        ]
+        if "effective_content" in cls.search_fields:
+            params.append(SearchFilter().search_param)
+        return tuple(params)
+
+    def _needs_effective_content_annotation(self) -> bool:
+        # effective_content is a per-row correlated subquery resolving each
+        # document's latest version. Filtering *on* it forces the database to
+        # evaluate it for every candidate row before reaching the LIMIT, which
+        # the root_document_id self-join makes pathological on MariaDB
+        # specifically once real candidate counts get large; otherwise the
+        # "versions" prefetch + Document.get_effective_content() resolves only
+        # the page that survives pagination. Every param here is deprecated in
+        # favor of the Tantivy-backed search endpoint (see filters.py's
+        # TitleContentFilter/EffectiveContentFilter docs), so pay that cost
+        # only when one is actually used. Blank values don't count, matching
+        # how those filters themselves no-op on them -- an empty `?search=`
+        # applies no predicate.
+        params = self.request.query_params
+        return any(
+            params.get(param, "").strip() for param in self._content_filter_params()
         )
+
+    def _requested_fields(self) -> list[str] | None:
+        # The sparse-fieldset `fields` param, as DynamicFieldsModelSerializer
+        # wants it: None means "no restriction, serialize everything", which
+        # a blank value means too. get_queryset() and get_serializer() both
+        # branch on this, and they have to read it identically -- a queryset
+        # that skips the content prefetch for a response that still
+        # serializes content reintroduces get_effective_content()'s
+        # per-instance fallback.
+        fields_param = self.request.query_params.get("fields")
+        return fields_param.split(",") if fields_param else None
+
+    def _needs_effective_content_prefetch(self) -> bool:
+        # The prefetch spares get_effective_content() a per-instance fallback
+        # query, but only earns itself when content can reach the response.
+        fields = self._requested_fields()
+        return fields is None or "content" in fields
+
+    def get_queryset(self):
         # A correlated subquery avoids the LEFT JOIN + Count() this used to
         # be, which forced a GROUP BY aggregate over every matching document
         # before the query could even be sorted or limited.
@@ -1108,40 +1158,43 @@ class DocumentViewSet(
         # ObjectFilter.filter(). A blanket .distinct() here forces the
         # database to fully sort and dedupe every visible document before
         # it can apply LIMIT, which is disastrous at scale.
-        return (
+        prefetches = [
+            Prefetch(
+                "versions",
+                queryset=Document.objects.only(
+                    "id",
+                    "added",
+                    "checksum",
+                    "version_label",
+                    "root_document_id",
+                    "version_index",
+                ),
+            ),
+            "tags",
+            Prefetch(
+                "custom_fields",
+                queryset=CustomFieldInstance.objects.select_related("field"),
+            ),
+            # NotesSerializer nests the author, this avoids query per note
+            Prefetch("notes", queryset=Note.objects.select_related("user")),
+        ]
+        if self._needs_effective_content_prefetch():
+            prefetches.append(latest_version_content_prefetch())
+        queryset = (
             Document.objects.filter(root_document__isnull=True)
             .order_by("-created", "-id")
-            .annotate(effective_content=Coalesce(latest_version_content, F("content")))
             .annotate(num_notes=Coalesce(note_count, 0))
             .select_related("correspondent", "storage_path", "document_type", "owner")
-            .prefetch_related(
-                Prefetch(
-                    "versions",
-                    queryset=Document.objects.only(
-                        "id",
-                        "added",
-                        "checksum",
-                        "version_label",
-                        "root_document_id",
-                        "version_index",
-                    ),
-                ),
-                "tags",
-                Prefetch(
-                    "custom_fields",
-                    queryset=CustomFieldInstance.objects.select_related("field"),
-                ),
-                # NotesSerializer nests the author, this avoids query per note
-                Prefetch("notes", queryset=Note.objects.select_related("user")),
-            )
+            .prefetch_related(*prefetches)
         )
+        if self._needs_effective_content_annotation():
+            queryset = annotate_effective_content(queryset)
+        return queryset
 
     def get_serializer(self, *args, **kwargs):
-        fields_param = self.request.query_params.get("fields", None)
-        fields = fields_param.split(",") if fields_param else None
         truncate_content = self.request.query_params.get("truncate_content", "False")
         kwargs.setdefault("context", self.get_serializer_context())
-        kwargs.setdefault("fields", fields)
+        kwargs.setdefault("fields", self._requested_fields())
         kwargs.setdefault("truncate_content", truncate_content.lower() in ["true", "1"])
         try:
             full_perms = get_boolean(
@@ -1603,6 +1656,22 @@ class DocumentViewSet(
                     {"ai": [_("AI backend request timed out.")]},
                     status=status.HTTP_503_SERVICE_UNAVAILABLE,
                 )
+            except LLMProviderError:
+                logger.exception(
+                    "AI backend rejected the request for document %s",
+                    doc.pk,
+                )
+                return Response(
+                    {
+                        "ai": [
+                            _(
+                                "AI backend rejected the request. "
+                                "Check logs for details.",
+                            ),
+                        ],
+                    },
+                    status=status.HTTP_502_BAD_GATEWAY,
+                )
             set_llm_suggestions_cache(
                 doc.pk,
                 llm_suggestions,
@@ -2311,7 +2380,6 @@ class ChatStreamingView(GenericAPIView[Any]):
     serializer_class = ChatStreamingSerializer
 
     def post(self, request, *args, **kwargs):
-        request.compress_exempt = True
         ai_config = AIConfig()
         if not ai_config.ai_enabled:
             return HttpResponseBadRequest("AI is required for this feature")
@@ -4950,12 +5018,12 @@ class BulkEditObjectsView(PassUserMixin):
                     qs_owner_update.update(owner=owner)
 
                 if "permissions" in serializer.validated_data:
-                    for obj in qs:
-                        set_permissions_for_object(
-                            permissions=permissions,
-                            object=obj,
-                            merge=merge,
-                        )
+                    set_permissions_for_objects(
+                        permissions=permissions,
+                        model=object_class,
+                        pks=qs.values_list("pk", flat=True),
+                        merge=merge,
+                    )
 
             except Exception as e:
                 logger.warning(
@@ -5437,7 +5505,10 @@ class TrashView(ListModelMixin, PassUserMixin):
 
     model = Document
 
-    queryset = Document.deleted_objects.all()
+    # A version is listed separately only when its root is not in the trash.
+    queryset = Document.deleted_objects.exclude(
+        root_document_id__in=Document.deleted_objects.values("id"),
+    )
 
     def get(self, request: Request, format: str | None = None) -> Response:
         self.serializer_class = DocumentSerializer
@@ -5468,7 +5539,15 @@ class TrashView(ListModelMixin, PassUserMixin):
             return HttpResponseForbidden("Insufficient permissions")
         action = serializer.validated_data.get("action")
         if action == "restore":
-            restored = list(Document.deleted_objects.filter(id__in=doc_ids))
+            restored = list(self.get_queryset().filter(id__in=doc_ids))
+            if len(restored) != len(doc_ids):
+                raise ValidationError(
+                    {
+                        "documents": [
+                            "Restore the root document instead of one of its versions.",
+                        ],
+                    },
+                )
             for doc in restored:
                 doc.restore(strict=False)
             if restored:

src/paperless/middleware.py

@@ -1,8 +1,23 @@
+from compression_middleware.middleware import CompressionMiddleware
 from django.conf import settings
 
 from paperless import version
 
 
+class StreamAwareCompressionMiddleware(CompressionMiddleware):
+    """
+    Bypasses compression for server-sent streams (text/event-stream).
+
+    See https://github.com/friedelwolff/django-compression-middleware/pull/7
+    """
+
+    def process_response(self, request, response):
+        content_type = response.headers.get("Content-Type", "")
+        if content_type.startswith("text/event-stream"):
+            return response
+        return super().process_response(request, response)
+
+
 class ApiVersionMiddleware:
     def __init__(self, get_response):
         self.get_response = get_response

src/paperless/settings/__init__.py

@@ -10,7 +10,6 @@ from pathlib import Path
 from typing import Final
 from urllib.parse import urlparse
 
-from compression_middleware.middleware import CompressionMiddleware
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext_lazy as _
 from dotenv import load_dotenv
@@ -96,6 +95,13 @@ MODEL_FILE = get_path_from_env(
     "PAPERLESS_MODEL_FILE",
     DATA_DIR / "classification_model.pickle",
 )
+
+# Minimum confidence (0.0-1.0) for the ML classifier to assign a correspondent,
+# document type, or storage path. 0.0 disables the threshold.
+CLASSIFIER_MATCH_THRESHOLD: Final[float] = get_float_from_env(
+    "PAPERLESS_CLASSIFIER_MATCH_THRESHOLD",
+    0.6,
+)
 LLM_INDEX_DIR = DATA_DIR / "llm_index"
 LLM_INDEX_LOCK = LLM_INDEX_DIR / "index.lock"
 # Cross-process read/write lock guarding the LLM index compaction/migration
@@ -194,22 +200,10 @@ MIDDLEWARE = [
     "allauth.account.middleware.AccountMiddleware",
 ]
 
-# Optional to enable compression
+# Optional to enable compression. The subclass leaves server-sent events
+# uncompressed; see paperless.middleware.StreamAwareCompressionMiddleware.
 if get_bool_from_env("PAPERLESS_ENABLE_COMPRESSION", "yes"):  # pragma: no cover
-    MIDDLEWARE.insert(0, "compression_middleware.middleware.CompressionMiddleware")
-
-# Workaround to not compress streaming responses (e.g. chat).
-# See https://github.com/friedelwolff/django-compression-middleware/pull/7
-original_process_response = CompressionMiddleware.process_response
-
-
-def patched_process_response(self, request, response):
-    if getattr(request, "compress_exempt", False):
-        return response
-    return original_process_response(self, request, response)
-
-
-CompressionMiddleware.process_response = patched_process_response
+    MIDDLEWARE.insert(0, "paperless.middleware.StreamAwareCompressionMiddleware")
 
 ROOT_URLCONF = "paperless.urls"
 

src/paperless/tests/test_compression_middleware.py

@@ -0,0 +1,54 @@
+from django.http import HttpResponse
+from django.http import StreamingHttpResponse
+from django.test import RequestFactory
+from django.test import TestCase
+
+from paperless.middleware import StreamAwareCompressionMiddleware
+
+
+class TestStreamAwareCompressionMiddleware(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.factory = RequestFactory()
+        self.middleware = StreamAwareCompressionMiddleware(lambda request: None)
+
+    def _request(self):
+        return self.factory.get(
+            "/api/documents/chat/",
+            HTTP_ACCEPT_ENCODING="gzip, deflate, br, zstd",
+        )
+
+    def test_event_stream_is_not_compressed(self) -> None:
+        """
+        GIVEN:
+            - A server-sent event response produced chunk by chunk
+        WHEN:
+            - The compression middleware processes it
+        THEN:
+            - It is passed through unencoded, one wire chunk per source chunk
+        """
+        chunks = [f"token{i} ".encode() for i in range(40)]
+        response = StreamingHttpResponse(
+            iter(chunks),
+            content_type="text/event-stream",
+        )
+
+        response = self.middleware.process_response(self._request(), response)
+
+        assert not response.has_header("Content-Encoding")
+        assert list(response.streaming_content) == chunks
+
+    def test_regular_response_is_still_compressed(self) -> None:
+        """
+        GIVEN:
+            - An ordinary response large enough to be worth compressing
+        WHEN:
+            - The compression middleware processes it
+        THEN:
+            - It is compressed as before
+        """
+        response = HttpResponse(b"a" * 5000, content_type="application/json")
+
+        response = self.middleware.process_response(self._request(), response)
+
+        assert response.has_header("Content-Encoding")

src/paperless_ai/ai_classifier.py

@@ -4,21 +4,24 @@ from django.conf import settings
 from django.contrib.auth.models import User
 
 from documents.models import Document
-from documents.permissions import get_objects_for_user_owner_aware
+from documents.permissions import permitted_object_ids
+from documents.permissions import restrict_queryset_to_visible
+from documents.permissions import user_is_unrestricted
 from paperless.config import AIConfig
 from paperless_ai.base_model import ClassificationSuggestions
 from paperless_ai.base_model import TaxonomyChoiceDict
 from paperless_ai.base_model import classification_suggestions_to_model
 from paperless_ai.client import AIClient
 from paperless_ai.db import db_connection_released
-from paperless_ai.indexing import _node_document_ids
 from paperless_ai.indexing import retrieve_similar_nodes
 from paperless_ai.indexing import truncate_content
 from paperless_ai.prompts.context import ClassificationPromptContext
 from paperless_ai.prompts.context import LocalizationPromptContext
 from paperless_ai.prompts.context import RagContextPromptContext
 from paperless_ai.prompts.render import render_prompt
+from paperless_ai.taxonomy import SimilarDocument
 from paperless_ai.taxonomy import TaxonomyCandidates
+from paperless_ai.taxonomy import _node_document_weights
 from paperless_ai.taxonomy import build_taxonomy_candidates
 from paperless_ai.taxonomy import empty_taxonomy_candidates
 from paperless_ai.taxonomy import format_taxonomy_for_prompt
@@ -37,6 +40,48 @@ logger = logging.getLogger("paperless_ai.rag_classifier")
 TAXONOMY_CANDIDATE_TOP_K = 15
 
 
+def _fulltext_similar_documents(
+    document: Document,
+    user: User | None,
+    top_k: int,
+) -> list[SimilarDocument]:
+    """Rank-based fallback when no embedding backend is configured. Uses
+    Tantivy's "More Like This" (term-overlap similarity) instead of vector
+    similarity - cruder, but far better than no candidates at all.
+    more_like_this_ids returns only a ranked ID list, no scores, so weight is
+    synthesized from rank (descending from top_k) rather than claiming a
+    similarity magnitude that doesn't exist. An unrestricted user (none, or an
+    active superuser - see user_is_unrestricted) is normalized to ``None``
+    before calling, since the backend's permission filter has no superuser
+    short-circuit of its own. Results are re-checked with
+    restrict_queryset_to_visible() since Tantivy's indexed permission fields
+    lag the DB via async reindexing.
+    """
+    from documents.search import get_backend
+
+    unrestricted = user_is_unrestricted(user)
+    search_user = None if unrestricted else user
+    backend = get_backend()
+    similar_ids = backend.more_like_this_ids(
+        document.pk,
+        user=search_user,
+        limit=top_k,
+    )
+    if not unrestricted:
+        allowed_ids = set(
+            restrict_queryset_to_visible(
+                Document.objects.filter(pk__in=similar_ids),
+                user,
+                "view_document",
+            ).values_list("pk", flat=True),
+        )
+        similar_ids = [doc_id for doc_id in similar_ids if doc_id in allowed_ids]
+    return [
+        SimilarDocument(document_id=doc_id, weight=float(top_k - rank))
+        for rank, doc_id in enumerate(similar_ids)
+    ]
+
+
 def get_language_name(language_code: str) -> str:
     normalized_language_code = language_code.lower()
     for code, name in settings.LANGUAGES:
@@ -136,43 +181,52 @@ def get_taxonomy_context(
     user: User | None = None,
     max_docs: int = 5,
 ) -> tuple[TaxonomyCandidates, str]:
-    """One retrieval feeds both taxonomy candidates and RAG text context.
-    On any retrieval failure, degrades to empty candidates/context rather than
-    propagating the exception - a vector-store outage should not block
-    classification, only its RAG-assisted enrichment.
+    """One retrieval feeds both taxonomy candidates and RAG text context. Uses
+    vector similarity when an embedding backend is configured, otherwise
+    falls back to Tantivy full-text "More Like This" similarity - see
+    _fulltext_similar_documents. On any retrieval failure, degrades to empty
+    candidates/context rather than propagating the exception - neither a
+    vector-store outage nor a search-index issue should block classification,
+    only its context-assisted enrichment.
     """
+    ai_config = AIConfig()
     try:
-        # None means "no restriction" to retrieve_similar_nodes. A superuser
-        # (like no user at all) can see every document, so skip materializing
-        # every visible pk into a Python list and passing it through as an IN
-        # filter: for a large library that is a wasted quadratic scan in the
-        # vector store at best, and past ~32,763 documents a hard
-        # sqlite3.OperationalError (SQLite's bound-parameter limit) at worst.
-        # get_objects_for_user_owner_aware() would return every Document for a
-        # superuser anyway (guardian's own with_superuser shortcut), so this
-        # changes nothing about which documents are considered -- only how we
-        # get there.
-        visible_document_ids = (
-            None
-            if user is None or user.is_superuser
-            else list(
-                get_objects_for_user_owner_aware(
-                    user,
-                    "view_document",
-                    Document,
-                ).values_list("pk", flat=True),
+        if ai_config.llm_embedding_backend:
+            # None means "no restriction" to retrieve_similar_nodes. An
+            # unrestricted user (no user at all, or an active superuser -- see
+            # user_is_unrestricted) can see every document, so skip
+            # materializing every visible pk into a Python list and passing it
+            # through as an IN filter: for a large library that is a wasted
+            # quadratic scan in the vector store at best, and past ~32,763
+            # documents a hard sqlite3.OperationalError (SQLite's
+            # bound-parameter limit) at worst.
+            # permitted_object_ids() has its own superuser shortcut that would
+            # return every Document's id anyway, so this changes nothing about
+            # which documents are considered -- only how we get there.
+            visible_document_ids = (
+                None
+                if user_is_unrestricted(user)
+                else list(permitted_object_ids(user, Document, "view_document"))
+            )
+            nodes = retrieve_similar_nodes(
+                document,
+                top_k=TAXONOMY_CANDIDATE_TOP_K,
+                document_ids=visible_document_ids,
+            )
+            similar_documents = _node_document_weights(nodes)
+        else:
+            # See _fulltext_similar_documents: it applies its own permission
+            # filter via `user`, so no visible-document-id list is needed here.
+            similar_documents = _fulltext_similar_documents(
+                document,
+                user,
+                top_k=TAXONOMY_CANDIDATE_TOP_K,
             )
-        )
-        nodes = retrieve_similar_nodes(
-            document,
-            top_k=TAXONOMY_CANDIDATE_TOP_K,
-            document_ids=visible_document_ids,
-        )
 
-        candidates = build_taxonomy_candidates(nodes, user)
+        candidates = build_taxonomy_candidates(similar_documents, user)
 
-        # ``nodes`` are already ordered by descending vector similarity; don't lose it.
-        similar_document_ids = list(dict.fromkeys(_node_document_ids(nodes)))
+        # similar_documents is already ordered by descending weight; don't lose it.
+        similar_document_ids = [s["document_id"] for s in similar_documents]
         similar_documents_by_id = Document.objects.in_bulk(similar_document_ids)
         similar_docs = [
             similar_documents_by_id[document_id]
@@ -186,8 +240,8 @@ def get_taxonomy_context(
             context_blocks.append(f"TITLE: {title}\n{text}")
     except Exception:
         logger.exception(
-            "Failed to retrieve RAG neighbours for document %s; continuing "
-            "without taxonomy candidates or similar-document context.",
+            "Failed to retrieve similar-document context for document %s; "
+            "continuing without taxonomy candidates or similar-document context.",
             document.pk,
         )
         return empty_taxonomy_candidates(), ""
@@ -241,17 +295,13 @@ def get_ai_document_classification(
 ) -> ClassificationSuggestions:
     ai_config = AIConfig()
 
-    if ai_config.llm_embedding_backend:
-        candidates, context = get_taxonomy_context(document, user)
-        prompt = build_prompt_with_rag(
-            document,
-            ai_config,
-            candidates=candidates,
-            context=context,
-        )
-    else:
-        candidates = empty_taxonomy_candidates()
-        prompt = build_prompt_without_rag(document, ai_config, candidates=candidates)
+    candidates, context = get_taxonomy_context(document, user)
+    prompt = build_prompt_with_rag(
+        document,
+        ai_config,
+        candidates=candidates,
+        context=context,
+    )
 
     client = AIClient()
     # Hand the pooled DB connection back while the (slow) LLM query runs so it

src/paperless_ai/client.py

@@ -22,6 +22,7 @@ from paperless.network import validate_outbound_http_url
 from paperless_ai.base_model import ClassificationSuggestions
 from paperless_ai.base_model import DocumentClassifierSchema
 from paperless_ai.base_model import model_to_classification_suggestions
+from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 
 logger = logging.getLogger("paperless_ai.client")
@@ -39,7 +40,6 @@ LLM_SYSTEM_PROMPT = (
 
 # openai-python rejects empty keys since 2.34.0, "fake" is the stand-in from
 # llama-index's own OpenAILike docs https://docs.llamaindex.ai/en/stable/api_reference/llms/openai_like/
-# TODO: remove pending resolution of https://github.com/openai/openai-python/issues/3224
 PLACEHOLDER_API_KEY: Final = "fake"
 
 
@@ -132,7 +132,7 @@ class AIClient:
         from llama_index.core.llms import ChatMessage
 
         if self.settings.llm_backend == LLMBackend.OLLAMA:
-            with self._normalize_timeouts():
+            with self._normalize_errors():
                 result = self.llm.chat(
                     [ChatMessage(role="user", content=prompt)],
                     format=DocumentClassifierSchema.model_json_schema(),
@@ -153,7 +153,7 @@ class AIClient:
             content=f"{prompt}\n\n"
             f"Answer by calling the {tool.metadata.name} tool. Do not write the answer as text.",
         )
-        with self._normalize_timeouts():
+        with self._normalize_errors():
             result = self.llm.chat_with_tools(
                 tools=[tool],
                 user_msg=user_msg,
@@ -173,7 +173,7 @@ class AIClient:
         )
 
     @contextmanager
-    def _normalize_timeouts(self) -> Iterator[None]:
+    def _normalize_errors(self) -> Iterator[None]:
         try:
             yield
         except httpx.TimeoutException as exc:
@@ -181,8 +181,23 @@ class AIClient:
         except Exception as exc:
             if self._is_openai_timeout(exc):
                 raise LLMTimeoutError from exc
+            if self._is_provider_error(exc):
+                raise LLMProviderError from exc
             raise
 
+    def _is_provider_error(self, exc: Exception) -> bool:
+        if self.settings.llm_backend == LLMBackend.OLLAMA:
+            from ollama import ResponseError
+
+            return isinstance(exc, ResponseError)
+
+        if self.settings.llm_backend == LLMBackend.OPENAI_LIKE:
+            from openai import APIStatusError
+
+            return isinstance(exc, APIStatusError)
+
+        return False
+
     def _is_openai_timeout(self, exc: Exception) -> bool:
         if self.settings.llm_backend != LLMBackend.OPENAI_LIKE:
             return False

src/paperless_ai/exceptions.py

@@ -1,2 +1,6 @@
 class LLMTimeoutError(Exception):
     pass
+
+
+class LLMProviderError(Exception):
+    """The LLM backend rejected the request."""

src/paperless_ai/indexing.py

@@ -721,20 +721,3 @@ def retrieve_similar_nodes(
             continue
         filtered.append(node)
     return filtered
-
-
-def _node_document_ids(nodes: list["NodeWithScore"]) -> list[int]:
-    document_ids: list[int] = []
-    for node in nodes:
-        document_id = node.metadata.get("document_id")
-        if document_id is None:  # pragma: no cover
-            # See the matching guard in retrieve_similar_nodes() above.
-            continue
-        try:
-            document_ids.append(int(document_id))
-        except ValueError:  # pragma: no cover
-            logger.warning(
-                "Skipping LLM index result with invalid document_id %r.",
-                document_id,
-            )
-    return document_ids

src/paperless_ai/taxonomy.py

@@ -31,6 +31,11 @@ class TaxonomyCandidate(TypedDict):
     weight: float
 
 
+class SimilarDocument(TypedDict):
+    document_id: int
+    weight: float
+
+
 class TaxonomyCandidates(TypedDict):
     tags: list[TaxonomyCandidate]
     document_types: list[TaxonomyCandidate]
@@ -49,10 +54,10 @@ def empty_taxonomy_candidates() -> TaxonomyCandidates:
     )
 
 
-def _node_document_weights(nodes: list["NodeWithScore"]) -> dict[int, float]:
-    """document_id -> that node's similarity score, summed if a document_id
-    appears more than once across the retrieved nodes (e.g. multiple chunks
-    of the same source document)."""
+def _node_document_weights(nodes: list["NodeWithScore"]) -> list[SimilarDocument]:
+    """Sum each node's similarity score into its document_id (a document can
+    appear via multiple chunks/nodes) and return one SimilarDocument per
+    distinct document_id."""
     weights: dict[int, float] = defaultdict(float)
     for node in nodes:
         document_id = node.metadata.get("document_id")
@@ -65,7 +70,14 @@ def _node_document_weights(nodes: list["NodeWithScore"]) -> dict[int, float]:
             weights[int(document_id)] += float(node.score or 0.0)
         except (TypeError, ValueError):  # pragma: no cover
             continue
-    return weights
+    return sorted(
+        (
+            SimilarDocument(document_id=document_id, weight=weight)
+            for document_id, weight in weights.items()
+        ),
+        key=lambda similar: similar["weight"],
+        reverse=True,
+    )
 
 
 def _visible_ranked_candidates(
@@ -101,21 +113,26 @@ def _visible_ranked_candidates(
 
 
 def build_taxonomy_candidates(
-    nodes: list["NodeWithScore"],
+    similar_documents: list[SimilarDocument],
     user: User | None,
 ) -> TaxonomyCandidates:
-    """Resolve each neighbour node's document_id to a live Document, read its
-    *current* tags/type/correspondent/storage_path via the ORM (never the
-    possibly-stale names cached in vector-index node metadata), weight each
-    distinct taxonomy object by aggregate neighbour similarity, permission-filter
+    """Resolve each similar document's id to a live Document, read its
+    *current* tags/type/correspondent/storage_path via the ORM (never any
+    possibly-stale names an adapter's source might have cached), weight each
+    distinct taxonomy object by aggregate similarity weight, permission-filter
     against what ``user`` can see, and return each category ranked by weight
-    and capped.
+    and capped. ``similar_documents`` may come from either the vector-RAG
+    adapter or the full-text fallback adapter - both produce this same shape.
     """
-
-    document_weights = _node_document_weights(nodes)
-    if not document_weights:
+    if not similar_documents:
         return empty_taxonomy_candidates()
 
+    # Both adapters guarantee at most one SimilarDocument per document_id, so
+    # this never silently drops a duplicate's weight.
+    document_weights: dict[int, float] = {
+        s["document_id"]: s["weight"] for s in similar_documents
+    }
+
     # Only .tags.all() needs prefetching (a reverse M2M, one extra query for
     # the whole batch). document_type/correspondent/storage_path are read
     # below via their *_id columns (neighbour.document_type_id, etc.), which

src/paperless_ai/tests/test_ai_classifier.py

@@ -1,4 +1,5 @@
 import datetime
+from collections.abc import Generator
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 from unittest.mock import patch
@@ -6,18 +7,24 @@ from unittest.mock import patch
 import pytest
 import pytest_mock
 from django.test import override_settings
+from guardian.shortcuts import assign_perm
+from guardian.shortcuts import remove_perm
 
 from documents.models import Document
+from documents.search import TantivyBackend
 from documents.tests.factories import DocumentFactory
 from documents.tests.factories import TagFactory
 from documents.tests.factories import UserFactory
 from paperless.config import AIConfig
+from paperless_ai.ai_classifier import TAXONOMY_CANDIDATE_TOP_K
+from paperless_ai.ai_classifier import _fulltext_similar_documents
 from paperless_ai.ai_classifier import build_localization_prompt
 from paperless_ai.ai_classifier import build_prompt_with_rag
 from paperless_ai.ai_classifier import build_prompt_without_rag
 from paperless_ai.ai_classifier import get_ai_document_classification
 from paperless_ai.ai_classifier import get_language_name
 from paperless_ai.ai_classifier import get_taxonomy_context
+from paperless_ai.taxonomy import SimilarDocument
 from paperless_ai.taxonomy import TaxonomyCandidate
 from paperless_ai.taxonomy import TaxonomyCandidates
 
@@ -220,12 +227,10 @@ def test_use_rag_if_configured(
 
 @pytest.mark.django_db
 @patch("paperless_ai.client.AIClient.run_llm_query")
-@patch("paperless_ai.ai_classifier.build_prompt_without_rag")
-@patch("paperless_ai.ai_classifier.AIConfig")
+@patch("paperless_ai.ai_classifier.build_prompt_with_rag")
 @override_settings(LLM_BACKEND="ollama", LLM_MODEL="some_model")
-def test_use_without_rag_if_not_configured(
-    mock_ai_config,
-    mock_build_prompt_without_rag,
+def test_use_rag_prompt_even_without_embedding_backend(
+    mock_build_prompt_with_rag,
     mock_run_llm_query,
     mock_document,
 ):
@@ -235,13 +240,13 @@ def test_use_without_rag_if_not_configured(
     WHEN:
         - get_ai_document_classification() is called
     THEN:
-        - The non-RAG prompt builder is used
+        - The RAG-context prompt builder is still used (fed by the full-text
+          fallback's context/candidates instead of the vector store's)
     """
-    mock_ai_config.return_value.llm_embedding_backend = None
-    mock_build_prompt_without_rag.return_value = "Prompt without RAG"
+    mock_build_prompt_with_rag.return_value = "Prompt with RAG"
     mock_run_llm_query.return_value = NESTED_SUGGESTIONS
     get_ai_document_classification(mock_document)
-    mock_build_prompt_without_rag.assert_called_once()
+    mock_build_prompt_with_rag.assert_called_once()
 
 
 @pytest.mark.django_db
@@ -320,6 +325,7 @@ def test_build_localization_prompt_preserves_unicode_characters():
 
 
 @pytest.mark.django_db
+@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
 def test_get_taxonomy_context_assembles_rag_text_and_candidates():
     """
     GIVEN:
@@ -354,6 +360,7 @@ def test_get_taxonomy_context_assembles_rag_text_and_candidates():
 
 
 @pytest.mark.django_db
+@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
 def test_get_taxonomy_context_preserves_similarity_order_and_distinct_documents():
     """
     GIVEN:
@@ -424,6 +431,7 @@ def test_get_taxonomy_context_preserves_similarity_order_and_distinct_documents(
 
 
 @pytest.mark.django_db
+@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
 def test_get_taxonomy_context_no_similar_docs():
     """
     GIVEN:
@@ -447,6 +455,67 @@ def test_get_taxonomy_context_no_similar_docs():
     }
 
 
+@pytest.mark.django_db
+def test_get_taxonomy_context_uses_fulltext_fallback_when_no_embedding_backend(
+    mocker: pytest_mock.MockerFixture,
+) -> None:
+    """
+    GIVEN:
+        - No LLM embedding backend is configured (the default test settings)
+    WHEN:
+        - get_taxonomy_context() is called
+    THEN:
+        - _fulltext_similar_documents() is called with the document, the user
+          and TAXONOMY_CANDIDATE_TOP_K
+        - retrieve_similar_nodes() (the vector path) is never called
+    """
+    document = DocumentFactory.create(content="Some content")
+    mock_fulltext = mocker.patch(
+        "paperless_ai.ai_classifier._fulltext_similar_documents",
+        return_value=[],
+    )
+    mock_retrieve = mocker.patch("paperless_ai.ai_classifier.retrieve_similar_nodes")
+
+    get_taxonomy_context(document, user=None)
+
+    mock_fulltext.assert_called_once_with(
+        document,
+        None,
+        top_k=TAXONOMY_CANDIDATE_TOP_K,
+    )
+    mock_retrieve.assert_not_called()
+
+
+@pytest.mark.django_db
+@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
+def test_get_taxonomy_context_uses_vector_path_when_embedding_backend_configured(
+    mocker: pytest_mock.MockerFixture,
+) -> None:
+    """
+    GIVEN:
+        - An LLM embedding backend is configured
+    WHEN:
+        - get_taxonomy_context() is called
+    THEN:
+        - retrieve_similar_nodes() (the vector path) is called
+        - _fulltext_similar_documents() (the no-embedding-backend fallback)
+          is never called
+    """
+    document = DocumentFactory.create(content="Some content")
+    mock_retrieve = mocker.patch(
+        "paperless_ai.ai_classifier.retrieve_similar_nodes",
+        return_value=[],
+    )
+    mock_fulltext = mocker.patch(
+        "paperless_ai.ai_classifier._fulltext_similar_documents",
+    )
+
+    get_taxonomy_context(document, user=None)
+
+    mock_retrieve.assert_called_once()
+    mock_fulltext.assert_not_called()
+
+
 class TestGetTaxonomyContextVisibility:
     """get_taxonomy_context must not materialize every visible document id
     for a user who can already see the whole library: a superuser (like no
@@ -459,6 +528,7 @@ class TestGetTaxonomyContextVisibility:
     """
 
     @pytest.mark.django_db
+    @override_settings(LLM_EMBEDDING_BACKEND="huggingface")
     def test_skips_permission_lookup_for_superuser(
         self,
         mocker: pytest_mock.MockerFixture,
@@ -477,17 +547,18 @@ class TestGetTaxonomyContextVisibility:
             "paperless_ai.ai_classifier.retrieve_similar_nodes",
             return_value=[],
         )
-        mock_get_objects = mocker.patch(
-            "paperless_ai.ai_classifier.get_objects_for_user_owner_aware",
+        mock_permitted = mocker.patch(
+            "paperless_ai.ai_classifier.permitted_object_ids",
         )
         user = UserFactory.create(is_superuser=True)
 
         get_taxonomy_context(document, user)
 
-        mock_get_objects.assert_not_called()
+        mock_permitted.assert_not_called()
         assert mock_retrieve.call_args.kwargs["document_ids"] is None
 
     @pytest.mark.django_db
+    @override_settings(LLM_EMBEDDING_BACKEND="huggingface")
     def test_skips_permission_lookup_when_no_user(
         self,
         mocker: pytest_mock.MockerFixture,
@@ -506,16 +577,17 @@ class TestGetTaxonomyContextVisibility:
             "paperless_ai.ai_classifier.retrieve_similar_nodes",
             return_value=[],
         )
-        mock_get_objects = mocker.patch(
-            "paperless_ai.ai_classifier.get_objects_for_user_owner_aware",
+        mock_permitted = mocker.patch(
+            "paperless_ai.ai_classifier.permitted_object_ids",
         )
 
         get_taxonomy_context(document, None)
 
-        mock_get_objects.assert_not_called()
+        mock_permitted.assert_not_called()
         assert mock_retrieve.call_args.kwargs["document_ids"] is None
 
     @pytest.mark.django_db
+    @override_settings(LLM_EMBEDDING_BACKEND="huggingface")
     def test_restricts_to_visible_documents_for_non_superuser(
         self,
         mocker: pytest_mock.MockerFixture,
@@ -526,7 +598,7 @@ class TestGetTaxonomyContextVisibility:
         WHEN:
             - get_taxonomy_context() is called
         THEN:
-            - The user's visible document ids are looked up and passed to
+            - The user's permitted document ids are looked up and passed to
               retrieve_similar_nodes() as a restriction
         """
         document = DocumentFactory.create(content="Some content")
@@ -534,21 +606,232 @@ class TestGetTaxonomyContextVisibility:
             "paperless_ai.ai_classifier.retrieve_similar_nodes",
             return_value=[],
         )
-        mock_queryset = mocker.MagicMock()
-        mock_queryset.values_list.return_value = [1, 2, 3]
-        mock_get_objects = mocker.patch(
-            "paperless_ai.ai_classifier.get_objects_for_user_owner_aware",
-            return_value=mock_queryset,
+        mock_permitted = mocker.patch(
+            "paperless_ai.ai_classifier.permitted_object_ids",
+            return_value=[1, 2, 3],
         )
         user = UserFactory.create(is_superuser=False)
 
         get_taxonomy_context(document, user)
 
-        mock_get_objects.assert_called_once_with(user, "view_document", Document)
+        mock_permitted.assert_called_once_with(user, Document, "view_document")
         assert mock_retrieve.call_args.kwargs["document_ids"] == [1, 2, 3]
 
 
 @pytest.mark.django_db
+class TestFulltextSimilarDocuments:
+    """_fulltext_similar_documents is the no-embedding-backend fallback: it
+    asks the Tantivy full-text index for "More Like This" neighbours instead
+    of the vector store, and synthesizes a rank-based weight since Tantivy's
+    more_like_this_ids returns only an ordered id list, no scores.
+    """
+
+    @pytest.fixture
+    def fulltext_backend(
+        self,
+        mocker: pytest_mock.MockerFixture,
+    ) -> Generator[TantivyBackend, None, None]:
+        """An in-memory Tantivy backend, wired up as the module-level
+        singleton _fulltext_similar_documents resolves via get_backend()."""
+        backend = TantivyBackend(path=None)
+        backend.open()
+        mocker.patch("documents.search.get_backend", return_value=backend)
+        try:
+            yield backend
+        finally:
+            backend.close()
+
+    def test_ranks_by_rank_based_weight_descending(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A source document and two similar documents indexed in Tantivy
+        WHEN:
+            - _fulltext_similar_documents() is called
+        THEN:
+            - Each result's weight reflects its rank (first result weighted
+              higher than the second), not a raw similarity score
+        """
+        source = DocumentFactory.create(content="quarterly financial report details")
+        first = DocumentFactory.create(content="quarterly financial report details")
+        second = DocumentFactory.create(content="financial report")
+        for doc in (source, first, second):
+            fulltext_backend.add_or_update(doc)
+
+        result = _fulltext_similar_documents(source, user=None, top_k=5)
+
+        assert len(result) == 2
+        weight_by_id = {s["document_id"]: s["weight"] for s in result}
+        assert weight_by_id[first.pk] > weight_by_id[second.pk]
+
+    def test_excludes_source_document(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A source document indexed in Tantivy with no other documents
+        WHEN:
+            - _fulltext_similar_documents() is called
+        THEN:
+            - An empty list is returned - the source document is never its
+              own similar document
+        """
+        source = DocumentFactory.create(content="unique unrelated content")
+        fulltext_backend.add_or_update(source)
+
+        result = _fulltext_similar_documents(source, user=None, top_k=5)
+
+        assert result == []
+
+    def test_empty_index_returns_empty_list(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A document that has never been indexed (fresh/empty Tantivy index)
+        WHEN:
+            - _fulltext_similar_documents() is called
+        THEN:
+            - An empty list is returned rather than raising
+        """
+        source = DocumentFactory.create(content="never indexed")
+
+        result = _fulltext_similar_documents(source, user=None, top_k=5)
+
+        assert result == []
+
+    def test_respects_top_k_limit(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A source document and four similar documents indexed
+        WHEN:
+            - _fulltext_similar_documents() is called with top_k=2
+        THEN:
+            - At most 2 results are returned
+        """
+        source = DocumentFactory.create(content="shared overlapping keyword text")
+        fulltext_backend.add_or_update(source)
+        for _ in range(4):
+            fulltext_backend.add_or_update(
+                DocumentFactory.create(content="shared overlapping keyword text"),
+            )
+
+        result = _fulltext_similar_documents(source, user=None, top_k=2)
+
+        assert len(result) == 2
+
+    def test_result_shape_is_similar_document(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A source document and one similar document indexed
+        WHEN:
+            - _fulltext_similar_documents() is called
+        THEN:
+            - Each result is a SimilarDocument (document_id + weight only)
+        """
+        source = DocumentFactory.create(content="shared content phrase")
+        other = DocumentFactory.create(content="shared content phrase")
+        fulltext_backend.add_or_update(source)
+        fulltext_backend.add_or_update(other)
+
+        result = _fulltext_similar_documents(source, user=None, top_k=5)
+
+        # rank 0 (the only/best result) with top_k=5 -> weight = top_k - rank = 5.0,
+        # per the "first result gets top_k, the last gets 1" formula.
+        assert result == [SimilarDocument(document_id=other.pk, weight=5.0)]
+
+    def test_superuser_sees_other_users_documents(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A source document owned by one user and a similar document
+              owned by a different user, with no sharing between them
+        WHEN:
+            - _fulltext_similar_documents() is called with a superuser
+        THEN:
+            - The other user's document is still returned as a similar
+              document - a superuser must not be narrowed by the backend's
+              owner-based permission filter
+        """
+        owner = UserFactory.create()
+        other_owner = UserFactory.create()
+        superuser = UserFactory.create(is_superuser=True)
+        source = DocumentFactory.create(
+            content="shared content phrase",
+            owner=owner,
+        )
+        other = DocumentFactory.create(
+            content="shared content phrase",
+            owner=other_owner,
+        )
+        fulltext_backend.add_or_update(source)
+        fulltext_backend.add_or_update(other)
+
+        result = _fulltext_similar_documents(source, user=superuser, top_k=5)
+
+        assert [s["document_id"] for s in result] == [other.pk]
+
+    def test_excludes_stale_permitted_document_for_regular_user(
+        self,
+        fulltext_backend: TantivyBackend,
+    ) -> None:
+        """
+        GIVEN:
+            - A regular (non-superuser) user
+            - A similar document the user is permitted to view, and another
+              similar document indexed while the user still had view
+              permission but which has since had that permission revoked in
+              the database, i.e. the Tantivy index has stale permission data
+        WHEN:
+            - _fulltext_similar_documents() is called with that user
+        THEN:
+            - Only the still-permitted document is returned - the DB
+              re-check via restrict_queryset_to_visible() must catch the
+              document Tantivy's stale index still thinks is visible
+        """
+        owner = UserFactory.create()
+        viewer = UserFactory.create(is_superuser=False)
+        source = DocumentFactory.create(
+            content="shared content phrase",
+            owner=owner,
+        )
+        permitted = DocumentFactory.create(
+            content="shared content phrase",
+            owner=owner,
+        )
+        now_private = DocumentFactory.create(
+            content="shared content phrase",
+            owner=owner,
+        )
+        assign_perm("view_document", viewer, permitted)
+        assign_perm("view_document", viewer, now_private)
+        fulltext_backend.add_or_update(source)
+        fulltext_backend.add_or_update(permitted)
+        fulltext_backend.add_or_update(now_private)
+
+        # Revoke access after indexing, without reindexing: the index still
+        # carries viewer as a permitted viewer for `now_private`.
+        remove_perm("view_document", viewer, now_private)
+
+        result = _fulltext_similar_documents(source, user=viewer, top_k=5)
+
+        assert [s["document_id"] for s in result] == [permitted.pk]
+
+
+@pytest.mark.django_db
+@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
 @patch("paperless_ai.ai_classifier.retrieve_similar_nodes")
 def test_get_taxonomy_context_retrieval_failure_degrades_to_no_hints(mock_retrieve):
     """
@@ -575,6 +858,7 @@ def test_get_taxonomy_context_retrieval_failure_degrades_to_no_hints(mock_retrie
 
 
 @pytest.mark.django_db
+@override_settings(LLM_EMBEDDING_BACKEND="huggingface")
 @patch("paperless_ai.ai_classifier.build_taxonomy_candidates")
 @patch("paperless_ai.ai_classifier.retrieve_similar_nodes")
 def test_get_taxonomy_context_candidate_building_failure_degrades_to_no_hints(

src/paperless_ai/tests/test_ai_indexing.py

@@ -1188,9 +1188,7 @@ class TestRetrieveSimilarNodesAgainstRealIndex:
 
         nodes = indexing.retrieve_similar_nodes(a, document_ids=[b.id])
 
-        assert all(
-            document_id == b.id for document_id in indexing._node_document_ids(nodes)
-        )
+        assert all(int(node.metadata["document_id"]) == b.id for node in nodes)
 
     def test_excludes_self(
         self,
@@ -1212,7 +1210,7 @@ class TestRetrieveSimilarNodesAgainstRealIndex:
 
         nodes = indexing.retrieve_similar_nodes(a, top_k=5)
 
-        assert set(indexing._node_document_ids(nodes)) == {b.id}
+        assert {int(node.metadata["document_id"]) for node in nodes} == {b.id}
 
     def test_excludes_self_with_multiple_chunks(
         self,
@@ -1235,4 +1233,4 @@ class TestRetrieveSimilarNodesAgainstRealIndex:
 
         nodes = indexing.retrieve_similar_nodes(a, top_k=3)
 
-        assert set(indexing._node_document_ids(nodes)) == {b.id}
+        assert {int(node.metadata["document_id"]) for node in nodes} == {b.id}

src/paperless_ai/tests/test_client.py

@@ -4,6 +4,7 @@ from unittest.mock import MagicMock
 from unittest.mock import patch
 
 import httpx
+import ollama
 import openai
 import pytest
 from llama_index.core.llms.llm import ToolSelection
@@ -11,6 +12,7 @@ from llama_index.core.llms.llm import ToolSelection
 from paperless_ai.client import LLM_SYSTEM_PROMPT
 from paperless_ai.client import PLACEHOLDER_API_KEY
 from paperless_ai.client import AIClient
+from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 
 
@@ -214,6 +216,52 @@ def test_run_llm_query_openai_timeout_raises_local_error(
         client.run_llm_query("test_prompt")
 
 
+def test_run_llm_query_openai_status_error_raises_provider_error(
+    mock_ai_config,
+    mock_openai_llm,
+):
+    mock_ai_config.llm_backend = "openai-like"
+    mock_ai_config.llm_model = "test_model"
+    mock_ai_config.llm_endpoint = "http://test-url"
+
+    request = httpx.Request("POST", "http://test-url/v1/chat/completions")
+    body = {"error": {"message": "Thinking mode does not support this tool_choice"}}
+    mock_openai_llm.return_value.chat_with_tools.side_effect = openai.BadRequestError(
+        "Error code: 400",
+        response=httpx.Response(400, request=request, json=body),
+        body=body,
+    )
+
+    client = AIClient()
+
+    with pytest.raises(LLMProviderError) as exc_info:
+        client.run_llm_query("test_prompt")
+    assert str(exc_info.value) == ""
+    assert isinstance(exc_info.value.__cause__, openai.BadRequestError)
+
+
+def test_run_llm_query_ollama_response_error_raises_provider_error(
+    mock_ai_config,
+    mock_ollama_llm,
+):
+    mock_ai_config.llm_backend = "ollama"
+    mock_ai_config.llm_model = "test_model"
+    mock_ai_config.llm_endpoint = "http://test-url"
+
+    response_error = ollama.ResponseError(
+        "confidential provider response",
+        status_code=400,
+    )
+    mock_ollama_llm.return_value.chat.side_effect = response_error
+
+    client = AIClient()
+
+    with pytest.raises(LLMProviderError) as exc_info:
+        client.run_llm_query("test_prompt")
+    assert str(exc_info.value) == ""
+    assert exc_info.value.__cause__ is response_error
+
+
 def test_run_llm_query_httpx_timeout_raises_local_error(
     mock_ai_config,
     mock_ollama_llm,

src/paperless_ai/tests/test_taxonomy.py

@@ -1,5 +1,4 @@
 import json
-from types import SimpleNamespace
 
 import pytest
 import pytest_mock
@@ -10,14 +9,14 @@ from documents.tests.factories import DocumentTypeFactory
 from documents.tests.factories import StoragePathFactory
 from documents.tests.factories import TagFactory
 from documents.tests.factories import UserFactory
+from paperless_ai.taxonomy import SimilarDocument
 from paperless_ai.taxonomy import TaxonomyCandidates
 from paperless_ai.taxonomy import build_taxonomy_candidates
 from paperless_ai.taxonomy import format_taxonomy_for_prompt
 
 
-def make_node(document_id: int, score: float) -> SimpleNamespace:
-    """A stand-in for NodeWithScore: only ``.metadata``/``.score`` are read."""
-    return SimpleNamespace(metadata={"document_id": str(document_id)}, score=score)
+def make_similar(document_id: int, weight: float) -> SimilarDocument:
+    return SimilarDocument(document_id=document_id, weight=weight)
 
 
 @pytest.mark.django_db
@@ -53,9 +52,9 @@ class TestBuildTaxonomyCandidates:
         doc_a.tags.add(tag)
         doc_b = DocumentFactory.create()
         doc_b.tags.add(tag)
-        nodes = [make_node(doc_a.pk, 0.9), make_node(doc_b.pk, 0.4)]
+        similar_documents = [make_similar(doc_a.pk, 0.9), make_similar(doc_b.pk, 0.4)]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["tags"]) == 1
         assert result["tags"][0]["id"] == tag.pk
@@ -80,9 +79,9 @@ class TestBuildTaxonomyCandidates:
         document.tags.add(tag)
         tag.name = "New Name"
         tag.save()
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert result["tags"][0]["name"] == "New Name"
 
@@ -102,9 +101,9 @@ class TestBuildTaxonomyCandidates:
         document = DocumentFactory.create()
         document.tags.add(tag)
         tag.delete()
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert result["tags"] == []
 
@@ -123,9 +122,12 @@ class TestBuildTaxonomyCandidates:
         strong_doc.tags.add(strong_tag)
         weak_doc = DocumentFactory.create()
         weak_doc.tags.add(weak_tag)
-        nodes = [make_node(strong_doc.pk, 0.9), make_node(weak_doc.pk, 0.1)]
+        similar_documents = [
+            make_similar(strong_doc.pk, 0.9),
+            make_similar(weak_doc.pk, 0.1),
+        ]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert [c["name"] for c in result["tags"]] == ["Strong", "Weak"]
 
@@ -141,9 +143,9 @@ class TestBuildTaxonomyCandidates:
         document = DocumentFactory.create()
         for i in range(15):
             document.tags.add(TagFactory.create(name=f"Tag{i}"))
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["tags"]) == 10
 
@@ -157,12 +159,12 @@ class TestBuildTaxonomyCandidates:
             - Only 5 correspondents are returned
         """
         correspondents = CorrespondentFactory.create_batch(7)
-        nodes = [
-            make_node(DocumentFactory.create(correspondent=c).pk, 0.5)
+        similar_documents = [
+            make_similar(DocumentFactory.create(correspondent=c).pk, 0.5)
             for c in correspondents
         ]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["correspondents"]) == 5
 
@@ -177,9 +179,9 @@ class TestBuildTaxonomyCandidates:
         """
         document_type = DocumentTypeFactory.create(name="Invoice")
         document = DocumentFactory.create(document_type=document_type)
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["document_types"]) == 1
         assert result["document_types"][0]["id"] == document_type.pk
@@ -195,12 +197,12 @@ class TestBuildTaxonomyCandidates:
             - Only 5 document_types are returned
         """
         document_types = DocumentTypeFactory.create_batch(7)
-        nodes = [
-            make_node(DocumentFactory.create(document_type=dt).pk, 0.5)
+        similar_documents = [
+            make_similar(DocumentFactory.create(document_type=dt).pk, 0.5)
             for dt in document_types
         ]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["document_types"]) == 5
 
@@ -215,9 +217,9 @@ class TestBuildTaxonomyCandidates:
         """
         storage_path = StoragePathFactory.create(name="Invoices")
         document = DocumentFactory.create(storage_path=storage_path)
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["storage_paths"]) == 1
         assert result["storage_paths"][0]["id"] == storage_path.pk
@@ -233,12 +235,12 @@ class TestBuildTaxonomyCandidates:
             - Only 5 storage_paths are returned
         """
         storage_paths = StoragePathFactory.create_batch(7)
-        nodes = [
-            make_node(DocumentFactory.create(storage_path=sp).pk, 0.5)
+        similar_documents = [
+            make_similar(DocumentFactory.create(storage_path=sp).pk, 0.5)
             for sp in storage_paths
         ]
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert len(result["storage_paths"]) == 5
 
@@ -258,14 +260,14 @@ class TestBuildTaxonomyCandidates:
         tag = TagFactory.create(name="Restricted")
         document = DocumentFactory.create()
         document.tags.add(tag)
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
         user = UserFactory.create()
         mocker.patch(
             "documents.permissions.permitted_object_ids",
             return_value=[],  # user cannot see this tag
         )
 
-        result = build_taxonomy_candidates(nodes, user=user)
+        result = build_taxonomy_candidates(similar_documents, user=user)
 
         assert result["tags"] == []
 
@@ -295,10 +297,10 @@ class TestBuildTaxonomyCandidates:
         tag.save()
         document = DocumentFactory.create()
         document.tags.add(tag)
-        nodes = [make_node(document.pk, 0.5)]
+        similar_documents = [make_similar(document.pk, 0.5)]
         spy = mocker.patch("documents.permissions.permitted_object_ids")
 
-        result = build_taxonomy_candidates(nodes, user=None)
+        result = build_taxonomy_candidates(similar_documents, user=None)
 
         assert result["tags"][0]["name"] == "Owned"
         spy.assert_not_called()

src/paperless_mail/tasks.py

@@ -1,7 +1,9 @@
 import logging
 
+from celery import Task
 from celery import shared_task
 
+from documents.models import PaperlessTask
 from paperless_mail.mail import MailAccountHandler
 from paperless_mail.mail import MailError
 from paperless_mail.models import MailAccount
@@ -10,8 +12,26 @@ from paperless_mail.models import MailRule
 logger = logging.getLogger("paperless.mail.tasks")
 
 
-@shared_task
-def process_mail_accounts(account_ids: list[int] | None = None) -> str:
+@shared_task(bind=True)
+def process_mail_accounts(self: Task, account_ids: list[int] | None = None) -> str:
+    # A scheduled check can still be running (or queued) when the next one
+    # ProcessedMail dedup only records a message once its
+    # handling has finished, so an overlapping run can still pick up the same
+    # not-yet-recorded message. Skip outright rather than race it.
+    other_mail_fetch_running = (
+        PaperlessTask.objects.filter(
+            task_type=PaperlessTask.TaskType.MAIL_FETCH,
+            status__in=[PaperlessTask.Status.PENDING, PaperlessTask.Status.STARTED],
+        )
+        .exclude(task_id=self.request.id)
+        .exists()
+    )
+    if other_mail_fetch_running:
+        logger.info(
+            "Mail account processing is already running; skipping this run.",
+        )
+        return "Skipped: mail account processing already in progress."
+
     total_new_documents = 0
     accounts = (
         MailAccount.objects.filter(pk__in=account_ids)

src/paperless_mail/tests/test_process_mail_accounts_overlap.py

@@ -0,0 +1,134 @@
+from typing import Final
+
+import pytest
+import pytest_mock
+
+from documents.models import PaperlessTask
+from documents.tests.factories import PaperlessTaskFactory
+from paperless_mail import tasks
+from paperless_mail.tests.factories import MailAccountFactory
+from paperless_mail.tests.factories import MailRuleFactory
+
+NO_DOCUMENTS_ADDED: Final = "No new documents were added."
+SKIPPED: Final = "Skipped: mail account processing already in progress."
+
+
+@pytest.mark.django_db
+@pytest.mark.usefixtures("account_with_rule")
+class TestProcessMailAccountsOverlap:
+    @pytest.fixture
+    def account_with_rule(self) -> None:
+        """An enabled mail account with a single enabled rule."""
+        account = MailAccountFactory.create()
+        MailRuleFactory.create(account=account, enabled=True)
+
+    @pytest.mark.parametrize(
+        ("status", "expected_result", "expected_call_count"),
+        [
+            pytest.param(
+                PaperlessTask.Status.PENDING,
+                SKIPPED,
+                0,
+                id="pending-task-blocks",
+            ),
+            pytest.param(
+                PaperlessTask.Status.STARTED,
+                SKIPPED,
+                0,
+                id="started-task-blocks",
+            ),
+            pytest.param(
+                PaperlessTask.Status.SUCCESS,
+                NO_DOCUMENTS_ADDED,
+                1,
+                id="finished-task-does-not-block",
+            ),
+        ],
+    )
+    def test_skips_only_while_another_mail_fetch_task_runs(
+        self,
+        mocker: pytest_mock.MockerFixture,
+        status: PaperlessTask.Status,
+        expected_result: str,
+        expected_call_count: int,
+    ) -> None:
+        """
+        GIVEN:
+            - An enabled mail account with a rule
+            - Another mail fetch task row in the given status
+        WHEN:
+            - Mail accounts are processed
+        THEN:
+            - Processing is skipped only if that other task is pending or running
+        """
+        PaperlessTaskFactory.create(
+            task_type=PaperlessTask.TaskType.MAIL_FETCH,
+            trigger_source=PaperlessTask.TriggerSource.SCHEDULED,
+            status=status,
+        )
+
+        mocked_handle = mocker.patch.object(
+            tasks.MailAccountHandler,
+            "handle_mail_account",
+            return_value=0,
+        )
+
+        result = tasks.process_mail_accounts()
+
+        assert mocked_handle.call_count == expected_call_count
+        assert result == expected_result
+
+    def test_runs_when_no_other_mail_fetch_task_exists(
+        self,
+        mocker: pytest_mock.MockerFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - An enabled mail account with a rule
+            - No other mail fetch task rows
+        WHEN:
+            - Mail accounts are processed
+        THEN:
+            - The account is handled
+        """
+        mocked_handle = mocker.patch.object(
+            tasks.MailAccountHandler,
+            "handle_mail_account",
+            return_value=0,
+        )
+
+        result = tasks.process_mail_accounts()
+
+        mocked_handle.assert_called_once()
+        assert result == NO_DOCUMENTS_ADDED
+
+    def test_does_not_skip_due_to_its_own_task_row(
+        self,
+        mocker: pytest_mock.MockerFixture,
+    ) -> None:
+        """
+        GIVEN:
+            - An enabled mail account with a rule
+            - A running mail fetch task row belonging to this very task
+        WHEN:
+            - Mail accounts are processed under that task id
+        THEN:
+            - The task does not skip itself and handles the account
+        """
+        PaperlessTaskFactory.create(
+            task_id="self-task-id",
+            task_type=PaperlessTask.TaskType.MAIL_FETCH,
+            trigger_source=PaperlessTask.TriggerSource.SCHEDULED,
+            status=PaperlessTask.Status.STARTED,
+        )
+
+        mocked_handle = mocker.patch.object(
+            tasks.MailAccountHandler,
+            "handle_mail_account",
+            return_value=0,
+        )
+
+        result = tasks.process_mail_accounts.apply(task_id="self-task-id").result
+
+        mocked_handle.assert_called_once()
+        assert result == NO_DOCUMENTS_ADDED